无锁非循环MPSC队列的最小内存序与编译器重排技术问询

场景说明 #

多个线程更新缓冲区中的bucket，单个消费者线程处理这些bucket，直至达到MAX_CAPACITY。

struct Bucket {
    std::atomic<int> lock{0};
    Data data;
};

std::atomic<int64_t> readIndex{0};
std::atomic<int64_t> writeIndex{0}; // 预期始终满足 writeIndex >= readIndex
std::array<Bucket, MAX_CAPACITY> buffer;

// 由多个更新线程调用
bool updateData(/* 参数 */) {
    while (true) {
        int64_t myWriteIndex = writeIndex.load(std::memory_order_relaxed);
        if (myWriteIndex == MAX_CAPACITY) {
            return false; // 缓冲区已满
        }
        Bucket& bucket = buffer[myWriteIndex];
        int expected = 0;
        if (bucket.lock.compare_exchange_strong(expected, 1, std::memory_order_relaxed)) { // 确保processData只有在成功获取CAS的写入者（尤其是更新bucket.data的那个）释放bucket.lock后，才能读取该bucket
            // 多个线程可能进入此处
            if (writeIndex.compare_exchange_strong(myWriteIndex, myWriteIndex + 1, std::memory_order_relaxed)) {
                // 只有一个线程能成功递增writeIndex
                // 更新bucket.data 
                bucket.lock.store(0, std::memory_order_release); // Sync1
                break;
            }
            bucket.lock.store(0, std::memory_order_release);
        }
    }
    return true;
}

// 仅由单个消费者线程调用
processData() {
    while (true) {
        int64_t myReadIndex = readIndex.load(std::memory_order_relaxed);
        if (myReadIndex == writeIndex.load(std::memory_order_relaxed)) {
            return false; // 没有更多bucket需要处理
        }
        ...
        Bucket& bucket = buffer[myReadIndex];
        if (bucket.lock.load(std::memory_order_acquire) != 0) { // Sync1；写入者在尝试更新时会锁定buffer
            continue;
        }
        // 处理bucket.data，随后销毁它
        readIndex.fetch_add(1, std::memory_order_relaxed);
        return true;
   }
}

核心问题1 #

在updateData函数中，是否真的需要为writeIndex使用acquire/release内存序？原作者认为不需要，理由如下：

1. bucket.lock已在updateData与processData之间建立acquire/release内存序，足以保证Bucket::data的更新对消费者可见；
2. 即使使用std::memory_order_relaxed，bucket.lock.compare_exchange_strong()也能确保所有更新线程看到bucket.lock的最新值；
3. updateData中的条件语句重排会违反“as-if”规则，因此编译器不会进行此类重排。恳请给出专业意见。

专业解答 #

原作者的结论正确，无需为writeIndex的操作使用acquire/release内存序，理由如下：

• 可见性保障已由bucket.lock覆盖：消费者仅在通过bucket.lock.load(acquire)读到0时才处理bucket.data，而写入者完成data更新后会以release语义存储0到lock。根据C++内存模型，release与后续acquire操作同步，已确保data修改对消费者可见，writeIndex的relaxed操作不影响这一核心同步逻辑。
• 更新线程间的lock一致性无需额外内存序：bucket.lock.compare_exchange_strong是原子操作，即使使用relaxed语义，硬件缓存一致性协议（如x86-64的MESI）会保证所有线程看到的lock值全局一致，不会出现过期值。
• 编译器重排受as-if规则约束：updateData的逻辑依赖“先加载writeIndex，再操作对应bucket的lock”，若编译器重排这些步骤，会导致线程错误操作非当前writeIndex的bucket，直接违反as-if规则（程序行为需与顺序执行一致），因此编译器不会做此类重排。

后续优化代码 #

借助ScopeLock类可更简洁地改写updateData：

struct ScopeLock {
    ScopeLock(std::atomic<int>& lock_) : lock(lock_) {}
    bool tryLock() {
        int expected = 0; // 原代码此处expected设为1错误，修正为0以正确尝试锁
        locked = lock.compare_exchange_strong(expected, 1, std::memory_order_relaxed);
        return locked;
    } 
    ~ScopeLock() {
        if (locked) {
            lock.store(0, std::memory_order_release); // Sync1
        }
    }
    bool locked;
    std::atomic<int>& lock;
};

bool updateData(/* 参数 */) {
    while (true) {
        int64_t myWriteIndex = writeIndex.load(std::memory_order_relaxed); 
        if (myWriteIndex == MAX_CAPACITY) {
            return false; // 缓冲区已满
        }
        Bucket& bucket = buffer[myWriteIndex];
        ScopeLock sl(bucket.lock);
        if (sl.tryLock() && 
            writeIndex.compare_exchange_strong(myWriteIndex, myWriteIndex + 1, std::memory_order_relaxed)) { // 短路求值保证编译时不会重排&&两侧操作
                // 更新bucket.data 
                break;
            }
        }
    }
    return true;
}

注：原ScopeLock的tryLock函数中，expected初始值错误设为1，修正为0才能正确尝试获取锁（预期lock为0时改为1）。

核心问题2 #

针对以下场景：

• 读者发现readIndex = writeIndex = n，进入循环；
• 一个writer为bucket n锁定Bucket::lock（使用relaxed语义的CAS），并成功自增writeIndex；
• 当writer仍在修改Bucket::data（即ScopeLock析构函数未执行）时，读者观测到writeIndex变为n+1，进而加载Bucket::lock（使用acquire内存序）。

在Bucket::lock的acquire/release内存序及读者已观测到writeIndex自增到n+1的前提下，C++标准或缓存一致性架构是否允许读者读到buffer[n].lock的值为0？

专业解答 #

不允许读者读到buffer[n].lock的值为0，具体分析如下：

1. C++内存模型层面：
   • 写入者线程的操作顺序是：CAS锁定bucket[n].lock（设为1）→ 成功递增writeIndex → 修改bucket.data → 析构时以release语义存储0到lock。
   • 读者观测到writeIndex=n+1，说明写入者的writeIndex.compare_exchange_strong已完成。根据程序顺序，写入者的CAS锁定操作先于递增writeIndex，因此lock=1的结果必须对后续操作可见。此时写入者尚未释放lock（仍在修改data），lock的值必然是1。
2. 缓存一致性架构层面（以x86-64为例）：
   • x86-64是强内存模型，所有存储默认带release语义，加载带acquire语义，且不会出现存储重排到加载之前的情况。写入者的CAS锁定lock（存储1）会先于递增writeIndex的存储完成，读者看到writeIndex=n+1时，必然能看到lock=1的结果，不可能读到0。

补充说明 #

代码面向x86-64平台，但上述结论基于C++标准定义的最小内存序规则，不依赖x86-64的强内存模型特性。相关CAS同步规则参考：std::atomic::compare_exchange_weak的同步语义——CAS操作成功执行时，操作前的所有内存操作对后续通过CAS同步的线程可见。

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