c++ - 等待逐步生成的 vector 的片段

Question

我想在 C++ 中实现以下内容，但我不知道哪种管理并发的方法更简单。

我有一个线程 Producer，它正在向 vector V 的后面添加元素。一旦添加了一个元素，它就被认为是只读的。为了简单起见，让我们假设我可以使用一个在增长时不会使迭代器无效的 vector ，或者我将使用读写互斥锁来处理锁​​定。但是 vector 的读者有一个问题:他们可能想要访问 V 的多个连续元素，其中一些可能尚未生成。

在任何给定时刻，V 都有一些元素，我将用“o”表示，并且 Producer 有可能添加更多元素，我将用“w”表示。因此，V 中的数据概念上 如下所示:

o o o o w w w w

我强调“概念上”，因为我不想在物理上放入尚未生成的 V 元素/虚拟对象。现在其中一位读者 R 对 V 的一段还没有完全生成感兴趣:

o o o o w w w w
    | | | |
    ---R---

因此 R 需要等待 V 增长，直到它包含 R 想要的所有元素。我可以在任何给定时刻使用 V 的最高生成元素的索引增加索引 j 。问题是，是否有一种简单的方法可以让 R 等待该索引的特定值？

Answer 1

好的，所以基于聊天，我相信这里的问题不是如何正确地同步，而是如何最小化仍然无法进行的线程的虚假唤醒.

(仅供引用，这根本不是我从原始问题中得到的印象)。

因此，我们可以进行一个简单的实现，它保留对调度哪个读取器的明确控制......

#include <queue>
#include <thread>

// associate a blocked reader's desired index with the CV it waits on
struct BlockedReadToken {
    int index_;
    std::condition_variable cv_;
    explicit BlockedReadToken(int index) : index_(index) {}
};
struct TokenOrder {
    bool operator() (BlockedReadToken const *a,
                     BlockedReadToken const *b)
    {
        return a->index_ < b->index_;
    }
};

class BlockedReaderManager
{
    std::priority_queue<BlockedReadToken*,
                        std::vector<BlockedReadToken*>, TokenOrder> queue_;
public:
    // wait for the actual index to reach the required value
    void waitfor(std::unique_lock<std::mutex> &lock,
                 int required, int const &actual)
    {
        // NOTE: a good pooled allocator might be useful here
        // (note we only allocate while holding the lock anyway,
        // so no further synchronization is required)
        std::unique_ptr<BlockedReadToken> brt(new BlockedReadToken(required));
        queue_.push(brt.get());
        while (actual < required)
            brt->cv_.wait(lock);
    }
    // release every reader blocked waiting for the new actual index
    // (don't wake any whose condition isn't satisfied yet)
    void release(std::unique_lock<std::mutex> &lock, int actual)
    {
        while (!(queue_.empty() || queue_.top()->index_ > actual)) {
            queue_.top()->cv_.notify_one();
            queue_.pop();
        }
    }
};

还有一些容器的包装器，它为读者使用了这种阻塞机制:

template <typename RandomAccessContainer>
class ProgressiveContainer
{
    int size_;
    std::mutex mutex_;
    BlockedReaderManager blocked_;
    RandomAccessContainer container_;
public:
    typedef typename RandomAccessContainer::size_type size_type;
    typedef typename RandomAccessContainer::value_type value_type;

    void push_back(value_type const &val) {
        std::unique_lock<std::mutex> guard(mutex_);
        container_.push_back(val);
        ++size_;
        blocked_.release(guard, size_);
    }
    void check_readable(int index) {
        // could optimistically avoid locking with atomic size here?
        std::unique_lock<std::mutex> guard(mutex_);
        if (size_ < index)
            blocked_.waitfor(guard, index, size_);
    }
    // allow un-locked [] access and require reader to call check_readable?
    value_type& operator[](int index) {
        return container_[index];
    }
    value_type& at(int index) {
        check_readable(index);
        return container_[index];
    }
};