用于推送和弹出的不同互斥锁答案

【问题标题】：different mutex for push and pop用于推送和弹出的不同互斥锁
【发布时间】：2011-04-08 08:38:50
【问题描述】：

我有一个名为“subscribedQueue”的类。此类通过其订阅的发布者（复数）调用其推送方法接收其数据。

在另一个线程中，调用此类的 pop 方法来接收该数据。因此在某种意义上，这个类是多个发布者和他们的订阅者之间的一种缓冲。对于实现，我基于找到的有关线程安全队列here 的信息。

现在我的问题是双重的：

如果我将使用相同的互斥锁来推送和弹出值（目前我正在使用两个不同的互斥锁），我的程序是否可能会卡住，等待被阻止的推送？
如果没有，push 和 pop 方法怎么可能通过'lock(the_same_mutex)'。

我的假设是，如果我将使用相同的互斥体并且程序进入 pop 方法，它将在 pop 中获取锁，检查队列是否为空并等待条件变量中永远无法设置push 方法（因为锁已经被 pop 获取）。

当前代码（使用两个不同的互斥锁）：

#include <boost/thread.hpp>
#include <queue>
#include "subscriber.h"
#include "pubdata.h"
#ifdef DEBUG
#include <iostream>
#include <boost/lexical_cast.hpp>
#endif

namespace PUBLISHSUBSCRIBE
{
  template<class T>
  class SubscribedQueue: public PUBLISHSUBSCRIBE::Subscriber<T>, private std::queue< PubData<T> >
  {
  public:
    PubData<T>  pop();   //removes the next item from the queue, blocks until the queue is not empty
    void push(const PubData<T> data); //method used by the publisher to push data onto the queue
  private:
    mutable boost::mutex writeMutex_; //only needed for publishing/pushing data
    mutable boost::mutex readMutex_;  //only needed for reading/popping data
    boost::condition_variable notify_;
  };

  template<class T>
  PubData<T> SubscribedQueue<T>::pop() { //Blocks until the queue is not empty
    boost::mutex::scoped_lock lock(readMutex_);
    while(std::queue< PubData<T> >::empty())
      notify_.wait(lock); //block until recieving a notification AND the queue is not empty
    PubData<T> head = std::queue< PubData<T> >::front();
    std::queue< PubData<T> >::pop();
#ifdef DEBUG
    std::string debugOut("pop: " + boost::lexical_cast<std::string>(head) + " - timestamp: " + boost::lexical_cast<std::string>(head.timestamp()) + " - from: " + boost::lexical_cast<std::string>(this) + "\n" );
    std::cout <<debugOut;
#endif
    lock.unlock();
    return head;
  }

  template<class T>
  void SubscribedQueue<T>::push(const PubData<T> data){
    boost::mutex::scoped_lock lock(writeMutex_);
#ifdef DEBUG
    std::cout << "published: " << data << std::endl;
#endif
    std::queue< PubData<T> >::push(data);
    lock.unlock();
    notify_.notify_one();
  }
}
#endif //SUBSCRIBEDQUEUE_H

[edit] 最让我担心的是：我有一个 boost::condition_variable notify_ 在 pop 中执行“等待直到通知”。但是 pop 必须首先锁定互斥锁，同样的互斥锁也必须在“推送”中锁定以“通知”条件变量。

所以不会导致死锁，为什么不呢？

【问题讨论】：

标签： c++ multithreading mutex boost-thread

【解决方案1】：

标准库容器不是线程安全的；如果您尝试同时从多个线程修改容器，那么会发生不好的事情。

如果您有一个单独的互斥锁用于推送和弹出操作，那么您就无法防止来自两个线程的同时推送和弹出，因此您根本没有真正保护集合。

我的假设是，如果我将使用相同的互斥体并且程序进入 pop 方法，它将在 pop 中获取锁，检查队列是否为空并等待条件变量中永远无法设置push 方法（因为锁已经被 pop 获取）。

当您在 pop 中等待条件变量时，wait() 解锁互斥体，因此等待时的 push() 将能够锁定它。 push() 调用 notify_one() 并通过 scoped_lock 在函数结束时超出范围来解锁互斥锁。然后，当 pop() 线程下一次调度时，它会立即重新锁定互斥锁并继续。

【讨论】：

您的描述正是我正在寻找但找不到自己的信息。感谢您为我解决这个问题！

【解决方案2】：

是的，您必须使用单个互斥体，否则表示推送已为某个值腾出空间并增加了大小，但尚未完成将该值复制到位...读者可以读取任何形式的垃圾。

您不必担心读取器和写入器死锁 - 条件变量旨在仲裁这种情况，允许推送线程在弹出线程等待通知时运行。

【讨论】：

【解决方案3】：

根据定义，“pop”是从列表中删除一个项目的行为。因此，如果您想从多个线程的同一列表中推送和弹出，您需要使用相同的互斥锁来保护该列表。

如果您使用不同的互斥锁，那么这意味着这些不同的线程可以同时添加/删除项目，从而破坏列表。

当线程 A 添加到受互斥体保护的列表中时，线程 B 试图从该列表中弹出一个项目，将不得不等到线程 A 完成添加项目并离开锁。

【讨论】：

【解决方案4】：

首先，您的代码现在不需要使用互斥锁 mutable - 它们似乎没有在任何 const 函数中使用。

其次，不是两个互斥锁，一个用于读取，一个用于写入，您应该只有一个互斥锁用于访问队列并将访问限制在可能的最窄范围内，不要持有锁一条比需要更长的指令。

【讨论】：

【解决方案5】：

您必须只使用一个 Mutex 来创建线程安全队列，因为您必须确保一次只有一个线程访问该队列（无论是推送还是弹出）。

回答您的问题： 1）使用您当前的 push 和 pop 实现，没有死锁的可能性（除非您在其他地方使用互斥锁），因为锁仅限于 push 和 pop 的范围，并且 scoped_lock 也会在出现异常时释放互斥锁. 2）当且仅当互斥锁当前未锁定时，推送和弹出可以越过锁定。否则线程将被挂起，直到锁定线程释放互斥体。

【讨论】：