Simple thread pool manager

• Download demo project - 15.9 Kb
• Download source - 3.93 Kb

Introduction

It is a simple thread pool management class, which gives flexibility of:

1. Queuing tasks in case no idle threads are present or the thread count has reached the maximum limit of allowed threads.
2. Altering thread pool at runtime. Which would mean increasing or decreasing the number of threads required by the system.
3. Creating pool at runtime to avoid unwanted resource usage. Maintaining a pool with idle threads becomes a costly affair in resource crunching moments, this would allow removing of idle threads and recreating them when required again.

Using the code

CThreadPoolManager is the class that handles the creation and management of the thread pool.

Here is the skeleton of the manager class:

class CThreadPoolManager
{
    static friend UINT ProcessRequestMgr(LPVOID);
    static friend UINT ProcessRequestWkr(LPVOID);
    static friend bool UpdateThreadPool(int&,int&);
public:
    static CThreadPoolManager* GetInstance();
    static void ReleaseInstance();
public:
    bool StartMgr(int,int);
    bool DelegateTask(CDelegator*);

public:
    const HANDLE GetMgrPort() const;
    int GetMaxAllowedThread() const 
    int GetMinAllowedThread() const 
    void GetTaskList(CStringArray&) const;
    const CThreadMap& GetThreadList() const;
private:
    CThreadPoolManager();
    virtual ~CThreadPoolManager();
private:
    bool ClearAllOperations();
private:
    static CThreadPoolManager* myInstance;
    CWinThread*    myManagerThrd;
    HANDLE         myManagerIOPort;
    int            myMaxThreadCount,myMinThreadCount;
};

Function Descriptions

static friend UINT ProcessRequestMgr(LPVOID)

Handler for the manager thread. Responsible for creating and managing the thread pool. Upon starting, the manager would wait for the request by calling GetQueuedCompletionStatus and taking actions accordingly.

Here are the events that the manager would respond to.

• MGR_UPDATE_POOL: Maintains the thread pool status, by adding or removing the required number of threads.
• MGR_INVOKE_WORKER: Checks the thread pool for idle threads. Associates the task if available or else creates the thread and associates the task in case if current thread count is less than the maximum number allowed or else puts the task in queue which would get associated to as soon as any of the thread gets idle.
• MGR_WORKER_JOB_STATUS: Reacts according to the status received from pool threads, if a pool thread has completed the task then assign with new one if anything is pending in the queue, else check minimum thread count, if acceptable then allow the thread to be idle or else terminate the thread.
• MGR_TERMINATE_POOL: Would send terminate request to the pool and finally terminate itself.

Here are the events that pool threads would respond to:

• WKR_START_JOB_PROCESSING: Would handle the execution of user request.

  OVERLAPPED aSendWrkIOPort;
  aSendWrkIOPort.Internal = reinterpret_cast<unsigned long>(aWkrIOPort);
  
  CDelegator* aProcessor  = reinterpret_cast<CDelegator*>(pStatus);
  
  if(!aProcessor)
  {
     TRACE("Invalid task pointer for worker thread!\n");
     if(!PostQueuedCompletionStatus(aMgIoPort, MGR_WORKER_JOB_STATUS,
                                                 0, &aSendWrkIOPort))
     {
    LOG("Unable to send job completion message to manager.\r\n");
    ASSERT(FALSE);
     }
     break;
  }
  
  aProcessor->ProcessData();
  delete aProcessor;
  aProcessor = NULL;
  
  if(!PostQueuedCompletionStatus(aMgIoPort, MGR_WORKER_JOB_STATUS,
                                              0, &aSendWrkIOPort))
  {
    LOG("Unable to send job completion message to manager.\r\n");
    ASSERT(FALSE);
  }
  break;
  

  All the tasks which are to be executed need to be derived from CDelegator whose virtual method is called by the thread to execute the assigned task.

  Skeleton of CDelegator is as follows:

  class CDelegator
  {
  public:
      CDelegator(){};
      virtual ~CDelegator(){};
      virtual void ProcessData()=0;
  };

• WKR_TERMINATE: Terminate pool thread.

About the demo code

Default maximum thread count is set to 25 and minimum to 0, this can be modified. On pressing of 'Start' button, the minimum number of threads specified would be created and get listed in the Inactive thread list as no task has been assigned to it currently.

Assigning of task could be done by pressing the 'Create Task' button. Each time when the button is pressed, an idle thread would be picked from the Inactive thread list and be shifted to the Active thread list. This would continue till the maximum number of permissible threads is reached.

Later tasks would be queued if the count exceeds the maximum thread count. Select a task from the active thread list and the button to complete the task would get activated, would terminate the task and free the thread (alternately, if you double click on the active thread, it would perform the same action). Immediately, one task from the pending task list would be picked and assigned to the thread. So still the thread would be shown in the active thread list instead as inactive.

On completion of all tasks, the threads would get into the inactive thread list. But there would not be any more than what we have specified in the minimum thread count. So the remaining threads would get terminated and would get active later if more tasks are generated.

At any point of time, the numbers in Min/Max can be changed, and on pressing the 'Start' button the change would get reflected provided the threads are not engaged in any active tasks.