Continuing tasks in TPL

This article will present code where a task is continued with another task in TPL. Let's look at some code for this:

  class Program
    {
        static void Main(string[] args)
        {
            var task = Task.Factory.StartNew(() =>
            {
                Console.WriteLine("Running the first task ...");
                Thread.Sleep(2000);
                Console.WriteLine("First task done");
            }).ContinueWith((antecendent) =>
             {
                 Console.WriteLine("Continuing the second task ...");
                 Thread.Sleep(3000);
                 Console.WriteLine("Second task done");
             }, TaskContinuationOptions.NotOnFaulted | TaskContinuationOptions.NotOnCanceled);

            task.Wait(); 

            Console.WriteLine("Press any key to continue ...");
            Console.ReadKey(); 

        }
    }

A task is started with Task.Factory.StartNew and the empty lambda contains a block following that contains the code that will run the work of the first task. Now, the ContinueWith extension method is used next to continue with the next task. I have specified TaskContinuationOptions here with NotOnFaulted and NotOnCanceled, this is an overload and one does not have to specify this always. An often used continuing TaskContinuationOption is OnlyOnFaulted. Let's look last at some code that actually returns some data and gives that data to the next task continuing the work. This makes it possible to partition the work performed in much like a pipeline, one calls this kind of parallelism as dataflow parallelism.

 class Program
    {
        static void Main(string[] args)
        {
            var task = Task.Factory.StartNew(() =>
            {
                Console.WriteLine("Running the first task ...");
                Thread.Sleep(2000);
                Console.WriteLine("First task done, returning some result");
                return 42;
            }).ContinueWith((antecendent) =>
             {
                 Console.WriteLine("Continuing the second task ...");
                 Thread.Sleep(3000);
                 Console.WriteLine("Got result from first task: {0}", antecendent.Result);
                 Console.WriteLine("Second task done");
                 return antecendent.Result * 2; 
             }).ContinueWith((antecendent) => {
                 Console.WriteLine("Continuing the third task ...");
                 Thread.Sleep(3000);
                 Console.WriteLine("Got result from second task: {0}", antecendent.Result);
                 Console.WriteLine("Second task done");
                 return antecendent.Result * 2; 
             });  

            task.Wait();

            Console.WriteLine("Final result: {0}", task.Result);

            Console.WriteLine("Press any key to continue ...");
            Console.ReadKey();

        }

In the code above, one not only continue a task with another task, but then continue with a third task. There can be arbitrary number of tasks being chained in this way. We propagate the result from the previous example by accessing the Result of the antecendent task (previous task). This makes it possible to partition a problem into stages - a pipeline of tasks doing individual processing to create the final result, much like how a factory works. It is possible to propagate values that are not only value types such as int, but also Reference types (arbitrary complex classes). This shows that TPL is able to help you compose the total amount of work to be processed into stages in a pipeline fashion in a simple way. Keep in mind that there are more optimized way of doing this following a more advanced Producer-Consumer approach if one wishes to implement a true pipeline with TPL.

Simple child tasks in TPL

This short article will present some code that displays how to perform the steps necessary to create a child task in the Task Parallel Library (TPL). To create a child task, simple specify the TaskCreationOptions set to TaskCreationOptions.AttachedToParent. The simple console application presents this next:

    var nums = ParallelEnumerable.Range(0, 10000);

    var parent = Task.Factory.StartNew(() =>
            {
                Console.WriteLine("Parent task starting");
                Thread.Sleep(2000);
                Console.WriteLine("Parent task done");

                   Task.Factory.StartNew(() =>
                    {
                        Console.WriteLine("Child task starting");
                        Thread.Sleep(5000);
                        Console.WriteLine("Child task done");
                    
                    }, TaskCreationOptions.AttachedToParent);               

            });

      parent.Wait(); 

      Console.WriteLine("Press any key to continue ...");
      Console.ReadKey(); 
                     

The console app code above waits for the parent task and will not continue to the next line until not only the parent task is performed and executed, but also its child tasks are finished. In other ways, a task can have multiple child tasks. However, only the child tasks that specifies a TaskCreationOptions set to AttachedToParent will actually be waited upon. Also note that creating a next Task is done using the Task.Factory.StartNew calls. This is the preferred option to create new tasks, but one can also instantiate a Task with the new keyword, but then you have to remember to start the task also. Task.Factory.StartNew not only instantiates a new task but also starts it immediately. In the next short article, the ContinueWith construct in TPL will be presented. This is a construct that allows one to pipeline tasks, one after another. This is known as dataflow parallelism.

Map and Reduce in PLinq

This article will present some code that can be used to apply a Map and a Reduce transformation of a provided input array of a given type of objects, using PLINQ. For developers familiar with LINQ, PLINQ is the parallel version of Linq. First, the source code will be presented, then the code will be explained.

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

namespace TestParallelMap
{
    class Program
    {

        static void Main(string[] args)
        {
            var nums = Enumerable.Range(0, 1001).ToArray(); 
            Func<int,TaggedResult<int,double>> mapFunction = n => new TaggedResult<int,double>(n, Math.Pow(n, 2)); 
            var result = ParallelMap(nums, mapFunction);

            foreach (var r in result)
            {
                Console.WriteLine("Got result {0} for the num {1}", r.OutputValue, r.InputtedValue);
            }

            Console.WriteLine("Press the any key to continue ...");
            Console.ReadKey();

            var resultSecond = ParallelReduce(nums, 0, (first, second) => first + second);

            Console.WriteLine("Got result {0}", resultSecond);  
         
            Console.WriteLine("Press the any key to continue ...");
            Console.ReadKey();

        }

        private static TValue ParallelReduce<TValue>(TValue[] source, TValue seedValue, Func<TValue, TValue, TValue> reduceFunction)
        {
            return source.AsParallel().AsOrdered().Aggregate(
                seedValue,
                (localResult, nextValue) => reduceFunction(localResult, nextValue),
                (overallResult, localResult) => reduceFunction(overallResult, localResult), 
                 overallResult => overallResult); 
                  
        }

        private static TOutput[] ParallelMap<TInput, TOutput>(TInput[] source, Func<TInput, TOutput> mapFunction) 
        {
            return source.AsParallel().AsOrdered().Select(x => mapFunction(x)).ToArray(); 
        }

        private class TaggedResult<TInput, TOutput>
        {
            public TaggedResult(TInput inputtedValue, TOutput outputtedValue)
            {
                InputtedValue = inputtedValue;
                OutputValue = outputtedValue; 
            }

            public TInput InputtedValue { get; set; }

            public TOutput OutputValue { get; set; }
        }

    }
}

The source code presented above is a simple console application. It is written in C# (obviously) and uses PLINQ to prosess the input array. The Map method will transform each value in the input array with a provided mapfunction. The source code uses Func to provide a function pointer to the method that will be executed to produce each item. The generic arguments control the type of the resulting array:

        private static TOutput[] ParallelMap<TInput, TOutput>(TInput[] source, Func<TInput, TOutput> mapFunction) 
        {
            return source.AsParallel().AsOrdered().Select(x => mapFunction(x)).ToArray(); 
        }

Note that PLINQ is very similar to LINQ and the main difference is the AsParallel() method. Also note that AsOrdered() will mean that the ordering will be maintained in the output array. If the order does not matter, there is a performance gain by omitting the call to AsOrdered().. Further, note that one uses the Select method of PLINQ to perform a projection where one at the same time performes the map method. The Reduce method is a bit more complicated. It uses PLINQ's Aggregate method, which really is a bit tough of understanding. Let's repeat the source code:

  private static TValue ParallelReduce<TValue>(TValue[] source, TValue seedValue, Func<TValue, TValue, TValue> reduceFunction)
        {
            return source.AsParallel().AsOrdered().Aggregate(
                seedValue,
                (localResult, nextValue) => reduceFunction(localResult, nextValue),
                (overallResult, localResult) => reduceFunction(overallResult, localResult), 
                 overallResult => overallResult); 
                  
        }

Actually, the PLINQ aggregate method is very versatile and powerful method when you want to process an entire array and perform a reduction method, in our case just a summation, but this could be anything, for example statistical methods and so on. The first argument is the seed value, the next parameter is a lambda expression that will perform the stepwise part of the reduction, the next parameter is the lambda expression that merges the stepwise result with the overall result - and the final fourth parameter is just a simple lambda which tells we do not want to do anything with the final overall result than propagate the value to the processed final result which then is returned. There are many more patterns that are a bit more complicated too, like MapReduce (which combines map and reduce), but this will be a relatively gentle introduction to the functionality that PLINQ presents. It is impressive to think that all the thread handling will be automatically be handled by PLINQ. From relatively simple code one can perform a multitude of different transformation from this source code with Map and Reduce, powered by PLINQ and generics.