Understanding Task Parallelism Framework (TPL) using C#

Objective : In my previous article Parallel Programming in C# at a glance I had briefed about Parallel Programming Framework. It can be used as a quick reference while writing your code. In this article you will understand the basic concepts of Task Parallelism. After completing this article, you should be able to create and manage tasks with parallel processing.

Task : It refers to an operation to be completed. It is a unit of work or processing.

Task Parallelism : It refers to one or more independent asynchronous tasks running concurrently i.e. in parallel.

You can create as many tasks as you want. All these tasks will be queued to the ThreadPool and managed by the framework. ThreadPool automatically adjust and maintain the number of threads maximizing throughput. It allocates the CPU cores to the tasks balancing load. It assures that all the CPU cores are utilized as much as possible. It keeps you distant from the complexities involved in creating, maintaining and disposing the threads.

Task Parallelism provides you the features to support waiting, cancellation, continuation, exception handling, detailed status and many more.

You can achieve task parallelism in your application either Implicitly or Explicitly.

Note : You can use delegate, anonymous method or lambda expression instead of Action/Func. In this article, I will be using any of them. I may also provide partial code, if possible, to avoid unnecessary code.

Create and run tasks implicitly

• Tasks are created and maintained implicitly by the framework.
• You should use this technique when you need to process some tasks which do not return any value and do not need more control over tasks.

Parallel.Invoke() – Executes each of the provided tasks/actions, possibly in parallel.

Example

static void Main()
{
 // Retrieve Darwin's "Origin of the Species" from Gutenberg.org.
 string[] words = CreateWordArray(@"http://www.gutenberg.org/files/2009/2009.txt");
 #region ParallelTasks
 // Perform three tasks in parallel on the source array
 Parallel.Invoke(() =>
 {
     Console.WriteLine("Begin first task...");
     GetLongestWord(words);
 }, // close first Action
 () =>
 {
     Console.WriteLine("Begin second task...");
     GetMostCommonWords(words);
 }, //close second Action
 () =>
 {
    Console.WriteLine("Begin third task...");
    GetCountForWord(words, "species");
 } //close third Action
 ); //close parallel.invoke
    Console.WriteLine("Returned from Parallel.Invoke");
 #endregion
    Console.WriteLine("Press any key to exit");
    Console.ReadKey();
}

Parallel.Invoke(ParallelOptions, Action[]) – Executes each of the provided tasks/actions, possibly in parallel, unless the operation is cancelled by the user.

This method enables you to configure the behavior of the operation. Using ParallelOptions, you can configure CancellationToken, MaxDegreeOfParallelism – maximum number of concurrent tasks enabled, and TaskScheduler. Here one thing is to note that if you configure the value of MaxDegreeOfParallelism as say n then framework will require maximum n or less threads to process the task.

Example

static void Main(string[] args)
        {
            try
            {
                CancellationTokenSource cts = new CancellationTokenSource();
                CancellationToken ct = cts.Token;
                ParallelOptions po = new ParallelOptions { CancellationToken = ct, MaxDegreeOfParallelism = System.Environment.ProcessorCount };

                Parallel.Invoke(po,
                        new Action(() => DoWork(1, ct)),
                        new Action(() => DoWork(2, ct)), 
                        new Action(() => DoWork(3, ct)),
                        new Action(() => DoWork(4, ct)),
                        new Action(() => DoWork(5, ct)),
                        new Action(() => DoWork(6, ct)),
                        new Action(() => { cts.Cancel(); }), 
                        new Action(() => DoWork(7, ct)),
                        new Action(() => DoWork(8, ct))
                    );                
            }
            catch (OperationCanceledException e)
            {
                Console.WriteLine(e.Message);
            }
            Console.ReadKey();
        }

For both of the above methods, you find the words “possibly in parallel”. It means that the framework will check if the tasks can be executed using multiple threads. However if it is not able to get multiple threads, the tasks may be executed using single thread as well.

Create and run tasks explicitly

• Tasks are created and maintained explicitly by the developer.
• You should use this technique when either you need more control over the tasks or process the return values.

Task Parallelism Framework provides you different flavours of Task class object creation. In this way, you can create different types of Tasks for specific scenarios.

Task constructors which can be used to create task with no return value. However you have control over task behaviour.

Task(Action) – Initializes a new Task with the specified action.
Task(Action, CancellationToken) – Initializes a new Task with the specified action and CancellationToken.
Task(Action, TaskCreationOptions) – Initializes a new Task with the specified action and creation options.
Task(Action<Object>, Object) – Initializes a new Task with the specified action and state.
Task(Action, CancellationToken, TaskCreationOptions) – Initializes a new Task with the specified action and creation options.
Task(Action<Object>, Object, CancellationToken) – Initializes a new Task with the specified action, state, and options.
Task(Action<Object>, Object, TaskCreationOptions) – Initializes a new Task with the specified action, state, and options.
Task(Action<Object>, Object, CancellationToken, TaskCreationOptions) – Initializes a new Task with the specified action, state, and options.

Example

public static void Main()
{
    // use an Action delegate and named method
    Task task1 = new Task(new Action(printMessage));
    // use an anonymous delegate
    Task task2 = new Task(delegate { printMessage(); });
    // use a lambda expression and a named method
    Task task3 = new Task(() => printMessage());
    // use a lambda expression and an anonymous method
    Task task4 = new Task(() => { printMessage(); });

    task1.Start();
    task2.Start();
    task3.Start();
    task4.Start();
    Console.WriteLine("Main method complete. Press <enter> to finish.");
    Console.ReadLine();
}

Task<TResult> constructors which can be used to create task with return value. You have control over task behavior as well. TResult is the type of value returned by the delegate or task.

Task<TResult>(Func<TResult>) – Initializes a new Task<TResult> with the specified function.
Task<TResult>(Func<TResult>, CancellationToken) – Initializes a new Task<TResult> with the specified function.
Task<TResult>(Func<TResult>, TaskCreationOptions) – Initializes a new Task<TResult> with the specified function and creation options.
Task<TResult>(Func<Object, TResult>, Object) – Initializes a new Task<TResult> with the specified function and state.
Task<TResult>(Func<TResult>, CancellationToken, TaskCreationOptions) – Initializes a new Task<TResult> with the specified function and creation options.
Task<TResult>(Func<Object, TResult>, Object, CancellationToken) – Initializes a new Task<TResult> with the specified action, state, and options.
Task<TResult>(Func<Object, TResult>, Object, TaskCreationOptions) – Initializes a new Task<TResult> with the specified action, state, and options.
Task<TResult>(Func<Object, TResult>, Object, CancellationToken, TaskCreationOptions) – Initializes a new Task<TResult> with the specified action, state, and options.

Example

public static void Main()
{
    int[] nums = Enumerable.Range(0, 1000).ToArray();

    var myTasks = nums.Select(num => Task.Factory.StartNew<Tuple<int, long>>(() => new Tuple<int, long>(num, num * num))).ToList();
    // Instead of simple calculation like num*num, you can write code to say download content from some site url

    DateTime threadStartTime = DateTime.Now;
    while (myTasks.Count > 0)
    {
        try
        {
            var waitTime = 2000;
            var index = Task.WaitAny(myTasks.Cast<Task>().ToArray(), (int)waitTime); // Wait for any task to complete and get the index of completed task
            if (index < 0) break;
            var currentTask = myTasks[index];
            myTasks.RemoveAt(index); // Remove the task from list since it is processed
            if (currentTask.IsCanceled || currentTask.IsFaulted) continue;
            if (currentTask.Result.Item1 == 0) continue;

            var input = currentTask.Result.Item1;
            var output = currentTask.Result.Item2;

            // You can process the downloaded content here one by one, which ever gets downloaded first.

            Console.WriteLine("Square root of {0} is {1}", input, output);
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
    }
    Console.WriteLine("Main method complete. Press <enter> to finish.");
    Console.ReadLine();
}

Important parameters, properties and methods which can be used for managing task

Here I will enlist all the important Task parameters, properties and methods which you will be using frequently while creating and managing tasks.

Some important parameters which can be used while creating task.

• CancellationToken – This token is used to cancel the task while task is already running.
• TaskCreationOptions – This is used to customize the task’s behavior. Important possible values for TaskCreationOptions are AttachedToParent, DenyChildAttach, LongRunning,PreferFairness.
• State object – It represents data to be used by the Action/Func/Task.

Some important properties which can be used to monitor current task.

• AsyncState – Gets the state object supplied when the Task was created, or null if none was supplied.
• Exception – Gets the AggregateException that caused the Task to end prematurely. If the Task completed successfully or has not yet thrown any exceptions, this will return null.
• IsCanceled – Gets whether this Task instance has completed execution due to being cancelled.
• IsFaulted – Gets whether the Task completed due to an un-handled exception.
• IsCompleted – Gets whether this Task has completed.
• Result – Gets the result value of this Task<TResult>. The data-type of Result would be TResult.
• Status – Gets the TaskStatus of this task. Possible values are Created, WaitingForActivation, WaitingToRun, Running, Cancelled, Faulted, WaitingForChildrenToComplete, RanToCompletion.

Some important methods which can be used to control current task.

• Start() – starts the Task, scheduling it for execution to the current TaskScheduler.
• Wait() - waits for the Task to complete execution. This methods has many overloads which accepts CancellationToken, time in milliseconds, or TimeSpan.

Example

// Wait on a single task with no timeout specified.
Task taskA = Task.Factory.StartNew(() => DoSomeWork(10000000));
taskA.Wait();
Console.WriteLine("taskA has completed.");
// Wait on a single task with a timeout specified.
Task taskB = Task.Factory.StartNew(() => DoSomeWork(10000000));
taskB.Wait(1000); //Wait for 1000 ms.

if (taskB.IsCompleted)
Console.WriteLine("taskB has completed.");
else
Console.WriteLine("Timed out before taskB completed.");

// Wait for all tasks to complete.
Task[] tasks = new Task[10];
for (int i = 0; i < 10; i++)
{
    tasks[i] = Task.Factory.StartNew(() => DoSomeWork(10000000));
}
Task.WaitAll(tasks);

// Wait for first task to complete.
Task<double>[] tasks2 = new Task<double>[3];

// Try three different approaches to the problem. Take the first one.
tasks2[0] = Task<double>.Factory.StartNew(() => DoAnotherWork(3000));
tasks2[1] = Task<double>.Factory.StartNew(() => DoAnotherWork(2000));
tasks2[2] = Task<double>.Factory.StartNew(() => DoAnotherWork(1000));
int index = Task.WaitAny(tasks2);
double d = tasks2[index].Result;
Console.WriteLine("task[{0}] completed first with result of {1}.", index, d);

• ContinueWith() – start new task after the current task has completed. It provides many constructors suitable for specific requirement.

Example

// The antecedent task. Can also be created with Task.Factory.StartNew.
Task<DayOfWeek> taskA = new Task<DayOfWeek>(() => DateTime.Today.DayOfWeek);

// The continuation. Its delegate takes the antecedent task as an argument and can return a different type.
Task<string> continuation = taskA.ContinueWith((antecedent) =>
{
    return String.Format("Today is {0}.", antecedent.Result);
    // antecedent.Result gives you the result of previous task in chain
});

// Start the antecedent.
taskA.Start();

// Use the contuation's result.
Console.WriteLine(continuation.Result);

• RunSynchronously() – runs the Task synchronously on the current TaskScheduler. By default, tasks are run asynchronously.

I hope it helps you understand the basic concepts of Task Parallelism. Please comment your query or suggestion which will help improve the article.