2

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

我有一个线程Producer正在将元素添加到 vector 的后面V。一旦添加了一个元素，它就被认为是只读的。为了简单起见，假设我可以使用一个不会在增长时使迭代器失效的向量，或者我将使用读写互斥锁来处理锁​​定。但是向量的读者有一个问题：他们可能想要访问 的多个连续元素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 想要的所有元素。我可以在任何给定时刻j使用最高生成元素的索引来增加索引。V问题是，是否有一种简单的方法R可以等待该索引的特定值？

4

2 回答 2

1

假设您有一个消费者，您可以执行以下操作：

    ...
    pthread_mutex_lock(&mtx);
    while (array.size() < targetSize) {
        pthread_cond_wait(&cv, &mtx);
    }
    // read data from array
    // remove data from array consumed
    pthread_mutex_unlock(&mtx);
    ...
于 2013-01-07T20:05:09.157 回答
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];
    }
};
于 2013-01-07T21:05:36.613 回答