What’s New For Parallel Programming in .NET 4.5?

Hi folks,

Today I would like to talk with you about parallel programming and new feature related to it in .NET 4.5, let us rock'n'roll.

 In order to talk about Parallel programming I would like to do a short overview of existing synchronous programming patterns existing in .net framework:

1) Asynchronous Programming Model (APM) (legacy)

The main item is IAsyncResult design pattern(not interface), it provides two implementations of Begin[OperationName] and End[OperationName] for [OperationName].

 - The Begin[OperationName] starts asynchronous [OperationName] operation and return IAsyncResult

 - End[OperationName] ends asynchronous[OperationName] operation and return same result as synchronous [OperationName].

// Create delegate

<Method>Delegate caller = new <Method>Delegate(<class>.<Method>);

// Initiate the asychronous call.IAsyncResult result = caller.BeginInvoke(3000,out <returnMethodValue>);

Thread.Sleep(0);

Console.WriteLine("Main thread {0} does some work.", Thread.CurrentThread.ManagedThreadId);

// Wait for the WaitHandle to become signaled.
result.AsyncWaitHandle.WaitOne();

// Perform additional processing here. 
// Call EndInvoke to retrieve the results. 
string returnValue = caller.EndInvoke(out returnMethodValue, result);

// Close the wait handle.
result.AsyncWaitHandle.Close();

Console.WriteLine("The call executed on thread {0}, with return value \"{1}\".",
                threadId, returnValue);

2) Event-based Asynchronous Pattern (EAP) (legacy) 

The simplest implementation of EAP consists from three items [MethodName]Async method, [MethodName]Completed event and [MethodName]AsyncCancel method. There is a possibility to track progress and its changes by custom statuses, however it is out of the current topic.

I haven't added code example here cause it is obviously - subscribed to event, execute method, waiting for events: changing state or complete execution. Manually stop execution in case of any issues.

 3) Task-based Asynchronous Pattern (TAP) (recommended for new development)

TAP bases  on the System.Threading.Tasks.Task and System.Threading.Tasks.Task<TResult>

Shortly parallel programmings tasks can be divided at next groups:
- Data Parallelism (Task Parallel Library) : describes how to manipulate with data (Parallel for and foreach)
- Task Parallelism (Task Parallel Library) : manipulation with tasks using parallel invoke and task.
- Dataflow (Task Parallel Library) - creation flow for data manipulation using TPL
- Parallel LINQ (PLINQ): describes how to achieve data parallelism with PLINQ

So first things first:

Data Parallelism (Task Parallel Library)

Data parallelism is submitted by Parallel   class. It contains two methods For and Foreach, let us check how we can use these methods instead of usual sequential For and Foreach:

First simple sample

// Sequential version             
foreach (var item in sourceCollection)
{
    Process(item);
}

// Parallel equivalent
Parallel.ForEach(sourceCollection, item => Process(item));

Ha, nothing special, here is more interesting example  let us move next.

Second sample

Here we will try to calculate summary of the huge int range, we are going to pass subtotal and add it to Total using Intrerlocked

int[] nums = Enumerable.Range(0, 1000000).ToArray();

long total = 0;

// Use type parameter to make subtotal a long, not an int
Parallel.For<long>(0, nums.Length, 

 () => 0, //Delegate to init local value is used for summary in our case

(j, loop, subtotal) => //For body
{
    subtotal += nums[j];
    return subtotal;
},
    (x) => Interlocked.Add(ref total, x) //Code executes after one of the "body" loop has completed work
);

Console.WriteLine("The total is {0:N0}", total);
Console.WriteLine("Press any key to exit");
Console.ReadKey();

Third sample

In addition about cancellation loop. There is an option to cancel loop by supply CancellationToken to the method in the ParallelOptions parameter and then enclose the parallel call in <try catch> block. Agree, looks not so easy as <break> or <continue> in sequential <for/foreach>

 int[] nums = Enumerable.Range(0, 10000000).ToArray();

CancellationTokenSource cts = new CancellationTokenSource();

// Use ParallelOptions instance to store the CancellationToken
ParallelOptions po = new ParallelOptions();
po.CancellationToken = cts.Token;
po.MaxDegreeOfParallelism = System.Environment.ProcessorCount;
Console.WriteLine("Press any key to start. Press 'c' to cancel.");
Console.ReadKey();

// Run a task so that we can cancel from another thread.
Task.Factory.StartNew(() =>
{
       if (Console.ReadKey().KeyChar == 'c')
               cts.Cancel();
       Console.WriteLine("press any key to exit");
});

try
{
      Parallel.ForEach(nums, po, (num) =>
      {
            double d = Math.Sqrt(num);
            Console.WriteLine("{0} on {1}", d, Thread.CurrentThread.ManagedThreadId);
            po.CancellationToken.ThrowIfCancellationRequested();
      });
}
catch (OperationCanceledException e)
{
     Console.WriteLine(e.Message);
}

Console.ReadKey();

Looks really weird, however make sense in terms of Parallelism + threads approach. Hope guys from MS will wrap it in some "syntax sugar".

Dataflow (Task Parallel Library) 

Dataflow components are very useful when you design system which has asynchronous data availability and further manipulation with data. In another words it is asynchronous chain of object which start its work, once input data is available.

// Creates the image processing dataflow network and returns the 
// head node of the network.
ITargetBlock<string> CreateImageProcessingNetwork()
{
   // 
   // Create the dataflow blocks that form the network. 
   // 

   // Create a dataflow block that takes a folder path as input 
   // and returns a collection of Bitmap objects. 
   var loadBitmaps = new TransformBlock<string, IEnumerable<Bitmap>>(path =>
      {
         try
         {
            return LoadBitmaps(path);
         }               
         catch (OperationCanceledException)
         {
            // Handle cancellation by passing the empty collection 
            // to the next stage of the network. 
            return Enumerable.Empty<Bitmap>();
         }
      });           

   // Create a dataflow block that takes a collection of Bitmap objects 
   // and returns a single composite bitmap. 
   var createCompositeBitmap = new TransformBlock<IEnumerable<Bitmap>, Bitmap>(bitmaps =>
      {
         try
         {
            return CreateCompositeBitmap(bitmaps);
         }
         catch (OperationCanceledException)
         {
            // Handle cancellation by passing null to the next stage  
            // of the network. 
            return null;
         }               
      });

   // Create a dataflow block that displays the provided bitmap on the form. 
   var displayCompositeBitmap = new ActionBlock<Bitmap>(bitmap =>
      {
         // Display the bitmap.
         pictureBox1.SizeMode = PictureBoxSizeMode.StretchImage;
         pictureBox1.Image = bitmap;

         // Enable the user to select another folder.
         toolStripButton1.Enabled = true;
         toolStripButton2.Enabled = false;
         Cursor = DefaultCursor;
      },
      // Specify a task scheduler from the current synchronization context 
      // so that the action runs on the UI thread. 
      new ExecutionDataflowBlockOptions
      {
          TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()
      });

   // Create a dataflow block that responds to a cancellation request by  
   // displaying an image to indicate that the operation is cancelled and  
   // enables the user to select another folder. 
   var operationCancelled = new ActionBlock<object>(delegate
      {
         // Display the error image to indicate that the operation 
         // was cancelled.
         pictureBox1.SizeMode = PictureBoxSizeMode.CenterImage;             
         pictureBox1.Image = pictureBox1.ErrorImage;

         // Enable the user to select another folder.
         toolStripButton1.Enabled = true;
         toolStripButton2.Enabled = false;
         Cursor = DefaultCursor;
      },
      // Specify a task scheduler from the current synchronization context 
      // so that the action runs on the UI thread. 
      new ExecutionDataflowBlockOptions
      {
         TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()
      });

   // 
   // Connect the network. 
   // 

   // Link loadBitmaps to createCompositeBitmap.  
   // The provided predicate ensures that createCompositeBitmap accepts the  
   // collection of bitmaps only if that collection has at least one member.
   loadBitmaps.LinkTo(createCompositeBitmap, bitmaps => bitmaps.Count() > 0);

   // Also link loadBitmaps to operationCancelled. 
   // When createCompositeBitmap rejects the message, loadBitmaps  
   // offers the message to operationCancelled. 
   // operationCancelled accepts all messages because we do not provide a  
   // predicate.
   loadBitmaps.LinkTo(operationCancelled);

   // Link createCompositeBitmap to displayCompositeBitmap.  
   // The provided predicate ensures that displayCompositeBitmap accepts the  
   // bitmap only if it is non-null.
   createCompositeBitmap.LinkTo(displayCompositeBitmap, bitmap => bitmap != null);

   // Also link createCompositeBitmap to operationCancelled.  
   // When displayCompositeBitmap rejects the message, createCompositeBitmap  
   // offers the message to operationCancelled. 
   // operationCancelled accepts all messages because we do not provide a  
   // predicate.
   createCompositeBitmap.LinkTo(operationCancelled);

   // Return the head of the network. 
   return loadBitmaps;
}

Parallel LINQ (PLINQ)

Parallel LINQ can significantly increase performance of LINQ. Most common simple transformation is adding <AsParallel()> extension method call to source collection. Let define when PLINQ can improve performance:

1) Computational cost of the overall work.

PLINQ query must have enough parallel work to offset the overhead. 

var queryA = from num in numberList.AsParallel()
             select ExpensiveFunction(num); //good for PLINQ

var queryB = from num in numberList.AsParallel()
             where num % 2 > 0
             select num; //not as good for PLINQ

2) The number of logical cores on the system (degree of parallelism).

The overall amount of speedup depends on what percentage of the overall work of the query is parallelizable.

3) The number and kind of operations.

PLINQ provides the AsOrdered operator for situations in which it is necessary to maintain the order of elements in the source sequence. There is a cost associated with ordering, but this cost is usually modest. GroupBy and Join operations likewise incur overhead. PLINQ performs best when it is allowed to process elements in the source collection in any order, and pass them to the next operator as soon as they are ready.

4) The form of query execution.

Calling ToArray or ToList involves an unavoidable computational cost, actually the same for "foreach", to improve "foreach" case please use "ForAll.



In addition, please take a look on following list describes the query shapes that PLINQ by default will execute in sequential mode:

• Queries that contain a Select, indexed Where, indexed SelectMany, or ElementAt clause after an ordering or filtering operator that has removed or rearranged original indices.
• Queries that contain a Take, TakeWhile, Skip, SkipWhile operator and where indices in the source sequence are not in the original order.
• Queries that contain Zip or SequenceEquals, unless one of the data sources has an originally ordered index and the other data source is indexable (i.e. an array or IList(T)).
• Queries that contain Concat, unless it is applied to indexable data sources.
• Queries that contain Reverse, unless applied to an indexable data source.

I have read a lot of articles about LINQ vs PLINQ, so I may say that there always plays to use PLINQ, I haven't found any notes that it can decrease performance in compare to LINQ, due to way how we get parllelism by adding <AsParallel()> extension method which transforms from LINQ to PLINQ.
What about how PLINQ increases performance, so for
- Processor: Intel 2 cores by 2GHz
- Query: var query = from num in source.AsParallel() select new { Num = num, IsPrime = IsPrime(num.Value) };

Result looks like:



So only with dramatically increasing items in collection we receive improving, However in it only current case, result can be various due to processor type and query.

That all , folks. Please comment and share your thoughts below :)