System.Threading.Thread

The Thread class was, originally, a 1:1 mapping to an operating system thread. It is typically used for long-running or specialized work such as monitoring a device or executing code with a low priority. Using the Thread class leaves us with a lot of control over the thread, so let’s see how the API works.

static void Main(string[] args)
{
    Thread monitorThread = new Thread(new ThreadStart(MonitorNetwork));
 
    monitorThread.Start();
}
 
static void MonitorNetwork()
{
    // ...
}

class Program
{
    static void Main(string[] args)
    {
        AsyncSignal h = new AsyncSignal();
        while (!h.Terminate) ;
    }
 
    class AsyncSignal
    {
        public bool Terminate;
 
        public AsyncSignal()
        {
            Thread monitorThread = new Thread(new ThreadStart(MonitorNetwork));
            monitorThread.Start();
       }
       private void MonitorNetwork()
       {
           Thread.Sleep(3000);
           Terminate = true;
       }
    }
}

Coordinating Threads (Join)

If code spawns a thread, it may well want to know when that thread finishes; for example, to process the results of the thread’s work. The Thread class’s Join method allows an observer to wait for the thread to end. There are two forms of the Join method: one that takes no parameters and returns void, the other that takes a timeout and returns a Boolean. The first form will block until the thread completes, regardless of how long that might be. The second form will return true if the thread completes before the timeout or false if the timeout is reached first. You should always prefer waiting with a timeout, as it allows you to proactively detect when operations are taking longer than they should. Listing 2-3 shows how to use Join to wait for a thread to complete with a timeout. You should remember that when Join times out, the thread is still running; it is simply the wait that has finished.

Listing 2-3.  Using Join to Coordinate Threads

FileProcessor processor = new FileProcessor(file);
Thread t = new Thread(processor.Process);
t.Start();
 
PrepareReport();
 
if (t.Join(TimeSpan.FromSeconds(5)))
{
    RunReport(processor.Result);
}
else
{
    HandleError("Processing has timed out");
}

Controlling a Thread’s Interaction with COM

One common use of the Thread class that is still important even in .NET 4.0 and 4.5 is to control how that thread behaves when it performs COM work (see the “Threading and COM” sidebar to understand the issues). If a thread is going to perform COM work, you should try to ensure it is in the same apartment as the COM objects it is going to be invoking. By default, .NET threads will always enter the MTA. To change this behavior, you must change the thread’s ApartmentState. Originally, this was done by setting the ApartmentState property, but this was deprecated in .NET 2.0. From 2.0 onward you need to use the SetApartmentState method on the thread.

Issues with the Thread Class

The API for the Thread class is fairly simple, so why not use it for all asynchronous work? As discussed in Chapter 1, threads are not cheap resources: they are expensive to create; clean up; they consume memory for stack space and require attention from the thread scheduler. As a result, if you have regular asynchronous work to do, continuously creating and destroying the threads is wasteful. Also, uncontrolled creation of threads can end up consuming huge amounts of memory and causing the thread scheduler to thrash—neither of which is healthy for your application.

A more efficient model would be to reuse threads that have already been created, thus relieving the application code of control of thread creation. This would then allow thread management to be regulated. This is potentially highly complex code for you to maintain. Fortunately, .NET already comes with an implementation, out of the box, in the form of the system thread pool.

Worker and I/O Threads

It turns out there are two groups of threads in the thread pool: worker and I/O threads. Worker threads are targeted at work that is generally CPU based. If you perform I/O on these threads it is really  a waste of resources, as the thread will sit idle while the I/O is performed. A more efficient model would be to kick off the I/O (which is basically a hardware operation) and commit a thread only when the I/O is complete. This is the concept of I/O completion ports, and this is how the I/O threads in the thread pool work.

Getting Work on to the Thread Pool

We have seen the basic mechanics of how the thread pool works but how does work get enqueued? There are three mechanisms you can use:

• ThreadPool.QueueUserWorkItem
• Timers
• The Asynchronous Programming Model (APM)

SearchResults GetResults(int page, int pageSize, out int itemsReturned);

IAsyncResult BeginGetResults(int page,
                            int pageSize,
                            out int itemsReturned,
                            AsyncCallback callback,
                            object state);
 
SearchResults EndGetResults(out int itemsReturned, IAsyncResult iar);

private IAsyncResult asyncCall;
private Timer asyncTimer;

private void OnPerformSearch(object sender, RoutedEventArgs e)
{
    int dummy;

    asyncCall = BeginGetResults(1, 50, out dummy, null, null);

    asyncTimer = new Timer();
    asyncTimer.Interval = 200;
    asyncTimer.Tick += OnTimerTick;
    asyncTimer.Start();
}

private void OnTimerTick(object sender, ElapsedEventArgs e)
{
    if (asyncCall.IsCompleted)
    {
        int resultCount;
        try
        {
            SearchResults results =EndGetResults(out resultCount, asyncCall);
            DisplayResults(results,  resultCount);
        }
        catch(Exception x)
        {
            LogError(x);
        }
 
        asyncTimer.Dispose();
    }
}

int dummy;

IAsyncResult iar = BeginGetResults(1, 50, out dummy, null, null);

// The async operation is now in progress and we can get on with other work
ReportDefaults defaults = GetReportDefaults();

// We now can't proceed without the results so wait for the operation to complete
// we're prepared to wait for up to 5 seconds
if (iar.AsyncWaitHandle.WaitOne(5000))
{
    int resultCount;
    try
    {
        SearchResults results = EndGetResults(out resultCount, asyncCall);

        GenerateReport(defaults, results);
    }
    catch(Exception x)
    {
        LogError(x);
    }
 
}
else
{
    throw new TimeoutException("Async GetResults timed out");
}

delegate void AsyncCallback(IAsyncResult iar);

private void OnPerformSearch(object sender, RoutedEventArgs e)
{
    int dummy;
 
    BeginGetResults(1, 50, out dummy, new AsyncCallback(Callback), null);
}
 
private void Callback(IAsyncResult iar)
{
    try
    {
        int resultCount;
        SearchResults results = EndGetResults(out resultCount, asyncCall);
 
        DisplayResults(results, resultCount);
    }
    catch(Exception x)
    {
        LogError(x);
    }
}

WebResponse GetResponse();
IAsyncResult BeginGetResponse(AsyncCallback callback, object state);
WebResponse EndGetResponse(IAsyncResult iar);

private void Callback(IAsyncResult iar)
{
}
 
private void OnPerformSearch(object sender, RoutedEventArgs e)
{
    WebRequest req = WebRequest.Create("http://www.google.com/#q=weather");
 
    req.BeginGetResponse(new AsyncCallback(Callback), null);
}

private void Callback(IAsyncResult iar)
{
    WebRequest req = (WebRequest) iar.AsyncState;
    try
    {
        WebResponse resp = req.EndGetResponse(iar);
 
        ProcessResponse(resp);
    }
    catch(Exception x)
    {
        LogError(x);
    }
}
 
private void OnPerformSearch(object sender, RoutedEventArgs e)
{
    WebRequest req = WebRequest.Create("http://www.google.com/#q=weather");
 
    req.BeginGetResponse(new AsyncCallback(Callback), req);
}

delegate bool CurrencyParser(string text, out decimal result);

private void OnDecrypt(object sender, RoutedEventArgs e)
{
    ClearText = Decrypt(CipherText);
}

delegate string EncryptorDecryptor(string inputText);
         
private void DecryptCallback(IAsyncResult iar)
{
    EncryptorDecryptor del = (EncryptorDecryptor) iar.AsyncState;
     
    // remember we must protect the call to EndInvoke as an exception
    // would terminate the process as this is running on a worker thread
    try
    {
        // for a UI application this would potentially have to be marshalled on to
        // the UI thread if the setting of the ClearText property changed the UI
        ClearText = del.EndInvoke(iar);
    }
    catch(Exception x)
    {
        LogError(x);
    }
}
 
private void OnDecrypt(object sender, RoutedEventArgs e)
{
    EncryptorDecryptor del = new EncryptorDecryptor(Decrypt);
 
    del.BeginInvoke(CipherText, new AsyncCallback(DecryptCallback), del);
}

Logical and Physical Separation

SQL Server had very specific requirements of the CLR to allow it to become a host for managed code. It needed to be able to take a lot of control over the CLR execution, such as memory allocation, controlling when and if garbage collection (GC) takes place, what code is and isn’t allowed to do, how execution is mapped on to threads, and much more. For SQL Server to control the mapping of execution to threads, it has to control what the CLR views as a thread, which in theory might not be a thread in the unmanaged sense. To make this happen, the CLR team separated the notion of a managed thread from a physical thread. The method call that allowed you to get hold of the thread ID, (AppDomain.GetCurrentThreadId()), was deprecated and a new property of the thread class was created, called ManagedThreadId.

Although this separation is real in API terms, in reality, for all managed code, a managed thread still maps on to an unmanaged thread. However, in theory a hosting process could change that mapping, and so code should not rely on running on the same physical thread over the lifetime of a managed thread. If you call APIs that have native thread affinity (e.g., certain synchronization objects rely on being acquired and released on the same native thread), then the CLR needs to be forced to make the mapping from managed thread to native thread fixed. There are two methods on the Thread class that control this: BeginThreadAffinity and EndThreadAffinity. In normal circumstances you should never need to call these methods; however, the APIs that rely on thread affinity already wrap these calls up internally.

Passing Data into a Thread

We saw that previously, getting data into a thread in an encapsulated way involved creating a class to wrap the thread function. .NET 2.0 introduced a new constructor on the Thread class that took a ParameterizedThreadStart delegate instead of a ThreadStart delegate. The key difference is that the ParameterizedThreadStart delegate takes an object as a parameter:

delegate void ParameterizedThreadStart(object obj);

Now to get data into the thread you simply use a method with a matching signature (one that returns void and takes a single parameter of type object) and pass the data to the Thread's Start method. Listing 2-11 shows an example of this. Notice that, in the thread function, the code needs to cast the passed parameter to the type it actually needs. If the code is refactored and the code creating the thread changes the type of the parameter object, the compiler will not catch that there is a problem, and the failure will occur at runtime.

Listing 2-11.  Using ParameterizedThreadStart

void ProcessResults(object obj)
{
    SearchResults results = (SearchResults) obj;
 
    // use results
    // ...
}
         
private void OnPerformSearch(object sender, RoutedEventArgs e)
{
    int itemCount;
    SearchResults results = GetResults(1, 50, out itemCount);
    ParameterizedThreadStart proc = ProcessResults;
             
    Thread t = new Thread(proc);
    t.Start(results);
}

Closures

For C# developers, .NET 2.0 saw the introduction of anonymous methods with their associated feature of closure (capturing the state of variables that are in scope). In many ways closures provided the most natural way for many async tasks. For example, the code in Listing 2-11 becomes much simpler with closures, as can be seen in Listing 2-12.

Listing 2-12.  Using Closures with the Thread Class

private void OnPerformSearch(object sender, RoutedEventArgs e)
{
    int itemCount;
    SearchResults results = GetResults(1, 50, out itemCount);
    ParameterizedThreadStart proc = delegate
                                    {
                                        DisplayResults(results, itemCount);
                                    };
             
    Thread t = new Thread(proc);
    t.Start();
}

Closures also make life simpler when working with APM. As shown in Listing 2-13, it allows a simple model of accessing the right object to call the End method without having to smuggle it across in the async state.

Listing 2-13.  Using Closures with APM

WebRequest req= WebRequest.Create("http://www.google.com/#q=weather");
 
AsyncCallback callback = delegate(IAsyncResult iar)
                         {
                             WebResponse resp = req.EndGetResponse(iar);
 
                             ProcessResponse(resp);
                         };
 
req.BeginGetResponse( callback, null);

SynchronizationContext

As we saw earlier, there are special considerations when using async and UI because UI objects have thread affinity (they must be manipulated on the thread that created them). Because of this, UI frameworks provide mechanisms to specify work that needs to be carried out on the UI thread: Windows Forms has the BeginInvoke method on the Control class; WPF and Silverlight have the dispatcher. The problem is that building a component that uses async internally becomes difficult because how marshalling code on to the UI thread is done is dependent on the framework under which the component is running. The solution to this is to provide a common abstraction over the different underlying mechanisms, and .NET 2.0 introduced one in the form of SynchronizationContext (see Listing 2-14).

Listing 2-14.  The SynchronizationContext Class

public class SynchronizationContext
{
    public static SynchronizationContext Current { get; }
    public virtual void Post(SendOrPostCallback d, object state);
    public virtual void Send(SendOrPostCallback d, object state);
 
    // other members omitted for clarity
}

The programming model works as follows:

1. On the UI thread, get hold of the SynchronizationContext using the static Current property
2. On the background thread, use the SynchronizationContext Send or Post to perform the work wrapped in the SendOrPostCallback delegate on the UI thread (see Listing 2-15).

Listing 2-15.  Using SynchronizationContext

private void OnPerformSearch(object sender, EventArgs e)
{
    WebRequest req = WebRequest.Create("http://www.google.com/#q=weather");
 
    // Must make this call on the UI thread
    uiCtx = SynchronizationContext.Current;
     
    AsyncCallback callback = delegate(IAsyncResult iar)
                             {
                                 WebResponse resp = req.EndGetResponse(iar);
 
                                 ProcessResponse(resp);
                             };
    req.BeginGetResponse(callback, null);
}
 
private SynchronizationContext uiCtx;
 
private void ProcessResponse(WebResponse resp)
{
    // This code is on the threadpool thread
    StreamReader reader = new StreamReader(resp.GetResponseStream());
    SendOrPostCallback callback = delegate
                                  {
                                      // this runs on the UI thread
                                      UpdateUI(reader.ReadToEnd());
                                      // must Dispose reader here as this code runs async
                                      reader.Dispose();
                                  };
    uiCtx.Post(callback, null);
}

Post is a “fire and forget” where the UI thread work is performed asynchronously; Send is synchronous in that the call blocks until the UI thread work has been performed. In general you should prefer Post as it is easier to end up in a deadlock with Send; however, occasionally Post can lead to race conditions if the background thread continues processing, having made assumptions about the state of the UI.

SynchronizationContext brings sanity to performing async work within different frameworks. This is especially important if you want to provide infrastructure that can automate some of the work.

Event-Based Asynchronous Pattern

APM has had some issues in terms of structure of code, UI interaction, and error handling, so in .NET 2.0 Microsoft introduced another async pattern called the Event-Based Asynchronous Pattern (EAP). The idea is that each operation has a pair of members: a method to start the async work and an event that fires when the operation is complete. This pattern was evident on the proxies generated with Add Web Reference in Visual Studio and also on the WebClient class. As an example, let’s have a look at WebClient in Listing 2-16.

Listing 2-16.  EAP in WebClient

private void OnPerformSearch(object sender, RoutedEventArgs e)
{
    WebClient client = new WebClient();
 
    client. DownloadStringCompleted += ClientOnDownloadDataCompleted;
    client. DownloadStringAsync(new Uri("http://www.google.com/#q=weather"));
}
 
private void ClientOnDownloadDataCompleted(object sender, DownloadStringCompletedEventArgs e)
{
    UpdateUI(e.Result);
}

As Listing 2-16 shows, with EAP, before calling the async method you wire up an event (if you wired up the event afterward, there would be a race condition between the completion and the event wire-up). When the async operation completes, the event is fired. Notice that we can update the UI directly from this event handler, as it uses SynchronizationContext under the covers to fire the event on the UI thread. The result of the async operation is available as the Result property on the event’s event arguments.

private void ClientOnDownloadDataCompleted(object sender, DownloadStringCompletedEventArgs e)
{
    if (e.Error == null)
    {
        UpdateUI(e.Result);
    }
    else
    {
        LogError(e.Error);
    }
}

void DownloadStrinAsync(Uri address, Object userToken);

Lambda Expressions

.NET 2.0 introduced anonymous delegates and closure. However, it turned out that we were commonly telling the compiler things it could already work out for itself. Consider the following anonymous delegate:

Predicate<Person> test = delegate(Person p)
                         {
                             return p.Age >= 18;
                         };

Let’s start removing the things the compiler could work out. To begin with, the delegate keyword is superfluous as the compiler knows Predicate<T> is a delegate type:

Predicate<Person> test = (Person p)
                         {
                             return p.Age >= 18;
                         };

We don’t need to specify the type of p as the compiler knows that the parameter for Predicate<Person> is a Person:

Predicate<Person> test = (p)
                         {
                             return p.Age >= 18;
                         };

Predicate<T> returns a Boolean and the return expression evaluates to a Boolean. There is just one statement in the body, so it must be the return. So let's omit the return keyword:

Predicate<Person> test = (p)
                         {
                             p.Age >= 18;
                         };

We only have one statement—we surely don’t need braces:

Predicate<Person> test = (p) p.Age >= 18;

There is only one parameter, so we could omit the parentheses:

Predicate<Person> test = p p.Age >= 18;

Last, we do need to separate the parameter from the lambda body, so we use the symbol =>:

Predicate<Person> test = p => p.Age >= 18;

This is a lambda expression. There are other, more complex forms of lambda expression: multiple or no parameters require the use of parentheses and multiple statements in the body require the use of braces and an explicit return.

Why are lambda expressions often used to model asynchronous work? There are two main reasons: first, closure simplifies getting data into the asynchronous operation; and second, quite often the actual code that needs to run asynchronously is quite brief (in terms of syntax), so creating a whole new method for this clouds readability.

Thread Pool Heuristics in .NET 3.5

Remember from .NET 1.1 that the default maximum number of threads in the thread pool was 25 per core, and the algorithm for adding new threads was to pause for half a second, waiting for a thread to become idle, before adding a new thread into the pool. .NET 3.5 changed the maximum number of threads in the thread pool to 250 per core. On, say, an eight-core machine, this gives a maximum size to the thread pool of 2,000 threads. Each thread will consume 1MB of stack space, so simply creating these threads will consume 2GB of data. This seems to fly in the face of one of the purposes of using thread pools: namely, to constrain the number of threads—2,000 is not much of a constraint. However, the algorithm to add more threads to the thread pool was also changed to increase the wait time exponentially as the number of threads increases.

Why did the CLR team feel it necessary to increase the maximum number of threads tenfold? To understand this change, we need to look at what problem they were trying to solve. Consider the following code:

Func<SearchResults> resultRetriever = GetResults;
 
IAsyncResult iar = resultRetriever.BeginInvoke(null, null);
 
ReportDefaults defaults = GetReportDefaults();
 
SearchResults results = resultRetriever.EndInvoke(iar);

This code seems quite innocuous. However, what if this code were invoked from a thread pool thread? Remember that EndInvoke will block until the async method completes. Now, what if all threads in the thread pool were running this code? The async method would never start because no thread was available and no thread would ever become available, because they are all waiting for this code to finish. It has essentially deadlocked the thread pool. You might argue that this is a heavily contrived problem; that people generally don’t spawn thread pool work from thread pool threads. And in many cases that is true—unless all of your code is running within a framework that always runs on thread pool threads like ASP.NET. Thread pool deadlock was an issue that people were experiencing in the field, so the .NET team decided to solve it by making sure another thread could always be added to break the deadlock—hence the tenfold increase in the maximum number of threads in the thread pool.

Remodeling the Thread Pool Queue

Prior to .NET 4.0 the thread pool queue was a linked list of work items. Imagine lots of cores are generating lots of work items. There are two issues that are going to become apparent:

1. The thread pool will become a large data structure with a very large number of references. This is an expensive for the Garbage Collector to deal with during its mark phase.
2. A linked list is a terrible data structure for concurrent manipulation. Processing across the cores would have to be serialized as it updated the queue (see Figure 2-3).

In .NET 4.0 the thread pool queue was redesigned with the new requirements of PFx in mind. Instead of using a simple linked list, the queue was built with arrays of work items with the arrays connected into a linked list. This means that there are a lot fewer references, and that adding and moving work items often will simply involve array index manipulation, which is a much cheaper way to make thread safe (see Figure 2-4).

Work-Stealing Queues

As we explained in Chapter 1, processors have caches. If you need to spawn async work that uses the same data that you have been processing, it would be most efficient in terms of data access to execute the work on the same processor, as then the data may well be read from cache rather than main memory. So if all of the cores are generally busy, then trying to get related work on to the same core potentially would be a significant performance improvement. The internals of the thread pool have been remodeled to give each thread its own queue. If a thread pool thread creates async work using the new API, then by default it will be put on the thread’s “local” queue. When a thread finishes its current work item, it looks for new work items in the following order:

1. It looks on its local queue.
2. It looks on the global queue.
3. It steals work from other threads’ local queues. As a result this local queue is known as a work-stealing queue (see Figure 2-5).

Thread Pool Heuristics in .NET 4.0

As mentioned earlier, .NET 3.5 changed the maximum number of threads in the thread pool to 250 threads per core. However, we also noted that the main resource that the threads are consuming is memory. Therefore, in .NET 4.0 the maximum number of threads is determined by the amount of memory available (on most modern machines it will be 1,023 worker threads and 1,000 I/O threads).