dpq2.cpp

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

#include <stdio.h>
#include <string.h>
#include "CMcl.h"

class CJob {
    // this class has only pure virtual functions
    // and must be derived from to use with the
    // CJobList object...
public:
    // execute method will be called by the
    // deferred processing queue to process
    // this job...
    virtual void Execute(void) = 0;

    // cancel method will be called by the
    // deferred processing queue to process
    // this job...
    virtual void Cancel(void) = 0;
};

class CJobList : private CMclLinkedList<CJob *> {
public:
    BOOL QueueJob( CJob *pJob) {
        return PutOnTailOfList(pJob);
    };

    BOOL DequeueJob( CJob **pJob, CMclEvent *pInterrupt = NULL, DWORD dwTimeout = INFINITE) {
        return GetFromHeadOfList( *pJob, dwTimeout, pInterrupt);
    };

    BOOL RemoveJob( CJob *pJob) {
        // acquire the critical section lock...
        CMclAutoLock autoLock(m_cCritSec);

        // search the list for a matching job...
        CMclLinkedListNode *pNode = m_MasterNode.m_pNext;
        while (pNode != &m_MasterNode) {
            // since pNode is not the master node, we know that
            // it points to a data node, so this cast is safe...
            if ((static_cast<CMclLinkedListDataNode *>(pNode))->m_data == pJob)
                break;
        }

        // if we found a match, remove it...
        if (pNode != &m_MasterNode) {
            // remove the node...
            pNode->m_pPrev->m_pNext = pNode->m_pNext;
            pNode->m_pNext->m_pPrev = pNode->m_pPrev;

            // if the list is now empty, reset the event...
            if (m_MasterNode.m_pNext == &m_MasterNode) {
                m_cNotEmpty.Reset();
            }

            // return TRUE when we remove a node...
            return TRUE;
        }
        else {
            // return FALSE when there is no matching
            // node to remove...
            return FALSE;
        }
    };
};

class CDPQueue2 : public CMclThreadHandler {
private:
    CJobList m_cJobList; // list of pending jobs
    CMclThreadAutoPtr m_apWorkerThread; // auto-pointer to worker thread
    CMclEvent m_ceControl; // auto-reset event to control thread
    BOOLEAN m_bThreadExists; // flag is TRUE when worker thread exits
    BOOLEAN m_bJobPosted; // flag is TRUE between a job being queued and processed
    CMclCritSec m_CritSec; // synchronize access to m_bThreadExists and m_bJobPosted
    DWORD m_dwTimeout; // timeout value passed in the constructor

private:
    unsigned ThreadHandlerProc(void) {
        CJob *pJob;
        while (TRUE) {
            if (m_cJobList.DequeueJob( &pJob, &m_ceControl, m_dwTimeout)) {
                pJob->Execute();

                CMclAutoLock lock(m_CritSec);
                m_bJobPosted = FALSE;
            }
            else {
                CMclAutoLock lock(m_CritSec);
                if (!m_bJobPosted) {
                    m_bThreadExists = FALSE;
                    break;
                }
            }
        }

        printf("DPQ2 thread exiting.\n");

        return 0;
    };

public:
    CDPQueue2( DWORD dwTimeout = 1000) {
        m_bThreadExists = FALSE;
        m_bJobPosted = FALSE;
        m_dwTimeout = dwTimeout;
        return;
    };

    ~CDPQueue2() {
        m_ceControl.Set();
        m_apWorkerThread->Wait(INFINITE);
    }

    BOOL QueueJob( CJob *pJob) {
        BOOL bStatus = m_cJobList.QueueJob(pJob);
        // limit scope to hold auto-lock
        {        
            CMclAutoLock lock(m_CritSec);
            m_bJobPosted = TRUE;
            if (!m_bThreadExists) {
                // we need to create a worker thread...
                m_bThreadExists = TRUE;
                m_apWorkerThread = new CMclThread(this);

                printf("DPQ2 thread created.\n");
            }
        }

        return bStatus;
    };

    BOOL CancelJob( CJob *pJob) {
        if (m_cJobList.RemoveJob(pJob) == TRUE) {
            pJob->Cancel();
            return TRUE;
        }
        else {
            return FALSE;
        }
    };

    void CancelAllJobs(void) {
        CJob *pJob;
        while (m_cJobList.DequeueJob( &pJob, NULL, 0)) {
            pJob->Cancel();
        }
    };
};

class CPrintJob : public CJob {
private:
    TCHAR *m_lpString;
    BOOL m_bInUse;

public:
    // constructor...
    CPrintJob(LPCTSTR lpString) {
        // initialize the print job...
        m_lpString = NULL;
        if (lpString) {
            m_lpString = new TCHAR[strlen(lpString) + 1];
            strcpy( m_lpString, lpString);
        }
    };

    // pure virtual destructor...
    virtual ~CPrintJob() {
        if (m_lpString) {
            delete [] m_lpString;
        }
    };

    void Execute(void) {
        // execute the print job...
        if (m_lpString) {
            printf( "Executing job <%s>.\n", m_lpString);
        }

        // make the job take a small amount of time...
        Sleep(50);

        // jobs delete themselves when complete...
        delete this;
    };

    void Cancel(void) {
        // cancel the print job...
        if (m_lpString) {
            printf( "Canceling job <%s>.\n", m_lpString);
        }

        // jobs delete themselves when canceled...
        delete this;
    };
};

int main( int argc, char *argv[]) {
    CDPQueue2 dpq(10);
    CPrintJob *pPrintJob;
    TCHAR string[32];

    // this DPQ starts automatically...

    // post 100 jobs at irregular intervals 
    // between 1 and 100 milliseconds...
    for (int index = 0; index < 100; index++) {    
        Sleep((rand() % 99) + 1);

        sprintf( string, "This is job %d", index);
        pPrintJob = new CPrintJob(string);
        dpq.QueueJob(pPrintJob);
    }

    // this DPQ stops automatically...

    // cancel any jobs remaining...
    dpq.CancelAllJobs();

    return 0;
}