Setting Merge Options
When you execute a PLINQ query, Tasks are assigned to process blocks of source data to produce a series of results, which are then consumed typically by a sequential enumeration. You can control how the results are passed from the Tasks to the consumer by using the WithMergeOptions method on an instance of ParallelQuery. This method takes a value from the ParallelMergeOptions enumeration, which is summarized in Table 6-4. Table 6-4. Values of the System.Linq.ParallelMergeOptions Enumeration...
Determining If a Task Was Cancelled
You can determine if a Task has been cancelled by checking the IsCancelled property, which will return true if the Task was cancelled. Listing 2-12 demonstrates the use of this property. Listing2-12. Using the Task.isCancelled Property create the cancellation token source CancellationTokenSource tokenSource1 new CancellationTokenSource create the cancellation token CancellationToken token1 tokenSource1.Token create the first task, which we will let run fully Task task1 new Task gt...
Description 1
Wait int, CancellationToken Wait TimeSpan, CancellationToken Reset the event using the original counter value or the value specified. Decrement the condition counter by one or by the amount specified. Try to increment the condition counter, and return true if the counter is incremented. Wait indefinitely for the event to be set. Wait for the event to be set, for a period of time to pass, or for a cancellation token to be cancelled. Return the number of times that Signal must be called before...
Using Custom Partitioning
PLINQ uses the same partitioning approach that you saw in the previous chapter. You can specify a customer partitioner for your PLINQ query by using the version of the AsParallel method that takes an instance of the class as an argument. The previous chapter showed you how to create a custom dynamic partitioner, where the number of partitions is not known in advance. Parallel loops support only dynamic partitioners. PLINQ supports the simpler static partitioners, which are much easier to write,...
Common Parallel Algorithms
In this chapter are 11 common parallel algorithms implemented using the TPL. You can download all of the source code for all of the algorithms from www.Apress.com. These are some of the most widely used building blocks in parallel programming, and I have included them as a reference for when you are facing a problem, a time-saver for when you need a quick answer, and a demonstration of how the TPL abstractions we covered throughout this book allow complex and rich functions to be developed with...
Using Wait Handles and the Mutex Class
Wait handles are wrappers around a Windows feature called synchronization handles. Several .NET synchronization primitives that are based on wait handles, and they all derive from the System.Threading.WaitHandle class. Each class has slightly different characteristics, and we'll walk through each of them in the next chapter when we explore coordinating Tasks. The wait handle class that has most relevance to avoiding data races and is System.Threading.Mutex. Listing 3-11 shows the basic use of...
AutoResetEvent
AutoResetEvent is similar to ManualResetEventSlim, but the event is reset automatically after each call to the Set method, and only one waiting worker Task is released each time the event is set. There is no lightweight alternative to AutoResetEvent, and being a classic primitive, it has no method that allows waiting using a CancellationToken. Table 4-9 describes the key members of the AutoResetEvent class. Table 4-9. Key Members of the System.Threading.AutoResetEvent Class
Configuring Interprocess Synchronization
Wait handles can be shared between processes. The Mutexes in the previous two listings were local, meaning that they were only usable in one process a local Mutex is created when you use the default constructor. You can also create a named system Mutex, which is the kind that can be shared between processes. You do this by using the overloaded constructors that take a name argument. When using a named Mutex, it is important to see if the Mutex you are looking for has already been created,...
Using a Custom Task Scheduler
The task scheduler is the point at which tasks and threads meet, and it is responsible for assigning threads to execute your Tasks and deciding how many Tasks will be executed in parallel and in which order. As Chapter 1 explained, the TPL is built on the foundation of the classic threading features, and you have to understand these to build a custom scheduler. I am not going to explain the classic features in this book, but I will show you how to use them to create a custom scheduler. This is...
Using the ReaderWriterLockSlim Class
The System.Threading.ReaderWriterLockSlim class provides a convenient implementation of reader-writer locking that takes care of managing the locks. This class is a lightweight alternative for the heavyweight System.Threading.ReaderWriter, which Microsoft no longer recommends using. The lightweight version is simpler to use, offers better performance, and avoids some potential deadlocks. You acquire and release the ReaderWriterLockSlim read lock by calling the EnterReadLock and ExitReadLock...
Writing a Contextual Partitioner
The default partitioner and the chunking partitioner both operate on any data type. One advantage of writing a customer partitioner is that you can tailor your strategy to the data type that you need to process. This section will demonstrate how to implement a contextual partitioner and, in doing so, explain how to extend the Partitioner class to implement a custom technique. To start our partitioner, we need a context some data type with characteristics that we are going to specialize. I have...
Cancelling PLINQ Queries
PLINQ supports cancellation using the CancellationTokens you have seen throughout this book. You associate a token with a query using the WithCancellation extension method. Listing 6-15 demonstrates cancellation by using a token that is cancelled by a Task while the results of a query are being enumerated. Listing 6-15. Cancelling PLINQ Queries using System using System.Collections.Generic using System.Linq using System.Threading using System.Threading.Tasks namespace Listing_15 class...
class ContextPartitioner OrderablePartitioner
the set of data items to partition protected WorkItem dataItems the target sum of values per chunk protected int targetSum the first un-chunked item private long sharedStartIndex 0 lock object to avoid index data races private object lockObj new object the object used to create enumerators private EnumerableSource enumSource public ContextPartitioner WorkItem data, int target base true, false, true set the instance variables from the parameters dataItems data targetSum target create the...
Creating ManytoOne and AnyToOne Continuations
The continuations we have seen so far have been one-to-one or one-to-many that is, one antecedent has one or more continuations. You can also perform many-to-one continuations using the ContinueWhenAll and ContinueWhenAny methods of the System.Threading.Tasks.TaskFactory class. You obtain an instance of TaskFactory through the static Task.Factory property. The ContinueWhenAll and ContinueWhenAny methods both take an array of Tasks argument. ContinueWhenAll schedules a continuation to be...
Using the Code Uio
The following code defines three very simple functions and chains them together with a Pipeline. Sequential values are generated and put into the pipeline, along with a callback that filters the results and prints out selected values to the Console. create a set of functions that we want to pipleline together Func lt int, double gt funcl input gt Math.Pow input, 2 Func lt double, double gt func2 input gt input 2 Func lt double, bool gt func3 input gt input 2 0 amp amp input gt 100 Action lt...
MemberDescription 1
QueueTask This method is called by the TPL when a Task has been created and TryExecuteTask This one is called by the scheduler to synchronously execute a Task that has been previously passed to the scheduler via the QueueTask method. The Task will not be executed if it is already executing or has already been executed. It returns true if the task was executed. TryExecuteTaskInLine This is called by the TPL to request that a Task be executed inline, and returns true if your scheduler executes...
Canceling Continuations
The techniques to handle cancellations for single Tasks, which we covered in the previous chapter, can be applied to continuations. The Task.ContinueWith , TaskFactory.ContinueWhenAll , and TaskFactory.ContinueWhenAny methods all have overloaded versions that accept a CancellationToken, which you can obtain by creating an instance of CancellationTokenSource. Listing 4-7 demonstrates canceling continuations. An antecedent Task is created and waits using a CancellationToken wait handle. When the...
Parallel vs Sequential Loops
For and foreach loops are cornerstones of C development, and it doesn't take long for a new programmer to reach the point where they become a frequently used and sometimes abused tool. The for and foreach keywords create sequential loops, so called because an iteration isn't started until the previous iteration has completed. The TPL contains support for parallel loops that is to say, loops in which the iterations are performed with some degree of concurrency. You can see a simple example in...
Using BlockingCollection as IEnumerable
The BlockingCollection class can be used with foreach loops to allow consumers to safely take items from the collection. However, the input to the loop must be the result of the method and not the collection itself. If the collection is used, the foreach loop will immediately exit if it is executed before the first item has been added to the collection by the producer. The BlockingCollection class implements the IEnumerable lt gt interface, so the compiler will not generate an error for this...
Using Thread Local Storage in Parallel Loops
Chapter 3 described thread local storage TLS , where each thread has its own isolated version of a variable. The Parallel.For and Parallel.ForEach methods have versions that support TLS, which you can use to efficiently share data between parallel loop iterations and to get a single result value from a parallel loop. A simple parallel loop that requires synchronization follows. The total variable is shared between the iterations of the loop and, since the iterations may be performed...
Using the Ordered Default Partitioning Strategy
When you use the Parallel.ForEach method, the default partitioning strategy will be applied unless the first argument you provide is an instance of the Partitioner or OrderablePartitioner classes in the System.Collections.Concurrent namespace. In the previous example, we applied the chunking strategy by creating an instance of OrderablePartitioner and passing it to Parallel.ForEach . But now, it is time for me to admit that I fudged things slightly to make the example simpler. The...
ConcurrentStack
The class implements a last in, first out LIFO queue taking an item from the queue returns the most recently added item. Items are added to the stack using the Push and PushRange methods and inspected and retrieved using the TryPeek , TryPop , and TryPopRange methods. Listing 3-19 demonstrates the use of the ConcurrentStack collection. Listing3-19. Using the ConcurrentStack Class using System using System.Collections.Concurrent using System.Threading using System.Threading.Tasks ConcurrentStack...
Parallel LINQ
Listing 6-1. A Parallel LINQ Query using System using System.Collections.Generic using System.Linq namespace Listing_0l class Listing_0l for int i 0 i lt sourceData.Length i sourceData i i from item in sourceData.AsParallel where item 2 0 select item Console.WriteLine Item 0 , item Console.WriteLine Press enter to finish Language Integrated Query, or LINQ pronounced link adds native data querying features to .NET. Parallel LINQ, known as PLlNQ, is a parallel implementation of LINQ to objects...
Debugging Program State
The best approach is to use the Visual Studio debugger, which contains some very useful new parallel features to support parallel programming. To look at these features, we are going to use the code in Listing 7-4. This program has no useful value other than it causes a number of Tasks to call a number of methods to demonstrate the debugger features. The CountDownEvent is used so that the main application thread can wait until all of the Tasks have been created and scheduled. The SemaphoreSlims...
ConcurrentQueue
The ConcurrentQueue class implements a first in, first out FIFO queue, which means that when you take items from the queue, you get them in the same order in which they were added. To place an item into a ConcurrentQueue, you call the Enqueue method. To take the first item in the queue, you call TryDequeue and to get the first item in the queue without taking it, you call TryPeek . TryDequeue and TryPeek take a parameter of the collection type, modified by the out keyword and return a bool...
Using Spin Waiting
The spin waiting technique is included in this chapter for completeness, but I recommend against using it. When you use the other two sleep techniques, the thread that is performing your task gives up its turn in the schedule when its sleeping, so any other threads can have a turn. The scheduler, which is responsible for managing the threads running at any given time, has to do some work to determine which thread should go next and make it happen. You can avoid the scheduler having to do this...
Writing an Orderable Contextual Partitioner
Having written a regular partitioner, it is a simple matter to create an orderable version. We won't go through this example member by member, because it is so similar to the previous listings. In fact, only four key things need to change to make our Partioner into an OrderablePartitioner. The first thing we have to change is the return types. Instead of returning enumerations of WorkItem, we are required to operate around instances of KeyValuePair lt long, WorkItem gt , where the long value is...
Monitoring Cancellation with a Delegate
You can register a delegate with a CancellationToken, which will be invoked when the CancellationTokenSource.Cancel method is called. You can use this as an alternative to the method shown in Listing 2-7 for checking cancellation, which can be useful if your task relies on other asynchronous operations, such as I O reads. You can also use the delegate feature to be notified when a cancellation happens this can be useful in UI applications. Listing 2-8 shows how the delegate feature can be used....
SemaphoreSlim
The System.Threading.SemaphoreSlim class allows you to specify how many waiting worker Tasks are released when the event is set, which is useful when you want to restrict the degree of concurrency among a group of Tasks. The supervisor releases workers by calling the Release method. The default version releases one Task, and you can specify how many Tasks are released by providing an integer argument. The constructor requires that you specify how many calls to the Wait method can be made before...
Creating Selective Continuations
By default, continuation Tasks are automatically scheduled when the antecedent Task completes. We can be selective about scheduling continuations by using the values of the enumeration when calling the Task.ContinueWith method. Table 4-2 details the enumeration values. Table 4-2. Key Values of the Enumeration Table 4-2. Key Values of the Enumeration This is equivalent to not specifying a value that is, the continuation will be scheduled to run when the antecedent completes. The continuation...
Setting Parallel Loop Options
You can influence the behavior of parallel loops by supplying an instance of the ParallelOptions class as an argument to Parallel.For or Parallel.ForEach . There are three properties in ParallelOptions, and they are described in Table 5-2. Table 5-2. Properties of the System.Threading.Tasks.ParallelOptions class Table 5-2. Properties of the System.Threading.Tasks.ParallelOptions class Get or set a CancellationToken see the Canceling Parallel Loops section Get or set the maximum concurrency for...
Using Parallel MapReduce
The final piece in the mapping and reduction set is where a Map algorithm and a Reduce algorithm are combined to create a MapReduce algorithm. There are three stages to MapReduce. First, a sequence of values is mapped to a sequence of intermediate results typically, each source value is mapped to multiple intermediate values, but this is not always the case. Second, the intermediate values are grouped together using a key. Finally, each group of intermediate values is reduced to a result value.
Creating a BlockingCollection instance
The first step in the listing is creating the BlockingCollection, which I have done using the default constructor. The class is strongly typed, so I have had to declare the type that will be in the collection, in this case Deposit. BlockingCollection lt Deposit gt blockingCollection new The default constructor creates unbounded collections, which means that there is no limit to the number of outstanding work items in the collection. The producers can put items in the collection at a much faster...
Getting Loop Results
The result returned by the Parallel.For and Parallel.ForEach methods is an instance of the structure. ParallelLoopResult contains two properties that you can use to determine if Break or Stop was called. Table 5-4 summarizes the properties. Table 5-4. Properies of the structure Table 5-4. Properies of the structure Return true if all of the loop iterations were completed without Stop and Return the index of the lowest iteration in which the Break method was Listing 5-9 shows the use of the...
Reading the Task Properties
An alternative to catching the exceptions is to use the properties of the Task class, in particular, the IsCompleted, IsFaulted, IsCancelled, and Exception properties. You still have to catch AggregateException when you call any of the trigger methods, but you can use the properties to determine if a task has completed, thrown an exception, or been cancelled, and if an exception was thrown, you can get the details of the exception. Listing 2-21 shows you how to use the Task properties. Listing...
Handling Exceptions
There are no special features for propagating exceptions through a continuation chain. Any exceptions thrown by any Task in a continuation chain must be processed, or they will be treated as unhandled exceptions when the finalizer for the Task is performed. See the previous chapter for details of processing Task exceptions. Listing 4-8 illustrates the problem with exceptions in chains. Listing 4-8. Unhandled Exceptions in Continuation Chains using System using System.Threading.Tasks namespace...
Using a Cancellation Token Wait Handle
The best way to put Tasks to sleep is to use the wait handle of a CancellationToken, which you saw earlier in the Cancelling Tasks section. Create an instance of CancellationTokenSource, and read the Token property to obtain the CancellationToken instance. Use the WaitHandle property, and call the overloaded WaitOne method. In the Cancelling Tasks section, you saw the version that takes no arguments, which causes the calling thread to wait until the CancellationTokenSource.Cancel method is...
Monitoring Cancellation with a Wait Handle
The third way to monitor task cancellation is to call the WaitOne method of the CancellationToken. WaitHandle property. I cover wait handles in depth later in this book, but for this chapter, it is enough to know that when you call the WaitOne method it blocks until the Cancel method is called on the CancellationTokenSource that was used to create the token whose wait handle you are using. Listing 2-9 demonstrates the use of the wait handle for cancellation monitoring. Two Tasks are created,...
Cancelling Tasks
One of the new areas of standardization in the TPL is cancelling tasks. This may seem like an odd thing to regard as useful, especially if you are accustomed to writing your own cancellation code using classic .NET threads. The new approach makes parallel code simpler and more consistent and reduces the risk of encountering some of the most commonly encountered problems when performing a cancellation, as you will see when we discuss putting a thread to sleep later in this chapter. Creating a...
Using Spin Locking
We encountered spinning in the last chapter. Typically, when waiting to acquire a regular synchronization primitive, your Task is taken out of the schedule waits until it has acquired the primitive and can run again. Spinning takes a different approach the Task enters a tight execution loop, periodically trying to acquire the primitive. Spinning avoids the overhead of rescheduling the Task because it never stops running, but it doesn't allow another Task to take its place. Spinning is useful if...
ConcurrentDictionary
The ConcurrentDictionary class implements a collection of key-value pairs. Like the other collection classes in the System.Collections.Concurrent namespace, ConcurrentDictionary provides methods whose names are prefixed with Try and returns bool results if they operate successfully. Table 3-9 describes the key members of the ConcurrentDictionary class. Table3-9. Key members of Try to add a new key-value pair to the collection. Return true if the pair was added successfully. Try to get the value...
Using Recursion and Upgradable Read Locks
Listing 3-15 separates the code that reads the shared data from the code that modifies it. Often, you will want to read data and make a change only if some condition is met. You could acquire the write lock to do this, but that requires exclusive access. Because you don't know in advance if you actually need to make changes, that would be a potential performance problem. But you are thinking, Aha I can acquire the nonexclusive read lock, perform the test, and then acquire the exclusive write...
Creating Child Tasks
A child Task, sometimes known as a nested Task, is one that is created inside the Task body of another. The Task in which the child is created is known as the parent. There are two kinds of child Task detached and attached. A detached Task, as demonstrated in Listing 4-10, has no special relationship with its parent the child will be scheduled and can be performed concurrently with the parent but has no impact on the parent itself. Create a new Task within the body of an existing one. Create a...
Description Juy
int Timespan int, CancellationToken Add an item of type T to any one of the BlockingCollection lt T gt instances specified, blocking until the item has been added to a collection, the specified time period has elapsed, or the CancellationToken has been cancelled. Return an int representing the array index of the BlockingCollection to which the item was added, or return -1 if the item could not be added to any of the collections. out T out T, CancellationToken Take an item from any one of the...
Performing Parallel Analysis with Visual Studio
Visual Studio 2010 includes the Concurrency Visualizer, which allows you to examine the behavior of your parallel program, albeit it with some significant limitations and frustrations. To use the Concurrency Visualizer, Visual Studio must be started with Administrator privileges. To do this, find Visual Studio in your Start menu, right-click it, and select Run as administrator. Load your project, and select Start Performance Analysis from the Debug menu. Note If you are running Windows 64-bit,...
Creating a Composite Cancellation Token
You can create a token that is composed from several CancellationTokens that will be cancelled if any of the underlying tokens is cancelled. You do this by calling the System.Threading. method and passing in the CancellationTokens that you want to link. The result is a new CancellationToken that you can use normally. Listing 2-11 demonstrates creating and using a composite cancellation token. Listing 2-11. Using a Composite Cancellation Token create the cancellation token sources...