Simple mutex-locked work queue thread pool implementation (win32)

#include <Windows.h>
#include <stdio.h>

namespace Thread
{
	typedef void(*threadFunction)(void *aParam);

	struct ThreadHandleData;
	typedef ThreadHandleData* ThreadHandle;

	void * createMutex();
	void destroyMutex(void *aHandle);
	void lockMutex(void *aHandle);
	void unlockMutex(void *aHandle);

	ThreadHandle createThread(threadFunction aThreadFunction, void *aParameter);

	void sleep(int aMSec);
	void wait(ThreadHandle aThreadHandle);
	void release(ThreadHandle aThreadHandle);

	struct ThreadHandleData
	{
		HANDLE thread;
	};

	void * createMutex()
	{
		CRITICAL_SECTION * cs = new CRITICAL_SECTION;
		InitializeCriticalSectionAndSpinCount(cs, 100);
		return (void*)cs;
	}

	void destroyMutex(void *aHandle)
	{
		CRITICAL_SECTION *cs = (CRITICAL_SECTION*)aHandle;
		DeleteCriticalSection(cs);
		delete cs;
	}

	void lockMutex(void *aHandle)
	{
		CRITICAL_SECTION *cs = (CRITICAL_SECTION*)aHandle;
		if (cs)
		{
			EnterCriticalSection(cs);
		}
	}

	void unlockMutex(void *aHandle)
	{
		CRITICAL_SECTION *cs = (CRITICAL_SECTION*)aHandle;
		if (cs)
		{
			LeaveCriticalSection(cs);
		}
	}

	struct soloud_thread_data
	{
		threadFunction mFunc;
		void *mParam;
	};

	static DWORD WINAPI threadfunc(LPVOID d)
	{
		soloud_thread_data *p = (soloud_thread_data *)d;
		p->mFunc(p->mParam);
		delete p;
		return 0;
	}

	ThreadHandle createThread(threadFunction aThreadFunction, void *aParameter)
	{
		soloud_thread_data *d = new soloud_thread_data;
		d->mFunc = aThreadFunction;
		d->mParam = aParameter;
		HANDLE h = CreateThread(NULL, 0, threadfunc, d, 0, NULL);
		if (0 == h)
		{
			return 0;
		}
		ThreadHandleData *threadHandle = new ThreadHandleData;
		threadHandle->thread = h;
		return threadHandle;
	}

	void sleep(int aMSec)
	{
		Sleep(aMSec);
	}

	void wait(ThreadHandle aThreadHandle)
	{
		WaitForSingleObject(aThreadHandle->thread, INFINITE);
	}

	void release(ThreadHandle aThreadHandle)
	{
		CloseHandle(aThreadHandle->thread);
		delete aThreadHandle;
	}


	
	void poolWorker(void *aParam);

	class PoolTask
	{
	public:
		virtual void work() = 0;
	};

#define MAX_THREADPOOL_TASKS 1024

	class Pool
	{
	public:
		volatile int mRunning; // running flag, used to flag threads to stop
		int mThreadCount; // number of threads
		ThreadHandle *mThread; // array of thread handles
		void *mWorkMutex; // mutex to protect task array/maxtask
		PoolTask *mTaskArray[MAX_THREADPOOL_TASKS]; // pointers to tasks
		int mMaxTask; // how many tasks are pending
		int mRobin; // cyclic counter, used to pick jobs for threads

		// Initialize and run thread pool. For thread count 0, work is done at addWork call.
		void init(int aThreadCount);
		// Ctor, sets known state
		Pool();
		// Dtor. Waits for the threads to finish. Work may be unfinished.
		~Pool();
		// Add work to work list. Object is not automatically deleted when work is done.
		void addWork(PoolTask *t);
		// Called from worker thread to get a new task. Returns null if no work available.
		PoolTask *getWork();
	};

	void Pool::init(int aThreadCount)
	{
		if (aThreadCount > 0)
		{
			mMaxTask = 0;
			mWorkMutex = createMutex();
			mRunning = 1;
			mThreadCount = aThreadCount;
			mThread = new ThreadHandle[aThreadCount];
			int i;
			for (i = 0; i < mThreadCount; i++)
			{
				mThread[i] = createThread(poolWorker, this);
			}
		}
	}

	Pool::Pool()
	{
		mRunning = 0;
		mThreadCount = 0;
		mThread = 0;
		mWorkMutex = 0;
		mRobin = 0;
		mMaxTask = 0;
	}

	Pool::~Pool()
	{
		mRunning = 0;
		int i;
		for (i = 0; i < mThreadCount; i++)
		{
			wait(mThread[i]);
			release(mThread[i]);
		}
		delete[] mThread;
		if (mWorkMutex)
			destroyMutex(mWorkMutex);
	}

	void Pool::addWork(PoolTask *aTask)
	{
		if (mThreadCount == 0)
		{
			aTask->work();
		}
		else
		{
			if (mWorkMutex) lockMutex(mWorkMutex);
			if (mMaxTask == MAX_THREADPOOL_TASKS)
			{
				// If we're at max tasks, do the task on calling thread 
				// (we're in trouble anyway, might as well slow down adding more work)
				if (mWorkMutex) unlockMutex(mWorkMutex);
				aTask->work();
			}
			else
			{
				mTaskArray[mMaxTask] = aTask;
				mMaxTask++;
				if (mWorkMutex) unlockMutex(mWorkMutex);
			}
		}
	}

	PoolTask * Pool::getWork()
	{
		PoolTask *t = 0;
		if (mWorkMutex) lockMutex(mWorkMutex);
		if (mMaxTask > 0)
		{
			int r = mRobin % mMaxTask;
			mRobin++;
			t = mTaskArray[r];
			mTaskArray[r] = mTaskArray[mMaxTask - 1];
			mMaxTask--;
		}
		if (mWorkMutex) unlockMutex(mWorkMutex);
		return t;
	}

	void poolWorker(void *aParam)
	{
		Pool *myPool = (Pool*)aParam;
		while (myPool->mRunning)
		{
			PoolTask *t = myPool->getWork();
			if (!t)
			{
				sleep(1);
			}
			else
			{
				t->work();
			}
		}
	}
}

// -----------------------------------------------------------
// -----------------------------------------------------------
// -----------------------------------------------------------
// -----------------------------------------------------------
// -----------------------------------------------------------
// -----------------------------------------------------------

int mande(int x, int y)
{
	double xx, yy, x0, y0, x1, y1, s;
	double divisor;
	int n;
	divisor = 1.6 * 0.001;

	xx = (x - 1024 + 128) * divisor;
	yy = (y - 1024 / 2) * divisor;
	x0 = xx;
	y0 = yy;
	n = 1;
	do
	{
		n++;
		x1 = x0 * x0;
		y1 = y0 * y0;
		y0 = (x0 * y0) * 2 + yy;
		x0 = x1 - y1 + xx;
		s = x1 + y1;
	} while ((s < 4) && (n < 2048));
	if (n == 15)
		return 0;
	return n;
}

int output[1024 * 1024];

class manderender : public Thread::PoolTask
{
public:
	int xofs, yofs;	
	virtual void work()
	{
		int i = 0, j = 0;
		for (i = 0; i < 32; i++)
			for (j = 0; j < 32; j++)
				output[xofs + i + (yofs + j) * 1024] = mande(xofs + i, yofs + j);				
	}
};

void main()
{
	manderender mandepiece[1024];
	int i, j;
	for (i = 0; i < 32; i++)
	{
		for (j = 0; j < 32; j++)
		{
			mandepiece[i * 32 + j].xofs = i*32;
			mandepiece[i * 32 + j].yofs = j*32;
		}
	}
	for (j = 0; j < 8; j++)
	{
		int t1 = GetTickCount();
		Thread::Pool *p = new Thread::Pool;
		p->init(j);
		for (i = 0; i < 1024; i++)
			p->addWork(&mandepiece[i]);

		while (p->mMaxTask) { Thread::sleep(1); }

		delete p;
		int t2 = GetTickCount();
		printf("%d threads, %d ms\n", j, t2 - t1);	
	}

}