C++ 11线程：多线程等待条件变量

Question

我目前正在研究一个模拟扩展Producer-Worker模型的问题。在这个问题中，有3个工人和3个工具可供使用，并且为了工作人员他们需要2个工具（和材料，但这些工具是无关紧要的）。如果Vault中有> = 2个工具，则工作人员将采用2.否则，他们将等待条件变量，该条件变量在> = 2时将发出信号。

这对2名工人来说很好：一个人会工作，然后将工具返回到保险库，另一个等待的工作人员将被唤醒并采取2个工具。问题是，有3名工人，总会有一个饥饿的工具。

经过一些测试后，我发现等待条件变量的线程是以栈的形式构造的。无论如何可以让它排队吗？ （1等待，2等待，3等待，当1被唤醒并且想要再次等待时，他必须等待2和3）

这里是一个样例输出。代码太长，所以我会发布它，如果它真的有必要。有3个工作线程和1个工具互斥。无论谁挨饿，每隔一段时间都会有所不同。

1 Tools taken. Remaining: 1 
2 Waiting on tools... 
3 Waiting on tools... 
1 Operator Product made. Tools returned. Tools now:3 
3 Tools taken. Remaining: 1 
1 Waiting on tools... 
3 Materials returned for switch. 
3 Operator Product made. Tools returned. Tools now:3 
1 Tools taken. Remaining: 1 
3 Waiting on tools... 
1 Materials returned for switch. 
1 Operator Product made. Tools returned. Tools now:3 
3 Tools taken. Remaining: 1 
1 Waiting on tools... 
3 Materials returned for switch. 
3 Operator Product made. Tools returned. Tools now:3 
1 Tools taken. Remaining: 1 
3 Waiting on tools... 
1 Materials returned for switch. 
1 Operator Product made. Tools returned. Tools now:3 
3 Tools taken. Remaining: 1 
1 Waiting on tools... 
3 Materials returned for switch. 
3 Operator Product made. Tools returned. Tools now:3 
1 Tools taken. Remaining: 1 
3 Waiting on tools... 
1 Materials returned for switch. 
... 

（正如你可以看到2个从来没有得到的工具）

更新：2013年7月5日 我加入一些代码。

int tools = 3; //global 
string last; //current last product on output buffer 
mutex toolsMutex; 
mutex matSearchMutex; 

int main(){ 
//Initializing Producers 
    Producer prod1(1); 
    Producer prod2(2); 
     Producer prod3(3); 



    thread p1(processor,1); 
    thread p2(processor,2); 
    thread p3(processor,3); 

    p1.detach(); 
    p2.detach(); 
    p3.detach(); 

    while(true){//forever running 

    } 

    return 0; 
} 

处理器：

//Processor method 
void processor(int i){ 
    srand(time(NULL)); 

    while (true){ //forever running 


    bool hasTools = false; 
    bool productMade = false; 
    while (productMade == false){ //while product has yet to be made. 
     //choose what to make... 



     if (hasTools == false){ 
      thread matT(getMaterials,whatToMake); 
      thread toolT(getTools,i); 
      toolT.join();   
      matT.join(); 
      hasTools = true; 
     } 
     else{ //tools acquired but no materials 
      thread matT(getMaterials,whatToMake); 
      matT.join(); 
     } 

     if (recordedLast.compare(last) != 0){ 

      //return materials and acquire new ones the next run 

      continue; 
     } 
     else { 
      makeProduct(whatToMake); 
      unique_lock<mutex> locker(toolMutex); 
      tools = tools + 2; 
      cout << i << " Operator Product made. Tools returned. Tools now:" << tools << endl; 
      productMade = true; 
      if (tools >=2) toolsCV.notify_one(); 
     } 

    //done processing 

    } 


} 

} 

makeProducts：

void makeProduct(int i){ 
    unique_lock<mutex> mainMatLock(matSearchMutex); 
    // make product according to i 
    this_thread::sleep_for(chrono::milliseconds(rand() % 1000 + 10)); 
} 

GETTOOLS：

void getTools(int i){ 
    unique_lock<mutex> locker(toolMutex); 
    if (tools <2){ 
     cout << i << " Waiting on tools..." << endl; 
     toolsCV.wait(locker);} 
    tools = tools - 2;//tools acquired 
    cout << i <<" Tools taken. Remaining: " << tools << endl; 

} 

感谢那些谁回答。我会尝试使用多个条件变量实现一个等待队列。

（PS有一些更好的办法做到代码格式化这里Stack Overflow上？除了四个空格...

Answer 1

std::condition_variable不指定在致电notify_one时哪个等待线程被唤醒。因此，您应该编写不关心哪个线程被唤醒的代码。标准模式是无论哪个线程被唤醒，该线程都应该完成需要完成的工作。

如果您需要按特定顺序唤醒线程，请使用其他机制。例如，您可以为每个线程分别设置std::condition_variable，然后在需要工具时将线程放入队列中。当一个线程递交工具时，它可以发信号通知与队列前面的线程相对应的条件变量。那个线程会被唤醒，其他线程将保持睡眠状态（模仿虚假唤醒）。

Answer 2

当有多个线程等待条件，顺序 它们被唤醒（notify_all）或哪一个是 唤醒（notify_one）是不确定的。如果你需要某种排序的 ，你需要使用notify_all，并执行它 自己可以一直等待的线程队列：。 之前等待（但是在获得互斥锁之后），将线程ID推到队列末尾的 。在循环中，循环开启“这个线程在队列的前面 和可用的必要工具”。当您获得 工具时，请从队列前面删除该ID，然后再次拨打 notify_all。

Answer 3

一种方法可能是使用线程之间共享的完全成熟的信号量，而不是条件变量。这样，你可以等待一个特定的计数。

#include <mutex> 
#include <thread> 
#include <condition_variable> 

using std::mutex; 
using std::condition_variable; 

class Semaphore 
{ 
public: 
    /** 
    * Construct a counting semaphore with an initial value 
    * @param cnt The value of the initial semaphore count 
    */ 
    Semaphore(unsigned int cnt); 

    /** 
    * acquire a semaphore count 
    * @param numRes The number of count ressources to acquire 
    */ 
    void acquire(unsigned int numRes = 1); 

    /** 
    * try to acquire a semaphore count. 
    * @param numRes The number of count ressources that the method tries to acquire 
    * @return true, if numRes could be aquired 
    *   false, otherwise 
    */ 
    bool tryAcquire(unsigned int numRes = 1); 

    /** 
    * release one semaphore cnt 
    * @param numRes The number of count ressources to release 
    */ 
    void release(unsigned int numRes = 1); 

private: 
    unsigned int cnt; 
    mutex mut; 
    condition_variable cond; 
}; 

实施的样子：

void Semaphore::acquire(unsigned int numRes) 
{ 
    unique_lock<mutex> lock(mut); 
    while (cnt < numRes) 
    { 
     cond.wait(lock); 
    } 

    cnt-=numRes; 
} 

bool Semaphore::tryAcquire(unsigned int numRes) 
{ 
    unique_lock<mutex> lock(mut); 
    if (cnt>=numRes) 
    { 
     cnt -= numRes; 
     return true; 
    } 
    return false; 
} 

void Semaphore::release(unsigned int numRes) 
{ 
    { 
     unique_lock<mutex> lock(mut); 
     cnt += numRes; 
    } 
    // notify <numRes> waiting entities 
    for (unsigned int i = 0; i<numRes; ++i) 
    { 
     cond.notify_one(); 
    } 
} 

Answer 4

这里真正的问题是，如果你有工作线程和所需要的资源量有限，你不应该关心这实际上线程被激活，你应该只关心工作完成了。唯一的区别在于日志记录。您定义的线程数是可以并行运行的线程数，这受限于资源的数量。

如果这对你没有好处，那么你需要自己采取行动，如其他答案中所述。