aes.cpp

《Web编程专家指南》

/* AES.cpp */

#include "AES.h"

#include <process.h>
#include <limits.h>
#include <crtdbg.h>

unsigned __stdcall AES::PoolThread::Worker(void *ref)
{
	PoolThread& This = *(PoolThread *)ref;

	while (!This.dying)
	{
		Event *event = This.event;
		This.event = 0;
		if (!event)
			WaitForSingleObject(This.sem, INFINITE);
		else
		{
			event->Work();
			delete event;
		}
	}
	return 0;
}

AES::PoolThread::PoolThread(int stack /* = 0 */ ) : dying(FALSE), 
		sem(0), handle(0), event(0), InitState(0) 
{
	if (!(sem = CreateSemaphore(0, 1, 0, 0))
			|| !(handle = (HANDLE)_beginthreadex(0, stack, 
					Worker, 0, 0, 0)))
		InitState = ERR_CANT_CREATE_HANDLE;
	return;
}

AES::PoolThread::~PoolThread(void)
{
	dying = TRUE;
	if (sem)
	{
		if (handle)
		{
			ReleaseSemaphore(sem, 1, 0);
			WaitForSingleObject(handle, INFINITE);
			CloseHandle(handle);
		}
		CloseHandle(sem);
	}
	return;
}

unsigned __stdcall AES::Monitor(void *ref)
{
	AES &This = *(AES*)ref;

	_CrtSetReportMode(_CRT_WARN, _CRTDBG_MODE_DEBUG);

	while (!This.dying)
	{
		int eventsWaiting = 0;
		ULONG now = ULONG(clock() / CLOCKS_PER_SEC);
		ULONG nextEventTime = ULONG_MAX, lastComplaintTime = 0;
		EnterCriticalSection(&This.critSec);
		for (EventList::iterator eli = This.eventList.begin(); 
				eli != This.eventList.end(); /**/ )
		{
			for (int i = 0; i < This.size && This.pool[i].Busy(); i++)
				;	// find next available worker
			if (i < This.size && (*eli)->When() <= now)
			{
				Event *event = *eli;
				eli = This.eventList.erase(eli);
				This.pool[i].Alloc(event);
			}
			else
			{
				if ((*eli)->When() <= now)
					eventsWaiting++;
				else if ((*eli)->When() < nextEventTime)
					nextEventTime = (*eli)->When();
				eli++;
			}
		}
		LeaveCriticalSection(&This.critSec);
		if (!eventsWaiting)
			WaitForSingleObject(This.sem, (nextEventTime - now) * 1000);
		else
		{
			if (lastComplaintTime + 3 < now)
			{
				_CrtDbgReport(_CRT_WARN, 0, 0, 0,
						"Thread pool overflow. %d events delayed.", 
						eventsWaiting);
				lastComplaintTime = now;
			}
			Sleep(100);
		}
	}
	return 0;
}

AES::AESHandle AES::ScheduleEvent(ULONG ulDelay, void (*pFunc)(void*), void *pData)
{
	Event *event;

	_ASSERT(pFunc);
	
	if (!(event = new Event(ulDelay, pFunc, pData)))
		return AESHandle(0);
	
	EnterCriticalSection(&critSec);
	eventList.push_front(event);
	LeaveCriticalSection(&critSec);
	ReleaseSemaphore(sem, 1, 0);
	return AESHandle(event);
}

BOOL AES::CancelEvent(AESHandle hEvent)
{
	Event *event = (Event *)hEvent;
	BOOL cancelled = FALSE;

	EnterCriticalSection(&critSec);
	for (EventList::iterator eli = eventList.begin(); 
			eli != eventList.end(); eli++)
		if (*eli == event)
		{
			eventList.erase(eli);
			delete event;
			cancelled = TRUE;
			break;
		}
	LeaveCriticalSection(&critSec);
	return cancelled;
}

AES::AES(int threads /* = 10 */) : dying(FALSE), sem(0), 
		handle(0), size(threads), pool(0), InitState(0)
{
	// Initialize event list critical section 
	InitializeCriticalSection(&critSec);

	// Allocate thread pool
	if (!(pool = new PoolThread[size]))
	{
		InitState = ERR_INSUFFICIENT_MEMORY;
		return;
	}

	// Check init state of threads in pool
	for (int i = 0; i < size; i++)
		if (pool[i].InitState)
		{
			InitState = pool[i].InitState;
			return;
		}

	// Create monitor semaphore and thread
	if (!(sem = CreateSemaphore(0, 0, 1, 0))
			|| !(handle = (HANDLE)_beginthreadex(0, 0, 
					Monitor, this, 0, 0)))
		InitState = ERR_CANT_CREATE_HANDLE;

	return;
}

AES::~AES(void)
{
	dying = TRUE;
	if (sem)
	{
		if (handle)
		{
			ReleaseSemaphore(sem, 1, 0);
			WaitForSingleObject(handle, INFINITE);
			CloseHandle(handle);
		}
		CloseHandle(sem);
	}
	delete [] pool;
	DeleteCriticalSection(&critSec);
	return;
}

/*===========================================================================*/