如何减少SPSC队列pop函数中的缓存缺失？

问题描述 #

我正在优化C++中的单生产者单消费者（SPSC）队列，以下是我的实现示例代码：

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

struct Queue {
    alignas(64) std::atomic<size_t> write_index; // Align to cache line size
    alignas(64) std::atomic<size_t> read_index;  // Align to cache line size
    int* data;
    size_t capacity;

    Queue(size_t cap) : write_index(0), read_index(0), capacity(cap + 1) {
        data = new int[capacity];
    }

    ~Queue() {
        delete[] data;
    }

    bool push(int val) {
        size_t write_index_local = write_index.load(std::memory_order_relaxed);
        size_t next_write_index = (write_index_local + 1) % capacity;

        if (next_write_index == read_index.load(std::memory_order_acquire)) {
            // Queue is full
            return false;
        }

        data[write_index_local] = val;
        write_index.store(next_write_index, std::memory_order_release);
        return true;
    }

    bool pop(int& val) {

        const size_t write_index_local = write_index.load(std::memory_order_acquire);
        const size_t read_index_local = read_index.load(std::memory_order_relaxed);

        if (read_index_local == write_index_local) {
            // Queue is empty
            return false;
        }


        val = data[read_index_local];
        size_t next_read_index = (read_index_local + 1) % capacity;
        read_index.store(next_read_index, std::memory_order_release);

        return true;
    }
};

void producer(Queue& queue) {
    for (int i = 0; i < 20; ++i) {
        while (!queue.push(i)) {
            // Busy-wait until there is space in the queue
        }
        std::cout << "Pushed: " << i << std::endl;
    }
}

void consumer(Queue& queue) {
    for (int i = 0; i < 20; ++i) {
        int val;
        while (!queue.pop(val)) {
            // Busy-wait until there is an item to pop
        }
        std::cout << "Popped: " << val << std::endl;
    }
}

int main() {
    Queue queue(10); // Create a queue with capacity 10 (internally 11)

    std::thread producer_thread(producer, std::ref(queue));
    std::thread consumer_thread(consumer, std::ref(queue));

    producer_thread.join();
    consumer_thread.join();

    return 0;
}

我发现在pop函数中存在两处缓存缺失：获取write_index时，以及获取read_index对应的数据时。我希望对此进行优化以减少缓存缺失次数。每当write_index更新时，pop函数需要同时获取write_index和对应数据，理想情况下希望能同时获取这两者，避免获取数据时的缓存缺失。我考虑过使用预取技术，但不确定该场景下的最佳实现方式。请问能否同时获取write_index和read_index对应的数据？是否有特定的数据布局策略或内存序考量有助于减少该场景下的缓存缺失？

优化方案 #

1. 数据布局优化：强化缓存对齐与预取友好性 #

你的当前布局已经通过alignas(64)规避了write_index和read_index之间的伪共享，可进一步优化数据数组的对齐：

• 将data数组的起始地址对齐到缓存行（64字节），确保每个数据块完整占据缓存行，避免跨缓存行的元素访问带来的额外缓存缺失。
• 若队列元素较小（如int），可按缓存行大小划分数据块，让CPU预取时一次性加载多个元素，提升后续访问效率。

修改后的队列构造/析构函数示例：

Queue(size_t cap) : write_index(0), read_index(0), capacity(cap + 1) {
    // 分配对齐到64字节的内存
    void* ptr = operator new[](capacity * sizeof(int), std::align_val_t(64));
    data = static_cast<int*>(ptr);
}

~Queue() {
    operator delete[](data, std::align_val_t(64));
}

2. 主动预取：消除数据访问的缓存延迟 #

无法做到原子性同时获取write_index和对应数据，但可以通过CPU预取指令，在确认队列非空后立即启动数据加载，让数据在真正读取前进入缓存：

• 使用_mm_prefetch指令（需包含<xmmintrin.h>），选择_MM_HINT_T0告知CPU这是即将使用的数据，优先加载到L1缓存。
• 可在空队列的busy-wait循环中提前预取下一个可能访问的位置，进一步减少等待后的缓存缺失。

优化后的pop函数：

#include <xmmintrin.h>

bool pop(int& val) {
    const size_t write_index_local = write_index.load(std::memory_order_acquire);
    const size_t read_index_local = read_index.load(std::memory_order_relaxed);

    if (read_index_local == write_index_local) {
        return false;
    }

    // 预取目标数据到缓存
    _mm_prefetch(reinterpret_cast<const char*>(&data[read_index_local]), _MM_HINT_T0);

    val = data[read_index_local];
    size_t next_read_index = (read_index_local + 1) % capacity;
    read_index.store(next_read_index, std::memory_order_release);

    return true;
}

优化后的consumer busy-wait逻辑：

void consumer(Queue& queue) {
    for (int i = 0; i < 20; ++i) {
        int val;
        while (!queue.pop(val)) {
            // 提前预取下一个可能读取的位置
            size_t read_idx = queue.read_index.load(std::memory_order_relaxed);
            _mm_prefetch(reinterpret_cast<const char*>(&queue.data[read_idx]), _MM_HINT_T0);
            // 加入短暂暂停减少空转CPU占用
            _mm_pause();
        }
        std::cout << "Popped: " << val << std::endl;
    }
}

3. 内存序与操作顺序优化 #

当前内存序的选择是合理的，但可微调操作顺序减少无效预取：

• 先加载read_index，再加载write_index，若检测到队列空则直接返回，避免不必要的数据预取。
• 保持write_index.load(std::memory_order_acquire)确保能看到生产者写入的最新数据，read_index的读写用memory_order_relaxed即可（仅消费者自身修改）。

4. 批量操作：减少索引访问与缓存缺失次数 #

若业务允许，实现批量push/pop操作：

• 生产者一次写入多个元素后再更新write_index，消费者一次读取多个元素后再更新read_index。
• 连续的内存访问会触发CPU自动预取，同时减少原子索引的访问次数，降低缓存一致性开销。

批量pop示例：

size_t pop_batch(int* buf, size_t max_count) {
    size_t write_idx = write_index.load(std::memory_order_acquire);
    size_t read_idx = read_index.load(std::memory_order_relaxed);
    size_t available = write_idx >= read_idx ? write_idx - read_idx : write_idx + capacity - read_idx;
    size_t take = std::min(available, max_count);

    if (take == 0) return 0;

    // 预取所有待读取的数据块
    for (size_t i = 0; i < take; ++i) {
        size_t pos = (read_idx + i) % capacity;
        _mm_prefetch(reinterpret_cast<const char*>(&data[pos]), _MM_HINT_T0);
    }

    // 批量拷贝数据
    for (size_t i = 0; i < take; ++i) {
        size_t pos = (read_idx + i) % capacity;
        buf[i] = data[pos];
    }

    read_index.store((read_idx + take) % capacity, std::memory_order_release);
    return take;
}

内容的提问来源于stack exchange，提问作者user26135499