Does Parallel.ForEach limit the number of active threads?

Question

Given this code:

var arrayStrings = new string[1000];
Parallel.ForEach<string>(arrayStrings, someString =>
{
    DoSomething(someString);
});

Will all 1000 threads spawn almost simultaneously?

Answer 1

No, it won't start 1000 threads - yes, it will limit how many threads are used. Parallel Extensions uses an appropriate number of cores, based on how many you physically have and how many are already busy. It allocates work for each core and then uses a technique called work stealing to let each thread process its own queue efficiently and only need to do any expensive cross-thread access when it really needs to.

Have a look at the PFX Team Blog for loads of information about how it allocates work and all kinds of other topics.

Note that in some cases you can specify the degree of parallelism you want, too.

Answer 2

On a single core machine... Parallel.ForEach partitions (chunks) of the collection it's working on between a number of threads, but that number is calculated based on an algorithm that takes into account and appears to continually monitor the work done by the threads it's allocating to the ForEach. So if the body part of the ForEach calls out to long running IO-bound/blocking functions which would leave the thread waiting around, the algorithm will spawn up more threads and repartition the collection between them. If the threads complete quickly and don't block on IO threads for example, such as simply calculating some numbers, the algorithm will ramp up (or indeed down) the number of threads to a point where the algorithm considers optimum for throughput (average completion time of each iteration).

Basically the thread pool behind all the various Parallel library functions, will work out an optimum number of threads to use. The number of physical processor cores forms only part of the equation. There is NOT a simple one to one relationship between the number of cores and the number of threads spawned.

I don't find the documentation around the cancellation and handling of synchronizing threads very helpful. Hopefully MS can supply better examples in MSDN.

Don't forget, the body code must be written to run on multiple threads, along with all the usual thread safety considerations, the framework does not abstract that factor... yet.

Answer 3

Great question. In your example, the level of parallelization is pretty low even on a quad core processor, but with some waiting the level of parallelization can get quite high.

// Max concurrency: 5
[Test]
public void Memory_Operations()
{
    ConcurrentBag<int> monitor = new ConcurrentBag<int>();
    ConcurrentBag<int> monitorOut = new ConcurrentBag<int>();
    var arrayStrings = new string[1000];
    Parallel.ForEach<string>(arrayStrings, someString =>
    {
        monitor.Add(monitor.Count);
        monitor.TryTake(out int result);
        monitorOut.Add(result);
    });

    Console.WriteLine("Max concurrency: " + monitorOut.OrderByDescending(x => x).First());
}

Now look what happens when a waiting operation is added to simulate an HTTP request.

// Max concurrency: 34
[Test]
public void Waiting_Operations()
{
    ConcurrentBag<int> monitor = new ConcurrentBag<int>();
    ConcurrentBag<int> monitorOut = new ConcurrentBag<int>();
    var arrayStrings = new string[1000];
    Parallel.ForEach<string>(arrayStrings, someString =>
    {
        monitor.Add(monitor.Count);

        System.Threading.Thread.Sleep(1000);

        monitor.TryTake(out int result);
        monitorOut.Add(result);
    });

    Console.WriteLine("Max concurrency: " + monitorOut.OrderByDescending(x => x).First());
}

I haven't made any changes yet and the level of concurrency/parallelization has jumped drammatically. Concurrency can have its limit increased with ParallelOptions.MaxDegreeOfParallelism.

// Max concurrency: 43
[Test]
public void Test()
{
    ConcurrentBag<int> monitor = new ConcurrentBag<int>();
    ConcurrentBag<int> monitorOut = new ConcurrentBag<int>();
    var arrayStrings = new string[1000];
    var options = new ParallelOptions {MaxDegreeOfParallelism = int.MaxValue};
    Parallel.ForEach<string>(arrayStrings, options, someString =>
    {
        monitor.Add(monitor.Count);

        System.Threading.Thread.Sleep(1000);

        monitor.TryTake(out int result);
        monitorOut.Add(result);
    });

    Console.WriteLine("Max concurrency: " + monitorOut.OrderByDescending(x => x).First());
}

// Max concurrency: 391
[Test]
public void Test()
{
    ConcurrentBag<int> monitor = new ConcurrentBag<int>();
    ConcurrentBag<int> monitorOut = new ConcurrentBag<int>();
    var arrayStrings = new string[1000];
    var options = new ParallelOptions {MaxDegreeOfParallelism = int.MaxValue};
    Parallel.ForEach<string>(arrayStrings, options, someString =>
    {
        monitor.Add(monitor.Count);

        System.Threading.Thread.Sleep(100000);

        monitor.TryTake(out int result);
        monitorOut.Add(result);
    });

    Console.WriteLine("Max concurrency: " + monitorOut.OrderByDescending(x => x).First());
}

I reccommend setting ParallelOptions.MaxDegreeOfParallelism. It will not necessarily increase the number of threads in use, but it will ensure you only start a sane number of threads, which seems to be your concern.

Lastly to answer your question, no you will not get all threads to start at once. Use Parallel.Invoke if you are looking to invoke in parallel perfectly e.g. testing race conditions.

// 636462943623363344
// 636462943623363344
// 636462943623363344
// 636462943623363344
// 636462943623363344
// 636462943623368346
// 636462943623368346
// 636462943623373351
// 636462943623393364
// 636462943623393364
[Test]
public void Test()
{
    ConcurrentBag<string> monitor = new ConcurrentBag<string>();
    ConcurrentBag<string> monitorOut = new ConcurrentBag<string>();
    var arrayStrings = new string[1000];
    var options = new ParallelOptions {MaxDegreeOfParallelism = int.MaxValue};
    Parallel.ForEach<string>(arrayStrings, options, someString =>
    {
        monitor.Add(DateTime.UtcNow.Ticks.ToString());
        monitor.TryTake(out string result);
        monitorOut.Add(result);
    });

    var startTimes = monitorOut.OrderBy(x => x.ToString()).ToList();
    Console.WriteLine(string.Join(Environment.NewLine, startTimes.Take(10)));
}

Answer 4

It works out an optimal number of threads based on the number of processors/cores. They will not all spawn at once.

Answer 5

See Does Parallel.For use one Task per iteration? for an idea of a "mental model" to use. However the author does state that "At the end of the day, it’s important to remember that implementation details may change at any time."

Answer 6

Does Parallel.ForEach limit the number of active threads?

If you don't configure the Parallel.ForEach with a positive MaxDegreeOfParallelism, the answer is no. With the default configuration, and assuming a source sequence with sufficiently large size, the Parallel.ForEach will use all the threads that are immediately available in the ThreadPool, and will continuously ask for more. By itself the Parallel.ForEach imposes zero restrictions to the number of threads. It is only limited by the capabilities of the associated TaskScheduler.

• The default ParallelOptions.MaxDegreeOfParallelism is -1, which means unlimited parallelism.
• The default ParallelOptions.TaskScheduler is the ThreadPoolTaskScheduler, better known as TaskScheduler.Default.

So if you want to understand the behavior of an unconfigured Parallel.ForEach, you have to know the behavior of the ThreadPool. Which is simple enough so that it can be described in a single paragraph. The ThreadPool satisfies immediately all requests for work, by starting new threads, up to a soft limit which be default is Environment.ProcessorCount. Upon reaching this limit, further requests are queued, and new threads are created with a rate of one new thread per second to satisfy the demand¹. There is also a hard limit on the number of threads, which in my machine is 32,767 threads. The soft limit is configurable with the ThreadPool.SetMinThreads method. The ThreadPool also retires threads, in case there are too many and there is no queued work, at about the same rate (1 per second).

Below is an experimental demonstration that the Parallel.ForEach uses all the available threads in the ThreadPool. The number of available threads is configured up front with the ThreadPool.SetMinThreads, and then the Parallel.ForEach kicks in and takes all of them:

ThreadPool.SetMinThreads(workerThreads: 100, completionPortThreads: 10);
HashSet<Thread> threads = new();
int concurrency = 0;
int maxConcurrency = 0;
Parallel.ForEach(Enumerable.Range(1, 1500), n =>
{
    lock (threads) maxConcurrency = Math.Max(maxConcurrency, ++concurrency);
    lock (threads) threads.Add(Thread.CurrentThread);
    // Simulate a CPU-bound operation that takes 200 msec
    Stopwatch stopwatch = Stopwatch.StartNew();
    while (stopwatch.ElapsedMilliseconds < 200) { }
    lock (threads) --concurrency;
});
Console.WriteLine($"Number of unique threads: {threads.Count}");
Console.WriteLine($"Maximum concurrency: {maxConcurrency}");

Output (after waiting for ~5 seconds):

Number of unique threads: 102
Maximum concurrency: 102

Online demo.

The number of completionPortThreads is irrelevant for this test. The Parallel.ForEach uses the threads designated as "workerThreads". The 102 threads are:

• The 100 ThreadPool threads that were created immediately on demand.
• One more ThreadPool thread that was injected after 1 sec delay.
• The main thread of the console application.

¹ _{The injection rate of the ThreadPool is not documented. As of .NET 7 it is one thread per second, but this could change in future .NET versions.}