Parallel.For

Listing 11-8 shows a normal C# loop, which when executed prints out all values from 0 to 19 inclusive. If the goal of this loop is to simply output all values from 0 to 19 inclusive and you don’t care about the order, then you can parallelize it.

Listing 11-8.  Regular C# Loop

for( int i = 0 ; i < 20 ; i++ )
{
  Console.WriteLine(i);
}

To parallelize this loop, remove the use of the C# for statement and replace it with a call to Parallel.For. In this example Parallel.For takes the start and end values for the loop along with an Action<int> delegate representing the body of the loop. The code in Listing 11-9 shows such an implementation; when executed it will display all the values between 0 and 19 inclusive, but each time you run it you may well see varying orders. Even a program as simple as this exposes chaotic behavior, as shown in Figure 11-1.

Listing 11-9.  Simple Parallel.For Loop

Parallel.For(0,20,i =>
{
  Console.WriteLine(i);
} );

The refactored code still resembles the original single-core implementation. The use of lambdas/anonymous methods has allowed you to express the loop body inline, thus resembling the original structure, and hopefully maintaining its readability.

All of the Parallel.XXX methods utilize TPL tasks to execute pieces of work in parallel. One obvious mapping in the case of loops would be to create as many tasks as there are loop iterations, so in the case of Listing 11-9, it would create 20 tasks. If you have at least 20 cores this may well be a good idea, but if you only have 4, not a great idea. The effort required to create a task is not free—it adds additional workload above and beyond the single-core implementation. Therefore it makes sense to create many tasks but to allow many tasks to do many iterations. Listing 11-10 has a modified parallel loop that is printing out the loop index and also the task ID associated with that loop iteration. As you can see from Figure 11-2, many tasks are involved in producing the final result, but not 20.

Listing 11-10.  Parallel.For and Associated Tasks

Parallel.For(0, 20, i =>
 {
     Console.WriteLine("{0} : {1}",Task.CurrentId,i);
 });

The Parallel.For method can’t know the perfect number of tasks to create; it has to guess and keep adjusting its guess based on observed throughput of the loop. The longer and longer a task takes to process each iteration, the more likely it is that creating more tasks to handle a smaller and smaller range of the loop makes sense. This can be illustrated by adding a Thread.SpinWait into each iteration of the loop (to simulate more work). Listing 11-11 has a bit more work per iteration, and Figure 11-3 shows the effect of increased work on the number of tasks created, now that a lot more tasks have been created, allowing for a greater opportunity for parallelism.

Listing 11-11.  A Bit More Work per Iteration

Parallel.For(0, 20, i =>
{
    Console.WriteLine("{0} : {1}", Task.CurrentId, i);
    Thread.SpinWait(100000000);
});

This approach taken by the Parallel class certainly won’t always be optimal but will often provide a respectable speedup (a moderate free lunch). To reach a more optimal solution requires an understanding of the environment the code is running in, whether each iteration of the loop has the same cost, what resources it consumes, and their availability. This level of knowledge is hard to build into a general purpose algorithm. One simple element of tuning you can provide is stating via ParallelOptions and setting MaxDegreeOfParallelis, which will restrict the number of tasks active at any one time, and hence control the level of concurrency. Note that it doesn’t restrict the number of tasks that will be created, but rather limits the number of tasks that will ever be submitted to the scheduler at any moment of time. Throughout this chapter we will look at various ways of refactoring our use of Parallel.XXX to get better performance through our knowledge of the algorithm and the process resources available to us.

Time Taken: 01.9607748 seconds Result: 3.14159265158926

// 4 * ( -1.0 / 3.0 + 1.0/5.0 - 1.0/7.0 + 1.0/9.0 - 1.0/11.0 ... )
private static double CalculatePi(int iterations)
{
    double pi = 1;
    double multiplier = -1;
    for (int i = 3; i < iterations; i+=2)
    {
        pi += 1.0/(double) i*multiplier;
        multiplier *= -1;
    }
 
    return pi*4.0;
}

Time Taken: 05.1148 seconds Result: 3.13983501709191

private static double ParallelCalculatePi(int iterations)
{
    double pi = 1;
    double multiplier = -1;
    Parallel.For(0, (iterations - 3)/2, loopIndex =>
        {
            int i = 3 + loopIndex*2;
            pi += 1.0/(double) i*multiplier;
            multiplier *= -1;
        });
 
    return pi * 4.0;
}

public static ParallelLoopResult For<TLocal>(int fromInclusive, int toExclusive,
                                             Func< TLocal> localInit,
                                             Func<int,ParallelLoopState, TLocal, TLocal> body,
                                             Action< TLocal> localFinally);

private static double ParallelCalculatePi(int iterations)
{
    double pi = 1;
     
    Parallel.For(0, (iterations - 3) / 2,
        InitialiseLocalPi,
        (int loopIndex, ParallelLoopState loopState, double localPi) =>
        {
            double multiplier = loopIndex%2 == 0 ? -1 : 1;
            int i = 3 + loopIndex*2;
            localPi+= 1.0/(double) i*multiplier;
            
            return localPi;
        },
    (double localPi) =>
    {
        pi += localPi;
    });
 
    return pi * 4.0;
}
 
private static double InitialiseLocalPi()
{ return 0.0;}

Time Taken: 01.05 seconds Result: 4.00000021446738

Time Taken: 01.09 seconds Result: 3.14159265247552
Time Taken: 01.0571 seconds Result: 3.14159266368947
Time Taken: 01.0283 seconds Result: 3.14159265565936

private static double ParallelCalculatePi(int iterations)
{
    double pi = 1;
    object combineLock = new object();
 
    Parallel.For(0, (iterations - 3) / 2,
        InitialiseLocalPi,
        (int loopIndex, ParallelLoopState loopState, double localPi) =>
        {
            double multiplier = loopIndex%2 == 0 ? -1 : 1;
            int i = 3 + loopIndex*2;
            localPi += 1.0/(double) i*multiplier;
            
            return localPi;
        },
    (double localPi) =>
    {
        lock (combineLock)
        {
            pi += localPi;
        }
    });
    return pi * 4.0;
}

Time Taken: 01.02 seconds Result: 3.14159265558957
Time Taken: 01.04 seconds Result: 3.1415926555895
Time Taken: 01.07 seconds Result: 3.14159265558942

Parallel.ForEach

As you can probably guess, Parallel.ForEach provides the parallel equivalent of the C# foreach loop. Parallel.ForEach works by consuming an IEnumerable<T> and farming out elements consumed through the enumerator to multiple tasks, thus processing the elements in parallel. Listing 11-17 shows the equivalent Parallel.ForEach of the initial Parallel.For example in Listing 11-9.

Listing 11-17.  Simple Parallel.ForEach

IEnumerable<int> items = Enumerable.Range(0, 20);
 
Parallel.ForEach(items, i =>
{
    Console.WriteLine(i);
});

Parallel.ForEach can therefore be applied to anything that implements IEnumerable (Lists, Arrays, Collections, etc.). One limiting factor with Parallel.For was that you were constrained to int or long and had to increment by 1s. This can be worked around by using an iterator method and Parallel.ForEach instead of Parallel.For as shown in Listing 11-18.

Listing 11-18.  Parallel Loop from 1 to 2 in Steps of 0.1

Parallel.ForEach(FromTo(1, 2, 0.1), Console.WriteLine);
         
. . .
private static IEnumerable<double> FromTo(double start, double end,double step)
{
    for (double current = start; current < end; current += step)
    {
        yield return current;
    }
}

Parallel.ForEach has similar overloads to that of Parallel.For, allowing for the use of local loop state and combining logic in the same way they were used for Parallel.For. What makes life harder for Parallel.ForEach compared to Parallel.For is that it potentially won’t know how many items there are to process, since a pure IEnumerable object has no way of yielding how many items there are to consume. Therefore Parallel.ForEach has to start off consuming a few items per task so as to allow many tasks to work on a small data set. As it learns that the IEnumerable stream keeps yielding an initial block size, it asks for larger and larger block sizes. This process is known as partitioning. If Parallel.ForEach could know the size of the IEnumerable stream, it could do a better job of partitioning. Therefore, to give itself the best possible chance internally, it attempts to discover at runtime if the IEnumerable<T> is in fact an array or an IList<T>. Once it knows it is a list or an array, it can use the Count and Length properties respectively to determine the size, and hence do a better job of initial partitioning. Parallel.ForEach therefore works best with arrays or lists.

ParallelLoopState

With a regular loop in C#, you can leave the loop early through the use of a break statement. Since the body of a parallel loop is represented by a delegate, you can’t use the break keyword. To allow this style of behavior for parallel loops, both Parallel.For and Parallel.ForEach provide an overload that allows the loop body delegate to take, in addition to the loop index, a ParallelLoopState object. This loop state object is used to both signal and observe if the loop is to terminate early. Listing 11-19 shows an example of a loop that for each iteration has a 1 in 50 chance of terminating.

Listing 11-19.  Early Termination using ParallelLoopState.Break

Random rnd = new Random();
Parallel.For(0, 100, (i, loopState) =>
{
    if (rnd.Next(1, 50) == 1)
    {
        Console.WriteLine("{0} : Breaking on {1}",Task.CurrentId, i);
        loopState.Break();
        return;
 
    }
    Console.WriteLine("{0} : {1}", Task.CurrentId, i);
});

The big difference between break in a regular loop to a parallel loop is that the break can be acted upon immediately and no further iterations will occur. With a parallel loop other iterations could be in progress when a particular iteration requests the break, in which case these iterations can’t be stopped without some cooperation of the iteration. Figure 11-5 shows an example of an early loop termination with break.

Although Task 5 has requested termination, Tasks 7, 3, and 1 continue to execute many more iterations. The reason for this is that ParallelLoopState.Break is trying to recreate the semantics of a regular loop break. With a regular loop break, if you break on Iteration 20 you know that all iterations prior to that will have executed. When you issue a ParallelLoopState.Break() the parallel loop ensures that all lower-value iterations will be executed before the loop ends. Studying the output shown in Figure 11-5, we can see that once Task 5 has issued the break apart from Tasks 7, 3, and 1, all other tasks complete their current iteration and end. Tasks 7, 3, and 1 keep running to complete all iterations less than 41. Obviously if iterations have already completed that represent values greater than 41, then there will be no way of rolling back. The behavior is thus not an exact match to the regular break, but it is as close as you can get considering the loop is being executed out of order.

If you do want to terminate the loop as soon as possible and don’t care about guaranteeing that all prior iterations have completed, then ParallelLoopState has another method called Stop. The Stop method attempts to end the loop as soon as possible—once issued, no loop task will start a new iteration. Listing 11-20 shows the same example using Stop instead of a Break. Figure 11-6 shows an example of a run where Stop is issued by Task 9: all remaining tasks finish their current iteration and don’t process any more.

Listing 11-20.  Early Loop Termination with ParallelLoopState.Stop()

Random rnd = new Random();
Parallel.For(0, 100, (i, loopState) =>
{
    if (rnd.Next(1, 50) == 1)
    {
        Console.WriteLine("{0} : Stopping on {1}", Task.CurrentId, i);
        loopState.Stop();
        return;
 
    }
 
    Console.WriteLine("{0} : {1}", Task.CurrentId, i);
});

In addition to initiating termination of the loop, ParallelLoopState can also be used to determine if another iteration of the loop has initiated termination, or has thrown an unhandled exception. In situations where the loop body takes a considerable amount of time, it may be desirable for the loop body to observe the fact that the loop should end, and assist in shutting down the loop early. Table 11-1 lists the properties available on ParallelLoopState to assist in detection of the need to terminate early.

Listing 11-21 shows an example of a loop body detecting a stop request and ending the loop body prematurely. Early termination can also be observed outside the loop body by examining the return value from Parallel.For or Parallel.ForEach. The return value is of type ParallelLoopResult and it provides a property called IsCompleted. If the loop did not complete all intended iterations due to a Stop or Break being issued, this will be set to false. Figure 11-7 shows an example run where early termination is being observed.

Listing 11-21.  Observing Stop Requests

Random rnd = new Random();
 
ParallelLoopResult loopResult= Parallel.For(0, 100, (i, loopState) =>
{
    if (rnd.Next(1,50) == 1)
    {
        Console.WriteLine("{0} : Stopping on {1}", Task.CurrentId, i);
        loopState.Stop();
        return;
    }
 
    Thread.Sleep(10);
 
    if ( loopState.IsStopped)
    {
        Console.WriteLine("{0}:STOPPED",Task.CurrentId);
        return;
    }
 
    Console.WriteLine("{0} : {1}", Task.CurrentId, i);
});
 
 
Console.WriteLine("Loop ran to completion {0}",loopResult.IsCompleted);

If a loop body were to throw an unhandled exception, this would be similar to issuing a Stop request. The exception would be caught by Parallel.For or Parallel.ForEach and then be re-thrown as part of an AggregateException, in order to inform the caller that the loop failed to complete due to exceptions.

Parallel loops are not the same as synchronous loops but, where possible, similar constructs are provided to allow for early termination of a loop. If a loop has completed its said purpose early, use Break and Stop for early graceful termination of the loop. If the loop fails to complete its purpose, then throw an exception, making the caller fully aware that the loop failed to complete its task.

Nested Loops

Earlier in Listing 11-16 we parallelized an algorithm to approximate the value of pi. While it was faster than the single-threaded version, the gain wasn’t earth shattering. The reason suggested for the lack of boost was the additional overhead introduced through the use of Parallel.For. The  fact that the loop body is now represented by a delegate, and as such each iteration of the loop now involves a delegate invocation this will drastically increase the amount of work for each iteration of the loop. With the amount of work per delegate invocation being so small the effect of parallelization is further lessened. This is a common problem with lots of loops, and as such we often find that one of the better structures to parallelize is not a single loop but nested loops. The code in Listing 11-22 has two loops; each iteration of the outer loop results in 100 iterations of the inner loop. The inner loop could therefore represent a reasonable amount of work. If the outer loop were to be parallelized, leaving the inner loop as the body, this could then represent a reasonable amount of work to compensate for the overhead of delegate invocation.

Listing 11-22.  Simple Nested Loop

for (int i = 0; i < 500; i++)
{
     for (int j = 0; j < 100; j++)
     {
         // Do Stuff
     }
}

Listing 11-23 shows how to parallelize Listing 11-22.

Listing 11-23.  Parallelized nested loop

Parallel.For(0, 500, i =>
{
    for (int j = 0; j < 100; j++)
    {
        // Do Stuff
    }
});

“All well and good,” you might say, “but what if my algorithm just has a single loop?” Any single loop can always be turned into an equivalent outer and inner loop as shown in Listing 11-24.

Listing 11-24.  Single Loop and Equivalent Nested Loop

for (int i = 0; i < 50000; i++)
{
    // Do Stuff
}
 
for (int i = 0; i < 500; i++)
{
     for (int j = 0; j < 100; j++)
     {
         // Do Stuff
     }
}

So the first task is to turn the single loop into two loops. Before we dive into how to produce the parallel version, see Listing 11-25, which shows a refactored nonparallel version of CalculatePI that now uses two loops. The outer loop is a foreach iterating over a stream of Tuple<int,int>. Each Tuple<int,int> represents a range of values for the inner loop to iterate over.

Listing 11-25.  CalculatePi Turned into Two Loops

public static double CalculatePi(int iterations)
{
  IEnumerable<Tuple<int, int>> ranges = Range(3, iterations, 10000);
  double pi = 1;
 
  foreach (var range in ranges)
  {
    double multiplier = range.Item1 % 2 == 0 ? 1 : -1;
 
    for (int i = range.Item1; i < range.Item2; i+=2)
    {
        pi += 1.0/(double) i*multiplier;
        multiplier *= -1;
    }
  }
  pi *= 4.0;
  return pi;
}
 
private static IEnumerable<Tuple<int, int>> Range(int start, int end, int size)
{
    for (int i = start; i < end; i+=size)
    {
        yield return Tuple.Create(i, Math.Min(i + size,end));
    }
}

Now that the algorithm has been refactored into two loops, you can parallelize the outer loop, leaving the inner loop to run synchronously and thus generate sufficient work to justify the overhead of delegate invocation, as shown in Listing 11-26.

Listing 11-26.  ParallelCalculatePi with Nested Loops

private static double TwoLoopsParallelCalculatePi(int iterations)
{
    IEnumerable<Tuple<int, int>> ranges = Range(3, iterations, 10000);
    double pi = 1;
    object combineLock = new object();
     
    Parallel.ForEach(ranges.ToList(),
        () => 0.0,
        (range, loopState, localPi) =>
        {
            double multiplier = range.Item1%2 == 0 ? 1 : -1;
 
            for (int i = range.Item1; i < range.Item2; i += 2)
            {
                localPi += 1.0/(double) i*multiplier;
                multiplier *= -1;
            }
 
            return localPi;
        },
        localPi =>
        {
            lock (combineLock)
            {
                pi += localPi;
            }
        });
    pi *= 4.0;
 
    return pi;
}

As explained earlier, when Parallel.ForEach executes it has to partition the IEnumerable stream, and it does this by combination of guessing and inspecting the actual source type of the stream. It is very hard to write a partitioner that is perfect for every scenario. Rather than Parallel.ForEach consuming an IEnumerable and also creating and using the default partitioner, you can supply it with a partitioner (see Listing 11-27). You can create your own partitioner that has explicit knowledge of how to partition your data source, but we won’t discuss that in this book. TPL ships with some out-of-the-box partitioners that you can take advantage of, made available through Create methods on the System.Collections.Concurrent.Partitioner class.

Listing 11-27.  Using Parallel.ForEach Using Partitioner Directly

private static double OptimizedParallelCalculatePiWithPartitioner(int iterations)
{
    //IEnumerable<Tuple<int, int>> ranges = Range(3, iterations, 10000);
    var ranges = Partitioner.Create(3, iterations, 10000);
 
    double pi = 1;
    object combineLock = new object();
    Parallel.ForEach(ranges, () => 0.0,
        (range, loopState, localPi) =>
        {
            double multiplier = range.Item1 % 2 == 0 ? 1 : -1;
 
            for (int i = range.Item1; i < range.Item2; i += 2)
            {
                localPi += 1.0 / (double)i * multiplier;
                multiplier *= -1;
            }
 
            return localPi;
        },
        localPi =>
        {
            lock (combineLock)
            {
                pi += localPi;
            }
        });
    pi *= 4.0;
 
    return pi;
}

Table 11-2 presents all the times from running the implementations of CalculatePi. The two-loops implementation gets us very close to linear scaling, albeit with some refactoring of the algorithm, so it’s not exactly a free lunch.

Partitioning

In the same way that Parallel.ForEach found it harder to partition up the data if it was given a pure IEnumerable<T>, the same is true for PLINQ. PLINQ works best with arrays and IList<T>. Listing 11-31 benchmarks the filtering of just even numbers from a range of numbers. Running this code produce an overall time of 1.09 seconds. The code is then modified as per Table 11-3 and rerun to obtain the time taken to execute that variant.

Listing 11-31.  Timed LINQ Query

static void Main(string[] args)
{
 
    IEnumerable<int> numbers =
                  Enumerable.Range(0, 100000000);
 
    TimeIt(() =>
    {
        var evenNumbers = from number in numbers
            where number%2 == 0
            select number;
 
        Console.WriteLine(evenNumbers.Count());
    });
}
 
private static void TimeIt(Action func)
{
    Stopwatch timer = Stopwatch.StartNew();
    func();
    Console.WriteLine("{0}() and took {1}",
        func.Method.Name, timer.Elapsed);
}

As shown in Table 11-3, for small, simple queries parallelism only works if PLINQ is able to partition efficiently. This very simple example hopefully tells you not to go through your entire code base modifying all LINQ statements to use AsParallel. Remember: Always benchmark code before attempting to optimize. Just because it is easy to throw on an AsParallel doesn’t mean you should.

Does Order Matter?

With conventional LINQ, items are processed in order as they are yielded from the IEnumerable<T> stream. With PLINQ, the processing order and the result order are unknown, just as it was with parallel loops. If you don’t care about order, all is good, but if you do need the results to come out in the same order as the source processing, then you need to tell PLINQ to preserve order. For example, consider a piece of PLINQ processing a CSV file row by row: its job is to take each row in turn, add a row total column, and then output the row to a new file. The order of the output CSV file should match that of the input CSV file. If you were to process this with just AsParallel(), then the output order could be different from that of the input. To preserve its order, call AsOrdered() as per Listing 11-32.

Listing 11-32.  Ordered Parallel Query

IEnumerable<string[]> csvRows = GetRows(@"....datastockData.csv").Skip(1);
 
var rowWithTotalColumn =
    from row in csvRows.AsParallel().AsOrdered()
    let total = row.Skip(1).Select(c => decimal.Parse(c)).Sum()
    select row.Concat(Enumerable.Repeat(total.ToString(), 1)).ToArray();
 
WriteCsvRows(rowWithTotalColumn, "RowsWithTotal.csv");

Preserving order comes at a cost, since it means the partitioner is more constrained about how it can partition and has to buffer up results before publishing them. Use AsOrdered for algorithms that need to stream data in, process it, and stream it out in the same order. If only part of the query needs to be ordered, then you can revert back to more efficient behavior by using the AsUnOrdered() method.

If all you need to care about is knowing which input index produced a given output, then a more efficient model is simply to flow an index with the data as per Listing 11-33. An example of running the code is shown in Figure 11-8. While the output order does not match that of the input, the output value does now contain a corresponding index of the input that produced the said output.

Listing 11-33.  Associating Result with Input Index

IEnumerable<string> values = new List<string>() { "one","two","three","four","five","six"};
 
var upperCases = values
    .Select((value, index) => new {Index = index, Value = value})
    .AsParallel()
    .Select(valueIndexPair => new {valueIndexPair.Index,Value=valueIndexPair.Value.ToUpper()});
       
foreach (var upperCase in upperCases)
{
    Console.WriteLine("{0}:{1}",upperCase.Index,upperCase.Value);
}