c++ - 如何同步三个线程？

我的应用程序由主进程和两个线程组成，它们同时运行并利用三个fifo队列:

fifo-q是Qmain，Q1和Q2。每个队列内部使用一个计数器，当将一个项目放入队列时，该计数器增加；当从队列中“获取”一个项目时，该计数器减小。

处理涉及两个线程，
从Q1和Q2获得的QMaster放入Qmain，
第二季度投入使用的显示器
以及从Qmain获得并放入Q1的主要过程。

QMaster线程循环连续检查Q1和Q2的计数，如果q中有任何项目，它将得到它们并将它们放入Qmain。

Monitor-thread循环从外部源获取数据，将其打包并将其放入Q2。

应用程序的主进程还运行一个循环，检查Qmain的计数，如果有任何项目，则获取一个项目
从Qmain在循环的每个迭代中进行处理。在此过程中，偶尔
将项目放入Q1中，以便稍后处理(当从Qmain依次获取时)。

问题:
如上所述，我已经实现了所有功能，并且它在随机(短时间)时间内工作，然后挂起。
我设法确定了崩溃的根源是在
fifo-q的计数(可能在任何一个中发生)。

我尝试过的
使用三个互斥锁:QMAIN_LOCK，Q1_LOCK和Q2_LOCK，只要有任何get / put操作，它们就会锁定
在相关的fifo-q上完成。结果:该应用程序无法运行，只是挂起。

主进程必须一直持续运行，并且不能在“读取”时被阻塞(命名管道失败，套接字对失败)。

有什么建议吗？
我认为我没有正确实现互斥锁，应该怎么做？
(也欢迎任何有关改进上述设计的评论)

[编辑]以下是流程和fifo-q-template:
我应该在何处以及如何放置互斥锁以避免上述问题？

main-process:
...
start thread QMaster
start thread Monitor
...
while (!quit)
{
    ...
    if (Qmain.count() > 0)
    {
        X = Qmain.get();
        process(X)
            delete X;
    }
    ...
    //at some random time:
    Q2.put(Y);
    ...
}

Monitor:
{
    while (1)
    {
        //obtain & package data
        Q2.put(data)
    }
}

QMaster:
{
    while(1)
    {
        if (Q1.count() > 0)
            Qmain.put(Q1.get());

        if (Q2.count() > 0)
            Qmain.put(Q2.get());
    }
}

fifo_q:
template < class X* > class fifo_q
{
    struct item
    {
        X* data;
        item *next;
        item() { data=NULL; next=NULL; }
    }
    item *head, *tail;
    int count;
public:
    fifo_q() { head=tail=NULL; count=0; }
    ~fifo_q() { clear(); /*deletes all items*/ }
    void put(X x) { item i=new item(); (... adds to tail...); count++; }
    X* get() { X *d = h.data; (...deletes head ...); count--; return d; }
    clear() {...}
};

最佳答案

我将如何调整设计并通过posix方式锁定队列的示例。
请注意，我将包装互斥锁以使用RAII或使用boost-threading，并且将STL::deque或STL::queue用作队列，但要尽可能地靠近代码:

main-process:
...
start thread Monitor
...
while (!quit)
{
    ...
    if (Qmain.count() > 0)
    {
        X = Qmain.get();
        process(X)
            delete X;
    }
    ...
    //at some random time:
    QMain.put(Y);
    ...
}

Monitor:
{
    while (1)
    {
        //obtain & package data
        QMain.put(data)
    }
}

fifo_q:
template < class X* > class fifo_q
{
    struct item
    {
        X* data;
        item *next;
        item() { data=NULL; next=NULL; }
    }
    item *head, *tail;
    int count;
    pthread_mutex_t m;
public:
    fifo_q() { head=tail=NULL; count=0; }
    ~fifo_q() { clear(); /*deletes all items*/ }
    void put(X x)
    {
      pthread_mutex_lock(&m);
      item i=new item();
      (... adds to tail...);
      count++;
      pthread_mutex_unlock(&m);
    }
    X* get()
    {
      pthread_mutex_lock(&m);
      X *d = h.data;
      (...deletes head ...);
      count--;
      pthread_mutex_unlock(&m);
      return d;
    }
    clear() {...}
};

还要注意，仍然需要像示例here中一样初始化互斥锁，并且count()也应该使用互斥锁