c++ - 带 bo​​ost::atomic 的无锁队列——我这样做对吗？

Question

简短版:

我正在尝试替换来自 here 的无锁、单一生产者、单一消费者队列实现中使用的 C++11 中的 std::atomic .如何将其替换为 boost::atomic ？

长版:

我正在尝试使用工作线程从我们的应用程序中获得更好的性能。每个线程都有自己的任务队列。我们必须在每个任务出队/入队之前使用锁进行同步。

然后我找到了Herb Sutter关于无锁队列的文章。这似乎是一个理想的替代品。但代码使用了 C++11 的 std::atomic，我目前无法将其引入到项目中。

更多的谷歌搜索导致了一些例子，例如 this one for Linux (echelon's) , 和 this one for Windows (TINESWARE's) .两者都使用平台的特定构造，例如 WinAPI 的 InterlockedExchangePointer 和 GCC 的 __sync_lock_test_and_set。

我只需要支持 Windows 和 Linux，所以也许我可以使用一些 #ifdef。但我认为使用 boost::atomic 提供的功能可能会更好。 Boost Atomic 还不是官方 Boost 库的一部分。所以我从http://www.chaoticmind.net/~hcb/projects/boost.atomic/下载了源码并在我的项目中使用包含文件。

这是我目前得到的:

#pragma once

#include <boost/atomic.hpp>

template <typename T>
class LockFreeQueue
{
private:
    struct Node
    {
        Node(T val) : value(val), next(NULL) { }
        T value;
        Node* next;
    };
    Node* first; // for producer only
    boost::atomic<Node*> divider;  // shared
    boost::atomic<Node*> last; // shared

public:
    LockFreeQueue()
    {
        first = new Node(T());
        divider = first;
        last= first;
    }

    ~LockFreeQueue()
    {
        while(first != NULL) // release the list
        {
            Node* tmp = first;
            first = tmp->next;
            delete tmp;
        }
    }

    void Produce(const T& t)
    {
        last.load()->next = new Node(t); // add the new item
        last = last.load()->next;
        while(first != divider) // trim unused nodes
        {
            Node* tmp = first;
            first = first->next;
            delete tmp;
        }
    }

    bool Consume(T& result)
    {
        if(divider != last) // if queue is nonempty
        {
            result = divider.load()->next->value; // C: copy it back
            divider = divider.load()->next;
            return true;  // and report success
        }
        return false;  // else report empty
    }
};

需要注意的一些修改:

boost::atomic<Node*> divider;  // shared
boost::atomic<Node*> last; // shared

和

    last.load()->next = new Node(t); // add the new item
    last = last.load()->next;

和

        result = divider.load()->next->value; // C: copy it back
        divider = divider.load()->next;

我在这里正确地应用了来自 boost::atomic 的 load()(和隐含的 store())吗？我们可以说这等同于 Sutter 的原始 C++11 无锁队列吗？

附言。我研究了 SO 上的许多线程，但似乎没有一个提供 boost::atomic 和无锁队列的示例。

Answer 1

你试过了吗Intel Thread Building Blocks' atomic<T> ？跨平台且免费。

还有...

单一生产者/单一消费者 使您的问题变得容易得多，因为您的线性化点可以是单一运算符(operator)。如果您准备好接受有界队列，事情会变得更容易。

有界队列为缓存性能提供了优势，因为您可以保留缓存对齐的内存块以最大化命中率，例如:

#include <vector>
#include "tbb/atomic.h"
#include "tbb/cache_aligned_allocator.h"    

template< typename T >
class SingleProdcuerSingleConsumerBoundedQueue { 
    typedef vector<T, cache_aligned_allocator<T> > queue_type;

public:
    BoundedQueue(int capacity):
        queue(queue_type()) {
        head = 0;
        tail = 0;
        queue.reserve(capacity);
    }

    size_t capacity() {
        return queue.capacity();
    }

    bool try_pop(T& result) {
        if(tail - head == 0)
            return false;
        else {
            result = queue[head % queue.capacity()];
            head.fetch_and_increment(); //linearization point
            return(true);
        }
    }

    bool try_push(const T& source) {
        if(tail - head == queue.capacity()) 
            return(false);
        else {
            queue[tail %  queue.capacity()] = source;
            tail.fetch_and_increment(); //linearization point
            return(true);
        }
    }

    ~BoundedQueue() {}

private:
    queue_type queue;
    atomic<int> head;
    atomic<int> tail;
};