Creating your own thread pool in .NET

• Download source and demo files - 13.6 Kb

Introduction

Suppose we are writing a server program that processes requests from clients. If multiple requests can come into the server simultaneously, we can usually get better overall performance by using multiple worker threads to process these requests. In ASP.NET applications, the .NET framework maintains a thread pool to process user requests. Typically, we don't have to do anything with the thread pool itself. In other applications, we may need to create our own thread pool.

There is a ThreadPool class in .NET and several people have published articles about it on this site. I agree with Marc Clifton and others that this class (or its documentation) is kind of confusing and not exactly easy to use. What I found out is, it is easy to write your own thread pool class in .NET.

Here is a list of the features I want in my own thread pool class.

1. A StartThreadPool method that creates a minimal number of threads and also restricts the maximal number of threads in this thread pool. The maximal number of threads should not be limited to 25 (this is a limit in the .NET built-in thread pool that is not easy to change).
2. An InsertWorkItem method that puts a work item in a queue to be processed by a thread in the thread pool. Work items will be processed on a first come first served order.
3. The work item in the above contains a function delegate (pointer) and a number of input parameters. There should be no restriction on the signature of the function.
4. An ExtractWorkItem method to get back a work item that is already processed. The return value of the function specified in the work item should be available for future use.
5. The number of threads in the thread pool should change dynamically and automatically within the specified range according to the number of currently unprocessed work items.
6. The user should be able to specify an error handler that is invoked automatically to process any exception happened within a worker thread.
7. The thread pool class should be thread-safe at the instance level and multiple instances of the thread pool should be able to co-exist in the same process.
8. The user should be able to stop and restart the thread pool at any time.

You may think that I am too ambitious, however, the XYThreadPool class described below does satisfy all the requirements listed above. The best part is, the implementation is not complicated at all.

The XYThreadPool class

The class XYThreadPool creates worker threads to process work items in an internal queue. First, we need to know what a work item looks like. Here are the public data members of the ThreadPoolWorkItem class.

// the name string can be anything assigned by the user 
public String m_sName; 

// a function to be called to process this item 
public Delegate m_pMethod; 

// an array of input parameters 
public Object[] m_pInput; 

// the output value 
public Object m_oOutput; 

// a boolean flag indicating whether the output of this item is stored 
public bool m_bStoreOutput;

// an exception object (error for this work item)
Exception m_oException;

After creating an XYThreadPool object in your application, the first method you need to call is StartThreadPool. Here is the signature of this function:

public bool StartThreadPool( int nMinThreads, int nMaxThreads) ;

This method creates nMinThreads threads immediately. It also restricts the maximal number of threads in this thread pool to nMaxThreads. Calling StartThreadPool the second time will do nothing unless you call StopThreadPool to terminate all existing threads first.

The threads created by the above method will be idle until you call InsertWorkItem to add work items. Here is the signature of the InsertWorkItem method.

public void InsertWorkItem(String sName,
     Delegate pMethod, Object[] pInput,  bool bStoreOutput);

This method should only be called after the StartThreadPool method, otherwise it won't do anything. The sName parameter is a user assigned string, it can be used to distinguish different work items returned by the ExtractWorkItem method. pMethod is a delegate (pointer) for the function to be used to process this work item. pInput is an array of input parameters for the function, you can use null if the function does not take any parameter. bStoreOutput is a boolean flag, true means you want to store the work item after it is processed for later retrieval with the ExtractWorkItem method, false means the work item will not be stored.

Since there is no restriction on the signature of the function specified in a work item, threads in the same thread pool can process different types of work items. For example, some work items can be database queries while others file I/O operations.

The ExtractWorkItem method will return a ThreadPoolWorkItem object that is already processed. You can check the output field or the error string field of the returned object. If no work item has been processed yet, the return value of this method will be null.

Suppose you have a class named Math with a static method Add which takes two integers and returns their sum. Here is the sample code that uses the XYThreadPool class to process 100 additions.

public delegate int myAdd(int Input1, int Input2);
...
XYThreadPool myPool =  new XYThreadPool();
myPool.StartThreadPool(10, 30);
...
for( int i=0; i<100; i++)
{
    myTool.InsertWorkItem("Add", new myAdd(Math.Add),
                         new Object[2]{i, i+1}, true);
}
...
for(int j=0; j<100; j++)
{
    ThreadPoolWorkItem myItem = myPool.ExtractWorkItem();
  
  
  
    if(myItem!= null)
    {
      System.Console.WriteLine("The result is: " +
                        myItem.m_oOutput.ToString());
   }
}

The above code creates 10 threads initially and no more than 30 threads will be allowed by this thread pool object. The test code included in this article is a C# console application that does similar things and also demonstrates the use of two thread pool objects. You can start the test application and watch the number of threads changing in the task manager.

Thread management and Error handling

One of my requirements is that the number of threads should change dynamically and automatically within the specified range. Here is how this requirement is implemented. The StartThreadPool method only creates a minimal number of threads. When the InsertWorkItem method is called, the code will compare the number of unprocessed work items with the number of active threads. If there is more unprocessed work items and the number of active threads is still below the maximum specified in StartThreadPool, then a new thread will be created in the pool. The GetThreadCount method returns the number of threads in the thread pool.

For each thread in the thread pool, it will fetch an unprocessed work item from an internal queue and process it. This processing will loop forever unless there is no more unprocessed work item left. If there is no more unprocessed work item and the number of active threads is greater than the minimum specified in StartThreadPool, the current thread will be terminated.

Calling the StopThreadPool method will cause all threads to be terminated eventually, all unprocessed work items in the internal queue will be lost when this method is called. You can use the SetShutdownPause method to specify how long (in milliseconds) the StopThreadPool method will wait before terminating each active thread, the default is 200 milliseconds If the time span you specified is 0 or negative, then an active thread will not be terminated forcefully, instead, it will exit when the processing of the current work item is done.

It is possible for errors to occur while the work items are being processed. We need a way for the user to handle these errors, of course. The user can call the method SetThreadErrorHandler to specificy a delegate to an error handling function. The error handler will be invoked automatically when an exception is thrown while the method specified in a work item is being executed. Here is the signature of the error handling function.

public delegate void ThreadErrorHandlerDelegate(ThreadPoolWorkItem
        oWorkItem, Exception oError);

The error handler has two parameters, the first one is the ThreadPoolWorkItem object currently being processed, the second one is the Exception object. If no error handler is specified, the default implementation will store the exception object in the work item.

Thank you for reading my articles.

Recent updates

• 12/16/2003
  • Added the SetShutdownPause method to control how long the system will wait before terminating each active thread when the StopThreadPool is called. If you call this method with 0 or negative input, then active threads won't be terminated forcefully when the thread pool is shutdown.
  • Added the SetServerPause method to fine tune the thread pool, see code for details.
• 12/04/2003
  • Modified code so that the users don't have to derive a class and override a virtual method to provide custom error handling.
  • Changed the article text to reflect the above code change.

This article has no explicit license attached to it but may contain usage terms in the article text or the download files themselves. If in doubt please contact the author via the discussion board below.

A list of licenses authors might use can be found here