避免 SPSC 队列索引的错误共享

Question

让我们想象一个无锁并发 SPSC（单生产者/单消费者）队列。

• 生产者线程 读取 head、tail、cached_tail 和写入 head、cached_tail。
• 消费者线程 读取 head、tail、cached_head 和写入 tail、cached head。

注意，cached_tail 只能被生产者线程访问，就像 cached_head 只能被消费者线程访问一样。它们可以被认为是私有线程局部变量，因此它们是不同步的，因此没有定义为原子的。

队列的数据布局如下：

#include <atomic>
#include <cstddef>
#include <thread>

struct spsc_queue
{
    /// ...

    // Producer variables
    alignas(std::hardware_destructive_interference_size) std::atomic<size_t> head; // shared
    size_t cached_tail; // non-shared

    // Consumer variables
    alignas(std::hardware_destructive_interference_size) std::atomic<size_t> tail; // shared
    size_t cached_head; // non-shared

    std::byte padding[std::hardware_destructive_interference_size - sizeof(tail) - sizeof(cached_head)];
};

由于我想避免错误共享，我将head 和tail 与L1 缓存行大小对齐。

push/pop 操作的伪代码实现如下：

bool push(const void* elems, size_t n)
{
    size_t h = atomic_load(head, relaxed);

    if (num_remaining_storage(h, cached_tail) < n)
    {
        cached_tail = atomic_load(tail, acquire);

        if (num_remaining_storage(h, cached_tail) < n)
            return false;
    }

    // write from elems

    atomic_store(head, h + n, release);
    return true;
}

bool pop(void* elems, size_t n)
{
    size_t t = atomic_load(tail, relaxed);

    if (num_stored_elements(cached_head, t) < n)
    {
        cached_head = atomic_load(head, acquire);

        if (num_stored_elements(cached_head, t) < n)
            return false;
    }

    // read to elems

    atomic_store(tail, t + n, release);
    return true;
}

void wait_and_push(const void* elems, size_t n)
{
    size_t h = atomic_load(head, relaxed);

    while (num_remaining_storage(h, cached_tail) < n)
        cached_tail = atomic_load(tail, acquire);

    // write from elems

    atomic_store(head, h + n, release);
}

void wait_and_pop(void* elems, size_t n)
{
    size_t t = atomic_load(tail, relaxed);

    while (num_stored_elements(cached_head, t) < n)
        cached_head = atomic_load(head, acquire);

    // write to elems

    atomic_store(tail, t + n, release);
}

在初始化时（此处未列出），所有索引都设置为0。 函数 num_remaining_storage 和 num_stored_elements 是 const 函数，它们根据传递的参数和不可变的队列容量执行简单的计算 - 它们不执行任何原子读取或写入。

现在的问题是：我是否还需要对齐 cached_tail 和 cached_head 以完全避免错误共享任何索引，或者它是可以的。由于cached_tail 是生产者私有的，cached_head 是消费者私有的，我认为cached_tail 可以与head（生产者缓存行）在同一缓存行中，就像cached_head 与@ 在同一缓存行中一样987654350@（消费者缓存行）不会发生错误共享。

我错过了什么吗？

Answer 1

感谢您提供伪代码 - 它仍然缺少一些细节，但我想我明白了基本概念。您有一个有界 SPSC 队列，其中索引可以环绕，并且您使用 push 中的 cached_tail 变量来检查是否有空闲插槽，因此您可以避免从可能无效的缓存行加载 tail（反之亦然pop 反之亦然。

我建议将head 和cached_tail 放在一起（即在同一缓存行上），将tail 和cached_head 放在不同的缓存行上。 push always 读取 both 变量 - head 和 cached_tail，因此将它们靠近在一起是有意义的。 cached_tail 仅在没有更多空闲插槽时更新，我们必须重新加载 tail。

您的代码在细节上有点薄，但似乎还有一些优化空间：

bool push(const void* elems, size_t n)
{
    size_t h = atomic_load(head);

    if (num_remaining_storage(h, cached_tail) < n)
    {
        auto t = atomic_load(tail);
        if (t == cached_tail)
          return false;

       // we only have to update our cached_tail if the reloaded value
       // is different - and in this case it is guaranteed that there
       // is a free slot, so we don't have to perform a recheck.
        cached_tail = t;
    }

    // write from elems

    atomic_store(head, h + n);
    return true;
}

这样cached_tail 只有在head 也更新时才会更新，所以这是它们位于同一缓存行的另一个原因。当然，同样的优化也可以应用于pop。

这正是我想看一些代码的原因，因为访问模式对于确定哪些变量应该共享一个缓存行和哪些不应该是至关重要的。