Looking for an optimum multithread message queue

Question

I want to run several threads inside a process. I'm looking for the most efficient way of being able to pass messages between the threads.

Each thread would have a shared memory input message buffer. Other threads would write the appropriate buffer.

Messages would have priority. I want to manage this process myself.

Without getting into expensive locking or synchronizing, what's the best way to do this? Or is there already a well proven library available for this? (Delphi, C, or C# is fine).

Answer 1

This is hard to get right without repeating a lot of mistakes other people already made for you :)

Take a look at Intel Threading Building Blocks - the library has several well-designed queue templates (and other collections) that you can test and see which suits your purpose best.

Answer 2

If you are going to work with multiple threads, it is hard to avoid synchronisation. Fortunately it is not very hard.

For a single process, a Critical Section is frequently the best choice. It is fast and easy to use. For simplicity, I normally wrap it in a class to handle initialisation and cleanup.

#include <Windows.h>

class CTkCritSec
{
public:
    CTkCritSec(void)
    {
        ::InitializeCriticalSection(&m_critSec);
    }
    ~CTkCritSec(void)
    {
        ::DeleteCriticalSection(&m_critSec);
    }
    void Lock()
    {
        ::EnterCriticalSection(&m_critSec);
    }
    void Unlock()
    {
        ::LeaveCriticalSection(&m_critSec);
    }

private:
    CRITICAL_SECTION m_critSec;
};

You can make it even simpler using an "autolock" class you lock/unlock it.

class CTkAutoLock
{
public:
    CTkAutoLock(CTkCritSec &lock)
    : m_lock(lock)
    {
        m_lock.Lock();
    }
    virtual ~CTkAutoLock()
    {
        m_lock.Unlock();
    }
private:
    CTkCritSec &m_lock;
};

Anywhere you want to lock something, instantiate an autolock. When the function finishes, it will unlock. Also, if there is an exception, it will automatically unlock (giving exception safety).

Now you can make a simple message queue out of an std priority queue

#include <queue>
#include <deque>
#include <functional>
#include <string>

struct CMsg
{
    CMsg(const std::string &s, int n=1)
    : sText(s), nPriority(n) 
    {
    }
    int         nPriority;
    std::string sText;

    struct Compare : public std::binary_function<bool, const CMsg *, const CMsg *>
    {
        bool operator () (const CMsg *p0, const CMsg *p1)
        { 
            return p0->nPriority < p1->nPriority; 
        }
    };
};

class CMsgQueue : 
         private std::priority_queue<CMsg *, std::deque<CMsg *>, CMsg::Compare >
{
public:
    void Push(CMsg *pJob)
    {
        CTkAutoLock lk(m_critSec);
        push(pJob);
    }
    CMsg *Pop()
    {
        CTkAutoLock lk(m_critSec);

        CMsg *pJob(NULL);
        if (!Empty())
        {
            pJob = top();
            pop();
        }
        return pJob;
    }
    bool Empty()
    {
        CTkAutoLock lk(m_critSec);

        return empty();
    }

private:
    CTkCritSec m_critSec;
};

The content of CMsg can be anything you like. Note that the CMsgQue inherits privately from std::priority_queue. That prevents raw access to the queue without going through our (synchronised) methods.

Assign a queue like this to each thread and you are on your way.

Disclaimer The code here was slapped together quickly to illustrate a point. It probably has errors and needs review and testing before being used in production.