MontoyaTech/Parallel.Net
MontoyaTech/Parallel.Net
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Bumped package version to 1.0.1, added more extensions and helper functions to process a list of items without getting a result back. Cleaned up the code, improved documentation.

Browse Source
This commit is contained in:
MattMo 2023-07-24 11:42:33 -07:00
parent a78f796f89
commit ffaee93aa6
5 changed files with 259 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
Parallel.Net.Tests/ParallelTests.cs
View File

@ -16,7 +16,7 @@ namespace MontoyaTech.Parallel.Net.Tests
{
var inputs = new List<int>() { 0, 1, 2, 3, 4 };
var results = inputs.Parallel(input => input * 5);
var results = inputs.ForEachParallel(input => input * 5);
results.Count.Should().Be(inputs.Count);
@ -32,7 +32,7 @@ namespace MontoyaTech.Parallel.Net.Tests
{
var inputs = new List<int>() { 0, 1, 2, 3, 4 };
var results = inputs.Parallel(input => input * 5, maxTasks: 2);
var results = inputs.ForEachParallel(input => input * 5, maxTasks: 2);
results.Count.Should().Be(inputs.Count);
@ -48,7 +48,7 @@ namespace MontoyaTech.Parallel.Net.Tests
{
var inputs = new List<int>() { 0, 1, 2, 3, 4 };
var results = inputs.Parallel(input => input * 5, maxTasks: 0);
var results = inputs.ForEachParallel(input => input * 5, maxTasks: 0);
results.Count.Should().Be(inputs.Count);
@ -64,7 +64,7 @@ namespace MontoyaTech.Parallel.Net.Tests
{
var inputs = new List<int>() { 0, 1, 2, 3, 4 };
var action = new Action(() => inputs.Parallel(input => input * 5, timeout: TimeSpan.FromSeconds(0)));
var action = new Action(() => inputs.ForEachParallel(input => input * 5, timeout: TimeSpan.FromSeconds(0)));
action.Should().Throw<Exception>();
}
@ -74,7 +74,7 @@ namespace MontoyaTech.Parallel.Net.Tests
{
var inputs = new List<int>() { 0, 1, 2, 3, 4 };
var action = new Action(() => inputs.Parallel(input => input * 5, timeout: TimeSpan.FromSeconds(0), throwOnTimeout: false));
var action = new Action(() => inputs.ForEachParallel(input => input * 5, timeout: TimeSpan.FromSeconds(0), throwOnTimeout: false));
action.Should().NotThrow<Exception>();
}
@ -84,7 +84,7 @@ namespace MontoyaTech.Parallel.Net.Tests
{
var inputs = new List<int>() { 0, 1, 2, 3, 4 };
var action = new Action(() => inputs.Parallel(input => input * 5));
var action = new Action(() => inputs.ForEachParallel(input => input * 5));
action.Should().NotThrow<Exception>();
}
@ -96,7 +96,7 @@ namespace MontoyaTech.Parallel.Net.Tests
var results = new List<int>();
var action = new Action(() => results = inputs.Parallel(input => {
var action = new Action(() => results = inputs.ForEachParallel(input => {
if (input == 0)
throw new Exception("Testing");
@ -121,7 +121,7 @@ namespace MontoyaTech.Parallel.Net.Tests
var results = new List<int>();
var action = new Action(() => results = inputs.Parallel(input => {
var action = new Action(() => results = inputs.ForEachParallel(input => {
if (input == 0)
throw new Exception("Testing");
@ -132,5 +132,20 @@ namespace MontoyaTech.Parallel.Net.Tests
results.Count.Should().BeGreaterThan(0);
}
[Fact]
public void Parallel_Inputs_Only_Should_Work()
{
var inputs = new List<int>() { 0, 1, 2, 3, 4 };
int results = 0;
inputs.ForEachParallel(input =>
{
results++;
});
results.Should().Be(5);
}
}
}

2
Parallel.Net/Parallel.Net.csproj
View File

@ -14,7 +14,7 @@
<PackageIcon>Logo_Symbol_Black_Outline.png</PackageIcon>
<RepositoryUrl>https://code.montoyatech.com/MontoyaTech/Parallel.Net</RepositoryUrl>
<GenerateDocumentationFile>True</GenerateDocumentationFile>
<Version>1.0.0</Version>
<Version>1.0.1</Version>
</PropertyGroup>
<ItemGroup>

145
Parallel.Net/ParallelTask.cs
View File

@ -12,7 +12,63 @@ namespace MontoyaTech.Parallel.Net
public class ParallelTask
{
/// <summary>
/// Takes a set of inputes and creates a set of parallel tasks and runs them returning the results. The results will be out of order.
/// Whether or not this parallel task is currently running.
/// </summary>
public bool Running
{
get
{
return this.running && !this.completed && !this.failed;
}
}
/// <summary>
/// Whether or not this parallel task completed successfully.
/// </summary>
public bool Completed
{
get
{
return this.completed;
}
}
/// <summary>
/// Whether or not this parallel task failed due to an exception.
/// </summary>
public bool Failed
{
get
{
return this.failed;
}
}
/// <summary>
/// The exception that was thrown while running this parallel task if it failed.
/// </summary>
public Exception ThrownException
{
get
{
return this.thrownException;
}
}
protected bool running = false;
protected bool completed = false;
protected bool failed = false;
protected bool exceptionStopped = false;
protected ManualResetEvent handle;
protected Exception thrownException = null;
/// <summary>
/// Takes a set of inputs and creates a set of parallel tasks and runs them returning the results. The results will be out of order.
/// </summary>
/// <typeparam name="InputT">The input type</typeparam>
/// <typeparam name="ResultT">The result type</typeparam>
@ -88,6 +144,69 @@ namespace MontoyaTech.Parallel.Net
return results;
}
/// <summary>
/// Takes a set of inputs and creates a set of parallel tasks and runs them.
/// </summary>
/// <typeparam name="InputT">The input type</typeparam>
/// <param name="inputs">The list of inputs to process</param>
/// <param name="action">The action to run against each input</param>
/// <param name="maxTasks">The max number of tasks to run in parallel. If null the number of inputs becomes the number of tasks. Default is null.</param>
/// <param name="timeout">If set this function will return early if the timeout is reached even if the tasks are not done. Default is null.</param>
/// <param name="throwOnTimeout">If true an exception will be thrown if a timeout is reached if one was set. Default is true.</param>
/// <param name="bubbleExceptions">If true, rethrows any exceptions from the running tasks. Default is true.</param>
public static void ForEach<InputT>(IList<InputT> inputs, Action<InputT> action, int? maxTasks = null, TimeSpan? timeout = null, bool throwOnTimeout = true, bool bubbleExceptions = true)
{
if (inputs == null || inputs.Count == 0)
return;
var tasks = new List<ParallelTask<List<InputT>>>();
if (!maxTasks.HasValue)
maxTasks = inputs.Count;
else if (maxTasks <= 0)
maxTasks = 1;
int itemsPerChunk = (inputs.Count / maxTasks.Value) + 1;
int itemIndex = 0;
for (int i = 0; i < maxTasks.Value; i++)
{
var chunk = new List<InputT>();
for (int p = 0; p < itemsPerChunk; p++)
{
chunk.Add(inputs[itemIndex++]);
if (itemIndex >= inputs.Count)
break;
}
if (chunk.Count > 0)
{
tasks.Add(new ParallelTask<List<InputT>>(inputs =>
{
if (inputs != null)
for (int i = 0; i < inputs.Count; i++)
action(inputs[i]);
}, chunk));
}
if (itemIndex >= inputs.Count)
break;
}
if (tasks.Count == 0)
return;
WhenAll(tasks, timeout, throwOnTimeout);
if (bubbleExceptions)
for (int i = 0; i < tasks.Count; i++)
if (tasks[i].Failed && tasks[i].ThrownException != null)
throw tasks[i].ThrownException;
}
/// <summary>
/// Runs a list of tasks and waits until they have been completed or failed and returns the results.
/// </summary>
@ -120,5 +239,29 @@ namespace MontoyaTech.Parallel.Net
return results;
}
/// <summary>
/// Runs a list of tasks and waits until they have been completed or failed.
/// </summary>
/// <typeparam name="InputT">The input type</typeparam>
/// <param name="tasks">The list of tasks to run</param>
/// <param name="timeout">If set this function will leave early if the timeout is met. Default is null.</param>
/// <param name="throwOnTimeout">If set an exception will be thrown if a timeout is reached. Default is true.</param>
/// <exception cref="Exception">A timeout exception is thrown if throwOnTimeout is true and a timeout is reached.</exception>
public static void WhenAll<InputT>(IList<ParallelTask<InputT>> tasks, TimeSpan? timeout = null, bool throwOnTimeout = true)
{
if (tasks == null || tasks.Count == 0)
return;
var countDown = new CountdownEvent(tasks.Count);
foreach (var task in tasks)
task.Run(timeout, false, throwOnTimeout, countDown);
if (!timeout.HasValue)
countDown.Wait();
else if (timeout.HasValue && !countDown.Wait(timeout.Value) && throwOnTimeout)
throw new Exception($"ParallelTasks did not complete within {timeout.Value.TotalMilliseconds} miliseconds");
}
}
}

17
Parallel.Net/ParallelTaskExtensions.cs
View File

@ -20,9 +20,24 @@ namespace MontoyaTech.Parallel.Net
/// <param name="throwOnTimeout">Whether or not to throw an exception on timeout. Default is true</param>
/// <param name="bubbleExceptions">If true, rethrows any exceptions from the running tasks. Default is true.</param>
/// <returns>A list of results from the items processed.</returns>
public static List<ResultT> Parallel<InputT, ResultT>(this IList<InputT> list, Func<InputT, ResultT> func, int? maxTasks = null, TimeSpan? timeout = null, bool throwOnTimeout = true, bool bubbleExceptions = true)
public static List<ResultT> ForEachParallel<InputT, ResultT>(this IList<InputT> list, Func<InputT, ResultT> func, int? maxTasks = null, TimeSpan? timeout = null, bool throwOnTimeout = true, bool bubbleExceptions = true)
{
return ParallelTask.ForEach<InputT, ResultT>(list, func, maxTasks, timeout, throwOnTimeout, bubbleExceptions);
}
/// <summary>
/// Processes a list of items in parallel and runs an action against them.
/// </summary>
/// <typeparam name="InputT">Input type</typeparam>
/// <param name="list">The list of items to process in parallel</param>
/// <param name="action">The action to take an input and perform an operation on it</param>
/// <param name="maxTasks">The max tasks to run in parallel if specified. Default is null. If null the number of items will be the number of tasks</param>
/// <param name="timeout">If set a timeout will be thrown if this doesnt complete in time. Default is null</param>
/// <param name="throwOnTimeout">Whether or not to throw an exception on timeout. Default is true</param>
/// <param name="bubbleExceptions">If true, rethrows any exceptions from the running tasks. Default is true.</param>
public static void ForEachParallel<InputT>(this IList<InputT> list, Action<InputT> action, int? maxTasks = null, TimeSpan? timeout = null, bool throwOnTimeout = true, bool bubbleExceptions = true)
{
ParallelTask.ForEach<InputT>(list, action, maxTasks, timeout, throwOnTimeout, bubbleExceptions);
}
}
}

138
Parallel.Net/ParallelTaskT.cs
View File

@ -11,41 +11,8 @@ namespace MontoyaTech.Parallel.Net
/// </summary>
/// <typeparam name="InputT"></typeparam>
/// <typeparam name="ResultT"></typeparam>
public class ParallelTask<InputT, ResultT>
public class ParallelTask<InputT, ResultT> : ParallelTask
{
/// <summary>
/// Whether or not this parallel task is currently running.
/// </summary>
public bool Running
{
get
{
return this.running && !this.completed && !this.failed;
}
}
/// <summary>
/// Whether or not this parallel task completed successfully.
/// </summary>
public bool Completed
{
get
{
return this.completed;
}
}
/// <summary>
/// Whether or not this parallel task failed due to an exception.
/// </summary>
public bool Failed
{
get
{
return this.failed;
}
}
/// <summary>
/// The result of this parallel task if it completed successfully.
/// </summary>
@ -57,43 +24,20 @@ namespace MontoyaTech.Parallel.Net
}
}
/// <summary>
/// The exception that was thrown while running this parallel task if it failed.
/// </summary>
public Exception ThrownException
{
get
{
return this.thrownException;
}
}
protected bool running = false;
protected bool completed = false;
protected bool failed = false;
protected Func<InputT, ResultT> work = null;
protected InputT input = default(InputT);
protected ResultT result = default(ResultT);
protected bool exceptionStopped = false;
protected Func<InputT, ResultT> func = null;
protected ManualResetEvent handle;
protected Exception thrownException = null;
protected InputT input = default(InputT);
/// <summary>
/// Creates a new parallel task with the worker function and input.
/// </summary>
/// <param name="work"></param>
/// <param name="func"></param>
/// <param name="input"></param>
public ParallelTask(Func<InputT, ResultT> work, InputT input)
public ParallelTask(Func<InputT, ResultT> func, InputT input)
{
this.work = work;
this.func = func;
this.input = input;
}
@ -117,7 +61,7 @@ namespace MontoyaTech.Parallel.Net
{
try
{
this.result = this.work((InputT)state);
this.result = this.func((InputT)state);
this.completed = true;
}
@ -146,4 +90,72 @@ namespace MontoyaTech.Parallel.Net
return this.result;
}
}
/// <summary>
/// The outline of a ParallelTask that takes an input and processes it when ran.
/// </summary>
/// <typeparam name="InputT"></typeparam>
public class ParallelTask<InputT> : ParallelTask
{
protected Action<InputT> action = null;
protected InputT input = default(InputT);
/// <summary>
/// Creates a new parallel task with the worker function and input.
/// </summary>
/// <param name="action"></param>
/// <param name="input"></param>
public ParallelTask(Action<InputT> action, InputT input)
{
this.action = action;
this.input = input;
}
/// <summary>
/// Runs this parallel task and returns when it completes.
/// </summary>
/// <param name="timeout">Default is null, if set, this function will return if the timeout is reached regardless if the task is complete.</param>
/// <param name="blocking">Default is true, whether or not to block the caller function and not return until this task is complete.</param>
/// <param name="throwOnTimeout">Whether or not to throw an exception if a timeout is reached, if one was specified. Default is true.</param>
/// <param name="countDown">An optional count down that if specified is signaled when the task completes.</param>
/// <exception cref="Exception">Throws a timeout exception if the task doesn't complete in time.</exception>
public void Run(TimeSpan? timeout = null, bool blocking = true, bool throwOnTimeout = true, CountdownEvent countDown = null)
{
this.running = true;
this.handle = new ManualResetEvent(false);
ThreadPool.QueueUserWorkItem((object state) =>
{
try
{
this.action((InputT)state);
this.completed = true;
}
catch (Exception ex)
{
this.failed = true;
this.thrownException = ex;
}
this.handle.Set();
if (countDown != null)
countDown.Signal();
}, this.input);
if (blocking)
{
if (!timeout.HasValue)
this.handle.WaitOne();
else if (!this.handle.WaitOne(timeout.Value) && throwOnTimeout)
throw new Exception($"ParallelTask did not complete within {timeout.Value.TotalMilliseconds} miliseconds");
}
}
}
}