Top: Multithreading: Examples

Example 1. This simple example shows the use of mutex objects to synchronize console output. The function showdiag() can be safely called from any thread, the output from different threads will never be mixed.

mutex diagsync;

void showdiag(char severity, const char* module, const char* msg)
{
    diagsync.enter();
    try
    {
        cerr << severity << ' ' << module << ' ' << msg << '\n';
		cerr.flush();
    }
    catch(...)
    {
        diagsync.leave();
        throw;
    }
    diagsync.leave();
}

Example 2. In this bigger example we use thread, semaphore, message and msgqueue objects. Here you will find a better idea on how to synchronize diagnostics output: we create a separate msgqueue object to send messages to console asynchronously, thus freeing all threads from waiting for a mutex object. Besides dealing with diagnostics output a single message queue can handle many tasks in your application, such like serving as an intermediary between threads, receiving commands to start new threads, etc.

#include <iostream.h>
#include <ptypes.h>
#include <pasync.h>

USING_PTYPES

const int MSG_DIAG = MSG_USER + 1;


//
// class diagmessage
//

class diagmessage: public message
{
protected:
    string module;
    string diagstr;
    friend class diagthread;
public:
    diagmessage(string imodule, string idiagstr)
        : message(MSG_DIAG), module(imodule),
          diagstr(idiagstr)  {}
};


//
// class diagthread
//

class diagthread: public thread, protected msgqueue
{
protected:
    virtual void execute();     // override thread::execute()
    virtual void cleanup();     // override thread::cleanup()
    virtual void msghandler(message& msg);  // override msgqueue::msghandler()
public:
    diagthread(): thread(false), msgqueue()  { }
    inline void postdiag(string module, string diagstr);
    void postquit();
};


inline void diagthread::postdiag(string module, string diagstr)
{  
    msgqueue::post(new diagmessage(module, diagstr));
}


void diagthread::postquit()
{ 
    msgqueue::post(MSG_QUIT); 
}


void diagthread::execute()
{
    // starts message queue processing; calls
    // msghandler for each message
    msgqueue::run();
}


void diagthread::cleanup()
{
}


void diagthread::msghandler(message& msg)
{
    switch (msg.id)
    {
    case MSG_DIAG:
        cout << char(msg.param) << " ["
            << ((diagmessage&)msg).module << "] "
            << ((diagmessage&)msg).diagstr << '\n';
        cout.flush();
        break;
    default:
        defhandler(msg);
    }
}


//
// class testthread
//

class testthread: public thread
{
protected:
    diagthread& diag;
    string myname;
    virtual void execute();
    virtual void cleanup();
public:
    semaphore sem;
    testthread(diagthread& idiag)
        : thread(false), diag(idiag), myname("testthread"), sem(0)  {}
};


void testthread::execute()
{
    diag.postdiag(myname, "starts and enters sleep for 1 second");
    psleep(1000);

    diag.postdiag(myname, "releases the semaphore");
    sem.post();

    diag.postdiag(myname, "enters sleep for 1 more second");
    psleep(1000);
}


void testthread::cleanup()
{
    diag.postdiag(myname, "terminates");
}


//
// main
//

int main()
{
    diagthread diag;
    testthread thr(diag);

    string myname = "main";
    
    diag.start();
    thr.start();

    diag.postdiag(myname, "waits for the semaphore");
    thr.sem.wait();

    diag.postdiag(myname, "now waits for testthread to terminate");

    // waitfor() is necessary for static or local thread objects
    thr.waitfor();

    // can't post messages anymore, diag thread goes down...
    diag.postquit();
    diag.waitfor();

    return 0;
}

See also: thread, semaphore, mutex, rwlock, trigger, msgqueue, message