[2008.08.06] The Task Parallel Library (TPL) [Part II/IV]

Using System.Threading.Tasks.Task

• in the TPL there are only two abstractions that are being exposed: Task and Parallel (which are replicable tasks) and all other abstractions are built on top of these two
• the System.Threading.Parallel class is useful for solving common data and task parallel problems
• it is implemented on top of the lower-level System.Threading.Tasks.Task class, which can be used directly to solve parallel problems with greater flexibility and control over the way work is partitioned
• Task queues work in a similar manner as the ThreadPool

Example: comparing the similarity in calling the Task and ThreadPool as both code snippets cause the provided delegate to be queued for asynchronous execution – but that is where the similarity ends

// using the ThreadPool

ThreadPool.QueueUserWorkItem(delegate { ... });

 

// using Task

Task.Create(delegate { ... });

• with the ThreadPool, referencing a queued work item in order to do things like wait on it is a manual and laborious process
• the Task class makes this operation much simpler

Example:

// [1] to create a task, call the static Create

// method and pass in and Action delegate

// which deals with the work to be done

static Task Create (Action action);

 

// [2] to the get the results of the work, call Wait

void Wait();

NOTE: between the creation point of the task and the point you call Wait(), the action will be potentially executed in parallel

• the static Create methods return a Task object that represents the newly created asynchronous operation and this instance then provides methods such as Wait
• so why the executed potentially in parallel and not a guaranteed parallelism?
  • the TPL is for speeding up a lot of work on multiple cores but it is not about fairness or asynchronous programming
  • as a programmer you are exposing potential parallelism in that the computations can be speeded up but the library does not guarantee this parallelism because it would imply a heavier weight mechanism in its design and implementation

Example: want to execute three methods and wait for them to complete

Task t1 = Task.Create(delegate { A(); });

Task t2 = Task.Create(delegate { B(); });

Task t3 = Task.Create(delegate { C(); });

t1.Wait();

t2.Wait();

t3.Wait();

 

// can  replace t1-t2.Wait with WaitAll

Task.WaitAll(t1,t2,t3)

 

// it can be made even simpler using the

//  Parallel.Invoke as it takes an array of actions

Parallel.Invoke(() => A(), () => B(), () => C());

• unlike the Invoke method, however, using the Task class directly allows for control over how the Task instances are created (using various overloads of Task.Create )
• it also allows for work to be done between the asynchronous invocations of each of the methods
• Task also provides cancellation support
• calling Cancel on a Task may remove the Task from the scheduler’s queue if the Task has not already begun execution
• for reliability reasons, if the Task has started execution, it will not be aborted
• instead, its IsCancelled property will be set to return true, which the Task itself can check during execution at periodic intervals
• the static Task.Current property will return the currently executing Task

[2008.07.29] The Task Parallel Library (TPL) [Part I/IV]

The Task Parallel Library (TPL) is exposed from two namespaces in the System.Threading.dll assembly

• System.Threading with focus on the Parallel type.
• System.Threading.Tasks with focus on the Task and Future<T> types.

Using System.Threading.Parallel

• this class is useful for solving data parallel problems, as it provides support for parallelizing loops and regions in .NET applications
• this functionality is exposed through a set of static methods on the Parallel type, namely For, ForEach, and Invoke

[1] Parallel.For

• in many loops, the iterations of the loop are independent, meaning that one iteration does not rely on results from or interfere with any other iteration
• the System.Threading.Parallel class supports the parallelization of such loops, whereby a developer can take a loop that executes sequentially and convert it to one where every iteration of the loop has the potential to run in parallel, provided enough processing cores are available
• Note: loops whose iterations are dependent or contain side-effecting operations will be incorrect when run in parallel without the addition of proper custom synchronization
• here is a comparison of the syntax for sequential and parallel versions of the for loop:

Sequential for loop

for (Int32 i = 0; i < N; i++)

{

    results[i] = Compute(i);

}

Parallel.For using a delegate

Parallel.For(0, N, delegate(int i)

{

    results[i] = Compute(i);

});

Parallel.For using an anonymous function

Parallel.For(0, N, i =>

{   // using a lambda instead of a

    // lengthier anonymous function

    results[i] = Compute(i);

});

• here is a side-by-side comparison of the differences between the sequential and parallel versions; note that <=> means "is replaced with"

for	<=>		Parallel.For
Int32 i = 0	<=>		0
i < N	<=>		N
i++	<=>
results[i] = Compute(i);	<=>		delegate(Int32 i) {results[i] = Compute(i);}); OR i => { results[i] = Compute(i);

• the type of N in the parallel version is inferred to be of type Int32 based on the value of the first parameter which is 0, the default for an integer
• notice that in both cases, sequential or parallel, the actual body of the for loop is identical – that has not changed at all
• if the machine that this code is running on has multiple cores, then the TPL will run the loop in parallel, but if it is just a single core, it will automatically run sequentially
• the Parallel.For construct is just a shortcut and the implementation of Parallel.For is built on top of other abstractions such as the Task API
  • Parallel.For spawns a bunch of parallel work and it does not return until all that work is done
  • this is possible because it is built on top of these Task abstractions that allow waiting and cancellations

Example:

XAML Code:

<Grid>

        <Grid.RowDefinitions>

            <RowDefinition Height="Auto" />

            <RowDefinition Height="Auto" />

            <RowDefinition Height="Auto" />

            <RowDefinition Height="Auto" />

        </Grid.RowDefinitions>

        <StackPanel Grid.Row="0" Orientation="Horizontal">

            <TextBlock FontSize="12" FontWeight="Bold">

                Time Taken with sequential for:</TextBlock>

            <TextBlock x:Name="textBlock1"></TextBlock>

        </StackPanel>

        <Polyline x:Name="pL"

                Grid.Row="1"

                Stroke="Red"

                StrokeThickness="1">

        </Polyline>

 

        <StackPanel Grid.Row="2" Orientation="Horizontal">

            <TextBlock FontSize="12" FontWeight="Bold">

                Time Taken with Parallel.For:</TextBlock>

            <TextBlock x:Name="textBlock2"></TextBlock>

        </StackPanel>

 

        <Polyline x:Name="pL2"

                Grid.Row="3"

                Stroke="Blue"

                StrokeThickness="1">

        </Polyline>

        <TextBlock x:Name="tb"></TextBlock>

 

    </Grid>

public partial class Polyline : Window

    {

        public Polyline()

        {

            InitializeComponent();

 

            // number of times to loop

            Int32 N = 40000;

 

            // create a Stopwatch to take time measurements

            Stopwatch sw1 = Stopwatch.StartNew();

            // create an array of points to plot

            Point[] temp1 = new Point[N];

            // compute each point in the Sine curve

            //  using sequential for

            for (Int32 i = 0; i < N; i++)

            {

                Double x = i * Math.PI;

                Double y = 40 + 30 * Math.Sin(x / 10);

                temp1[i] = new Point(x, y);

            }

            // get the time taken to run this loop

            Int64 t1 = sw1.ElapsedTicks;

 

 

            Stopwatch sw2 = Stopwatch.StartNew();

            Point[] temp2 = new Point[N];

            // use the Parallel.For construct

            Parallel.For(0, N,

                i =>

                {

                    Double x = i * Math.PI;

                    Double y = 40 + 30 * Math.Sin(x / 10);

                    temp2[i] = new Point(x, y);

 

                });

            Int64 t2 = sw2.ElapsedTicks;

 

            // write the elapsed time taken for

            //  each loop

            textBlock1.Text = t1.ToString();

            textBlock2.Text = t2.ToString();

 

            // draw the sine curve for each

            //  loop

            for (int i = 0; i < N; i++)

            {

                pL.Points.Add(temp1[i]);

                pL2.Points.Add(temp2[i]);

            }

        }

    }// end class

[2] Parallel.ForEach

• parallelism is supported for IEnumerable<T> types using the foreach operator

Sequential foreach loop

foreach (MyClass c in data)

{

    Compute(c);

}

Parallel.ForEach using a delegate

Parallel.ForEach(data, delegate(MyClass c)

{

    Compute(c);

});

Parallel.ForEach using an anonymous function

Parallel.ForEach(data, c =>

{

    Compute(c);

});

• using Parallel.ForEach is generally less efficient than Parallel.For, because many threads must access the same underlying enumerator
• Parallel.ForEach is, however, intelligent enough to detect and access IList<T> instances in a more efficient manner

Example: consider enumerating all of the image files in a directory and processing the images found using Parallel.ForEach

Parallel.ForEach(Directory.GetFiles(path, "*.jpg"),

    imagePath =>

    {

        ProcessImage(imagePath);

    });

[3] Parallel.Invoke

• the Invoke static method of the supports type the parallelization of blocks of statements
• it accepts an array of actions to execute
• use this to execute a sequence of statements in parallel when each statement block is independent of each other and the order of execution is not important
• can be used for recursive divide-and-conquer algorithms such as walking a tree

[2008.07.18] Coordinate Data Structures in System.Threading.Collections [Part II/II]

[3] System.Threading.Collections.BlockingCollection<T>

• a blocking queue is a classic solution to many producer/consumer problems where you have some threads that are producing work and other threads are consuming that work
• producers generate data and store it into a queue; consumers remove data from the queue and process it
• such a queue is typically thread-safe so that producers and consumers can access it concurrently from multiple threads

Example:

// create in instance of BlockingCollection

var blockColl = new BlockingCollection<T>();

 

// now somewhere a thread can call Add to append

// items to the underlying data structure

// NOTE: by default, the BlockingCollection

// uses a ConcurrentQueue<T> FIFO structure

// as the underlying data storage structure

blockColl.Add(data);

 

// somewhere else another thread can call Remove

// this is guaranteed to return data because it

// will block until it has data to return

blockColl.Remove();

• additionally, such a queue has blocking functionality built into it, so that consumers requesting data can block until data arrives
• there are additional features one might want in a blocking collection:
  • just as consumers may want to block, waiting for data to arrive, producers may also want to block waiting for space to be available in the queue, a technique useful for throttling data production
  • another useful capability is signaling that no more data will be produced, such that consumers waiting for data to arrive don’t wait indefinitely if no more data is inbound
  • it’s also useful for some scenarios, such as in pipelining scenarios, to be able to add to or remove from any one of several blocking queues
• there are some overloads to the BlockingCollection<T>:
  • [1] one overload lets you create an upperbound on the number of items you want in the collection, so for instance, if you want to have 10 items, you can instantiate it as:
    

    var blockColl = new BlockingCollection<T>(10);

    • now if another thread comes along and try to add the 11th item to the collection, it will block
    • essentially it throttles the producers in a producer/consumer relationship where the producer is running faster than the consumer
    • in this way, throttling does not let producer use up all the system resources
  • [2] another overload lets you choose your own underlying storage mechanism
    • the default implementation of the BlockingCollection uses the first-in-first-out (FIFO) ConcurrentQueue<T> data structure
    • however, it also acknowledges that the FIFO behavior of queues isn’t always the most desirable and can easily convert this to a last-in-first-out (LIFO) structure using a ConcurrentStack<T>

      var blockColl =

          new BlockingCollection<T>

              (new ConcurrentStack());

• BlockingCollection<T> acts as a wrapper, that is, accepts as a parameter around any concurrent collection that implements the System.Threading.Collections.IConcurrentCollection<T> interface, providing blocking and bounding capabilities on top of such a collection
• both the ConcurrentStack<T> and ConcurrentQueue<T> types implement IConcurrentCollection<T>, allowing them to be used with BlockingCollection<T>
  • however, custom implementations of IConcurrentCollection<T>, allowing them to be used with can also be used

Example: consider creating a blocking queue of strings:

private BlockingCollection<string> _data =

    new BlockingCollection<string>();

 

// can also be done by explicitly providing

//  the underlying collection to be used:

 

private BlockingCollection<string> _data =

          new BlockingCollection<string>

              (new ConcurrentQueue<string>());

producer threads can now add data to the queue

private void Producer()

{

    while(true)

    {

        string s = ...;

        _data.Add(s);

    }

}

while one or more consumer threads are removing data from the queue, blocking as necessary until data is available

private void Consumer()

{

    while(true)

    {

        string s = _data.Remove();

        UseString(s);

    }

}

• such a consumer loop can be made simpler by taking advantage of BlockingCollection<T>’s GetComsumingEnumerable method
• this method returns an IEnumerable<T> that calls Remove under the covers and removes the next element from the collection on each call to System.Threading.Collections.IEnumerable<T>.MoveNext
• this allows for a BlockingCollection<T> to be consumed in standard constructs like a foreach loop and PLINQ

[2008.07.10] Coordinate Data Structures in System.Threading.Collections [Part I/II]

• the IConcurrentCollection<T> interface represents a collection which in a thread safe way you can add or remove from
• some implementations of this are:
  • ConcurrentStack<T>
  • ConcurrentQueue<T>

[1] System.Threading.Collections.ConcurrentQueue<T>

• NOTE: I will drop the System namespace below to avoid overflow but it precedes all the namespaces listed
• Threading.Collections.ConcurrentQueue<T> is a thread-safe and scalable queue data structure
• as with its Collections.Generics.Queue<T> counterpart, ConcurrentQueue<T> provides an Enqueue method for adding an element to the queue
• unlike Queue<T>, however, ConcurrentQueue<T> does not provide a Dequeue method for removing an item from the queue
• instead, it provides a TryDequeue method that returns a Boolean value indicating whether an item could be dequeued and an out parameter containing the dequeued element if one could be retrieved
• while Queue<T> provides Enqueue/Dequeue methods, ConcurrentQueue<T> provides the Enqueue/TryDequeue methods

Example: consider a loop meant to remove each element from the queue and process it

Using Queue<T>

while (queue.Count > 0) {     Data d = stack.Dequeue();     Process(d); }

Using ConcurrentQueue<T>

Data d; while (queue.TryDequeue(out d)) {     Process(d); }

[2] System.Threading.Collections.ConcurrentStack<T>

• ConcurrentStack<T> serves as a thread-safe alternative to Collections.Generics.Stack<T>
• while Stack<T> provides Push/Pop methods, ConcurrentStack<T> provides the Push/TryPop methods

Example:

Using Stack<T>

Stack<T> stack = new Stack<T>(); while (stack.Count > 0) {     Use(stack.Pop()); } // in this stack implmentation one would // have had to lock around the axis // of both stack.Count and stack.Pop

Using ConcurrentStack<T>

ConcurrentStack<T> cStack = new ConcurrentStack<T>(); T data; while (cStack.Count > 0) {     Use(cStack.TryPop()); } // TryPop says try and get something and //  if its available give it to me

[2008.07.09] Coordinate Data Structures in System.Threading [Part III/III]

[6] System.Threading.LazyInit<T>

• provides support for several common patterns of thread-safe lazy initialization
• lazy initialization is a commonly-used tactic for delaying data initialization until the data is actually needed
• LazyInit<T> can be used to lazily initialize an expensive data structure especially in situations where as a developer you do not want to figure out if it has already been created or deal with locking for multithreaded access
• in single-threaded applications, this frequently takes a form similar to using properties

Example: using properties to model lazy initialization

private MyData _data;

public MyData Data

{

    get

    {

        // check is what you want exists

        if (_data == null)

            // if not, create it when needed

            _data = new MyData();

            // now have a lazily initialized

            //  data structure

            return _data;

    }

}

• however, multithreaded applications, where multiple threads may be accessing the lazily-initialized data simultaneously, need sophisticated, thread-safe constructs

Example: with LazyInit<T> the previous example can be written as:

private LazyInit<MyData> _data;

public MyData Data

{

    get

    { return _data.Value; }

}

• there are several modes in which LazyInit<T> can operate:
• 
  [1] if the type of data being lazily initialized has a public, parameterless constructor, LazyInit<T> can use that constructor to initialize the instance of the type being passed into LazyInit

Example: using LazyInit<T> where T has a public, parameterless constructor:

// create an instance of LazyInit<T> where

//  T has a public, parameterless constructor

LazyInit<MyType> _myType;   // this is all that is needed

 

// to get back the lazily initialized instance of the

//  MyType that was created, call the Value

//  property like:

_myType.Value;

• 
  [2] if the type doesn’t provide a public, parameterless constructor, or if different logic is needed, a delegate can be provided to LazyInit<T> that contains the initialization logic

Example: the CreateMyData function is used to create and return an instance of MyData when requested to do so by LazyInit<T>

// use a lambda to pass in the method that will

//  create the type

private LazyInit<MyData> _data =

    new LazyInit<MyData>(() => CreateMyData());

// get the type when needed

public MyData Data

{

    get

    { return _data.Value; }

}

• given that you are working in a multithreaded or multi processor environment, then it is possible that multiple threads will race to create the lazily-initialized instance
• even though many instances may be created, only one instance will ever be published through LazyInit<T>.Value property
• LazyInit<MyData> provides an additional constructor that accepts a System.Threading.LazyInitMode which configures this behavior

Example: the mode is passed as a second parameter to the constructor of LazyInit<T>

private LazyInit<MyData> _data =

    new LazyInit<MyData>(

        () => CreateMyData(),

        LazyInitMode.XXX);

• XXX can take on three values:
  • AllowMultipleExecution
  • this is the default value that will be used if this version of the LazyInit<T> constructor is used
  • this means that even though multiple threads may race to initialize the value only one value will be published for all threads to access
  
  
  • EnsureSingleExecution
  • the initialization function will only be executed once so only one instance will ever be created, even if multiple threads race to initialize the value
  • this value will be published for all threads to access
  
  
  • ThreadLocal
  • each thread will get its own published value
  • basically, when _data.Value is called, you will get an instance of MyData because the CreateMyData method will be called to do the initialization

• the type is very small with little overhead
• if it is defined like LazyInit<MyData> _mt; then you are not doing any allocations at all until it is actually invoked and the cost of using this in a class is very negligible

[7] System.Threading.WriteOnce<T>

• WriteOnce<T> similar to a readonly construct, but inverted
• a readonly field is one which you can set once in a constructor and outside the constructor you can only read and not modify it
• a WriteOnce<T> is a single-assignment variable is a variable that can be written to only once and these types of variables are more relevant in concurent applications
• the WriteOnce<T>.Value property has get and set accessors; however, Value may only be retrieved after it has been set, and it may only be set once
• any accesses that violate these rules throw exceptions
• attempting to access the Value of a WriteOnce<T> before it’s been set is a significant programming error and as a result, such an action will invalidate the WriteOnce<T> instance for all future accesses

[2008.07.07] Coordinate Data Structures in System.Threading [Part II/III]

[4] System.Threading.SemaphoreSlim

• the System.Threading.SemaphoreSlim class provides functionality that allows a developer to limit the number of threads that can access concurrently a resource or pool of resources as well as limit the costs associated with the .NET 2.0 version of a Semaphore
• like ManualResetEvent, Semaphore is a thin wrapper around underlying kernel objects

Example: use a SemaphoreSlim to build a blocking queue

class BlockingQueue<T>

{ // start Class BlockingQueue<T>

    // create a generic Queue to model the Blocking Queue

    private Queue<T> _queue = new Queue<T>();

    // create a SemaphoreSlim object and specify the

    //  initial number of requests that can be granted

    //  concurrently

    private SemaphoreSlim _semaphore =

        new SemaphoreSlim(0);

 

    public void Enqueue(T data)

    {

        if (data == null) throw new

            ArgumentNullException("data");

        // lock the queue to prevet other threads from

        //  accessing it during this operation

        lock (_queue)

            _queue.Enqueue(data); // add data to queue

        // exits the semaphore and returns the value

        //  of SemaphoreSlim.CurrentCount which

        //  basically counts the number of operations

        //  performed thus far

            _semaphore.Release();

    }

 

    public T Dequeue()

    {

        // if CurrentCount > 0, decrement it by one

        //  and perform the Dequeue operation as this means

        //  we have data in the queue we can process

        // if CurrentCount = 0, blocks the current

        //  thread until it's greater than zero as this means

        //  that the queue does not have data to Dequeue so

        //  we will just block the thread this operation

        //  is running on, waiting until more data is

        //  added to the queue

        _semaphore.Wait();

        // lock the queue to prevet other threads from

        //  accessing it during this operation

        lock (_queue)

            return _queue.Dequeue();

    }

} // endClass BlockingQueue<T>

[5] System.Threading.CountdownEvent

• like ManualResetEvent and AutoresetEvent, System.Threading.CountdownEvent is a synchronization primitive that allows threads to signal the event and other threads to wait for it to be set
• waits for a number of things to happen
• the Wait() method blocks the current thread until the System.Threading.CountdownEvent is set
• CountdownEvent is set when a certain number of threads have signaled the event, counting down from a predetermined value
• so you may have 10 work items and can initialize the CountdownEvent item to 10
  • now every time something completes, it calls Decrement() and only at zero does this event get set
• this is frequently useful in fork/join operations, where a number of asynchronous operations may happen in the background, and only when those operations have completed is some main thread of execution allowed to proceed

Example:

int n = 10;

using(var ce = new CountdownEvent(n))

{

    for(int i=0; i<n; i++)

    {

        ThreadPool.QueueUserWorkItem(delegate

        {

            ...

            // registers a signal with the CountdownEvent by

            //  decrementing its count

            ce.Decrement();   

 

        });

    }

        // blocks the current thread until the CountdownEvent

        //  is set

        ce.Wait();

}

• CountdownEvent is not limited to counting down from a preset value using Decrement()
• it also provides an Increment() method that can be used to increment its current count if the count hasn't already reached 0

Example:

IEnumerable<T> src = ...;

using(CountdownEvent ce = new CountdownEvent(1))

{

    foreach(var element in src)

    {

        ce.Increment();

        ThreadPool.QueueUserWorkItem(state =>

        {

            Handle((T)state);

            ce.Decrement();

        }, element);

    }

    ce.Decrement();

    // block the main thread (i.e. the thread running

    //  the for loop) until all asynchronous work is done

    ce.Wait();              

}

• in this example, CountdownEvent is initialized to a value of 1, representing the main thread (which runs the for loop) that’s spawning off the asynchronous work items
• before each work item, Increment the count by 1, and when each work item finishes, it's decreased by one using the Decrement method
• when the main thread finishes spawning work items, it also decrements the count by 1, and then Wait on the event for all work items to complete