Proper thread pooling and parallelization with an execution graph in .NET .4.5.1

Question

I have a current algorithm that goes like this.

public class Executor
{ 
    private ParallelOptions options = new ParallelOptions();
    private IList<Step> AllSteps;
    public void Execute()
    {
        options.MaxDegreeOfParallelism = 4;
        var rootSteps = AllSteps.Where(s => !s.Parents.Any());
        Parallel.Foreach(rootSteps, options, RecursivelyExecuteStep);
    }   
    private void RecursivelyExecuteStep(Step step)
    {
        ExecuteStep();
        var childSteps = AllSteps.Where(s=>s.Parents.Contains(step) 
            && step.Parents.All(p=>p.IsComplete);
        Parallel.ForEach(childSteps, options, RecursivelyExecuteStep);
    }
}

ParallelOptions.MaxDegreeOfParallelism will be an input variable (but left it out of the code example for brevity).

I was wondering if thread pooling is handled for me automatically or if this creates new threads every time. Also what's the best way to optimize this, is thread pooling something I want. How do I use thread pooling. I'm fairly new to multithreading and what's new in 4.5[.1]

Will this not limit the algorithm to only 4 threads because each Parallel.Foreach would have it's own MaxDegreeOfParallelism of 4 thus not limiting all the threads in the app to 4? How do I achieve limiting all threading in the app to 4?

Edit: MaxDegreeOfParallelism

Answer 1

Parallel.Foreach basically is a nice way to queue up work items to the .NET ThreadPool.

Your application (process) has only one instance of the ThreadPool, and it tries to be as smart as possible regarding how many concurrent threads it uses, taking things like number of available cores and virtual memory into account.

So yes, the .NET ThreadPool handles thread pooling for you, and in many cases you don't need to worry about it, use Parallel.Foreach and let it get on with it.

EDIT: As noted by others, you should be careful in overusing the ThreadPool since it is a shared resource and it may disturb other parts of your application. It will also start creating new threads if your items are blocking or very long-running, which often is wasteful. A rule of thumb is that the work items should be relatively quick and preferably non-blocking. You should test and benchmark, and if it works for your use case then it is very convenient.

You can control the max number of concurrent threads used by the ThreadPool in your application if you want explicit control, by calling ThreadPool.SetMaxThreads. I'd advice against that unless you really have to though, and know what you are doing. The ThreadPool already tries to avoid using more concurrent threads than you have cores for example.

What you can do with ParallellOptions.MaxDegreeOfParallelism is only to further limit the number of concurrent ThreadPool threads that are used to execute that particular call to Parallel.Foreach.

If you need more explicit control of how many concurrent threads an invocation of your algorithm uses, here are some possible alternatives (in, arguably, increasing implementation complexity):

• With the default ThreadPool you can't limit concurrency while calling Parellel.Foreach recursively. You could for example consider using Parallel.Foreach only on the top level (with a ParellelOptions.MaxDegreeOfParallelism argument) and let RecursivelyExecuteStep use a standard foreach.
• Modify (or replace) the ThreadPool for your algorithm to limit concurrency by setting ParallelOptions.TaskScheduler to an instance of QueuedTaskScheduler from Parallel Extension Extras as described here.
• As suggested by @VMAtm, you can use TPL Dataflow to get more explicit control of how your computations are performed, including concurrency (this can also be combined with a custom task scheduler if you really want to knock yourself out).

Answer 2

You can solve this problem with TPL DataFlow library (you can get it via NuGet). As it is said in other answer, Parallel class is using the ThreadPool internally, and you should not be bothered with that.

With the TPL Dataflow the only thing you need is create an TransformManyBlock<TInput,TOutput> linked on itself (or link BufferBlock with ActionBlock with Encapsulate extension), and set the MaxDegreeOfParallelism = 4 or whatever constant you think it should be.

Answer 3

A simple straightforward implementation could look like the following:

ParallelOptions Options = new ParallelOptions{MaxDegreeOfParallelism = 4};
IList<Step> AllSteps;

public void Execute()
{
    var RemainingSteps = new HashSet<Step>(AllSteps);

    while(RemainingSteps.Count > 0)
    {
        var ExecutableSteps = RemainingSteps.Where(s => s.Parents.All(p => p.IsComplete)).ToList();

        Parallel.ForEach(ExecutableSteps, Options, ExecuteStep);

        RemainingSteps.ExceptWith(ExecutableSteps);
    }   
}

Granted, this will execute steps in phases, so you will not always have maximum concurrency. You may only be executing one step at the end of each phase, since the next steps to execute are only realized after all steps in the current phase complete.

If you want to improve concurrency, I would suggest using a BlockingCollection. You'll need to implement a custom partitioner to use Parallel.ForEach against the blocking collection in this case. You'll also want a concurrent collection of the remaining steps, so that you don't queue the same step multiple times (the race condition previously commented on).

public class Executor
{ 
    ParallelOptions Options = new ParallelOptions() { MaxDegreeOfParallelism = 4 };

    IList<Step> AllSteps;

    //concurrent hashset of remaining steps (used to prevent race conditions)
    ConcurentDictionary<Step, Step> RemainingSteps = new ConcurentDictionary<Step, Step>();

    //blocking collection of steps that can execute next
    BlockingCollection<Step> ExecutionQueue = new BlockingCollection<Step>();

    public void Execute()
    {
        foreach(var step in AllSteps)
        {
            if(step.Parents.All(p => p.IsComplete))
            {
                ExecutionQueue.Add(step);
            }
            else
            {
                RemainingSteps.Add(step, step);
            }
        }

        Parallel.ForEach(
            GetConsumingPartitioner(ExecutionQueue),
            Options,
            Execute);
    }

    void Execute(Step step)
    {
        ExecuteStep(step);

        if(RemainingSteps.IsEmpty)
        {
            //we're done, all steps are complete
            executionQueue.CompleteAdding();
            return;
        }

        //queue up the steps that can execute next (concurrent dictionary enumeration returns a copy, so subsequent removal is safe)
        foreach(var step in RemainingSteps.Values.Where(s => s.Parents.All(p => p.IsComplete)))
        {
            //note, removal only occurs once, so this elimiates the race condition
            Step NextStep;
            if(RemainingSteps.TryRemove(step, out NextStep))
            {
                executionQueue.Add(NextStep);
            }
        }
    }

    Partitioner<T> GetConsumingPartitioner<T>(BlockingCollection<T> collection)
    {
        return new BlockingCollectionPartitioner<T>(collection);
    }

    class BlockingCollectionPartitioner<T> : Partitioner<T>
    {
        readonly BlockingCollection<T> Collection;

        public BlockingCollectionPartitioner(BlockingCollection<T> collection)
        {
            if (collection == null) throw new ArgumentNullException("collection");

            Collection = collection;
        }

        public override bool SupportsDynamicPartitions { get { return true; } }

        public override IList<IEnumerator<T>> GetPartitions(int partitionCount)
        {
            if (partitionCount < 1) throw new ArgumentOutOfRangeException("partitionCount");

            var Enumerable = GetDynamicPartitions();

            return Enumerable.Range(0, partitionCount)
                             .Select(i => Enumerable.GetEnumerator()).ToList();
        }

        public override IEnumerable<T> GetDynamicPartitions()
        {
            return Collection.GetConsumingEnumerable();
        }
    }
}