PLINQ: Parallel Queries in .NET

So far, you've seen the AsParallel method prefixed to only the Select clause. However, Select is only one of the LINQ clauses that can take the AsParallel method as a prefix. Starting at that clause, the remainder of the query is conducted in parallel. Methods preceding the AsParallel clause in the query statement execute sequentially. In this example, the Select clause executes sequentially, but the GroupBy and OrderBy clauses execute in parallel:

numbers.Select(/* selection */ )
  .AsParallel()
  .GroupBy( /* categorize */ )
  .OrderBy( /* sort*/ );

AsSequential is the opposite of the AsParallel clause. AsSequential serializes portions of your LINQ query. You might choose this to resolve dependencies in a PLINQ query. You can then use AsSequential to isolate the dependency and make a part of a PLINQ query sequential.

You might also decide that a portion of a PLINQ query is more efficiently run in parallel as opposed to sequentially.

Use AsParallel and AsSequential as gates for parallel and sequential execution, as shown in Figure 5. Although it is not common, a single PLINQ query can have multiple AsParallel and AsSequential clauses. Similar to the AsParallel clause, AsSequential can be used to prefix a LINQ method. From that position of the query forward, the remainder of the LINQ query executes sequentially — at least until it encounters an AsParallel clause. Figure 5 illustrates a PLINQ query with both AsParallel and AsSequential clauses. The Select and Groupby clauses execute in parallel, while the OrderBy clause is sequential.

Figure 5: Execution in a PLINQ query with both AsParallel and AsSequential clauses.

Using AsOrdered

For some, an orderly universe is important. Unfortunately, this is contrary to the default behavior of PLINQ. The following code squares and renders the values of an integer list:

int[] numbers = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
var result = from number in numbers.AsParallel()
  select number * number;

You might be surprised by the results, which are indeed squared but do not appear in the same order as the underlying list:

0 4 16 25 36 49 64 81 1 9

PLINQ creates tasks for the query. Each task is then scheduled and placed on a queue in the .NET Framework 4 thread pool. The tasks are then scheduled on processor cores and executed. But PLINQ does not guarantee the order in which tasks will execute, so it is likely, if not probable, that the list iteration is unordered.

If you prefer ordered results, use the AsOrdered clause. The PLINQ query still executes in an unordered fashion to improve performance and fully utilize the available processor cores. However, the results are buffered and then reordered at the completion of the query. This localizes the performance degradation to the AsOrdered clause. Here is the modified query:

int[] numbers = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
var result = from number in numbers.AsParallel()
  .AsOrdered() select number * number;

The results are now ordered:

0 1 4 9 16 25 36 49 64 81

Using WithDegreeOfParallelism

By default, PLINQ uses the available processor cores, up to a maximum of 64. The goal, of course, is to keep the available processor cores busy with active work, as shown in the graph in Figure 6. The graph depicts 100% utilization, which is ideal. This graph was taken during a PLINQ query where the parallel clauses were compute bound.

Figure 6: Full CPU utilization.

You can explicitly set the maximum number of processor cores with the WithDegreeOfParallelism clause. There are two primary reasons for using this clause. First, this is useful when operations are I/O bound. I/O-bound threads are sometimes suspended, which causes processor cores to be underutilized. In this circumstance, you want to increase the degree of parallelism to exceed the number of processor cores. Conversely, you can decrease the number of tasks used in a PLINQ query with the WithDegreeOfParallelism clause. For example, you could create a more cooperative environment for other running applications by purposely reducing the degree of parallelism to less than the number of available cores. Assuming eight available processor cores, the following code reduces the degree of parallelism. The amount of reduction depends on the number of available cores.

int[] numbers = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
var result = 
  numbers.AsParallel()
  .WithDegreeOfParallelism(2)
  .Select(number=>PerformSelect(number))
  .ToList();

Handling Exceptions

Unhandled exceptions raised in a PLINQ query are propagated to the joining thread. Because there can be one or more parallel tasks, multiple unhandled exceptions could occur concurrently. For that reason, unhandled exceptions are raised as an AggregateException. You can enumerate the AggregateException.InnerExceptions property to retrieve the original exceptions raised. This exception model is used in the following code to catch unhandled exceptions in PLINQ. In this example, you iterate an array of integers. Each element is then used as a divisor in a calculation. Unfortunately, a couple of the values in the array are zero. This throws the expected divide-by-zero exceptions. You will successfully catch and display the divide-by-zero exceptions.

Catch an exception in a PLINQ query as an AggregateException and display the results

1. Create a console application for C# in Visual Studio 2010. In the Main method, define an integer array. Notice that there are two zeros in the array list.

int [] intArray = { 5, 1, 2, 7, 4, 0, 6, 2, 9, 0 };

2. Perform a PLINQ query that iterates the integer array. In the Select clause, return a 1000/nth calculation.

var results = intArray.AsParallel()
  .Select(item => (int)1000 / (int) item);

3. Define a try and catch block. In the try block, iterate over the results by using a ForAll method.

try
{
  results.ForAll((item) => Console.WriteLine(item));
}

4. In the catch statement, catch an AggregateException. Then iterate over its InnerExceptions property and display the original underlying exceptions.

catch(AggregateException ex)
{
  foreach (var inner in ex.InnerExceptions)
  {
    Console.WriteLine(inner.Message);
  }
}

Here is the entire code:

using System;
using System.Collections.Generic; 
using System.Linq;
using System.Text;

namespace Exception
{
  class Program
  {
    static void Main(string[] args)
    {
      int [] intArray = { 5, 1, 2, 7, 4, 0, 6, 2, 9, 0 };
      var results = intArray.AsParallel()
        .Select(item => (int)1000 / (int) item);
      try
      {
        results.ForAll((item) =>
           Console.WriteLine(item));
      }
      catch(AggregateException ex)
      {
        foreach (var inner in ex.InnerExceptions)
        {
          Console.WriteLine(inner.Message);
        }
      }
      Console.WriteLine("Press enter to exit");
      Console.ReadLine();
    }
  }
}

Figure 7 shows the output from the application, which shows two unhandled exceptions. Why? There is one exception for each zero in the integer list. Each will cause an unhandled exception in a separate task.

Figure 7: Output from the AggregateException example.

Cancellation

You cancel a PLINQ query with a CancellationTokenSource and CancellationToken object. CancellationToken is a property of the CancellationTokenSource class. Using the WithCancellation clause, you provide a cancellation token to a PLINQ query. You can then call CancellationToken.Cancel to cancel the query operation. When you cancel the operation, it throws an OperationCanceledException exception. Here are the basic steps of the cancellation model with PLINQ:

1. Define a CancellationTokenSource.
2. Execute the PLINQ query within a try block.
3. Add the WithCancellation clause to the PLINQ query. The only parameter is the CancellationToken.
4. Call the CancellationTokenSource.Cancel method to assert a cancellation. This will throw an OperationCanceledException in the PLINQ query.
5. In a catch block, catch and handle the OperationCanceledException.

Review the following example code. Note that it catches both the OperationCanceledException and AggregateException. A PLINQ query consists of parallel tasks, so it is quite possible for both an unhandled exception and a cancellation exception to occur on separate tasks of the same query. Therefore, you should be prepared to catch both exceptions. This example uses a separate thread to invoke the cancellation request.

static CancellationTokenSource cs
  = new CancellationTokenSource();
static void Main(string[] args)
{
  int[] numbers = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
  (new Thread(new ThreadStart(Cancellation))).Start();
  try
  {
    var result = numbers.AsParallel()
      .WithCancellation(cs.Token)
      .Select(number => PerformSelect(number))
      .ToList();
  }
  catch (OperationCanceledException ex)
  {
    Console.WriteLine(ex.Message);
  }
  catch (AggregateException ex)
  {
    foreach (var inner in ex.InnerExceptions)
    {
      Console.WriteLine(inner.Message);
    }
  }
}

static void Cancellation()
{
  Thread.Sleep(4000); cs.Cancel();
}