станд :: conditional_variable :: notify_all не просыпаются все темы

Question

У меня есть простой пример здесь:

Проект можно назвать академическим, так как я пытаюсь научиться C++ 11 потоков. Описание того, что происходит.

Представьте себе очень большой std::string с много сборки исходного кода внутри как

MOV EBX, ECX; \ г \ nmov EAX, ECX; \ г \ п ....

Parse() функция принимает эту строку и находит все позиции линии, отмечая начало и конец строки и сохраняя их как string::const_iterators в очереди заданий.

После этого 2 рабочих потока выкладывают эту информацию из очереди и анализируют подстроку в объект класса Intstuction. Они push_back получившегося экземпляра класса Инструкции в 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 и видит его неверным.

Тогда Parse комплекты signaled true и делает notify_all; на данный момент Func не ждет, поэтому не видит уведомление.

Func затем ждет от переменной состояния и блоков.

Вы можете избежать этого, поставив замок condMutex вокруг задания на signaled.

Это нормальный шаблон для правильного использования переменных состояния - вам необходимо как проверить, так и изменить условие, в котором вы хотите ожидать, в рамках одного и того же мьютекса.