0

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

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>());
}
4

2 回答 2

2

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

什么是“线程锁”？很难看出你的代码试图完成什么。首先考虑您的 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();

您的布尔值以 false 开始，并且仅在线程锁定 mutexwritingData[idx]时才暂时变为 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]回到了假，因此永远不会进入 while 循环（并在 CV 上等待）。

完全对称的分析适用于布尔值，在互斥锁之外readingData[idx]也总是错误的。

所以你的条件变量永远不会被等待。您需要彻底重新考虑您的设计。

从每个队列一个互斥锁开始（双端队列对于简单地传递数据来说是多余的），并为每个队列关联一个条件变量与非空队列。因此，该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_;
};
于 2013-01-13T07:49:26.357 回答
1

是的。你需要两个互斥锁。您的死锁几乎可以肯定是单个互斥锁争用的结果。如果您使用调试器闯入正在运行的程序，您将看到线程挂起的位置。我也不明白你为什么需要布尔值。（编辑：可能会提出一种使用单个互斥锁的设计，但每个共享数据结构使用一个互斥锁更简单、更安全）

一个经验法则是，每个要保护的共享数据结构都有一个互斥锁。该互斥锁保护数据结构免受并发访问并提供线程安全。在您的情况下，每个双端队列一个互斥锁。例如：

class SafeQueue
{
private:
  std::deque<Data> q_;
  boost::mutex m_;
  boost::condition_variable v_;

public:
  void push_back(Data newData)
  {
    boost::lock_guard<boost::mutex> lock(m_);
    q_.push_back(newData);
    // notify etc.
  }
// ...
};

关于通过条件变量的通知，请参见此处：

在生产者-消费者情况下使用条件变量

condition_variable所以每个对象也会有一个，生产者会通知，消费者会等待。现在您可以创建两个这样的队列来进行双向通信。请记住，如果两个线程都被阻塞（等待数据）并且两个队列都是空的，那么只有两个线程仍然可以死锁。

于 2013-01-13T07:24:30.837 回答