ch07l05.txt

《Web编程专家指南》

Listing 7.5 AES C++ class.

#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;
}