c++ - 使用 boost::thread 发生线程锁。我的条件变量有什么问题？

Question

我编写了一个 Link 类，用于在网络中的两个节点之间传递数据。我用两个双端队列实现了它(一个用于从节点 0 到节点 1 的数据，另一个用于从节点 1 到节点 0 的数据)。我正在尝试对应用程序进行多线程处理，但我遇到了线程锁。我试图防止同时读取和写入同一个双端队列。在阅读有关我最初如何实现它的更多信息时，我认为我错误地使用了条件变量(也许不应该使用 bool 变量？)。我应该有两个互斥体，每个双端队列一个吗？如果可以的话请帮忙。谢谢!

class Link {
public:
// other stuff...
void push_back(int sourceNodeID, Data newData);
void get(int destinationNodeID, std::vector<Data> &linkData);

private:
// other stuff...
std::vector<int> nodeIDs_;
// qVector_ has two deques, one for Data from node 0 to node 1 and 
// one for Data from node 1 to node 0
std::vector<std::deque<Data> > qVector_; 
void initialize(int nodeID0, int nodeID1);

boost::mutex mutex_;
std::vector<boost::shared_ptr<boost::condition_variable> > readingCV_;
std::vector<boost::shared_ptr<boost::condition_variable> > writingCV_;
std::vector<bool> writingData_;
std::vector<bool> readingData_;
};

push_back 函数:

void Link::push_back(int sourceNodeID, Data newData)
{
int idx;
if (sourceNodeID == nodeIDs_[0]) idx = 1;
else 
{
    if (sourceNodeID == nodeIDs_[1]) idx = 0;
    else throw runtime_error("Link::push_back: Invalid node ID");
}

boost::unique_lock<boost::mutex> lock(mutex_);
// pause to avoid multithreading collisions
while (readingData_[idx]) readingCV_[idx]->wait(lock);

writingData_[idx] = true;
qVector_[idx].push_back(newData);
writingData_[idx] = false;
writingCV_[idx]->notify_all();
}

获取函数:

void Link::get(int destinationNodeID,
std::vector<Data> &linkData)
{
int idx;
if (destinationNodeID == nodeIDs_[0]) idx = 0;
else 
{
    if (destinationNodeID == nodeIDs_[1]) idx = 1;
    else throw runtime_error("Link::get: Invalid node ID");
}

boost::unique_lock<boost::mutex> lock(mutex_);
// pause to avoid multithreading collisions
while (writingData_[idx]) writingCV_[idx]->wait(lock);
readingData_[idx] = true;

std::copy(qVector_[idx].begin(),qVector_[idx].end(),back_inserter(linkData));
qVector_[idx].erase(qVector_[idx].begin(),qVector_[idx].end());
readingData_[idx] = false;
readingCV_[idx]->notify_all();
return;
}

这里是初始化(以防有帮助)

void Link::initialize(int nodeID0, int nodeID1)
{
readingData_ = std::vector<bool>(2,false);
writingData_ = std::vector<bool>(2,false);
for (int i = 0; i < 2; ++i)
{
    readingCV_.push_back(make_shared<boost::condition_variable>());
    writingCV_.push_back(make_shared<boost::condition_variable>());
}
nodeIDs_.reserve(2);
nodeIDs_.push_back(nodeID0);
nodeIDs_.push_back(nodeID1);
qVector_.reserve(2);
qVector_.push_back(std::deque<Data>());
qVector_.push_back(std::deque<Data>());
}

Answer 1

我正在尝试对应用程序进行多线程处理，但我遇到了线程锁。

什么是“线程锁”？很难看出您的代码试图完成什么。首先考虑您的 push_back() 代码，其同步部分如下所示:

boost::unique_lock<boost::mutex> lock(mutex_);

while (readingData_[idx]) readingCV_[idx]->wait(lock);

writingData_[idx] = true;
qVector_[idx].push_back(newData);
writingData_[idx] = false;
writingCV_[idx]->notify_all();

您的 writingData[idx] bool 值开始时为 false，只有在线程锁定了互斥体 时才会暂时变为 true。当互斥量被释放时，它又是假的。因此，对于任何必须等待获取互斥量的其他线程，writingData[idx] 将永远为真。

但是在你的 get() 代码中，你有

boost::unique_lock<boost::mutex> lock(mutex_);
// pause to avoid multithreading collisions
while (writingData_[idx]) writingCV_[idx]->wait(lock);

当线程获得互斥锁时，writingData[idx] 返回 false，因此 while 循环(并等待 CV)从不 输入。

一个完全对称的分析适用于 readingData[idx] bool 值，它在互斥锁之外也总是为假。

所以您的条件变量从不等待。您需要彻底重新考虑您的设计。

从每个队列一个互斥锁开始(双端队列对于简单地传递数据来说有点矫枉过正)，并且为每个队列关联一个条件变量，该队列是非空的。 get() 方法将等待队列非空，这将在 push_back() 方法中发出信号。像这样的东西(未经测试的代码):

template <typename Data>
class BasicQueue
{
public:
    void push( Data const& data )
    {
        boost::unique_lock  _lock( mutex_ );
        queue_.push_back( data );
        not_empty_.notify_all();
    }

    void get ( Data& data )
    {
        boost::unique_lock  _lock( mutex_ );
        while ( queue_.size() == 0 )
            not_empty_.wait( _lock ); // this releases the mutex
        // mutex is reacquired here, with queue_.size() > 0
        data = queue_.front();
        queue_.pop_front();         
    }

private:
    std::queue<Data>            queue_;
    boost::mutex                mutex_;
    boost::condition_variable   not_empty_;
};