cfixedthreadpool.cpp

window下的多线程编程参考书。值得一读

//
// FILE: CFixedThreadPool.cpp
//
// Copyright (c) 1997 by Aaron Michael Cohen and Mike Woodring
//
/////////////////////////////////////////////////////////////////////////

#include <windows.h>
#include <stdio.h>
#include <CMclAutoLock.h>
#include "CFixedThreadPool.h"

// CFixedThreadPool implementation.
//
// Note - the following implementation artificially limits the
// number of threads in the pool to MAXIMUM_WAIT_OBJECTS (64)
// only for brevity.  The CGrowableThreadPool class provides
// a more sophisticated way to have more threads in the pool
// than can be waited on in one WaitForMultipleObjects operation.
//
CFixedThreadPool::CFixedThreadPool( long lMaxThreads )
    : m_DispatchQueue(min(lMaxThreads, MAXIMUM_WAIT_OBJECTS)),
      m_lMaxThreads(min(lMaxThreads, MAXIMUM_WAIT_OBJECTS)),
      m_ExitEvent(TRUE) // Manual reset event.
{
    printf(
        "[0x%08lx] CFixedThreadPool::CFixedThreadPool\n",
        GetCurrentThreadId()
    );

    for( long lThread = 0; lThread < m_lMaxThreads; lThread++ )
    {
        m_Threads[lThread] = new CMclThread(this);
    }
}

CFixedThreadPool::~CFixedThreadPool()
{
    printf(
        "[0x%08lx] CFixedThreadPool::~CFixedThreadPool entered\n",
        GetCurrentThreadId()
    );

    {
        CMclWaitableCollection  WaitSet;

        for( long lThread = 0; lThread < m_lMaxThreads; lThread++ )
        {
            WaitSet.AddObject(*m_Threads[lThread]);
        }

        m_ExitEvent.Set();
        WaitSet.Wait(TRUE, INFINITE);
    }

    printf(
        "[0x%08lx] CFixedThreadPool::~CFixedThreadPool returning\n",
        GetCurrentThreadId()
    );
}

BOOL CFixedThreadPool::DispatchThread( CMclThreadHandler *pHandler )
{
    CDispatchQueue::DISPATCHRECORD Job(pHandler);

    return m_DispatchQueue.Put(Job);
}

unsigned CFixedThreadPool::ThreadHandlerProc( void )
{
    printf(
        "[0x%08lx] CFixedThreadPool::ThreadHandlerProc started\n",
        GetCurrentThreadId()
    );

    CDispatchQueue::DISPATCHRECORD  Job;
    long                            lJobsHandled = 0;

    while( m_DispatchQueue.Get(Job, &m_ExitEvent) )
    {
        Job.pUserThreadHandler->ThreadHandlerProc();
        lJobsHandled++;
    }
    
    printf(
        "[0x%08lx] CFixedThreadPool::ThreadHandlerProc done, %d jobs handled\n",
        GetCurrentThreadId(),
        lJobsHandled
    );

    return(0);
}

// Dispatch Queue Implementation
//
CFixedThreadPool::CDispatchQueue::CDispatchQueue( long lMaxDepth )
    : m_CritSec(),
      m_pQueue(new DISPATCHRECORD[lMaxDepth]),
      m_SlotFull(0, lMaxDepth),
      m_SlotFree(lMaxDepth, lMaxDepth),
      m_lHeadIndex(0),
      m_lTailIndex(0),
      m_lNumSlots(lMaxDepth)
{
}

CFixedThreadPool::CDispatchQueue::~CDispatchQueue()
{
}

BOOL CFixedThreadPool::CDispatchQueue::Put( const DISPATCHRECORD& Node )
{
    BOOL fPutOk;

    fPutOk = CMclWaitSucceeded(m_SlotFree.Wait(0), 1);

    if( fPutOk )
    {
        CMclAutoLock Lock(m_CritSec);

        m_pQueue[m_lTailIndex] = Node;

        m_lTailIndex = (m_lTailIndex + 1) % m_lNumSlots;
        
        m_SlotFull.Release(1);
    }

    return(fPutOk);
}

BOOL CFixedThreadPool::CDispatchQueue::Get( DISPATCHRECORD& Node, CMclEvent *pInterrupt )
{
    CMclWaitableCollection  WaitSet;
    BOOL                    fGetOk;

    WaitSet.AddObject(pInterrupt);
    WaitSet.AddObject(m_SlotFull);

    fGetOk = WaitSet.Wait(FALSE, INFINITE);

    if( fGetOk )
    {
        CMclAutoLock Lock(m_CritSec);

        Node = m_pQueue[m_lHeadIndex];

        m_lHeadIndex = (m_lHeadIndex + 1) % m_lNumSlots;
        
        m_SlotFree.Release(1);
    }

    return(fGetOk);
}