如何在多线程C++中删除观察者关系？

Question

我有一个主题，它向客户提供Subscribe(Observer*)和Unsubscribe(Observer*)。Subject在自己的线程中运行(在订阅的观察者上调用Notify() )，互斥对象保护它的内部观察者列表。

我希望客户机代码--我无法控制--能够在未订阅后安全地删除一个观察者。如何才能做到这一点？

• 保持互斥锁--甚至是递归互斥锁--而我通知观察者并不是一种选择，因为存在死锁风险。
• 我可以在取消订阅调用中标记一个删除的观察者，并将其从主题线程中删除。然后客户端可以等待一个特殊的“安全删除”通知。这看起来很安全，但对客户来说却很麻烦。

编辑

下面是一些说明性代码。问题是如何在运行时防止取消订阅发生在“这里的问题”评论中。然后我就可以回电话给一个被删除的对象。或者，如果我始终保持互斥而不是复制，我可以锁住某些客户端。

#include <set>
#include <functional>
#include <boost/thread.hpp>
#include <boost/bind.hpp>

using namespace std;
using namespace boost;

class Observer
{
public:
    void Notify() {}
};

class Subject
{
public:
    Subject() : t(bind(&Subject::Run, this))
    {
    }

    void Subscribe(Observer* o)
    {
        mutex::scoped_lock l(m);
        observers.insert(o);
    }

    void Unsubscribe(Observer* o)
    {
        mutex::scoped_lock l(m);
        observers.erase(o);
    }

    void Run()
    {
        for (;;)
        {
            WaitForSomethingInterestingToHappen();
            set<Observer*> notifyList;
            {
                mutex::scoped_lock l(m);
                notifyList = observers;
            }
            // Problem here
            for_each(notifyList.begin(), notifyList.end(), 
                     mem_fun(&Observer::Notify));
        }
    }

private:
    set<Observer*> observers;
    thread t;
    mutex m;
};

编辑

由于死锁的风险，我不能在持有互斥锁时通知观察者。发生这种情况的最明显的方式--客户端调用订阅或从内部通知取消订阅--很容易通过使互斥递归来补救。更危险的是在不同线程上出现间歇死锁的风险。

我所处的是多线程环境，所以在线程执行的任何时候，它通常都会持有一系列锁( L1、L2、.在里面。另一个线程将持有锁K1，K2，.公里。正确编写的客户端将确保不同的线程总是以相同的顺序获得锁。但是当客户端与我的主题的互斥对象交互时--叫它X--这个策略就会中断:对订阅/取消订阅的调用按顺序L1，L2，.从我的主题线程发出通知的调用获得X，K1，K2，……的锁。公里。如果Li或Kj中的任何一个可以在任何调用路径上重合，客户端就会出现间歇性死锁，而调试它的可能性很小。因为我不控制客户端代码，所以我不能这样做。

Answer 1

“理想”解决方案将涉及使用shared_ptr和weak_ptr。但是，为了成为通用的，它还必须考虑到Subject在其某些Observer之前被删除的问题(是的，这也可能发生)。

class Subject {
public:
    void Subscribe(std::weak_ptr<Observer> o);
    void Unsubscribe(std::weak_ptr<Observer> o);

private:
    std::mutex mutex;
    std::set< std::weak_ptr<Observer> > observers;
};

class Observer: boost::noncopyable {
public:
    ~Observer();

    void Notify();

private:
    std::mutex;
    std::weak_ptr<Subject> subject;
};

通过这种结构，我们创建了一个循环图，但是明智地使用了weak_ptr，这样就可以在没有协调的情况下销毁Observer和Subject。

注意:为了简单起见，我假设Observer一次只观察一个Subject，但它可以很容易地观察多个主题。

现在，您似乎陷入了不安全的内存管理中。正如你所能想象的那样，这是一个相当困难的情况。在这种情况下，我建议进行一个实验:异步Unsubscribe。或者至少，对Unsubscribe的调用从外部是同步的，但是是异步实现的。

想法很简单:我们将使用事件队列来实现同步。这就是：

• 对Unsubscribe的调用在队列中发布一个事件(有效载荷Observer*)，然后等待
• 当Subject线程处理了Unsubscribe事件时，它会唤醒等待的线程。

您可以使用繁忙等待或条件变量，除非性能要求不同，否则我建议使用条件变量。

注意:这个解决方案完全不能解释Subject过早死亡的原因。

Answer 2

取消订阅()应该是同步的，这样它就不会返回，直到观察者被保证不再在Subject的列表中。这是唯一安全的方法。

ETA (请将我的意见转到答案)：

由于时间似乎不是一个问题，采取并释放之间的互斥通知每个观察者。您将无法以现在的方式使用for_each，您必须检查迭代器以确保它仍然有效。

for ( ... )
{
    take mutex
    check iterator validity
    notify
    release mutex
}

做你想做的事。

Answer 3

您可以更改订阅()的签名吗？用类似于shared_ptr的东西替换观察者*会使事情变得更容易。

编辑:将上面的“轻松”替换为“更容易”。有关这如何“正确”的示例，请参阅采纳-but-not-yet-in-the-distribution Boost.Signals2 (以前为Boost.ThreadSafeSignals)库的历史。