Return to Question

The data is meant to simulate a large set of expensive operations, which is the natural use-case for parallelism. All parameters and data are identity between the ThreadBlock and PLINQ tests.

The data is meant to simulate a large set of expensive operations, which is the natural use-case for parallelism. All parameters and data are identical between the ThreadBlock and PLINQ tests.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace GPS.SimpleThreading.Blocks
{
    /// <summary>
    /// Parallel thread block class that provides for
    /// thread warmup, execution, and continuation.
    /// </summary>
    /// <remarks>
    /// ## Features
    /// * Allows capture of results of thread executions
    /// * Allows warmup action per data item before spawning thread
    /// * Allows continuation action per data item after executing thread
    /// * Allows continuation of the entire set
    /// </remarks>
    public sealed class ThreadBlock<TData, TResult>
    {
        private readonly ConcurrentDictionary<TData, (TData data, TResult result)?> _results =
            new ConcurrentDictionary<TData,  (TData data, TResult result)?>();

        private readonly ConcurrentBag<TData> _baseList =
            new ConcurrentBag<TData>();

        private bool _locked;
        private readonly Func<TData, TResult> _action;
        private readonly Action<(TData data, TResult result)?[]> _continuation;

        /// <summary>
        /// Constructor accepting the action and block continuation.
        /// </summary>
        public ThreadBlock(
            Func<TData, TResult> action,
            Action<(TData data, TResult result)?[]> continuation = null)
        {
            _action = action;
            _continuation = continuation;
        }

        /// <summary>
        /// Add single data item.
        /// </summary>
        public void Add(TData item)
        {
            if(!_locked) _baseList.Add(item);
        }

        /// <summary>
        /// Adds range of data items from an IEnumerable
        /// </summary>
        public void AddRange(IEnumerable<TData> collection)
        {
            Parallel.ForEach(collection, Add);
        }

        /// <summary>
        /// Adds range of data items from an ICollection.
        /// </summary>
        public void AddRange(ICollection<TData> collection)
        {
            Parallel.ForEach(collection, Add);
        }

        /// <summary>
        /// Adds range of data items from an IProducerConsumerCollection.
        /// </summary>
        public void AddRange(IProducerConsumerCollection<TData> collection)
        {
            Parallel.ForEach(collection, Add);
        }

        /// <summary>
        /// Maximum number of concurrent threads (default = 1).
        /// </summary>
        public int MaxDegreeOfParallelism { get; set; } = 1;

        /// <summary>
        /// Removes a data item from the block.
        /// </summary>
        public bool Remove(TData item)
        {
            TData itemToRemove;

            if(!_locked)
                return _baseList.TryTake(out itemToRemove);

            return false;
        }

        /// <summary>
        /// Locks the data of the block, allowing processing.
        /// </summary>
        public void LockList()
        {
            _locked = true;
        }

        /// <summary>
        /// Executes the action over the set of data.
        /// </summary>
        public void Execute(
            int maxDegreeOfParallelization = -1,
            Action<TData> warmupItem = null,
            Action<Task, (TData data, TResult result)?> threadContinuation = null)
        {
            if (!_locked) throw new NotLockedException();

            if (maxDegreeOfParallelization == -1)
            {
                maxDegreeOfParallelization = MaxDegreeOfParallelism;
            }

            if (maxDegreeOfParallelization < 1)
            {
                throw new ArgumentOutOfRangeException(
                    "Must supply positive value for either " +
                    $"{nameof(maxDegreeOfParallelization)} or " +
                    $"this.{nameof(MaxDegreeOfParallelism)}.");
            }

            var padLock = new object();
            var queue = new Queue<TData>(_baseList);
            var allTasks = new Dictionary<TData, Task>();

            int depth = 0;

            while (queue.Any())
            {
                var item = queue.Dequeue();

                warmupItem?.Invoke(item);

                var task = new Task<TResult>(() => _action(item));

                task.ContinueWith((resultTask, data) =>
                        {
                            var returnValue = ((TData, TResult)?)(data, resultTask.Result);

                            if (threadContinuation != null)
                            {
                                threadContinuation(resultTask, returnValue);
                            }

                            _results.AddOrUpdate(item, returnValue,
                                (itemData, resultTaskResult) => resultTaskResult);

                            lock (padLock)
                            {
                                depth--;
                            }
                        }, item);


                allTasks.Add(item, task);
            }

            foreach (var task in allTasks)
            {
                int d = 0;
                lock (padLock)
                {
                    d = depth;
                }

                while (d >= maxDegreeOfParallelization)
                {
                    System.Threading.Thread.Sleep(5);
                    lock (padLock)
                    {
                        d = depth;
                    }
                }

                task.Value.Start(TaskScheduler.Default);

                lock (padLock)
                {
                    depth++;
                }
            }

            var dd = 0;

            lock (padLock)
            {
                dd = depth;
            }

            while (dd > 0)
            {
                Thread.Sleep(5);
                lock (padLock)
                {
                    dd = depth;
                }
            }

            _continuation?.Invoke(_results.Values.ToArray());
        }

        /// <summary>
        /// Point-in-time results providing a stable result set
        /// for processing results as the block runs.
        /// </summary>
        public ConcurrentDictionary<TData, (TData data, TResult result)?> Results
        {
            get
            {
                var results = new ConcurrentDictionary<TData, (TData data, TResult result)?>();

                foreach (var key in _results.Keys)
                {
                    var result = _results[key];
                    var value = key;

                    results.AddOrUpdate(value, result, (resultKey, resultValue) => resultValue);
                }

                return results;
            }
        }
    }
}

        [Fact]
        public void ContrivedTest()
        {
            void Warmup(string data) 
                { _log.WriteLine($"Contrived Warmup for {data}");}

            var block = new ThreadBlock<string, string>(
                s => {return $"contrived example output: {s}";},
                r => _log.WriteLine($"Results count: {r.Length}"));

            block.AddRange(new [] {"one", "two", "three"});

            block.LockList();

            block.Execute(
                2,
                Warmup,
                (t, r) => _log.WriteLine(r.Value.result));
    }

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace GPS.SimpleThreading.Blocks
{
    /// <summary>
    /// Parallel thread block class that provides for
    /// thread warmup, execution, and continuation.
    /// </summary>
    /// <remarks>
    /// ## Features
    /// * Allows capture of results of thread executions
    /// * Allows warmup action per data item before spawning thread
    /// * Allows continuation action per data item after executing thread
    /// * Allows continuation of the entire set
    /// </remarks>
    public sealed class ThreadBlock<TData, TResult>
    {
        private readonly ConcurrentDictionary<TData, (TData data, TResult result)?> _results =
            new ConcurrentDictionary<TData, (TData data, TResult result)?>();

        private readonly ConcurrentBag<TData> _baseList =
            new ConcurrentBag<TData>();

        private bool _locked;
        private readonly Func<TData, TResult> _action;
        private readonly Action<ICollection<(TData data, TResult result)?>> _continuation;

        /// <summary>
        /// Constructor accepting the action and block continuation.
        /// </summary>
        public ThreadBlock(
            Func<TData, TResult> action,
            Action<ICollection<(TData data, TResult result)?>> continuation = null)
        {
            _action = action;
            _continuation = continuation;
        }

        /// <summary>
        /// Add single data item.
        /// </summary>
        public void Add(TData item)
        {
            if (!_locked) _baseList.Add(item);
        }

        /// <summary>
        /// Adds range of data items from an IEnumerable
        /// </summary>
        public void AddRange(IEnumerable<TData> collection)
        {
            Parallel.ForEach(collection, Add);
        }

        /// <summary>
        /// Adds range of data items from an ICollection.
        /// </summary>
        public void AddRange(ICollection<TData> collection)
        {
            Parallel.ForEach(collection, Add);
        }

        /// <summary>
        /// Adds range of data items from an IProducerConsumerCollection.
        /// </summary>
        public void AddRange(IProducerConsumerCollection<TData> collection)
        {
            Parallel.ForEach(collection, Add);
        }

        /// <summary>
        /// Maximum number of concurrent threads (default = 1).
        /// </summary>
        public int MaxDegreeOfParallelism { get; set; } = 1;

        /// <summary>
        /// Removes a data item from the block.
        /// </summary>
        public bool Remove(TData item)
        {
            TData itemToRemove;

            if (!_locked)
                return _baseList.TryTake(out itemToRemove);

            return false;
        }

        /// <summary>
        /// Locks the data of the block, allowing processing.
        /// </summary>
        public void LockList()
        {
            _locked = true;
        }

        /// <summary>
        /// Executes the action over the set of data.
        /// </summary>
        public void Execute(
            int maxDegreeOfParallelization = -1,
            Action<TData> warmupItem = null,
            Action<Task, (TData data, TResult result)?> threadContinuation = null)
        {
            if (!_locked) throw new NotLockedException();

            if (maxDegreeOfParallelization == -1)
            {
                maxDegreeOfParallelization = MaxDegreeOfParallelism;
            }

            if (maxDegreeOfParallelization < 1)
            {
                throw new ArgumentOutOfRangeException(
                    "Must supply positive value for either " +
                    $"{nameof(maxDegreeOfParallelization)} or " +
                    $"this.{nameof(MaxDegreeOfParallelism)}.");
            }

            var padLock = new object();
            var queue = new Queue<TData>(_baseList);
            var allTasks = new Dictionary<TData, Task>();

            int depth = 0;

            while (queue.Count > 0)
            {
                var item = queue.Dequeue();

                if (warmupItem != null) warmupItem(item);

                var task = new Task<TResult>(() => _action(item));

                task.ContinueWith((resultTask, data) =>
                {
                    var returnValue = ((TData, TResult)?)(data, resultTask.Result);

                    if (threadContinuation != null)
                    {
                        threadContinuation(resultTask, returnValue);
                    }

                    _results.AddOrUpdate(item, returnValue,
                        (itemData, resultTaskResult) => resultTaskResult);

                    lock (padLock)
                    {
                        depth--;
                    }
                }, item);

                int d = 0;
                lock (padLock)
                {
                    d = depth;
                }

                while (d >= maxDegreeOfParallelization)
                {
                    System.Threading.Thread.Sleep(1);
                    lock (padLock)
                    {
                        d = depth;
                    }
                }

                task.Start(TaskScheduler.Current);

                lock (padLock)
                {
                    depth++;
                }
            }

            var dd = 0;

            lock (padLock)
            {
                dd = depth;
            }

            while (dd > 0)
            {
                Thread.Sleep(1);
                lock (padLock)
                {
                    dd = depth;
                }
            }

            _continuation?.Invoke(_results.Values);
        }

        /// <summary>
        /// Point-in-time results providing a stable result set
        /// for processing results as the block runs.
        /// </summary>
        public ConcurrentDictionary<TData, (TData data, TResult result)?> Results
        {
            get
            {
                var results = new ConcurrentDictionary<TData, (TData data, TResult result)?>();

                foreach (var key in _results.Keys)
                {
                    var result = _results[key];
                    var value = key;

                    results.AddOrUpdate(value, result, (resultKey, resultValue) => resultValue);
                }

                return results;
            }
        }
    }
}

        [Fact]
        public void ContrivedTest()
        {
            string Processor(int data)
            {
                System.Threading.Thread.Sleep(data);
                return $"Waiting {data} miliseconds";
            }

            void Warmup(int data)
            {
                _log.WriteLine($"Contrived Warmup for {data}");
            }

            void ThreadBlockContinuation(Task task, (int data, string result)? result)
            {
                _log.WriteLine($"Contrived Thread Continuation result: {result.Value.data}, {result.Value.result}");
            }

            void PLINQContinuation((int data, string result)? result)
            { 
                _log.WriteLine($"Contrived Thread Continuation result: {result.Value.data}, {result.Value.result}");
            }

            void BlockContinuation(ICollection<(int data, string result)?> results)
            {
                _log.WriteLine($"Results count: {results.Count}");
            }

            var dataSet = new int[500];

            var rand = new System.Random();

            for(int i = 0; i < dataSet.Length; ++i)
            { 
                dataSet[i] = rand.Next(250, 2500);
            }

            var block = new ThreadBlock<int, string>(
                Processor,
                BlockContinuation);

            block.AddRange(dataSet);

            block.LockList();

            var parallelism = 8;

            var sw = new System.Diagnostics.Stopwatch();
            sw.Start();

            block.Execute(parallelism, Warmup, ThreadBlockContinuation);

            sw.Stop();
            var blockElapsed = sw.Elapsed;

            sw = new System.Diagnostics.Stopwatch();

            sw.Start();
            
            var resultSet = dataSet
                .Select(data => { Warmup(data); return data; })
                .AsParallel()
                .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                .WithDegreeOfParallelism(parallelism)
                .Select(data => 
                {
                        return new Nullable<(int data, string result)>
                            ((data: data, result: Processor(data)));
                })
                .AsSequential()
                .Select(result => {
                    PLINQContinuation(result);
                    return result;
                }).ToList();

            BlockContinuation(resultSet.ToArray());

            sw.Stop();
            var plinqElapsed = sw.Elapsed;

            _log.WriteLine(
                $"block: {blockElapsed.TotalSeconds}, " + 
                $"PLINQ: {plinqElapsed.TotalSeconds}");

            Assert.Equal(dataSet.Length, block.Results.Count);
            Assert.Equal(dataSet.Length, resultSet.Count); 

            // This is here to force the test to fail
            // allowing dotnet test to output the log.
            Assert.Equal(blockElapsed, plinqElapsed);
        } 

Edit 2

Added a PLINQ equivalent to the test. Execution times are practically identitical. To me, the PLINQ version is a mess.

So it really comes down to what you like better.

Here's an example result from my system using the exact test above:

... lots of data ....
 Contrived Thread Continuation result: 1143, Waiting 1143 miliseconds
 Contrived Thread Continuation result: 1593, Waiting 1593 miliseconds
 Contrived Thread Continuation result: 2206, Waiting 2206 miliseconds
 Results count: 500
 block: 84.4324359, PLINQ: 85.2551954

The data is meant to simulate a large set of expensive operations, which is the natural use-case for parallelism. All parameters and data are identity between the ThreadBlock and PLINQ tests.

var block = new ThreadBlock<string, string>(
    s => return $"contrived example output: {s}",
    r => Console.WriteLine($"Results count: {r.Length}"));

block.AddRange(new [] "one", "two", "three");

block.Lock();

block.Execute(
    2,
    s => Warmup,
    (t, r) => Console.WriteLine(r.result));

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace GPS.SimpleThreading.Blocks
{
    /// <summary>
    /// Parallel thread block class that provides for
    /// thread warmup, execution, and continuation.
    /// </summary>
    /// <remarks>
    /// ## Features
    /// * Allows capture of results of thread executions
    /// * Allows warmup action per data item before spawning thread
    /// * Allows continuation action per data item after executing thread
    /// * Allows continuation of the entire set
    /// </remarks>
    public sealed class ThreadBlock<TData, TResult>
    {
        private readonly ConcurrentDictionary<TData, (TData data, TResult result)?> _results =
            new ConcurrentDictionary<TData,  (TData data, TResult result)?>();

        private readonly ConcurrentBag<TData> _baseList =
            new ConcurrentBag<TData>();

        private bool _locked;
        private readonly Func<TData, TResult> _action;
        private readonly Action<(TData data, TResult result)?[]> _continuation;

        /// <summary>
        /// Constructor accepting the action and block continuation.
        /// </summary>
        public ThreadBlock(
            Func<TData, TResult> action,
            Action<(TData data, TResult result)?[]> continuation = null)
        {
            _action = action;
            _continuation = continuation;
        }

        /// <summary>
        /// Add single data item.
        /// </summary>
        public void Add(TData item)
        {
            if(!_locked) _baseList.Add(item);
        }

        /// <summary>
        /// Adds range of data items from an IEnumerable
        /// </summary>
        public void AddRange(IEnumerable<TData> collection)
        {
            Parallel.ForEach(collection, Add);
        }

        /// <summary>
        /// Adds range of data items from an ICollection.
        /// </summary>
        public void AddRange(ICollection<TData> collection)
        {
            Parallel.ForEach(collection, Add);
        }

        /// <summary>
        /// Adds range of data items from an IProducerConsumerCollection.
        /// </summary>
        public void AddRange(IProducerConsumerCollection<TData> collection)
        {
            Parallel.ForEach(collection, Add);
        }

        /// <summary>
        /// Maximum number of concurrent threads (default = 1).
        /// </summary>
        public int MaxDegreeOfParallelism { get; set; } = 1;

        /// <summary>
        /// Removes a data item from the block.
        /// </summary>
        public bool Remove(TData item)
        {
            TData itemToRemove;

            if(!_locked)
                return _baseList.TryTake(out itemToRemove);

            return false;
        }

        /// <summary>
        /// Locks the data of the block, allowing processing.
        /// </summary>
        public void LockList()
        {
            _locked = true;
        }

        /// <summary>
        /// Executes the action over the set of data.
        /// </summary>
        public void Execute(
            int maxDegreeOfParallelization = -1,
            Action<TData> warmupItem = null,
            Action<Task, (TData data, TResult result)?> threadContinuation = null)
        {
            if (!_locked) throw new NotLockedException();

            if (maxDegreeOfParallelization == -1)
            {
                maxDegreeOfParallelization = MaxDegreeOfParallelism;
            }

            if (maxDegreeOfParallelization < 1)
            {
                throw new ArgumentOutOfRangeException(
                    "Must supply positive value for either " +
                    $"{nameof(maxDegreeOfParallelization)} or " +
                    $"this.{nameof(MaxDegreeOfParallelism)}.");
            }

            var padLock = new object();
            var queue = new Queue<TData>(_baseList);
            var allTasks = new Dictionary<TData, Task>();

            int depth = 0;

            while (queue.Any())
            {
                var item = queue.Dequeue();

                warmupItem?.Invoke(item);

                var task = new Task<TResult>(() => _action(item));

                task.ContinueWith((resultTask, data) =>
                        {
                            var returnValue = ((TData, TResult)?)(data, resultTask.Result);

                            if (threadContinuation != null)
                            {
                                threadContinuation(resultTask, returnValue);
                            }

                            _results.AddOrUpdate(item, returnValue,
                                (itemData, resultTaskResult) => resultTaskResult);

                            lock (padLock)
                            {
                                depth--;
                            }
                        }, item);


                allTasks.Add(item, task);
            }

            foreach (var task in allTasks)
            {
                int d = 0;
                lock (padLock)
                {
                    d = depth;
                }

                while (d >= maxDegreeOfParallelization)
                {
                    System.Threading.Thread.Sleep(5);
                    lock (padLock)
                    {
                        d = depth;
                    }
                }

                task.Value.Start(TaskScheduler.Default);

                lock (padLock)
                {
                    depth++;
                }
            }

            var dd = 0;

            lock (padLock)
            {
                dd = depth;
            }

            while (dd > 0)
            {
                Thread.Sleep(5);
                lock (padLock)
                {
                    dd = depth;
                }
            }

            _continuation?.Invoke(_results.Values.ToArray());
        }

        /// <summary>
        /// Point-in-time results providing a stable result set
        /// for processing results as the block runs.
        /// </summary>
        public ConcurrentDictionary<TData, (TData data, TResult result)?> Results
        {
            get
            {
                var results = new ConcurrentDictionary<TData, (TData data, TResult result)?>();

                foreach (var key in _results.Keys)
                {
                    var result = _results[key];
                    var value = key;

                    results.AddOrUpdate(value, result, (resultKey, resultValue) => resultValue);
                }

                return results;
            }
        }
    }
}

        [Fact]
        public void ContrivedTest()
        {
            void Warmup(string data) 
                { _log.WriteLine($"Contrived Warmup for {data}");}

            var block = new ThreadBlock<string, string>(
                s => {return $"contrived example output: {s}";},
                r => _log.WriteLine($"Results count: {r.Length}"));

            block.AddRange(new [] {"one", "two", "three"});

            block.LockList();

            block.Execute(
                2,
                Warmup,
                (t, r) => _log.WriteLine(r.Value.result));
    }