在使用互斥锁和等待数据时出现死锁

Question

我需要一些帮助来实现并发C++编程。 我有名字的文件，命名为"names.txt"，格式如下：

0 James 
1 Sara 
2 Isaac 

而且我有一个名为"op.txt"另一个文件，该文件载有关于地名的一些操作的文件，格式如下：

0 1 + // this means add Sara to James and store it in 0 position 
1 2 $ // this means swap values in position 1 and position 2 

和文件"output.txt"，其具有操作的输出，以此格式：

0 JamesSara 
1 Isaac 
2 Sara 

问题说创建一个用于读取names.txt和op.txt的线程并将其存储。接下来创建一些变量线程同时执行操作，最后在一个线程中执行output.txt。

这是我对这个问题的代码，并且当并发线程数大于012时，它可以正常工作。但是1和2线程的输出是不正确的。 我在这段代码中错过了什么？

#include <fstream> 
#include <iostream> 
#include <vector> 
#include <sstream> 
#include <cstdlib> 
#include <thread> 
#include <mutex> 
#include <condition_variable> 
#include <deque> 

using namespace std; 

std::mutex _opMutex; 
std::condition_variable _initCondition; 
std::condition_variable _operationCondition; 

int _counter = 0; 
int _initCounter = 0; 
int _doOperationCounter = 0; 

struct OperationStruct 
{ 
    int firstOperand; 
    int secondOperand; 
    char cOperator; 
}; 

const int THREADS = 5; 

std::deque<std::pair<int, string> > _nameVector; 
std::deque<OperationStruct> _opStructVec; 

void initNamesAndOperations() 
{ 
    ifstream infile; 

    std::pair<int, string> namePair; 

    infile.open("names.txt"); 
    if (!infile) 
    { 
     cout << "Unable to open file"; 
     exit(-1); 
    } 

    int id; 
    string value; 

    while (infile >> id >> value) 
    { 
     namePair.first = id; 
     namePair.second = value; 
     _nameVector.push_back(namePair); 
    } 
    infile.close(); 

    infile.open("op.txt"); 

    if (!infile) 
    { 
     cout << "Unable to open file"; 
     exit(-1); 
    } 

    int firstOperand; 
    int secondOperand; 
    char cOperator; 

    while (infile >> firstOperand >> secondOperand >> cOperator) 
    { 
     OperationStruct opSt; 
     opSt.firstOperand = firstOperand; 
     opSt.secondOperand = secondOperand; 
     opSt.cOperator = cOperator; 
     _opStructVec.push_back(opSt); 
     ++_initCounter; 
    } 
    infile.close(); 

    return; 
} 

void doOperationMath(int firstIndex, string firstValue, string secondValue, char cOp) 
{ 
    //basic mathematics 
    switch (cOp) 
    { 
    case '+': 
    { 
       for (int i = 0; i < _nameVector.size(); ++i) 
       { 
        std::pair<int, string> acc = _nameVector[i]; 
        if (acc.first == firstIndex) 
        { 
         acc.second = firstValue + secondValue; 
         _nameVector[i].second = acc.second; 
        } 
       } 
    } 
    break; 

    default: 
     break; 
    } 

    ++_doOperationCounter; 
} 

void doOperationSwap(int firstIndex, int secondIndex, string firstValue, string secondValue) 
{ 
    //swap 
    for (int i = 0; i < _nameVector.size(); ++i) 
    { 
     if (_nameVector[i].first == firstIndex) 
      _nameVector[i].second = secondValue; 

     if (_nameVector[i].first == secondIndex) 
      _nameVector[i].second = firstValue; 
    } 
    ++_doOperationCounter; 
} 

void doOperations() 
{ 
    while (_doOperationCounter < _initCounter) 
    { 
     std::unique_lock<mutex> locker(_opMutex); 
     _initCondition.wait(locker, [](){return !_opStructVec.empty(); }); 
     OperationStruct opSt = _opStructVec.front(); 
     _opStructVec.pop_front(); 
     locker.unlock(); 
     _operationCondition.notify_one(); 
     int firstId = opSt.firstOperand; 
     int secondId = opSt.secondOperand; 
     char cOp = opSt.cOperator; 

     string firstValue = ""; 
     string secondValue = ""; 

     for (int j = 0; j < _nameVector.size(); ++j) 
     { 
      std::pair<int, string> acc = _nameVector[j]; 
      if (firstId == acc.first) 
       firstValue = acc.second; 

      if (secondId == acc.first) 
       secondValue = acc.second; 
     } 

     if (cOp == '$') 
     { 
      doOperationSwap(firstId, secondId, firstValue, secondValue); 
     } 
     else 
     { 
      doOperationMath(firstId, firstValue, secondValue, cOp); 
     } 

    } 

    return; 
} 

void doOutputFile() 
{ 
    ofstream outfile; 

    outfile.open("sampleOutput.txt", std::ios::out | std::ios::app); 
    if (!outfile) 
    { 
     cout << "Unable to open the file"; 
     exit(-1); 
    } 

    while (_counter < _initCounter) 
    { 
     std::unique_lock<mutex> locker(_opMutex); 
     _operationCondition.wait(locker, [](){return !_nameVector.empty(); }); 
     auto accPair = _nameVector.front(); 
     _nameVector.pop_front(); 
     locker.unlock(); 

     outfile << accPair.first << " " << accPair.second << endl; 
     ++_counter; 
    } 

    return; 
} 

int main() 
{ 
    thread th1(initNamesAndOperations); 

    std::vector<thread> operationalThreads; 
    for (int i = 0; i < THREADS; ++i) 
    { 
     operationalThreads.push_back(thread(doOperations)); 
    } 

    thread th3(doOutputFile); 

    th1.join(); 

    for (auto& opthread : operationalThreads) 
     opthread.join(); 

    th3.join(); 

    return 0; 
} 

Answer 1

如果从多个线程修改变量，则可能必须使用一些同步来确保读取正确的值。最简单的方法可能是使用std::atomic作为变量，以确保操作正确排序。

此外，您的代码中没有任何内容可以确保您的doOperations线程在读取整个文件之前不会完成。

显然，您需要先读取整个数据，或者有办法等待某些数据变为可用（或达到数据的末尾）。如果读取初始数据的速度很快但处理速度较慢，那么较简单的解决方案是在开始处理线程之前读取数据。

可能发生的情况是，如果创建了大量线程，则在创建最后一个线程时，initNamesAndOperations应该会读取整个文件。

我强烈建议您购买并阅读C++并发行动作者：安东尼威廉姆斯。通过阅读这本书，你将会对现代C++多线程有一个很好的理解，它会帮助你写出正确的代码。