dpq1.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 CDPQueue1 {
private:
    class CDPWorkerThreadHandler : public CMclThreadHandler {
    private:
        CJobList *m_pcJobList; // pointer to job list passed from owner DPQ
        CMclEvent m_ceControl; // auto-reset event 
        BOOL m_bRun; // should the worker thread continue?

    public:
        CDPWorkerThreadHandler(CJobList *pcJobList) {
            m_pcJobList = pcJobList;
            m_bRun = TRUE;
        };

        void Stop(void) {
            m_bRun = FALSE;
            m_ceControl.Set();
        };

        unsigned ThreadHandlerProc(void) {
            // m_bRun will allow the handler is exit even if there is always
            // work in the queue...
            CJob *pJob;
            while (m_pcJobList->DequeueJob( &pJob, &m_ceControl, INFINITE)) {
                pJob->Execute();
                if (!m_bRun)
                    break;
            }
            return 0;
        };
    };

private:
    CJobList m_cJobList; // list of jobs to process
    CMclThreadAutoPtr m_apWorkerThread; // worker thread auto-pointer
    CDPWorkerThreadHandler m_dpHandler; // thread handler object for worker thread

public:
    CDPQueue1() : m_dpHandler( &m_cJobList) {
        return;
    };

    BOOL Start(void) {
        // check for redundant start...
        if (m_apWorkerThread.IsNull() == FALSE) {
            return FALSE;
        }

        // create the worker thread...
        m_apWorkerThread = new CMclThread( &m_dpHandler);
        printf("DPQ1 thread created.\n");
        return TRUE;
    };

    BOOL Stop(void) {
        // check for redundant stop...
        if (m_apWorkerThread.IsNull() == TRUE) {
            return FALSE;
        }

        // signal the worker thread to stop and wait
        // for it to exit...
        m_dpHandler.Stop();
        m_apWorkerThread->Wait(INFINITE);

        printf("DPQ1 thread exited.\n");

        return TRUE;
    };

    BOOL QueueJob( CJob *pJob) {
        // add the job to the queue...
        return m_cJobList.QueueJob(pJob);    
    };

    BOOL CancelJob( CJob *pJob) {
        // remove a particular job from the queue...
        if (m_cJobList.RemoveJob(pJob) == TRUE) {
            pJob->Cancel();
            return TRUE;
        }
        else {
            return FALSE;
        }
    };

    void CancelAllJobs(void) {
        // remove all jobs from the queue...
        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[]) {
    CDPQueue1 dpq;
    CPrintJob *pPrintJob;
    TCHAR string[32];

    // start the deferred processing queue...
    dpq.Start();

    // 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);
    }

    // stop the deferred processing queue...
    dpq.Stop();

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

    return 0;
}