环形缓冲区固定索引写入比递增索引快3倍的性能原因排查

我正在学习环形缓冲区相关知识，基于ringbuffer-spsc crate尝试改造为MPSC队列，但遇到了无法理解的性能差异。

MPMRRingBuffer结构体 #

pub struct MPMRRingBuffer<T: Clone, const N: usize> {
    buffer: UnsafeCell<[MaybeUninit<T>; N]>,
    idx_w: CachePadded<AtomicUsize>,
}


unsafe impl<T: Clone, const N: usize> Send for MPMRRingBuffer<T, N> {}
unsafe impl<T: Clone, const N: usize> Sync for MPMRRingBuffer<T, N> {}

impl<T: Clone, const N: usize> MPMRRingBuffer<T, N> {
    pub fn init() -> (MultipleRingBufferWriter<T, N>, MultipleRingBufferReader<T, N>) {
        assert!(
            N.is_power_of_two(),
            "RingBuffer requires the capacity to be a power of 2. {N} is not."
        );
        let rb = Arc::new(MPMRRingBuffer {
            buffer: UnsafeCell::new(array_init::array_init(|_| MaybeUninit::uninit())),
            idx_w: CachePadded::new(AtomicUsize::new(0)),
        });
        (
            MultipleRingBufferWriter {
                inner: rb.clone(),
            },
            MultipleRingBufferReader {
                inner: rb,
            },
        );
    }
}


pub struct MultipleRingBufferWriter<T: Clone, const N: usize> {
    inner: Arc<MPMRRingBuffer<T, N>>,
}

impl<T: Clone, const N: usize> MultipleRingBufferWriter<T, N> {

#[inline]
    pub fn push(&self, t: T) {
        // Let's increment the counter and let it grow indefinitely and potentially overflow resetting it to 0.
        let previous_write_idx = self.inner.idx_w.fetch_add(1, Ordering::Relaxed);

        // Insert the element in the ring buffer
        unsafe {
            let item = &mut (*self.inner.buffer.get())[previous_write_idx & (N - 1)];
            let _ = mem::replace(item, MaybeUninit::new(t));
        };
    }

    #[inline]
    pub fn push2(&self, t: T) {
        // Let's increment the counter and let it grow indefinitely and potentially overflow resetting it to 0.
        let _ = self.inner.idx_w.fetch_add(1, Ordering::Relaxed);

        // Insert the element in the ring buffer
        unsafe {
            let item = &mut (*self.inner.buffer.get())[1];
            let _ = mem::replace(item, MaybeUninit::new(t));
        };
    }
}

pub struct MultipleRingBufferReader<T: Clone, const N: usize> {
    inner: Arc<MPMRRingBuffer<T, N>>,
}
impl<T: Clone, const N: usize> MultipleRingBufferReader<T, N> {
    #[inline]
    pub fn peek_newest_write(&self) -> Option<(T, usize)> {
        // Check if the ring buffer is potentially empty
        let idx = self.inner.idx_w.load(Ordering::Relaxed);
        if idx == 0 {
            // No items yet
            return None;
        }

        let idx = idx - 1;
        let t = unsafe {
            self.inner.get_mut(idx).assume_init_ref().clone()
        };

        Some((t, idx))
    }

    #[inline]
    pub fn peek_newest_write2(&self) -> Option<(T, usize)> {
        // Check if the ring buffer is potentially empty
        let idx = self.inner.idx_w.load(Ordering::Relaxed);
        if idx == 0 {
            // No items yet
            return None;
        }

        let t = unsafe {
            self.inner.get_mut(1).assume_init_ref().clone()
        };

        Some((t, idx))
    }
}

测试场景为6个线程向容量2048的缓冲区写入500000个i32，push方法比push2慢3-4倍。在我的MacBook Pro M1上，push耗时31.189ms，push2耗时8.9905ms。

请问这是分支预测问题、写入idx+1时的缓存失效，还是其他原因？

注：我知道push2并非正确的环形缓冲区实现，仅用于对比性能。

基准测试代码（使用criterion） #

const NUM_MESSAGES: usize = 524_281;
const BUFFER_SIZE: usize = 2048;
const WRITER_THREADS: usize = 6;

fn benchmark_push2(c: &mut Criterion) {
    c.bench_function("push2_channel", |b| {
        b.iter(|| {
            let (sender, receiver) = MPMRRingBuffer::<i32, BUFFER_SIZE>::init();
            let mut handles = vec!();


            for i in 0..WRITER_THREADS {
                let arc_sender = sender.clone();
                let handle = thread::spawn(move || {
                    for j in 0..NUM_MESSAGES / WRITER_THREADS {
                        arc_sender.push2((i * j) as i32); // Send a message
                    }
                });

                handles.push(handle);
            }

            handles.into_iter().for_each(|f| f.join().unwrap()); // Wait for the thread to finish
            let a = receiver.peek_newest_write2();
            assert_eq!(a.unwrap().1, 524279);
            assert_ne!(a.unwrap().0, i32::MAX) // make sure value is not optimized away
        });
    });
}

fn benchmark_push(c: &mut Criterion) {
    c.bench_function("push_channel", |b| {
        b.iter(|| {
            let (sender, receiver) = MPMRRingBuffer::<i32, BUFFER_SIZE>::init();
            let mut handles = vec!();


            for i in 0..WRITER_THREADS {
                let arc_sender = sender.clone();
                let handle = thread::spawn(move || {
                    for j in 0..NUM_MESSAGES / WRITER_THREADS {
                        arc_sender.push((i * j) as i32); // Send a message
                    }
                });

                handles.push(handle);
            }

            handles.into_iter().for_each(|f| f.join().unwrap()); // Wait for the thread to finish
            let a = receiver.peek_newest_write();
            assert_eq!(a.unwrap().1, 524279);
            assert_ne!(a.unwrap().0, i32::MAX) // make sure value is not optimized away
        });
    });
}

性能差异的核心原因是多线程下的缓存行竞争（伪共享）以及内存写冲突，和分支预测无关，具体拆解如下：

1. push2的内存访问特性
   push2始终写入数组的第1个位置，多线程场景下所有线程都修改同一个内存地址。对于M1的ARM架构CPU，当多个核心同时向同一缓存行执行写操作时，缓存一致性协议会快速处理这种情况——虽然每个线程的写操作会覆盖前一个，但无需在不同缓存行之间切换，减少了缓存失效次数，甚至部分操作可被硬件合并，因此开销极低。

2. push的内存访问特性
   push根据计算出的索引循环写入环形缓冲区的不同位置。M1的缓存行是128字节，一个i32占4字节，单个缓存行可容纳32个i32元素。6个线程同时写入不同元素时，大概率会命中同一缓存行，触发伪共享：修改缓存行内的不同元素会导致整个缓存行在核心间强制同步，引发频繁的缓存失效和总线传输，大幅增加延迟。同时，环形写入的分散访问模式会让CPU预取器难以预测，进一步降低内存访问效率。

3. 原子操作的影响
   两个方法都调用了fetch_add，这部分开销一致，性能差异的根源不在原子操作本身，而在于后续的内存写入模式。

总结：push2虽为错误实现，但因所有线程写入同一内存地址，避免了伪共享和分散内存访问，性能反而更高；push的正确环形写入模式在多线程下触发了大量缓存行同步开销，导致性能下降。

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