Dr. Dobb's | Event-Driven Threads in C++

Event-Driven Threads in C++

Dan presents a powerful, multithreaded architecture that can be used by almost any application. Implemented in C++, this class library lets you quickly create and control threads.

June 01, 1995
URL:http://www.drdobbs.com/cpp/event-driven-threads-in-c/184409571

Figure 1

An object-oriented infrastructure for multithreaded apps

Dan is a software engineer in Hewlett-Packard's medical-products group. He can be contacted at [email protected].

For some years now, the superior performance and responsiveness of multithreaded applications has been anticipated by users and developers alike. However, as developers are finding, great care must be taken in designing multithreaded applications, or many expected benefits are not realized. Furthermore, multiple threads can greatly increase the complexity of an application, thereby increasing development and testing costs.

The designer of a high-caliber multithreaded application is faced with two challenges that go beyond traditional, single-threaded application development: First, inherent parallelisms in the system must be identified and translated into program segments that can execute independently; and second, effective interthread communication and synchronization strategies must be designed. Failure in the first area will result in a program that doesn't deliver on the promise of multithreading (by using CPU resources inefficiently); failure in the second will mire development in unnecessary complexity and overhead. In this article, I'll focus on the second issue and present a powerful, multithreaded architecture that can be used by almost any application--once the basic building blocks are available. Since the concepts presented are particularly useful to object-oriented programs, I'll also describe and implement a set of C++ classes.

Almost all threads must at some point coordinate their actions with events occurring in other threads. In this article, any occasion in which a thread must wait for an event to occur in another thread will be referred to as a "synchronization point." Two threads can implement a synchronization point with a semaphore. If one thread must wait until an event in another thread has occurred, then it simply blocks on a semaphore and waits for it to be cleared by the other thread. The two threads might look something like Example 1. Although this example is a bit simplistic (for example, it ignores some semaphore creation and initialization tasks), this common mechanism is easy and works well when the number of distinct events or synchronization points is small. However, it doesn't scale up well to applications with complex synchronization needs or those that contain a large number of threads communicating in different ways. Each synchronization point will need its own dedicated semaphore; as the number of threads increases, the potential number of dedicated semaphores increases dramatically. Other types of synchronization needs are more complicated, such as when a thread must wait for one of many events. Furthermore, threads using this type of synchronization are difficult to test and debug because they must be tested in isolation.

The mechanism in Example 1 doesn't scale up well because it is not structured on an event-driven model, despite the event-driven nature of the system. Thread synchronization is inherently event driven. An alternative structure for the thread is a message-passing architecture where, instead of threads that block on dedicated semaphores, message-driven threads depend on a single queue of messages. A skeleton message-driven thread is shown in Example 2.

While a thread is waiting for a message to arrive on its queue, it is blocked and consumes little or no CPU resources. As soon as the message arrives, the thread wakes up, retrieves the message, and carries out some action in response to it. Finally, it returns to wait for another message on the queue.

Message-driven threads have many advantages over the simpler type of threads described earlier:

They have a simple and predictable behavior, yet communication protocols of arbitrary complexity can be implemented with other threads.
All threads communicate with each other using the same interface. This frees up the threads to concentrate on their respective tasks. Valuable resources are saved since allocating semaphores for specific synchronization points is not necessary. Defining an interface to a thread consists of defining the set of messages the thread is prepared to respond to, the arguments that must accompany these messages, and what responses are expected.
Testing is easier because standard test beds can be developed. A message-driven thread can often be unit tested in isolation by sending it messages and trapping the responses. Similarly, if all threads use the same messaging mechanism, debugging can be facilitated by message-logging and tracking tools.
Message passing between threads of the same process can be implemented more efficiently than interprocess-communication mechanisms.

The Class Hierarchy

In Figure 1, an overview of the class-inheritance hierarchy, each box shows a class with the class name at the top, followed by the public and protected methods in the next two sections of the box. (Private methods are not shown.)

The base class of the hierarchy is a simple thread class. Although it is a no-frills class, it is not abstract; that is, you may create instances of this class. When you instantiate a simple thread, you must provide a thread function. A stack size can be supplied as well, but it is optional. The Start method must be invoked to cause the thread to begin. Start accepts an optional 32-bit argument that is passed to the thread. Once instantiated and started, the thread can be suspended, resumed, and stopped with the methods provided; see Listings One and Two; listings begin on page 98. There is also a method to get the OS/2 thread ID, which can be used for OS/2 API calls that require a thread ID.

The next class in the hierarchy, QThread, is abstract and defines the behavior and interface for the message-driven threads described earlier; see Listings Three and Four. The interface provides a method, SendMsg, which is used to send a message to the thread. Several of the methods are pure virtual, meaning that we must provide an implementation for them in the derived classes. QThread is abstract because we will want to implement at least two different types of message-driven threads. Therefore, this class serves only to formalize the behavior and interface of all message-driven threads, regardless of implementation.

At first glance, it might appear that the QThread class constructor is missing an argument--no thread procedure is required. How can you instantiate a thread object without providing a thread procedure? QThread contains a private static-thread procedure, threadProc, that is passed to the base thread class. This procedure enforces the behavior of all message threads derived from QThread. The behavior of threadProc consists of a three-step process:

It calls the Startup method, which will be implemented in the derived class. The derived class can perform any necessary initialization. This initialization is distinct from that which might be performed by the constructor for the derived class, because this startup method is called in the execution context of the new thread (that is, after the thread has been started). The constructor, on the other hand, is called in the execution context of the thread that instantiated the new thread. When the startup method is called, it is passed the initial argument supplied when the thread is started.
Next the MsgLoop method is called. Most of a typical, message-driven thread's life is spent in this simple loop, which looks something like Example 3. The methods GetMessage and DispatchMsg are pure virtual, so they will be implemented in the derived class. The GetMessage method returns FALSE to cause MsgLoop to terminate.
When MsgLoop terminates, the Shutdown method is called. This method is also pure virtual. The derived class can then perform any cleanup or notification tasks. For instance, when a message thread ends, it may wish to send messages to other threads to inform them that it will no longer be available. Just like the Startup method, this call is made in the same execution context as the rest of the thread.

An important consideration when designing the QThread class is the format of the messages. Messages can be as simple as a single short integer, or arbitrarily complex. However, three characteristics of a message format render it sufficiently flexible to meet the needs of a wide variety of applications:

It should uniquely identify a specific event or command. This is the message identifier, and usually consists of an unsigned integer.
It should allow arguments to be passed along with the message. The space allowed for arguments can vary, but at a minimum, enough room for a pointer should be allocated, since that allows passing pointers to arbitrarily complex data.
For some threads it is useful if the message carries with it additional addressing information. The thread can then use this addressing information to dispatch the message to a particular object or subsystem. Such a field is not mandatory, since that information could be passed in the argument list; however, if such a field is provided in the message structure, the overhead of packaging up a message can often be reduced.

For the design of the QThread class, I chose the same format OS/2 Presentation Manager (PM) uses for its user-interface messages--the QMSG structure. This allows the creation of two kinds of message-driven threads--OS/2 PM threads and non-OS/2 PM threads. The QMSG structure, as defined by OS/2, is shown in Example 4.

The first field is used by PM threads to hold the window handle for the window which is to receive the message (since a single PM thread may service many windows). Non-PM threads derived from QThread can use this field to contain any 32-bit value, since it is up to the thread to interpret the contents of the message; however, to follow the convention set by PM, this field could be used as a pointer to an object to which the message is addressed or on which the thread will operate. The second field of the QMSG structure is an unsigned 32-bit integer used for the message identifier. A message thread will use this value to decide what action to take. The third and fourth fields are general-purpose fields to hold arguments that accompany the message. They are 32-bit values, and their contents will depend on the message ID. In this implementation, the remaining fields will not be used.

The class hierarchy in Figure 1 shows two classes derived from QThread: MsgThread and PMThread. Although the source code for PMThread (an OS/2 PM thread class) is included (see Listings Nine and Ten), I won't discuss it in detail here. The MsgThread (Listings Five and Six) class encapsulates a message-driven thread useful for OS/2 background threads. MsgThread provides implementations for all the pure virtual methods defined by QThread, and is therefore not an abstract class.

When a MsgThread is instantiated, a message procedure must be provided as a parameter to the constructor. This user-supplied thread procedure is called each time a new message arrives on the message queue. It is only called to handle messages when they are available--the real thread procedure is the static method belonging to the QThread class, which drives the message loop. This message procedure resembles a PM window procedure, its counterpart in PM threads.

The message procedure expects five arguments: The first four are the four fields of the message (from the QMSG structure); the last argument is the initial argument passed to the MsgThread when it was started (using the Start method in the base Thread class). This argument is usually used to point to the message procedure's instance data. Looking at the source listings, you can see that the QThread class saved the Start argument as a protected member, so it is accessible to the MsgThread class, which adds it as the fifth argument to each message-procedure call.

MsgThread also provides implementations for the Startup and Shutdown methods (which were pure virtual in QThread). Startup simply calls the message procedure with a special message ID (MSG_THRD_STARTUP, defined in MsgThread.h). Shutdown calls the message procedure with another special message ID (MSG_THRD_SHUTDOWN, also defined in MsgThread.h). These two messages allow the message procedure to perform any necessary initialization and cleanup tasks. The Startup message also passes the this pointer so the message procedure can get access to the message-thread object. (This might be necessary, for example, if the message procedure needed to post messages to its own message queue.)

The final component of the MsgThread class is the actual implementation for the message queue, provided in a separate class; see Listings Seven and Eight. This allows the implementation for the message queues to change without impacting the MsgThread class. When designing the message-queue class, remember that many threads might attempt to post a message to the same queue simultaneously. Access to the queue is serialized (with a mutual exclusion semaphore) to prevent the queue from being corrupted. Another (event) semaphore is used to unblock, or wake up, the waiting thread when a message arrives.

Using the Classes

The structure of a message-driven application usually consists of a collection of communicating, message-driven threads, each with a particular responsibility. Usually, the main thread creates the other threads and then blocks until it's time to shut down. Sample.cpp and PMApp.cpp are two OS/2 programs that illustrate this; see "Availability," page 3. All of the sample code has been compiled with the Borland C++ compiler for OS/2 2.x. Sample.cpp creates several message threads, each of which forwards its messages to the next thread. Each time a thread receives a message, a bit in one of the message parameters is turned on to mark the message as having visited that thread. When the message makes a complete circle, visiting each thread, the message is posted to a display thread which prints a string indicating that the message has arrived. The display thread keeps a count of how many messages have arrived. When all messages have been accounted for, it clears a semaphore and terminates. The main thread then resumes execution and terminates the program. PMApp.cpp is a PM "Hello World" program in which the Hello World window is implemented in its own thread. In addition to the OS/2 version, a Windows NT implementation is provided electronically.

Messages are useful for command and control purposes. They are an efficient way to send requests, signal events, and synchronize activities. However, they are not usually a good choice for serializing access to shared data. A better strategy is to encapsulate shared data into objects that provide methods to access the data. These objects can contain their own (private) instances of semaphores that are used by their public methods to serialize access to the data. This way, the serialization is guaranteed without building in specific knowledge about any particular thread.

Enhancing the Message-Thread Concept

The framework presented here is a no-frills infrastructure for multithreaded applications. One obvious enhancement is to add functionality to the message queues. The queues described here do not prioritize messages. Also, some applications might wish to filter the message queue, or to peek at the messages on the queue to see if a particular message is waiting. Other enhancements could include methods to broadcast or forward messages.

Message-driven threads create exciting opportunities in the area of object-oriented design. Message-driven threads make it easy to encapsulate a thread into an otherwise ordinary class, making it possible to build a library of truly asynchronous classes. Objects in the real world tend to communicate asynchronously and perform their functions in parallel, so why not do the same with our software objects? One of the goals of object-oriented design is to minimize the conceptual leap that must be made when we map real-world objects into programming structures. By wrapping message threads into application classes and mapping events into messages, we approximate the real world we are trying to model. Both sending and receiving objects perform their duties in parallel.

Of course, mapping a problem into a system of objects operating in parallel requires rethinking interface design. Interfaces in such a system become less a collection of functions and more a protocol of messages and responses.

Conclusion

Threads are an exciting, powerful tool. However, like most other powerful new technologies, it is easy to use them inappropriately or carelessly. The class library presented here can provide ways to quickly create and control threads. In addition, message-driven threads facilitate the design and development of event-driven, multithreaded applications. Finally, I hope that these classes can aid in the development of new asynchronous classes, which can be used to better model the complexities of the real world.

Figure 1 The inheritance hierarchy using OMT notation.

Example 1: Two threads that implement a synchronization point with a semaphore.

ThreadA()
{
  // do some stuff
  block on semaphore(a)
// wait for a specific event
  // do some more stuff
}
ThreadB()
{
  // do some stuff
  clear semaphore(a)
// signals ThreadA
  // do more stuff
}

Example 2: Skeleton message-driven thread.

MsgThreadA()
{
  while (TRUE) {
    msg = getMsgFromQueue();
    // do something with msg
  }
}

Example 3: Typical message-driven thread.

while (this->GetMessage(qmsg))
  this->DispatchMsg(qmsg);

Example 4: QMSG structure as defined by OS/2.

struct QMSG {
  HWND   hwnd;
  ULONG    msg;
  MPARAM mp1;
  MPARAM mp2;
  ULONG    time;
  POINTL ptl;
  ULONG    reserved;
}

Listing One


// Thread.h
#if !defined(THREADS_INC)
#define THREADS_INC

//---------------------- Constants and Types -----------------------
const int THRDS_DEF_STACK       = 8192;     // default stack size

typedef void FNTHREADPROC (VOID * ulArg);   // thread procedure type
typedef FNTHREADPROC* PFNTHREADPROC;

//------------------------------ Class -----------------------------
class Thread {
public:
    Thread( PFNTHREADPROC pfnThread,        // Constructor
            ULONG ulStack=THRDS_DEF_STACK);
    virtual ~Thread();              // Destructor
    virtual VOID Start (ULONG arg=0L);
    VOID    Stop()      {   DosKillThread(idThread); }
    VOID    Resume()    {   DosResumeThread(idThread); }
    VOID    Suspend()   {   DosSuspendThread(idThread); }
    TID GetTID()        { return idThread; }
private:
    ULONG           ulStackSize;
    TID             idThread;
    PFNTHREADPROC   pfnThreadProc;
};
#endif

Listing Two


// Thread.cpp
//------------------------- Includes ------------------------------
#define INCL_DOS
#include <os2.h>
#include <Process.h>
#include "Thread.h"

//--------------------------- code --------------------------------
Thread::Thread(PFNTHREADPROC pfnThread, ULONG ulStack)
{
    ulStackSize = ulStack;
    pfnThreadProc = pfnThread;
}
Thread::~Thread() {}    // empty implementation
VOID Thread::Start (ULONG arg)
{
    idThread = _beginthread(pfnThreadProc, ulStackSize, (void*)arg);
}

Listing Three


// QThread.h
#if !defined(QTHREAD_INC)
#define QTHREAD_INC

//----------------------------- Includes ---------------------------
#include "Thread.h"
//----------------------------- defines ----------------------------
const int QTHRD_DEF_QSIZE       = 0L;
//------------------------------ Types -----------------------------
class QThread;      // forward declaration
typedef VOID FNQTHPROC (QThread *, ULONG);  // QThread Procedure type
typedef FNQTHPROC * PFNQTHPROC;

//------------------------------ Class -----------------------------
class QThread : public Thread {
public:
    QThread (   ULONG ulQueueSize=QTHRD_DEF_QSIZE,  // Constructor
                ULONG ulStackSize=THRDS_DEF_STACK);
    ~QThread();                         // Destructor
            VOID    Start(ULONG ulArg=0L);
    virtual VOID    SendMsg(ULONG objAddr, ULONG msg,
                        MPARAM mp1, MPARAM mp2) = 0;
protected:
    virtual VOID    MsgLoop();
    virtual BOOL    GetMessage(QMSG & qmsg) = 0;    // pure virtual
    virtual VOID    DispatchMsg (QMSG & qmsg) = 0;  // pure virtual
    virtual VOID    Startup(ULONG ulArg) = 0;       // pure virtual

    ULONG   ulQSize;
    ULONG ulParam;  // initial argument passed in when thread is started
private:
    static VOID threadProc(QThread*);       // static thread procedure
};
#endif

Listing Four


// QThread.cpp
//------------------------- Includes ------------------------------
#define INCL_WIN
#define INCL_DOS

#include <os2.h>
#include "QThread.h"
//--------------------------- code --------------------------------
QThread::QThread ( ULONG ulQueueSize, ULONG ulStack): 
         Thread((PFNTHREADPROC)this->threadProc, ulStack), ulQSize(ulQueueSize)
{}
QThread::~QThread() 
{}
VOID  QThread::Start(ULONG ulArg)
{
    ulParam = ulArg;
    this -> Thread::Start((ULONG)this);
}
VOID QThread::MsgLoop()
{
    QMSG  qmsg;
    while (this -> GetMessage(qmsg))
        this -> DispatchMsg(qmsg);
}
VOID QThread::threadProc (QThread* pQThrd)
{
    pQThrd->Startup(pQThrd->ulParam);
    pQThrd->MsgLoop();
    pQThrd->Shutdown(pQThrd->ulParam);
}

Listing Five

// MsgThrd.h
#if !defined(MSGTHREAD_INC)
#define MSGTHREAD_INC

//----------------------------- Includes ---------------------------
#include "QThread.h"
#include "MsgQ.h"
//----------------------------- defines ----------------------------
const USHORT MSG_DEF_QSIZE      = 10;

//----------------------------------------------------
// The following two values are reserved messages ID's. All MsgThread's must be
// prepared to receive them. All other message ID's are user defined.
const ULONG MSG_THRD_SHUTDOWN   = 0;  // Received during shutdown
const ULONG MSG_THRD_STARTUP    = 1;  // Received at startup
const ULONG MSG_THRD_USER   = 2;  // First user defined msg ID
//------------------------------ Types -----------------------------
typedef VOID FNMSGTHRDPROC (ULONG objAddr, ULONG msgID, MPARAM mp1, 
                                                   MPARAM mp2, ULONG ulParam);
typedef FNMSGTHRDPROC* PFNMSGTHRDPROC;
//------------------------------ Class -----------------------------
class MsgThread : public QThread {
public:
    MsgThread ( PFNMSGTHRDPROC pfn, USHORT usQSize=MSG_DEF_QSIZE,
                            ULONG ulStack=THRDS_DEF_STACK);
    ~MsgThread ();
    VOID SendMsg (ULONG objAddr, ULONG msgID, MPARAM mp1, MPARAM mp2)
            { pMsgQ->PostMsg(objAddr, msgID, mp1, mp2); }
protected:
    BOOL GetMessage (QMSG & qmsg) 
            { return pMsgQ->WaitMsg(qmsg); }
    VOID DispatchMsg (QMSG & qmsg) 
           { pfnMsg((ULONG)qmsg.hwnd,qmsg.msg,qmsg.mp1,qmsg.mp2,ulParam); }
    VOID Startup (ULONG ulArg)
            { pfnMsg((ULONG)this, MSG_THRD_STARTUP, 
                     (MPARAM)ulArg, (MPARAM)NULL,ulArg); }
    BOOL Shutdown(ULONG ulArg);
private:
    MsgQueue*       pMsgQ;      // pointer to msg queue
    PFNMSGTHRDPROC  pfnMsg;     // pointer to client thread proc
};
#endif

Listing Six


// MsgThrd.cpp
//------------------------- Includes ------------------------------
#define INCL_WIN
#define INCL_DOS

#include <os2.h>
#include "MsgThrd.h"

//--------------------------- code --------------------------------
MsgThread::MsgThread (  PFNMSGTHRDPROC pfn, USHORT usQSize, ULONG ulStack) :
                QThread(usQSize,ulStack), pfnMsg(pfn)
{
    pMsgQ = new MsgQueue(usQSize);
}
MsgThread::~MsgThread ()
{
    delete(pMsgQ);
}
BOOL MsgThread::Shutdown(ULONG ulArg)
{
    pfnMsg((ULONG)NULL, MSG_THRD_SHUTDOWN, 0L, 0L, ulArg);
    return TRUE;
}

Listing Seven


// Msgq.h
#if !defined(MSGQUEUE_INC)
#define MSGQUEUE_INC

//-------------------------- defines -------------------------------
const USHORT MQ_DEF_QSIZE   = 10;

//------------------------------ Class -----------------------------
class MsgQueue {
public:
    MsgQueue (USHORT usQSz=MQ_DEF_QSIZE);
    ~MsgQueue ();
    //--------------------------------------------------------------
    // This method blocks until it acquires the mutual exclusion 
    // semaphore for the queue. It then calls the private 
    // method QPut to add the message to the queue.
    VOID PostMsg (ULONG hobj, ULONG msg, MPARAM mp1, MPARAM mp2);
    //--------------------------------------------------------------
    // This method blocks until a message is available on the queue.
    // It then obtains the necessary mutual exclusion semaphores
    // before calling the private method QGet.
    BOOL WaitMsg(QMSG & qmsg);
private:
    BOOL    QEmpty();       // returns TRUE if queue is empty
    //--------------------------------------------------------------------
    // This function puts a message in the queue.  This function is private
    // because it assumes that the proper mutual exclusion semaphores have
    // already been acquired. If the queue is full it will automatically 
        // grow, so it cannot overflow until memory is exhausted.
    VOID    QPut(   ULONG hobj,  // hwnd or object handle
                    ULONG msg,      // msg ID
                    MPARAM mp1,     // parameter 1
                    MPARAM mp2);    // parameter 2
    //--------------------------------------------------------------------
    // This function extracts a waiting message from the queue and fills 
    // the QMSG structure.  This is a private function because it does 
    // no mutual exclusion and assumes a msg is indeed waiting at the 
    // Front of the queue (it returns whatever is there, valid or not).
    // This function does not block.
    VOID    QGet (QMSG & pqmsg);
    HEV hevItmRdy;      // Semaphore to indicate item ready
    HMTX    hmtx;           // Mutual exclusion semaphore
    USHORT  Front, Rear;            // Queue pointers
    USHORT  usQSize;        // Maximum number of queue entries
    QMSG    *msgs;          // Array of QMSG structures
};
#endif

Listing Eight


// MsgQ.cpp
//------------------------- Includes ------------------------------
#define INCL_WIN
#define INCL_DOS

#include <os2.h>
#include "MsgQ.h"

//-------------------------- defines ------------------------------
const USHORT MQ_INCREMENT   = 5;

MsgQueue::MsgQueue (USHORT usQSz) : usQSize(usQSz), Front(0), Rear(0)
{
    msgs = new QMSG[usQSize];
    DosCreateMutexSem (NULL, &hmtx, DC_SEM_SHARED, FALSE);
    DosCreateEventSem (NULL, &hevItmRdy, DC_SEM_SHARED, FALSE);
}
MsgQueue::~MsgQueue()
{
    DosCloseEventSem (hevItmRdy);
    DosCloseMutexSem (hmtx);
    delete msgs;
}
VOID MsgQueue::PostMsg (ULONG hobj, ULONG msg, MPARAM mp1, MPARAM mp2)
{
    DosRequestMutexSem (hmtx, SEM_INDEFINITE_WAIT);
    QPut(hobj, msg, mp1, mp2);
    DosReleaseMutexSem (hmtx);
    DosPostEventSem (hevItmRdy);    // wake up whoever is waiting for msgs
}
BOOL MsgQueue::WaitMsg(QMSG & qmsg)
{
    ULONG ulNPosts;
    
    DosWaitEventSem (hevItmRdy, SEM_INDEFINITE_WAIT);
    DosRequestMutexSem (hmtx, SEM_INDEFINITE_WAIT);
    QGet (qmsg);
    if (QEmpty())
        DosResetEventSem (hevItmRdy, &ulNPosts);
    DosReleaseMutexSem (hmtx);
    return (qmsg.msg);
}
BOOL MsgQueue::QEmpty()
{
    return (Front == Rear);
}
VOID MsgQueue::QPut(ULONG hobj, ULONG msg, MPARAM mp1,MPARAM mp2)
{
    USHORT  usNxtR, usNQSize, idxF, i;
    QMSG    *p;
    msgs[Rear].hwnd = (HWND)hobj;
    msgs[Rear].msg = msg;
    msgs[Rear].mp1 = mp1;
    msgs[Rear].mp2 = mp2;
    // If queue has filled up, then reallocate a larger queue
    // and transfer the contents to the new queue
    usNxtR = (Rear+1) % usQSize;
    if (usNxtR == Front) {
        usNQSize = usQSize + MQ_INCREMENT;
        p = new QMSG[usNQSize];
        idxF = Front;
        for (i=0; i < usQSize; i++) {
            p[i] = msgs[idxF++];
            if (idxF == usQSize)
                idxF = 0;
        }
        Front = 0;
        Rear = usQSize;
        delete msgs;
        usQSize = usNQSize;
        msgs = p;
    } else 
        Rear = usNxtR;
}
VOID MsgQueue::QGet (QMSG & qmsg)
{
    qmsg.hwnd = msgs[Front].hwnd;
    qmsg.msg = msgs[Front].msg;
    qmsg.mp1 = msgs[Front].mp1;
    qmsg.mp2 = msgs[Front].mp2;
    Front = (++Front % usQSize);
}

Listing Nine


// PMThread.h

#if !defined(PMTHREAD_INC)
#define PMTHREAD_INC

//----------------------------- Includes ---------------------------
#include "QThread.h"

//----------------------------- defines ----------------------------
const ULONG PMTHRD_DEF_STACKSIZE    = 8192;

//--------------------------- Public Types -------------------------
// Type for the procedure that is supplied to perform initialization and 
// shutdown for the PM thread.  Usually this proc registers user classes
// and/or creates the main window or windows.
class PMThread;     // forward declaration
typedef VOID FNPROC (BOOL start, ULONG ulArg, PMThread* pmThrd);
typedef FNPROC* PFNPROC;
//------------------------------ Class -----------------------------
class PMThread : public QThread {
public:
    PMThread (  PFNPROC pfn, USHORT usQSize=0,
                ULONG ulStackSize=PMTHRD_DEF_STACKSIZE);
    ~PMThread ();
    VOID Startup (ULONG ulArg);
    BOOL Shutdown(ULONG ulArg);
    VOID SendMsg(   ULONG objAddr, ULONG msg, MPARAM mp1, MPARAM mp2);
    BOOL GetMessage(QMSG & qmsg)
            { return WinGetMsg(hab, &qmsg, NULLHANDLE, 0,0); }
    VOID DispatchMsg (QMSG & qmsg)
            { WinDispatchMsg (hab, &qmsg); }
    HAB QueryHAB() { return hab; }
    HMQ QueryHMQ() { return hmq; }
private:
    HAB         hab;        // PM Anchor block handle
    HMQ         hmq;        // Message Queue handle
    PFNPROC             pfnProc;
};
#endif

Listing Ten


// PMThread.cpp

//------------------------- Includes ------------------------------
#define INCL_WIN
#define INCL_DOS

#include <os2.h>
#include "PMThread.h"

//--------------------------- code --------------------------------
PMThread::PMThread (PFNPROC pfn, USHORT usQSize, ULONG ulStackSize) :   
                                  QThread (usQSize, ulStackSize), pfnProc(pfn)
{}
PMThread::~PMThread()
{}
VOID PMThread::Startup(ULONG ulArg)
{
    hab = WinInitialize(0);
    hmq = WinCreateMsgQueue (hab, ulQSize);
    pfnProc(TRUE, ulArg, this);
}
BOOL PMThread::Shutdown(ULONG ulArg)
{
    pfnProc(FALSE, ulArg, this);
    WinDestroyMsgQueue(hmq);
    WinTerminate(hab);
    return TRUE;
}
VOID PMThread::SendMsg( ULONG objAddr, ULONG msg, MPARAM mp1, MPARAM mp2)
{
    if (objAddr)
        WinPostMsg ((HWND)objAddr, msg, mp1, mp2);
    else
        WinPostQueueMsg (hmq, msg, mp1, mp2);
}