循环缓冲区的线程安全实现

Question

来自 boost 库的 Circular_buffer 不是线程安全的。所以我将 boost::circular_buffer 对象包装在一个类中，如下所示。通过使用条件变量、互斥锁和锁获取/释放来实现线程之间的互斥（我认为）。这个实现线程安全吗？

#include <boost/thread/condition.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
#include <boost/circular_buffer.hpp>

// Thread safe circular buffer 
template <typename T>
class circ_buffer : private boost::noncopyable
{
public:
    typedef boost::mutex::scoped_lock lock;
    circ_buffer() {}
    circ_buffer(int n) {cb.set_capacity(n);}
    void send (T imdata) {
        lock lk(monitor);
        cb.push_back(imdata);
        buffer_not_empty.notify_one();
    }
    T receive() {
        lock lk(monitor);
        while (cb.empty())
            buffer_not_empty.wait(lk);
        T imdata = cb.front();
        cb.pop_front();
        return imdata;
    }
    void clear() {
        lock lk(monitor);
        cb.clear();
    }
    int size() {
        lock lk(monitor);
        return cb.size();
    }
    void set_capacity(int capacity) {
        lock lk(monitor);
        cb.set_capacity(capacity);
    }
private:
    boost::condition buffer_not_empty;
    boost::mutex monitor;
    boost::circular_buffer<T> cb;
};

编辑 现在这是一个模板类，它接受任何类型的对象（不仅仅是cv::Mat 对象）。

Answer 1

是的。
如果你用同一个锁锁定所有公共方法，它将是线程安全的。

您可以考虑使用 read-write locks，如果您有很多并发阅读器，它的性能可能会更好。

如果您没有很多读者，它只会增加开销，但可能值得检查选项和测试。

Answer 2

我认为它看起来不错，只是在send 中制作了一些毫无意义的 Mat 副本。不需要new，直接pushsend的参数到你的cb即可。

Answer 3

您的实现类似于blogger 所示的实现。您应该阅读该博客，看看您在实施过程中是否遗漏了什么。

如果您的 Mat 对象的创建/复制成本很高，则应避免不断地创建/复制/删除它们。相反，您应该有一个 Mat 对象池（又名空闲列表），这些对象在某种管道架构中不断回收。我在这个answer 中描述了这种类型的架构到一个相关的问题。

在那个答案中，我建议使用阻塞堆栈来实现池，但您也可以使用阻塞 circular_buffer。我建议使用堆栈的原因是因为我认为它可能对缓存更友好，但我从未实际测量过它是否会有所作为。

Answer 4

很老的问题:) 这是一个带有无锁实现的 disgin Link

这是 Link 的 BSD-2 库

Answer 5

乍一看看起来不错，只是您根本没有使用buffer_not_full 条件。您可能想要添加类似于buffer_not_empty 代码的代码。

Answer 6

//This implementation above is broken. You also need condition variable
//boost::condition buffer_not_full; and wait in send on available space in the circular buffer.
enter code here

void send (T imdata) {
   lock lk(monitor);
   while (cb.full())
      buffer_not_full.wait(lk);
   cb.push_back(imdata);
   buffer_not_empty.notify_one();
}

T receive() {
     lock lk(monitor);
     while (cb.empty())
        buffer_not_empty.wait(lk);
     T imdata = cb.front();
     cb.pop_front();
     buffer_not_full.notify_one();
     return imdata;
}