如何同步三个线程？ [英] How can I synchronize three threads?

问题描述

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

My app consist of the main-process and two threads, all running concurrently and making use of three fifo-queues:

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

The fifo-q's are Qmain, Q1 and Q2. Internally the queues each use a counter that is incremented when an item is put into the queue, and decremented when an item is 'get'ed from the queue.

涉及两个线程，

QMaster，它从Q1和Q2获取，并放入Qmain，

Monitor，放入Q2，

，

The processing involve two threads,
QMaster, which get from Q1 and Q2, and put into Qmain,
Monitor, which put into Q2,
and the main process, which get from Qmain and put into Q1.

QMaster线程循环连续检查Q1和Q2的计数，如果有任何项目在q中，

The QMaster-thread loop consecutively checks the counts of Q1 and Q2 and if any items are in the q's, it get's them and puts them into Qmain.

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

The Monitor-thread loop obtains data from external sources, package it and put it into Q2.

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

The main-process of the app also runs a loop checking the count of Qmain, and if any items, get's an item from Qmain at each iteration of the loop and process it further. During this processing it occasionally puts an item into Q1 to be processed later (when it is get'ed from Qmain in turn).

问题： br />
我已经实现了所有如上所述，它工作在随机（短）的时间，然后挂起。
我已经设法识别崩溃的源头发生在一个fifo-q的
计数的增量/减少（它可能发生在其中任何一个）。

The problem:
I've implemented all as described above, and it works for a randomly (short) time and then hangs. I've managed to identify the source of the crashing to happen in the increment/decrement of the count of a fifo-q (it may happen in any of them).

我试过的：

使用三个互斥量：QMAIN_LOCK，Q1_LOCK和Q2_LOCK，当任何get / put操作
在一个相关fifo-q。结果：应用程序不会进行，只是挂起。

What I've tried:
Using three mutex's: QMAIN_LOCK, Q1_LOCK and Q2_LOCK, which I lock whenever any get/put operation is done on a relevant fifo-q. Result: the app doesn't get going, just hangs.

主进程必须一直运行，不能被阻止在一个'读' -pipe失败，socketpair失败）。

The main-process must continue running all the time, must not be blocked on a 'read' (named-pipes fail, socketpair fail).

任何建议？

我想我没有正确实现互斥体，应该怎么做？

（任何改进上述设计的意见也欢迎）

Any advice?
I think I'm not implementing the mutex's properly, how should it be done?
(Any comments on improving the above design also welcome)

下面是过程和fifo-q模板：

其中&如何在这应该放置互斥体，以避免上述问题？

[edit] below are the processes and the fifo-q-template:
Where & how in this should I place the mutex's to avoid the problems described above?

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作为队列，但保持尽可能接近你的代码：

An example of how I would adapt the design and lock the queue access the posix way. Remark that I would wrap the mutex to use RAII or use boost-threading and that I would use stl::deque or stl::queue as queue, but staying as close as possible to your code:

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() {...}
};

---

还要注意的是，互斥锁还需要初始化如示例此处，并且count（）也应使用互斥体

---

Remark too that the mutex still needs to be initialized as in the example here and that count() should also use the mutex

这篇关于如何同步三个线程？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！