Win32 Thread Pool

#pragma once
// Filename		: ThreadPool.h
// Author		: Siddharth Barman
// Date			: 18 Sept 2005
// Refactored	: 29 Sept 2010 by Greg Labute
/* Description	: Defines ThreadPool class. How to use the Thread Pool. First
				create a ThreadPool object. The default constructor will 
				create 10 threads in the pool. To defer creation of the pool
				pass a pool size of 0 to the constructor. 

				You can use two approaches while working with the thread 
				pool. 

				1. To make use of the thread pool, you will need to first 
				create a function having the following signature
				DWORD WINAPI ThreadProc(LPVOID); Check the CreateThread 
				documentation in MSDN to get details. Add this function to	
				the pool by calling the Run() method and pass in the function 
				name and a void* pointer to any object you want. The pool will
				pick up the function passed into Run() method and execute as 
				threads in the pool become free. 

				2. Instead of using a function pointer, you can use an object
				of a class which derives from IRunObject. Pass a pointer to 
				this object in the Run() method of the thread pool. As threads
				become free, the thread pool will call the Run() method of you
				r class. You will also need to write a body for AutoDelete() f
				unction. If the return value is true, the thread pool will use
				'delete' to free the object you pass in. If it returns false,  
				the thread pool will not do anything else to the object after
				calling the Run() function.

				It is possible to destroy the pool whenever you want by 
				calling the Destroy() method. If you want to create a new pool
				call the Create() method. 

				If this code works, it was written by Siddharth Barman, email 
				siddharth_b@yahoo.com. 
				_____ _                        _  ______           _  
				|_   _| |                      | | | ___ \         | | 
				| | | |__  _ __ ___  __ _  __| | | |_/ /__   ___ | | 
				| | | '_ \| '__/ _ \/ _` |/ _` | |  __/ _ \ / _ \| | 
				| | | | | | | |  __/ (_| | (_| | | | | (_) | (_) | | 
				\_/ |_| |_|_|  \___|\__,_|\__,_| \_|  \___/ \___/|_|  
------------------------------------------------------------------------------*/

#include <list>
#include <vector>

class ThreadPool
{
public:
	// PoolSize is the number of worker threads that will be created.
	ThreadPool(int poolSize = 10, int waitTimeForThreadsToCompleteMS = 5000);
	virtual ~ThreadPool();

	// Use this method to create the worker threads if you specified a poolSize 
	// of 0 in the constructor, or if you want to change the number of worker threads.
	// Returns: true if no error.
	bool Create(int poolSize);

	// Use this method to destroy the worker threads. The destructor of
	// this class will destroy the threads automatically.
	void Destroy();

	// How many worker threads are in the pool?
	int GetPoolSize() { return m_threads.size(); }

	// This decides whether a job is added to the front or back of the queue.
	enum ThreadPriority { High, Low };

	// Run a job
	void Run(LPTHREAD_START_ROUTINE pFunc, LPVOID pData, ThreadPriority priority = Low)
		{ AddJobToQueue(pFunc, pData, NULL, priority); }
	struct IRunObject
	{
		virtual ~IRunObject() {}
		virtual void Run() = 0;
		virtual bool AutoDelete() = 0;
	};
	void Run(IRunObject* job, ThreadPriority priority = Low)
		{ AddJobToQueue(NULL, NULL, job, priority); }

	// This is function expected to be called by thread functions or IRunObject
	// derived. The expectation is that the code will check this 'property' of
	// the pool and stop its processing as soon as possible.
	bool CheckThreadStop() 
		return WaitForSingleObject(m_hNotifyShutdown, 0) != WAIT_TIMEOUT; }

	// How many threads are currently busy?
	int GetWorkingThreadCount();

private:
	void AddJobToQueue(
		LPTHREAD_START_ROUTINE pFunc, LPVOID pData, 
		IRunObject* runObject, ThreadPriority priority);

	// This method is called by the worker threads
	HANDLE GetWaitHandle(DWORD dwThreadId);

	// The worker thread proc
	static UINT __stdcall ThreadProc(LPVOID pParam);

	// info about worker threads will be stored here
	struct ThreadData
	{
		HANDLE hWait;
		HANDLE hThread;
		DWORD dwThreadId;
	};
	typedef std::vector<ThreadData, std::allocator<ThreadData> > ThreadList;

	// jobs passed in by clients will be stored here
	struct JobData
	{
		LPTHREAD_START_ROUTINE lpStartAddress;
		LPVOID pData;
		IRunObject* runObject;
	};
	typedef std::list<JobData, std::allocator<JobData> > JobList;

#pragma warning (push)
#pragma warning (disable : 4251) // non-dll interface, ok because member is private
	JobList m_jobList;
	ThreadList m_threads;
#pragma warning (pop)

	int m_nWaitForThreadsToDieMS; // In milli-seconds
	HANDLE m_hNotifyShutdown; // notifies worker threads to terminate immediately

	CRITICAL_SECTION m_cs;
};

#include "stdafx.h"
#include "ThreadPool.h"
#include <process.h>

ThreadPool::ThreadPool(int poolSize, int waitTimeForThreadsToCompleteMS)
{
	m_nWaitForThreadsToDieMS = waitTimeForThreadsToCompleteMS;
	m_hNotifyShutdown = NULL;

	InitializeCriticalSection(&m_cs); 

	if (poolSize > 0 && !Create(poolSize))
		throw "Unable to create thread pool";
}

ThreadPool::~ThreadPool()
{
	Destroy();
	DeleteCriticalSection(&m_cs);
}

bool ThreadPool::Create(int poolSize)
{
	Destroy();
	if (poolSize == 0)
		return true;

	// create the event which will signal the threads to stop
	m_hNotifyShutdown = CreateEvent(NULL, TRUE, FALSE, NULL);
	assert(m_hNotifyShutdown != NULL);
	if (m_hNotifyShutdown == NULL)
		return false;

	// create the threads
	for (int nIndex = 0; nIndex < poolSize; ++nIndex)
	{
		UINT uThreadId;	
		HANDLE hThread = (HANDLE)_beginthreadex(
			NULL, 0, ThreadPool::ThreadProc, this,  
			CREATE_SUSPENDED, (UINT*)&uThreadId);
		DWORD dwThreadId = uThreadId;
		assert(INVALID_HANDLE_VALUE != hThread);
		if (hThread == INVALID_HANDLE_VALUE)
			return false;

		// add the entry to the thread list
		ThreadData td;
		td.hWait = CreateEvent(NULL, TRUE, FALSE, NULL);
		td.hThread = hThread;
		td.dwThreadId = dwThreadId;
		m_threads.push_back(td);
		ResumeThread(hThread); 
	}
	return true;
}

void ThreadPool::Destroy()
{
	// Never created?
	if (!m_hNotifyShutdown)
		return;

	// tell all threads to shutdown.
	SetEvent(m_hNotifyShutdown);
	int timElapsed = 0;
	while (GetWorkingThreadCount() > 0 && timElapsed < m_nWaitForThreadsToDieMS)
	{
		Sleep(100); // give the threads a chance to complete
		timElapsed += 100;
	}

	// walk through the events and threads and close them all
	for (ThreadList::iterator iter = m_threads.begin(); iter != m_threads.end(); ++iter)
	{
		CloseHandle(iter->hWait);
		CloseHandle(iter->hThread);
	}

	// close the shutdown event
	CloseHandle(m_hNotifyShutdown);
	m_hNotifyShutdown = NULL;

	// free any jobs not released
	for (JobList::iterator funcIter = m_jobList.begin(); 
		funcIter != m_jobList.end(); funcIter++) 
		if (funcIter->runObject && funcIter->runObject->AutoDelete())
			delete funcIter->runObject;

	// empty all collections
	m_jobList.clear();
	m_threads.clear();
}

int ThreadPool::GetWorkingThreadCount()
{
	int nCount = 0;
	EnterCriticalSection(&m_cs);
	for (ThreadList::iterator iter = m_threads.begin(); iter != m_threads.end(); ++iter) 
		if (WaitForSingleObject(iter->hWait, 0) == WAIT_OBJECT_0)
			++nCount;
	LeaveCriticalSection(&m_cs);
	return nCount;
}

void ThreadPool::AddJobToQueue(
	LPTHREAD_START_ROUTINE pFunc, LPVOID pData, 
	IRunObject* runObject, ThreadPriority priority)
{
	assert(pFunc != NULL || runObject != NULL);
	JobData jobData;
	jobData.lpStartAddress = pFunc;
	jobData.pData = pData;
	jobData.runObject = runObject;

	// add it to the list
	EnterCriticalSection(&m_cs);
	{
		if (priority == Low)
			m_jobList.push_back(jobData);
		else
			m_jobList.push_front(jobData);

		// See if any threads are free
		for (ThreadList::iterator iter = m_threads.begin(); iter != m_threads.end(); ++iter)
		{
			if (WaitForSingleObject(iter->hWait, 0) == WAIT_TIMEOUT)
			{
				// here is a free thread, wake it up
				SetEvent(iter->hWait); 
				break;
			}
		}
	}
	LeaveCriticalSection(&m_cs);
}

// Called by the worker thread to get its wake-up event handle
HANDLE ThreadPool::GetWaitHandle(DWORD dwThreadId)
{
	HANDLE hWait = NULL;
	EnterCriticalSection(&m_cs);
	for (ThreadList::iterator iter = m_threads.begin(); !hWait && iter != m_threads.end(); ++iter)
		if (iter->dwThreadId == dwThreadId)
			hWait = iter->hWait;
	LeaveCriticalSection(&m_cs);
	return hWait;
}

// static method:
// This is the worker thread function which will run 
// continuously till the Thread Pool is deleted.
UINT __stdcall ThreadPool::ThreadProc(LPVOID pParam)
{
	ThreadPool* pool = static_cast<ThreadPool*>(pParam);
	assert(NULL != pool);
	if (pool == NULL)
		return 1;

	DWORD dwThreadId = GetCurrentThreadId();
	HANDLE hWaits[2] = { pool->GetWaitHandle(dwThreadId), pool->m_hNotifyShutdown };
	bool bContinue = true;
	while (bContinue)
	{	
		DWORD waitResult = WaitForMultipleObjects(2, hWaits, FALSE, INFINITE);
		if (waitResult == WAIT_FAILED)
			throw "Wait failed";
		if (waitResult == WAIT_OBJECT_0 || waitResult == WAIT_OBJECT_0 + 1)
		{
			// Shutting down?
			if (pool->CheckThreadStop())
				break;

			// We have been woken up.  Get the next available job
			EnterCriticalSection(&pool->m_cs);
			JobList::iterator iter = pool->m_jobList.begin();
			if (iter == pool->m_jobList.end())
			{
				// No job available, go back to sleep
				ResetEvent(hWaits[0]);
				LeaveCriticalSection(&pool->m_cs);
				continue;
			}
			else
			{
				// Get the job and remove it from the queue
				LPTHREAD_START_ROUTINE proc = iter->lpStartAddress;
				LPVOID data = iter->pData;
				IRunObject* runObject = iter->runObject;
				pool->m_jobList.pop_front();
				LeaveCriticalSection(&pool->m_cs);

				// Run the job
				if (proc)
					proc(data);
				else if (runObject)
				{
					bool autoDelete = runObject->AutoDelete();
					runObject->Run(); // may execute "delete this"
					if (autoDelete)
						delete runObject;
				}
			}
		}
	}
	// Thread is shutting down
	ResetEvent(hWaits[0]);
	return 0;
}

m_hNotifyShutdown = CreateEvent(NULL, TRUE, FALSE, SHUTDOWN_EVT_NAME + AppName);

if(ThreadData.bFree)
{
    // here is a free thread, put it to work
    m_threads[ThreadData.dwThreadId].bFree = false;
    SetEvent(ThreadData.WaitHandle);