While waiting for Parallel Extensions (wikipedia), that will be shipped with next .Net Framework release, I designed a couple of class that allow to quickly (and easily!) write code that make use of multi-threading and, therefore, speed-up many tasks. What I show here is a simple class able to collect pieces of code (in form of delegate) to be executed concurrently, and parallelize and synchronize the execution. The main goals I kept in mind while realizing the solution were:
- Ease of use, minimal requirement in terms of multi-threading skills
- Allowing single-threaded code to be ported to multi-threading with few changes
- Code readability
- Support for exception management
Here what I came up with:
The API exposes one class, ParallelExecutor, and one interface, IExecutionResult (plus a generic specialization of the interface and the necessary implementations). Before seeing the internals of the solution, I want to show the ease of use of the API.
Let’s start with this example:
1: void SerialExecution()
2: {
3: Uri uri; int value;
4:
5: CallWebService(uri);
6: PerformComputation(value);
7: }
CallWebService() and PerformComputation() represent the most classic example of methods that can (and should!) run in parallel as they perform respectively I/O and CPU computation. But usually programmers haven’t enough confidence with multi-threading execution and synchronization to do things in the right way. In any case they shouldn’t care about this stuff, they should focus on coding logic, and not on coding about how the logic get’s executed. Let’s see how ParallelExecutor comes in help:
1: void ParallelExecution
2: {
3: Uri uri; int value;
4:
5: ParallelExecutor executor = new ParallelExecutor();
6:
7: executor.Add(CallWebService, uri);
8: executor.Add(PerformComputation, value);
9:
10: executor.WaitAll();
11: }
Not too bad, right? With a couple of lines I got the two methods run concurrently. This is pretty simple: the Add() method takes as arguments the method to invoke and the arguments of that method. The WaitAll() causes the calling thread to wait for all the methods passed to Add() to complete. Take a (partial) look to the ParallelExecutor definition:
1: public class ParallelExecutor
2: {
3: public IExecutionInfo<object> Add(Action workItem)
4: public IExecutionInfo<object> Add<T>(Action<T> workItem, T arg)
5: //overloads for other Action versions
6:
7: public IExecutionInfo<TResult> Add<TResult>(Func<TResult> workItem)
8: public IExecutionInfo<TResult> Add<T, TResult>(Func<T, TResult> workItem, T arg)
9: //overloads for other Func versions
10:
11: public bool WaitAll():
12: }
As you can see, method to be parallelized are passed to Add() in form of variations of Action and Func generics delegate (there are also overloads that take MethodInfo and the standard Delegate). But what about the Add() return type, IExecutionResult? Let’s see the interface declaration:
1: public interface IExecutionInfo
2: {
3: object[] Args { get; }
4:
5: Exception Exception { get; }
6:
7: object Result { get; }
8: }
9:
10: public interface IExecutionInfo<TResult> : IExecutionInfo
11: {
12: TResult Result { get; }
13: }
At every invocation of the Add() method an instance of a class implementing IExecutionResult is created, and its Args property is filled with method invocation arguments. After the invocation completes, the Result and Exception property are filled too (in both case, if any). And the magic of generics allow the code to be type safe. So, returning to the example, if CallWebService() has a return value and if you want to check for exceptions, you can write:
1: void ParallelExecution
2: {
3: Uri uri; int value;
4:
5: ParallelExecutor executor = new ParallelExecutor();
6:
7: IExecutionInfo<int> execution = executor.Add<Uri, int>(CallWebService, uri);
8: executor.Add(PerformComputation, value);
9:
10: executor.WaitAll();
11:
12: if (execution.Exception != null)
13: ManageException(execution.Exception);
14: else
15: Console.WriteLine("Result: {0}", execution.Result);
16: }
Another couple of properties exposed by ParallelExecutor allow you to control when the execution starts (immediately after Add() or when WaitAll() gets called) and the aggressiveness of the threading policy (one new Thread for every method passed to Add() or threads got from ThreadPool).
Obviously this is a very simple implementation, far away from what Parallel Extensions will be. But its simplicity may suite well some implementation with no heavy requirements.
0