PLINQ (basics)

Language-Integrated Query (LINQ) has been implemented into C# to perform queries over any kind of data like eg. objects or databases. Parallel Language-Integrated Query (PLINQ) is a further approach to access objects. As the name tells us already, it has to do with parallelism. The evolution from sequential queries (LINQ) to parallel queries (PLINQ) was predictable. Extension methods for PLINQ are defined in the class System.Linq.ParallelEnumerable.

public static void PLINQ1() {
    int[] range = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };

    Console.WriteLine("old school");
    for (int i = 0, n = range.Length; i < n; i++) {         if (range[i] % 2 == 0) Console.WriteLine(range[i]);     }     Console.WriteLine("LINQ");     var linq = from i in range where (i % 2 == 0) select i;     foreach (int i in linq) Console.WriteLine(i);     Console.WriteLine("LINQ2");     var linq2 = range.Where(i => i % 2 == 0);
    foreach (int i in linq2) Console.WriteLine(i);

    Console.WriteLine("PLINQ1");
    var plinq = from i in range.AsParallel() where (i % 2 == 0) select i;
    foreach (int i in plinq) Console.WriteLine(i);

    Console.WriteLine("PLINQ2");
    var plinq2 = range.AsParallel().Where(i => i % 2 == 0);
    foreach (int i in plinq2) Console.WriteLine(i);

    Console.WriteLine("PLINQ3");
    var plinq3 = range.AsParallel().Where(i => { Thread.Sleep(1000); return (i % 2 == 0); });
    foreach (int i in plinq3) Console.WriteLine(i);

    Console.WriteLine("PLINQ3 sorted");
    var plinq3sorted = range.AsParallel().AsOrdered().Where(i => { Thread.Sleep(1000); return (i % 2 == 0); });
    foreach (int i in plinq3sorted) Console.WriteLine(i);

    Console.ReadLine();
} //

Interestingly the runtime decides by itself if it makes sense to execute your query as parallel. Thus parallel execution is not guaranteed unless you specify WithExecutionMode(ParallelExecutionMode.ForceParallelism).
You can even get more specific by using WithDegreeOfParallelism() and limit the number of processors being used.

public static void PLINQ2() {
    var range = Enumerable.Range(0, 25);
    var result = range.AsParallel().WithExecutionMode(ParallelExecutionMode.ForceParallelism).Where(i => i % 2 == 0);
    //var result = range.AsParallel().WithDegreeOfParallelism(4).Where(i => i % 2 == 0);
    //var result = range.AsParallel().WithExecutionMode(ParallelExecutionMode.ForceParallelism).WithDegreeOfParallelism(4).Where(i => i % 2 == 0);
    foreach (int i in result) Console.WriteLine(i);

    Console.ReadLine();
} //

And now we are facing something weird. We created a sorted result, but somehow the sorting gets lost! The reason is that unlike “foreach” ForAll() does not wait for all results before it starts executing.
Hence we see an unexpected result.

public static void PLINQ3() {
    var range = Enumerable.Range(0, 25);

    var plinq2sorted = range.AsParallel().AsOrdered().Where(i => { Thread.Sleep(1000); return (i % 2 == 0); });
    Console.WriteLine("sorted");
    foreach (int i in plinq2sorted) Console.WriteLine(i);
    Console.WriteLine("something is going wrong?");
    plinq2sorted.ForAll(i => Console.WriteLine(i));

    Console.ReadLine();
} //

example output:
sorted
0
2
4
6
8
10
12
14
16
18
20
22
24
something is going wrong?
6
0
4
2
14
10
12
8
22
18
16
20
24

Parallel queries can fail and throw exceptions. These exceptions will not stop the execution of the queries. They are collected and returned in one single Exception that can be caught by the usual try/catch block.

public static void PLINQ4() {
    var range = Enumerable.Range(0, 25);

    try {
        var plinq2sorted = range.AsParallel().AsOrdered().Where(i => {
            Thread.Sleep(500);
            if (i > 15) throw new ArgumentException("exception for number " + i);
            return (i % 2 == 0);
        });

        foreach (int i in plinq2sorted) Console.WriteLine(i);
    }
    catch (AggregateException e) {
        Console.WriteLine("Exception thrown for " + e.InnerExceptions.Count + " results");
        foreach (var innerException in e.InnerExceptions) Console.WriteLine(innerException.Message);
    }

    Console.ReadLine();
} //

example output:
Exception thrown for 8 results
exception for number 16
exception for number 20
exception for number 17
exception for number 21
exception for number 19
exception for number 23
exception for number 22
exception for number 18