I've created a library providing a ThreadBlock that has the following features:
- Aggregates results of all actions executed
- Provides for non-threaded warm up to prepare data for processing
- Provides for per-thread continuation actions
- Provides for per-block continuation actions
The code is located on GitHub.
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;
}
}
}
}
I would be very grateful for feedback on the features, design and implementation.
A simple usage is
[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 identical between the ThreadBlock and PLINQ tests.