固定大小无锁环形缓冲区可靠实现及Push双更新问题求解

你的代码中，推进tail后再写入数据的逻辑确实存在风险：当其他线程完成多次Push/Pop操作后，同一个数组位置会被复用，导致未完成写入的数据被覆盖。结合TaggedHandle解决ABA问题的同时，可通过以下两种方案解决双更新问题：

方案一：基于位置Tag标记的标准实现 #

给每个缓冲区位置增加一个原子Tag变量，标记该位置当前有效的数据版本，确保读取线程只读取已完成写入的对应版本数据。

核心逻辑 #

1. 预留位置：通过CAS推进tail（标记下一个可预留的位置），确保当前线程独占该位置的写入权；
2. 写入数据：安全写入缓冲区对应位置；
3. 标记完成：更新该位置的Tag为当前tail的Tag，通知读取线程数据已就绪；
4. 读取逻辑：读取线程需等待位置Tag与head的Tag匹配后，再读取数据并推进head。

代码实现 #

定义TaggedHandle

struct TaggedHandle {
    size_t index;
    size_t tag;
    
    TaggedHandle next() const {
        size_t next_idx = index + 1;
        size_t next_tag = tag;
        if (next_idx == BUFFER_SIZE) {
            next_idx = 0;
            next_tag += 1;
        }
        return {next_idx, next_tag};
    }
    
    bool operator==(const TaggedHandle& other) const {
        return index == other.index && tag == other.tag;
    }
};

缓冲区成员

constexpr size_t BUFFER_SIZE = 1024;
std::atomic<TaggedHandle> tail_{0, 0};
std::atomic<TaggedHandle> head_{0, 0};
std::array<T, BUFFER_SIZE> buffer_;
std::array<std::atomic<size_t>, BUFFER_SIZE> tag_;

// 初始化时给每个位置设置初始Tag
RingBuffer() {
    for (size_t i = 0; i < BUFFER_SIZE; ++i) {
        tag_[i].store(0, std::memory_order_relaxed);
    }
}

Push函数

bool push(const T& item) {
    TaggedHandle tail, head, next_tail;
    do {
        tail = tail_.load(std::memory_order_acquire);
        head = head_.load(std::memory_order_acquire);
        next_tail = tail.next();
        
        // 缓冲区满：下一个预留位置与head完全重合
        if (next_tail == head) {
            return false;
        }
    } while (!tail_.compare_exchange_weak(tail, next_tail, std::memory_order_release));
    
    // 安全写入数据（其他线程无法再预留该Tag版本的index）
    buffer_[tail.index] = item;
    // 标记该位置数据已写入完成
    tag_[tail.index].store(tail.tag, std::memory_order_release);
    
    return true;
}

Pop函数（配套实现）

bool pop(T& item) {
    TaggedHandle head, tail;
    do {
        head = head_.load(std::memory_order_acquire);
        tail = tail_.load(std::memory_order_acquire);
        
        // 缓冲区空
        if (head == tail) {
            return false;
        }
        
        // 等待数据写入完成：位置Tag需与head的Tag匹配
        if (tag_[head.index].load(std::memory_order_acquire) != head.tag) {
            continue;
        }
        
        // 读取数据
        item = buffer_[head.index];
        
        // CAS推进head，完成读取
        if (head_.compare_exchange_strong(head, head.next(), std::memory_order_release)) {
            return true;
        }
    } while (true);
}

方案二：复用TaggedHandle的状态标记（无额外内存开销） #

利用TaggedHandle的Tag字段的奇偶性标记位置状态：偶数Tag表示正常可用/已完成，奇数Tag表示已预留待写入。

Push函数示例 #

bool push(const T& item) {
    TaggedHandle tail, head, next_tail, reserved_tail;
    do {
        tail = tail_.load(std::memory_order_acquire);
        head = head_.load(std::memory_order_acquire);
        next_tail = tail.next();
        
        if (next_tail == head) {
            return false;
        }
        
        // 标记当前位置为预留状态（Tag+1变为奇数）
        reserved_tail = {tail.index, tail.tag + 1};
        if (!tail_.compare_exchange_weak(tail, reserved_tail, std::memory_order_acquire)) {
            continue;
        }
        
        // 写入数据
        buffer_[tail.index] = item;
        
        // 推进tail到下一个位置，恢复Tag为偶数
        if (!tail_.compare_exchange_strong(reserved_tail, next_tail, std::memory_order_release)) {
            // 理论上不会失败，仅做异常处理
            std::abort();
        }
        return true;
    } while (true);
}

关键原理 #

• TaggedHandle解决ABA：同一index的不同Tag代表不同版本，避免CAS时的ABA问题；
• 预留-写入-标记的流程：确保每个位置的写入权被当前线程独占，读取线程仅读取已就绪的数据；
• 内存屏障：使用memory_order_acquire/release保证操作的可见性和顺序性，避免指令重排导致的错误。

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