c++ - C++ : how to avoid race condition 上的 ManualResetEvent(来自 C#)实现

Question

我想学习更多关于多线程编程的知识，我认为尝试在 C++ 中实现一些 C# 同步原语会是一个很好的练习。
我从 ManualResetEvent 开始，这是我目前所拥有的:

class manual_reset_event
{
public:
    void wait_one()
    {
        if (cv_flag_.load() == false)
        {
            thread_local std::mutex mutex;
            std::unique_lock<std::mutex> lock(mutex);
            cond_var_.wait(lock, [this]() { return cv_flag_.load(); });
        }
    }

    void set()
    {
        cv_flag_.store(true);
        cond_var_.notify_all();
    }

    void reset()
    {
        cv_flag_.store(false);
    }

private:
    std::condition_variable cond_var_;
    std::atomic<bool> cv_flag_;
};

但是，这里存在竞争条件:您可以在一个线程上调用 wait_one()，通过 if (cv_flag) 检查，然后从另一个线程调用 set。这将导致 wait_one() 等待，即使 cv_flag_ 现在为真。
我可以通过在 wait_one 上使用锁、设置和重置来解决这个问题。
我想我也可以通过在 wait_one() 上的 cond_var_.wait() 之后立即调用 cond_var_.notify_all() 来解决这个问题，但我认为这不是一个好主意(尽管也许我错了)。
我想知道我是否可以在这里做其他事情(甚至可能是完全不同的方法而不使用条件变量)来避免这种竞争条件。

Answer 1

在大多数情况下，最简单的方法是使用互斥锁在对象内部使用顺序并忽略原子。只需确保对您数据的所有访问都受到锁的保护。

如果您只存储一个位，那么在之后快速执行 set 和 reset 时可能会导致丢失唤醒，因为等待的线程只会被调度reset 完成后。为了解决这个问题，我将使用一个计数器。计数器的最低位是它的“打开”状态。这种状态的每一次改变都是用一个增量来实现的。我使用 64 位计数器以防万一。 32 位不够用的可能性极小，即使它可能在长时间运行的程序中回绕。

class manual_reset_event
{
public:
    void wait_one()
    {
        std::unique_lock<std::mutex> lock(mutex_);
        uint64_t initial_value = value_;
        if(initial_value & 1)
        {
            return;
        }
        while (value_ == initial_value)
        {
            signalled_.wait(lock);
        }
    }

    void set()
    {
        std::unique_lock<std::mutex> lock(mutex);
        if((value_ & 1) == 0)
        {
            value_++;
            lock.release(); // optimization
            signalled_.notify_all();
        }
    }

    void reset()
    {
        std::unique_lock<std::mutex> lock(mutex);
        if(value_ & 1)
        {
            value_++;
        }
    }

private:
    std::mutex mutex_;
    std::condition_variable signalled_;
    uint64_t value_;
};

如果您坚持避免不必要的锁使用，您可以使用原子，但解决方案有点棘手，因为需要考虑更多的顺序。

class manual_reset_event
{
public:
    void wait_one()
    {
        uint64_t initial_value = value_;
        if(initial_value & 1)
        {
            return;
        }
        std::unique_lock<std::mutex> lock(mutex_);
        while (value_ == initial_value)
        { // !
            signalled_.wait(lock);
        }
    }

    void set()
    {
        uint64_t initial_value = value_;
        if(initial_value & 1)
        {
            return;
        }
        std::unique_lock<std::mutex> lock(mutex_);
        // Still need lock to prevent lost wakeup if atomic change happens when
        // other thread is on "// !" line.
        if(value.compare_exchange_strong(initial_value, initial_value + 1)) {
        // One strong attempt is enough. If it fails than someone else must have
        // succeeded. It's as if these two set() operations happened at the same time.
            lock.release();
            signalled_.notify_all();
        }
    }

    void reset()
    {
        uint64_t initial_value = value_;
        if((initial_value & 1) == 0)
        {
            return;
        }
        std::unique_lock<std::mutex> lock(mutex_);
        value.compare_exchange_strong(initial_value, initial_value + 1);
    }

private:
    std::mutex mutex_;
    std::condition_variable signalled_;
    std::atomic<uint64_t> value_;
};