c++ - std::conditional_variable::notify_all 不唤醒所有线程

Question

我这里有一个简单的例子:

该项目可以称为学术，因为我尝试学习 c++11 线程。以下是对正在发生的事情的描述。

想象一个很大的std::string里面有很多汇编源代码，比如

mov ebx,ecx;\r\nmov eax,ecx;\r\n....

Parse()函数获取此字符串并通过标记行的开头和结尾并将它们保存为 string::const_iterators 来查找所有行位置。在作业队列中。

之后，2 个工作线程从队列中弹出此信息，并将子字符串解析为 Intstuction 类对象。他们将 Instruction 类的结果实例推回 std::vector<Instruction> result

这是一个结构声明，用于保存要解析的子字符串的行号和迭代器

struct JobItem {
    int lineNumber;
    string::const_iterator itStart;
    string::const_iterator itEnd;
};

那是一个小记录器...

void ThreadLog(const char* log) {
    writeMutex.lock();
    cout << "Thr:" << this_thread::get_id() << " " << log << endl;
    writeMutex.unlock();
}

这是共享数据:

queue<JobItem> que;
vector<Instruction> result;

这里是同步的所有原语

condition_variable condVar;
mutex condMutex;
bool signaled = false;

mutex writeMutex;
bool done=false;
mutex resultMutex;
mutex queMutex;

每线程函数

void Func() {
    unique_lock<mutex> condLock(condMutex);
    ThreadLog("Waiting...");
    while (!signaled) {
        condVar.wait(condLock);
    }
    ThreadLog("Started");
    while (!done) {
        JobItem item;
        queMutex.lock();
        if (!que.empty()) {
            item = que.front(); que.pop();
            queMutex.unlock();
        }
        else {
            queMutex.unlock();
            break;
        }
        //if i comment the line below both threads wake up
        auto instr = ParseInstruction(item.itStart, item.itEnd);
        resultMutex.lock();
        result.push_back(Instruction());
        resultMutex.unlock();
    }

管理线程的管理器函数...

vector<Instruction> Parser::Parse(const string& instructionStream){
    thread thread1(Func);
    thread thread2(Func);

    auto it0 = instructionStream.cbegin();
    auto it1 = it0;
    int currentIndex = instructionStream.find("\r\n");
    int oldIndex = 0;
    this_thread::sleep_for(chrono::milliseconds(1000)); //experimental 


    int x = 0;
    while (currentIndex != string::npos){
        auto it0  = instructionStream.cbegin() + oldIndex;
        auto it1  = instructionStream.cbegin() + currentIndex;

        queMutex.lock();
        que.push({ x,it0,it1 });
        queMutex.unlock();
        if (x == 20) {//fill the buffer a little bit before signal
            signaled = true;
            condVar.notify_all();
        }
        oldIndex = currentIndex + 2;
        currentIndex = instructionStream.find("\r\n", oldIndex);
        ++x;
    }
    thread1.join();
    thread2.join();
    done = true;

    return result;
}

问题出现在Func()功能。如您所见，我在其中使用了一些日志记录。日志说:

Output:
Thr:9928 Waiting...
Thr:8532 Waiting...
Thr:8532 Started

意思是在主线程发送了notify_all()之后对于等待的线程，只有其中一个真正醒来。如果我注释掉对 ParseInstruction() 的调用Func() 内部那么两个线程都会唤醒，否则只有一个线程会唤醒。如果能得到一些建议就太好了。

Answer 1

假设 Func 读取 signaled 并发现它为 false。

然后 Parse 将 signaled 设置为 true 并执行 notify_all；此时 Func 没有等待，所以没有看到通知。

Func 然后等待条件变量并阻塞。

您可以通过在对 signaled 的赋值周围放置一个 condMutex 锁来避免这种情况。

这是正确使用条件变量的正常模式 - 您需要在同一个互斥体中测试和修改要等待的条件。