Explicitly Creating and Aborting a Thread

Although the last sample was a considerable improvement on the situation from the point of view of being able to wait until the asynchronous operations were complete, the sample still has a problem. What happens if one of the attempts to retrieve an address simply hangs? In that case, the TimerDemo sample will simply carry on polling the status of the results indefinitely, presumably until the user gives up and kills the process. Clearly that's not an acceptable situation for a real application - we need some way of aborting an operation that is clearly taking too long or going wrong in some way.

There are two possible approaches to this:

• Use WaitOrTimerCallback.WaitOrTimerCallback is a delegate that is provided in the System.Threading namespace which provides in-built support methods that are performed asynchronously, but which we might wish to abort after a fixed period of time. It is normally used in conjunction with the RegisterWaitForSingleObject() method. This method allows you to associate a delegate that is waiting for any object derived from WaitHandle (including ManualResetEvent) with a WaitOrTimerCallback delegate, allowing you to specify a maximum time to wait.

• Explicitly create a thread for the asynchronous task instead of using the thread pool - which means that we can abort the thread if necessary.

In this chapter we will take the second approach, because I want to demonstrate how to create and abort a thread explicitly. However, if you find yourself needing to write this kind of code, you should check out the WaitOrTimerCallback delegate in case it provides a better technique for your particular scenario.

Aborting a thread that appears to be taking too long to perform its task is quite easy in principle - the Thread class offers an Abort() method for exactly this purpose. The Thread.Abort() method has to be one of the brightest ideas in the .NET Framework - if you use it to tell a thread to abort, then the CLR responds by inserting an instruction to throw a ThreadAbortException into the code that the thread is executing. ThreadAbortException is a special exception that has a unique property: whenever it is caught, it automatically throws itself again - and keeps rethrowing itself, causing the flow of execution to repeatedly jump to the next containing catch/finally block - until the thread exits. This means that the thread aborts, but executes all relevant catch/finally blocks in the process. A completely clean abort that can be instigated from any other thread - this is something that was never possible before the days of .NET (with the Windows API, aborting a thread always meant killing the thread straight away, and was a technique to be avoided at all costs because this meant the aborted thread might leave any data it was writing in a corrupt state, or might leave file or database connections open, since it would have no chance to perform any cleanup). For managed code, there's even a Thread.ResetAbort() method that an aborting thread can call to cancel the abort!

The Thread.Abort() method and accompanying architecture is great news for anyone writing a software in which there is a chance that operations might need to be canceled. However, it does require a reference to the Thread object that encapsulates the thread you want to abort, which means you need to have created the thread explicitly. It's not sensible to do this with a thread-pool thread (besides, the whole point of the thread pool is that operations like creating and aborting threads is under the control of the CLR, not your code). This is just the trade-off I mentioned near the beginning of the chapter: the performance and scalability benefits of the thread pool come at the expense of the fine control of threads you get with explicit thread creation.

The AbortThread Sample

In this sample we are going to modify the TimerDemo sample so that if any of the address retrieval operations take too long, we can just abort the thread by calling Thread.Abort(), and display whatever results we have. This is going to involve quite a bit of rewriting of the sample, since it will be based on explicitly creating threads rather than on asynchronous delegate invocation.

To start with, let's see what we have to do with the DataRetriever class. The fields, the constructor, and the GetResults() method are unchanged. The GetAddress() method that actually returns the address is almost unchanged - except that I'm going to insert an extra delay for Wrox Press, to simulate an asynchronous request hanging:

 // In DataRetriever.GetAddress() Thread.Sleep(1000); if (name == "Simon")    return "Simon lives in Lancaster"; else if (name == "Wrox Press") {    Thread.Sleep(6000);    return "Wrox Press lives in Acocks Green"; } // etc. 

We can remove the GetAddressAsync() method that calls GetAddress() asynchronously, as well as the associated callback method - we won't be needing either of them now. Instead, there's a new GetAddressSync() method that will call GetAddress() synchronously, and wrap exception handling round it. Note that there is no delegate involved now - it's a straight method call:

 public void GetAddressSync() {    string address = null;    ResultStatus status = ResultStatus.Done;    try    {       address = GetAddress();       status = ResultStatus.Done;    }    catch(ThreadAbortException)    {       address = "Operation aborted";       status = ResultStatus.Failed;    }    catch(ArgumentException e)    {       address = e.Message;       status = ResultStatus.Failed;    }    finally    {       lock(this)       {          this.address = address;          this.status = status;       }    } } 

The point of GetAddressSync() is to call GetAddress() and to make sure that no matter what happens inside GetAddress(), even if an exception is thrown, some sensible results get stored back in the address and status fields. With this in mind, GetAddressSync() calls GetAddress() inside a try block and caches the results locally. However, it also catches two exceptions: ArgumentException (which we throw if the name wasn't found in the database) and ThreadAbortException (which we throw if the timer callback method decided to abort this thread because it was taking too long). Because this is only a sample, we don't catch any other exceptions - these are the only two exceptions we want to demonstrate. A finally block makes sure that whatever happens, some appropriate data is written to the address and status fields. Notice that we still take the trouble to synchronize access to these fields (this is necessary because these fields will be read on the timer callback thread, when the timer callback decides to call DataRetriever.GetResults() to retrieve the address).

Now let's examine the changes to the Main() method:

 public static void Main() {    ThreadStart workerEntryPoint;    Thread [] workerThreads = new Thread [3];    DataRetriever [] drs = new DataRetriever[3];    string [] names = { "Simon", "Julian", "Wrox Press" };    for (int i=0; i<3; i++)    {       drs[i] = new DataRetriever(names[i]);       workerEntryPoint = new ThreadStart.(drs[i].GetAddressSync);       workerThreads[i] = new Thread(workerEntryPoint);       workerThreads[i].Start();    }    ManualResetEvent endProcessEvent = new ManualResetEvent(false);    CheckResults resultsChecker = new CheckResults(endProcessEvent, drs,                                                   workerThreads);    resultsChecker.InitializeTimer();    endProcessEvent.WaitOne();    EntryPoint.OutputResults(drs); } 

There are quite a few changes here, mostly associated with the fact that Main() is explicitly creating threads and not invoking delegates asynchronously. We start by declaring an array that will hold the Thread references, as well as a delegate that will hold the worker thread entry points:

 ThreadStart workerEntryPoint; Thread [] workerThreads = new Thread[3]; 

Then, inside the loop in which the DataRetriever objects are initialized, we also instantiate the Thread objects, set the entry point of each and start each thread off:

 workerEntryPoint = new ThreadStart(drs[i].GetAddressSync); workerThreads[i] = new Thread(workerEntryPoint); workerThreads[i].Start(); 

There are a couple of differences around the instantiation of the ResultsChecker object that implements the timer callbacks. This is because ResultsChecker is going to play a more prominent role now and needs more information - so more parameters are passed to its constructor. Also, the Timer will be stored in the ResultsChecker class rather than the Main() method, because the ResultsChecker is going to need access to it later on. So the Main() method simply creates a ResultsChecker instance, asks it to start the timer off, and then - as in the previous sample - sits back and waits for the ManualResetEvent to be signaled.

Now for the CheckResults class. First its constructor and member fields, and the InitializeTimer() method:

 class CheckResults {    private ManualResetEvent endProcessEvent;    private DataRetriever[] drs;    private Thread[] workerThreads;    private int numTriesToGo = 10;    private Timer timer;    public CheckResults(ManualResetEvent endProcessEvent, DataRetriever[] drs,                        Thread[] workerThreads)    {       this.endProcessEvent = endProcessEvent;       this.drs = drs;       this.workerThreads = workerThreads;    }    public void InitializeTimer()    {       TimerCallback timerCallback = new TimerCallback(this.CheckResultStatus);       timer = new Timer(timerCallback, null, 0, 500);    } 

This code should be self-explanatory. Notice the numTriesToGo field - this field will be used to count down from 10, ensuring that the timer is canceled after 10 callbacks (5 seconds, since the interval is set in InitializeTimer() to half a second).

Next let's examine the timer callback method:

 public void CheckResultStatus(object state) {    Interlocked.Decrement(ref numTriesToGo);    int numResultsToGo = 0;    foreach(DataRetriever dr in drs)    {       string name;       string address;       ResultStatus status;       dr.GetResults(out name, out address, out status);       if (status == ResultStatus.Waiting)          ++numResultsToGo;    }    if (numResultsToGo == 0)    {       EntryPoint.OutputResults(drs);       endProcessEvent.Set();       return;    }    else    {       Console.WriteLine("{0} of {1} results returned",                         drs.Length - numResultsToGo, drs.Length);    }    if (numTriesToGo == 0)    {       timer.Change(Timeout.Infinite, Timeout.Infinite);       TerminateWorkerThreads();       endProcessEvent.Set();    } } 

There aren't too many changes to this method. The main difference is that we decrement numTriesToGo. We call the Interlocked.Decrement() method to do this - just in case execution of the timer callback takes longer than a timeslice, in which case this method could be executing concurrently on two different threads. There is then a new final if block, which catches the situation for which numTriesToGo hits zero but we are still waiting for one or more results. If this happens, the first thing we do is change the timer interval to infinity, effectively stopping the timer from firing again. We don't want the callback function being called again on another thread while the code to terminate the processing is being executed! We terminate any outstanding worker threads that are retrieving addresses, using a method called TerminateWorkerThreads(), which we will examine next. Then we output the results and signal the ManualResetEvent so that the main thread can exit and terminate the process.

The TerminateWorkerThreads() method is the one that might in theory take some time to execute:

 private void TerminateWorkerThreads() {    foreach (Thread thread in workerThreads)    {       if (thread.IsAlive)       {          thread.Abort();          thread.Joint);       }    } } 

This method loops through all the worker thread references we have. For each one, it uses another Thread property, IsAlive, which returns true if that thread is still executing, and false if that thread has already terminated (or if it hasn't started - but that won't happen in our sample). For each thread that is alive, it calls Thread.Abort().Thread.Abort() returns immediately, although the thread concerned will still be going through its abort sequence, executing finally blocks, and so on. We don't want TerminateWorkerThreads() to exit before all the worker threads have actually finished aborting, because if we do then we may end up displaying the results in the following call to EntryPoint.OutputResults() before the worker threads have finished writing the final data to their address and status fields. So we call another method, Thread.Join(). The Join() method simply blocks execution until the thread referenced in its parameter has terminated. So calling Thread.Join() ensures that TerminateWorkerThreads() doesn't itself return too early.

That completes the sample, and we are ready to try running it. It gives us the following:

 In GetAddress...  hash: 2, pool: False, backgrnd: False, state: Running In GetAddress...  hash: 3, pool: False, backgrnd: False, state: Running In GetAddress...  hash: 4, pool: False, backgrnd: False, state: Running 0 of 3 results returned 0 of 3 results returned 1 of 3 results returned 1 of 3 results returned 1 of 3 results returned 2 of 3 results returned 2 of 3 results returned 2 of 3 results returned 2 of 3 results returned 2 of 3 results returned Name: Simon, Status: Done, Result: Simon lives in Lancaster Name: Julian, Status: Failed, Result: The name Julian is not in the database Name: Wrox Press, Status: Failed, Result: Operation aborted 

This output shows the sample has worked correctly. The worker threads have returned the expected results for Simon and Julian. The Wrox Press thread has aborted, but the output shows that while aborting it did write the correct "Operation aborted" string to the address field - in other words, it performed the appropriate cleanup before terminating.