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

Question

我的应用程序由主进程和两个线程组成，所有线程都同时运行并使用三个 fifo 队列:

fifo-q 是 Qmain、Q1 和 Q2。在内部，每个队列都使用一个计数器，该计数器在将项目放入队列时递增，并在项目从队列中“取出”时递减。

处理涉及两个线程，
QMaster，从Q1和Q2得到，放入Qmain，
监控，投入Q2，
和主进程，从 Qmain 获取并放入 Q1。

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

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

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

问题:
我已经按照上面的描述实现了所有，它会随机(短)时间工作，然后挂起。我已经设法确定崩溃的来源发生在增量/减量中fifo-q 的计数(它可能发生在其中任何一个中)。

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

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

有什么建议吗？
我认为我没有正确实现互斥锁，应该如何完成？
(也欢迎对上述设计提出改进意见)

[edit] 下面是流程和 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() {...}
};

Answer 1

我将如何调整设计并以 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() 也应该使用互斥锁