并发队列

Question

最近，我编写了一个并发的、没有互斥锁(但不是无锁的)队列，并想知道它是否真的正确，以及是否有什么特别的改进可以做：

template <typename T>
class concurrent_queue
{
protected:
    T *storage;
    std::size_t s;
    std::atomic<T*> consumer_head, producer_head;

    union alignas(16) dpointer
    {
        struct
        {
            T *ptr;
            std::size_t cnt;
        };
        __int128 val;
    };

    dpointer consumer_pending, producer_pending;

public:
    concurrent_queue(std::size_t s): storage(nullptr)
    {
        storage = static_cast<T*>(::operator new((s+1)*sizeof(T)));

        consumer_head = storage;
        __atomic_store_n(&(consumer_pending.val), (dpointer{storage, 0}).val, __ATOMIC_SEQ_CST);

        producer_head = storage;

        __atomic_store_n(&(producer_pending.val), (dpointer{storage, 0}).val, __ATOMIC_SEQ_CST);
        this->s = s + 1;
    }
    ~concurrent_queue()
    {
        while(consumer_head != producer_head)
        {
            ((T*)consumer_head)->~T();
            ++consumer_head;
            if(consumer_head == storage + s)
                consumer_head = storage;
        }
        ::operator delete(storage);
    }

    template <typename U>
    bool push(U&& e)
    {
        while(true)
        {
            dpointer a;
            a.val = __atomic_load_n(&(producer_pending.val), __ATOMIC_ACQUIRE);
            auto b = consumer_head.load(std::memory_order_relaxed);
            auto next = a.ptr + 1;
            if(next == storage + s) next = storage;

            if(next == b) return false;
            dpointer newval{next, a.cnt+1};
            if(!__atomic_compare_exchange_n(&(producer_pending.val), &(a.val), (newval.val), true, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) continue;

            new (a.ptr) T(std::forward<U>(e));

            while(!producer_head.compare_exchange_weak(a.ptr, next, std::memory_order_release, std::memory_order_relaxed));
            return true;
        }
    }

    template <typename U>
    bool pop(U& result)
    {
        while(true)
        {
            dpointer a;
            a.val = __atomic_load_n(&(consumer_pending.val), __ATOMIC_ACQUIRE);
            auto b = producer_head.load(std::memory_order_relaxed);
            auto next = a.ptr + 1;
            if(next == storage + s) next = storage;

            if(a.ptr == b) return false;
            dpointer newval{next, a.cnt+1};
            if(!__atomic_compare_exchange_n(&(consumer_pending.val), &(a.val), (newval.val), true, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) continue;

            result = std::move(*(a.ptr));
            (a.ptr)->~T();

            while(!consumer_head.compare_exchange_weak(a.ptr, next, std::memory_order_release, std::memory_order_relaxed));
            return true;
        }
    }
};

这是一个特定于GCC的x86_64平台和一个支持双宽度CAS的CPU，但我想对其他平台进行调整并不难。

我强调用多线程推送和弹出式测试它，使用POD类型和内存管理类型(如std::vector )，并且没有任何问题，所以我把far...before放在双宽度的far...before中，我遇到了far...before问题，并且有分割错误，但是它似乎很好。

然而，我对所有这些多线程的东西都是新手，我希望有一个比我更有经验的人告诉我，这个方法是否能在一个内存模型薄弱的系统上工作。

Answer 1

对于push()和pop()中的条件返回类型，使用不同类型的无限循环(如for(;;) )可能会更清楚。这将更好地将注意力集中在循环内容上。

您也可以为单行语句使用大括号。其中一些行相当长，因此执行的代码处于末尾。这也应该有助于区分他们和繁忙的等待。