C++11 无锁单生产者单消费者 : how to avoid busy wait

Question

我正在尝试实现一个使用两个线程的类:一个用于生产者，一个用于消费者。当前的实现不使用锁:

#include <boost/lockfree/spsc_queue.hpp>
#include <atomic>
#include <thread>

using Queue =
        boost::lockfree::spsc_queue<
            int,
            boost::lockfree::capacity<1024>>;

class Worker
{
public:
    Worker() : working_(false), done_(false) {}
    ~Worker() {
        done_ = true;    // exit even if the work has not been completed
        worker_.join();
    }

    void enqueue(int value) {
        queue_.push(value);
        if (!working_) {
            working_ = true;
            worker_ = std::thread([this]{ work(); });
        }
    }

    void work() {
        int value;
        while (!done_ && queue_.pop(value)) {
            std::cout << value << std::endl;
        }
        working_ = false;
    }

private:
    std::atomic<bool> working_;
    std::atomic<bool> done_;
    Queue queue_;
    std::thread worker_;
};

应用程序需要将工作项排入队列一定时间，然后休眠等待事件。这是模拟行为的最小 main:

int main()
{
    Worker w;
    for (int i = 0; i < 1000; ++i)
        w.enqueue(i);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    for (int i = 0; i < 1000; ++i)
        w.enqueue(i);
    std::this_thread::sleep_for(std::chrono::seconds(1));
}

我很确定我的实现有问题:如果工作线程完成并且在执行 working_ = false 之前，另一个 enqueue 来了怎么办？是否可以在不使用锁的情况下使我的代码线程安全？

解决方案需要:

• 快速排队
• 即使队列不为空，析构函数也必须退出
• 没有忙等待，因为工作线程有很长一段时间是空闲的
• 尽可能不加锁

编辑

根据您的建议，我对 Worker 类做了另一个实现。这是我的第二次尝试:

class Worker
{
public:
    Worker()
        : working_(ATOMIC_FLAG_INIT), done_(false) { } 

    ~Worker() {
        // exit even if the work has not been completed
        done_ = true;
        if (worker_.joinable())
            worker_.join();
    }

    bool enqueue(int value) {
        bool enqueued = queue_.push(value);
        if (!working_.test_and_set()) {
            if (worker_.joinable())
                worker_.join();
            worker_ = std::thread([this]{ work(); });
        }
        return enqueued;
    }

    void work() {
        int value;
        while (!done_ && queue_.pop(value)) {
            std::cout << value << std::endl;
        }
        working_.clear();
        while (!done_ && queue_.pop(value)) {
            std::cout << value << std::endl;
        }
    }

private:
    std::atomic_flag working_;
    std::atomic<bool> done_;
    Queue queue_;
    std::thread worker_;
};

我在enqueue 方法中引入了worker_.join()。这会影响性能，但在极少数情况下(当队列变空并且在线程退出之前，另一个 enqueue 出现)。 working_ 变量现在是一个 atomic_flag，在 enqueue 中设置并在 work 中清除。 working_.clear() 之后的附加 while 是必需的，因为如果另一个值被推送，则在 clear 之前，但在 while 之后，该值不被处理。

这个实现是否正确？

我做了一些测试，实现似乎有效。

OT:将其作为编辑还是作为答案更好？

Answer 1

what if the worker thread completes and before executing working_ = false, another enqueue comes?

然后该值将被插入队列，但不会被处理，直到另一个值在设置标志后入队。您(或您的用户)可以决定这是否可以接受。使用锁可以避免这种情况，但它们违反了您的要求。

如果正在运行的线程即将完成并设置 working_ = false; 但在下一个值入队之前尚未停止运行，则代码可能会失败。在这种情况下，您的代码将调用 operator=根据链接文档，在运行的线程上调用 std::terminate。

在将 worker 分配给新线程之前添加 worker_.join() 应该可以防止这种情况发生。

另一个问题是，如果队列已满，queue_.push 可能会失败，因为它的大小是固定的。目前您只是忽略大小写，并且该值不会添加到完整队列中。如果您等待队列有空间，则不会快速入队(在边缘情况下)。您可以获取 push 返回的 bool(它表明它是否成功)并从 enqueue 返回它。这样调用者可以决定是等待还是丢弃该值。

或者使用非固定大小的队列。关于这个选择，Boost 是这样说的:

Can be used to completely disable dynamic memory allocations during push in order to ensure lockfree behavior. If the data structure is configured as fixed-sized, the internal nodes are stored inside an array and they are addressed by array indexing. This limits the possible size of the queue to the number of elements that can be addressed by the index type (usually 2**16-2), but on platforms that lack double-width compare-and-exchange instructions, this is the best way to achieve lock-freedom.