[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

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

The Coordinate Data Structures introduced in Parallel Extensions are used in all aspects of the library as seen in Fig.1. I have made notes based on the June 2008 CTP help file as well as from interviews with the people working on PE.

Fig.1 Parallel Extensions Hierarchy

[1] System.Threading.SpinWait

• spin waiting can be an effective low-level mechanism for avoiding unnecessary context switches and associated kernel transitions
• most of the data structures in the CDS use SpinWait in some form or the other
• is used in the Scheduler, Task Parallel Library and PLINQ
• SpinWait is waiting for something to happen by spinning (waiting in a loop)
• in an application there are several types of waiting such as:
  • wait by blocking a kernel primitive
  • polling (keep asking: are we there yet, are we there yet ...)
• know something will eventually happen and you don't want to take the full cost of a context switch or accessing the kernel
• the idea is that by spinning just a little bit, that by wasting a little work to do the spinning, you will save yourself from wasting a lot of work by doing a context switch or querying the kernel

[2] System.Threading.SpinLock

• the SpinLock data structure is built on top of the SpinWait
• idea behind SpinLock is not going into kernel at all – it is all user mode
• a SpinLock is a mutual exclusion lock primitive where a thread trying to acquire the lock waits in a loop repeatedly checking until the lock becomes available (in other words, it “spins”) [keep asking: is the lock available yet, is the lock available yet, ...]
• both SpinWait and SpinLock are designed for advanced scenarios as the thread remains active performing a non-useful task
• can think of SpinLock as a kind of busy waiting which consumes CPU resources without performing real work
• spin locks can be more efficient than other kinds of locks on multi-processor machines if threads are only likely to be blocked for a very short period of time as they avoid unnecessary context switches and associated kernel transitions, which might otherwise be more costly than just spinning
• in addition, both SpinWait and SpinLock are structs and as such they are cheaper to implement because if you wanted to use a Monitor or a Mutex, you will need to allocate a garbage collected object as these locks are all classes

[3] System.Threading.ManualResetEventSlim

• the ManualResetEvent in System.Threading of the mscorlib.dll assembly notifies one or more waiting threads that an event has occurred
• if one wants to use the Windows kernel event objects, there is a high overhead:
  • the cost of allocating the kernel object,
  • the cost of a new finalizable object for the garbage collector to track, and
  • the costs of the associated kernel transitions any time one wishes to wait or set the event
• ManualResetEventSlim type addresses many of these issues by trying to do everything in user mode if possible and only lazily instantiate the underlying kernel events if absolutely necessary
• it provides an interface similar to the original ManualResetEvent but unlike ManualResetEvent it:
  • does not provide the ability to create named events and thus cannot be used for cross-process synchronization
  • does not derive from WaitHandle

[2008.07.01] Parallel Extensions: Coordinate Data Structures Overview

• the June 2008 CTP introduces a set of coordination data structures (CDS) that complement PLINQ and the Task Parallel Library
• the System.Threading namespace of the .NET Framework 3.5 already contains a handful of synchronization primitives, such as Events, Monitor, ManualResetEvent, Mutex and Semaphore.
• these types are still very useful for developing multithreading applications, and in fact Parallel Extensions relies on many of them, but sometimes higher-level or lighter-weight primitives are required to facilitate communications between threads
• many parallel applications, however, would benefit greatly from higher-level or lighter-weight constructs are required to facilitate communications between threads such as thread-safe collections, more sophisticated locking primitives, data structures to facilitate work exchange, and types that control how variables are initialized
• Parallel Extensions adds several such constructs to the namespaces of System.Threading and System.Threading.Collections
• these new structures are as follows:

CDS in System.Threading

System.Threading.SpinWait System.Threading.SpinLock System.Threading.ManualResetEventSlim System.Threading.SemaphoreSlim System.Threading.CountdownEventm System.Threading.LazyInit System.Threading.WriteOnce

CDS in System.Threading.Collections

System.Threading.BlockingCollection System.Threading.ConcurrentQueue System.Threading.ConcurrentStack

[2008.06.29] Parallel Extensions

We are now living in the age of multicore CPU and there is great potential for this generation of hardware. But all this wonderful hardware is pointless if the software that runs on it is outdated and not scalable to take advantage of the ManyCore or MultiProc environment. As applications evolve and process more data, it is clear that we would like to do process all of the data both faster and safer.

Multithreading is nothing new but programming in a concurrent environment is probably a daunting task for developers (myself included). There are many little details that require constant attention in order to program a viable, multithreaded application. Microsoft Research is currently working on making concurrent and parallel programming much more straight forward by abstracting a lot of the underlying concepts to a much higher level.

Parallel Extensions [PE] is a managed library to the .NET Framework 3.5. It is designed to simplify parallel programming when using any of the .NET languages. Basically, with very little modification of your existing code base, you can convert your sequential processing application to one that is fully parallel and scalable. The PE library is built from the ground up and it is written in C#. It is composed of several parts:

• Coordinate Data Structures [CDS]
• Scheduler
• Task Parallel Library
• Parallel LINQ

You can download the June 2008 CTP of Parallel Extensions. Installing the CTP is very straight forward.

To access the types in the library, you have to add it to your project. So after you create your project, in the Solutions Explorer, right click and select Add Reference... This will bring up the list of references that can be added to the project. The image shows what you are looking for.

Access the PE library in the System.Threading namespace

I will continue this posting by first looking at the CDS, the TPL and then PLINQ. As stated, I am learning how to use this library and I will share what I know as I go along. Basically, I will take information from the June CTP help file as well as from interviews given by the team members so as to create a complete picture of what is going on. Stay tuned for more.