C++ 等待无关（Wait-free）算法：在大规模并行环境下的确定性延迟优化

各位同仁、技术爱好者们，大家好！

今天，我们将深入探讨一个在高性能、大规模并行计算领域至关重要，却又极具挑战性的话题：C++ 中的等待无关（Wait-free）算法。在追求极致性能和可预测性的现代系统中，尤其是在高频交易、实时控制、大规模数据处理等场景下，确定性延迟（Deterministic Latency）成为了衡量系统质量的关键指标。传统的基于锁的并发控制机制，尽管易于理解和实现，却常常成为性能瓶颈和延迟不确定性的根源。而等待无关算法，正是为了解决这些问题而生。

我将以一名编程专家的视角，为大家剖析等待无关算法的核心理念、实现技术、应用场景及其复杂性。我们将通过大量的代码示例，深入理解其工作原理，并探讨在实际项目中应用这类算法时所需面对的挑战和权衡。

一、并行计算的困境：为什么我们需要等待无关？

在多核处理器日益普及的今天，并行计算已成为提升应用性能的必由之路。然而，并行并非没有代价。当多个线程或进程需要访问和修改共享数据时，我们必须引入并发控制机制来维护数据的一致性。

最常见的并发控制机制是基于锁的（Lock-based）方法，例如互斥锁（std::mutex）、读写锁（std::shared_mutex）等。它们通过阻止多个线程同时访问临界区来确保数据安全。这种方法直观易懂，但在大规模并行环境下，却暴露出诸多弊端：

1. 非确定性延迟（Non-deterministic Latency）：线程在获取锁时可能被阻塞，等待其他线程释放锁。这种等待时间是不可预测的，取决于操作系统的调度、其他线程的执行速度以及锁的争用程度。在高并发场景下，一个线程可能被无限期地暂停，导致所谓的“饥饿”（Starvation）。
2. 死锁（Deadlock）：当多个线程互相等待对方释放资源时，就会发生死锁，导致系统完全停滞。
3. 活锁（Livelock）：线程不断地尝试获取资源，但由于同步冲突，它们总是失败，并不断地重试，导致所有线程都在忙碌地做无用功。
4. 优先级反转（Priority Inversion）：低优先级的线程持有高优先级线程所需的锁，导致高优先级线程被阻塞，这在实时系统中是灾难性的。
5. 上下文切换开销（Context Switching Overhead）：线程阻塞和唤醒涉及到操作系统调度器的介入，产生不小的上下文切换开销。
6. 缓存失效（Cache Invalidation）：锁操作可能导致缓存行在不同CPU核心之间频繁迁移，增加内存访问延迟。

所有这些问题都指向一个核心痛点：基于锁的算法无法提供可预测的、有界限的延迟。在某些对延迟敏感的系统中，哪怕是微秒级的抖动都可能造成严重后果。这就是等待无关算法的用武之地。

二、并发算法的层级：从阻塞到等待无关

为了更好地理解等待无关算法的独特之处，我们首先需要了解并发算法的不同层级。Herlihy 和 Shavit 在其经典著作中，根据算法的进步保证（Progress Guarantee）将并发算法分为以下几类：

算法类型	进步保证	饥饿（Starvation）	死锁（Deadlock）	活锁（Livelock）	典型实现	复杂度
阻塞（Blocking）	整体系统可能停滞，单个线程可能无限期阻塞。	可能	可能	可能	互斥锁（std::mutex）、信号量（std::semaphore）	较低
无阻塞（Non-blocking）	系统整体保证进步，但单个线程仍可能饥饿。	可能	不可能	不可能
– 无障碍（Obstruction-free）	如果没有其他线程干扰，当前线程保证进步。	可能	不可能	不可能	简单的CAS循环，不处理外部干扰	中等
– 无锁（Lock-free）	至少有一个线程在有限步内完成操作，系统整体保证进步。	可能	不可能	不可能	CAS循环，ABA问题处理，内存回收机制	较高
– 等待无关（Wait-free）	所有线程在有限步内完成操作，每个线程都保证进步。	不可能	不可能	不可能	复杂原子操作、通用构造，需要谨慎设计	极高

让我们逐一深入了解：

2.1 阻塞算法（Blocking Algorithms）

这是最常见的并发控制方法。当一个线程需要访问共享资源时，它尝试获取一个锁。如果锁已被其他线程持有，当前线程就会被阻塞，并被操作系统调度器挂起，直到锁被释放。

优点：

• 概念简单，易于理解和实现。
• 在低争用场景下，性能可能很好。

缺点：

• 上述提到的所有问题：非确定性延迟、死锁、活锁、饥饿、优先级反转、上下文切换开销。

2.2 无阻塞算法（Non-blocking Algorithms）

无阻塞算法的显著特点是，没有任何线程的失败或暂停会导致其他线程被阻塞。它消除了死锁和活锁的可能性，并且通常能减少上下文切换。

2.2.1 无障碍（Obstruction-free）

这是最弱的无阻塞保证。它保证如果一个线程独立运行（即没有其他线程干扰），它能够在有限步内完成其操作。然而，如果有其他线程同时操作，当前线程可能会无限期地重试，导致饥饿。它的优点是实现相对简单，通常是无锁算法的起点。

2.2.2 无锁（Lock-free）

无锁算法保证系统整体的进步。这意味着在任何时候，至少有一个线程能够在有限步内完成其操作。即使某些线程被操作系统暂停或出现故障，其他线程也能继续执行。无锁算法消除了死锁和活锁，但仍然不能保证单个线程的进步，即单个线程仍然可能饥饿。

2.2.3 等待无关（Wait-free）

等待无关是所有并发算法中最强的进步保证。它保证所有线程在有限步内完成其操作，无论其他线程的行为或速度如何。这意味着：

• 无死锁：所有线程都能继续执行。
• 无活锁：所有线程都能继续执行。
• 无饥饿：每个线程都保证能在有限步内完成操作，因此不会被无限期地推迟。
• 确定性延迟：由于每个操作的步数是有限且可预测的，因此可以为单个操作提供确定性的最坏情况延迟。

等待无关算法是实现确定性延迟的终极目标，但其实现难度也是最高的。

三、等待无关算法的基石：C++ 原子操作

在 C++ 中实现无阻塞和等待无关算法，离不开 std::atomic 库提供的原子操作。这些操作能够以原子性（不可中断性）的方式对共享数据进行读写，从而避免数据竞争。

3.1 std::atomic 基础

std::atomic 模板类是 C++11 引入的，用于封装一个类型 T 的值，并提供原子性的操作。

#include <atomic>
#include <iostream>
#include <thread>
#include <vector>

std::atomic<int> counter(0); // 初始化为0的原子计数器

void increment_counter() {
    for (int i = 0; i < 100000; ++i) {
        counter++; // 原子递增操作，等价于 counter.fetch_add(1, std::memory_order_seq_cst);
    }
}

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 10; ++i) {
        threads.emplace_back(increment_counter);
    }

    for (auto& t : threads) {
        t.join();
    }

    std::cout << "Final counter value: " << counter.load() << std::endl; // 原子读取
    return 0;
}

在这个例子中，counter++ 是一个原子操作，它保证了即使多个线程同时尝试递增 counter，最终结果也是正确的，不会出现丢失更新的情况。

3.2 compare_exchange 系列操作

compare_exchange_weak 和 compare_exchange_strong 是实现复杂无锁/等待无关算法的核心。它们执行一个原子性的“比较并交换”（CAS）操作：如果当前原子变量的值等于预期值，则将其设置为新值。

• bool compare_exchange_weak(Expected& expected, Desired desired, ...):
  • expected: 引用，用于存储原子变量的当前值。如果比较成功，它保持不变；如果失败，它会被更新为原子变量的实际当前值。
  • desired: 要设置的新值。
  • 如果原子变量的当前值等于 expected，则将其值更新为 desired 并返回 true。
  • 如果原子变量的当前值不等于 expected，则不进行更新，并将 expected 更新为原子变量的实际当前值，返回 false。
  • _weak 版本可能在比较成功时返回 false（称为“伪失败”Spurious Failure），但在循环中通常不是问题，因为它允许在某些处理器上生成更高效的代码。
• bool compare_exchange_strong(Expected& expected, Desired desired, ...):
  • 与 _weak 版本类似，但保证只在比较失败时返回 false，不会出现伪失败。通常用于不希望有伪失败的场景，或者在非循环中使用。

示例：一个简单的无锁自旋锁 (使用 std::atomic_flag)

std::atomic_flag 是最简单的原子类型，它只能存储 true 或 false，并提供 test_and_set 和 clear 操作。它非常适合实现自旋锁。

#include <atomic>
#include <thread>
#include <vector>
#include <iostream>

class SpinLock {
public:
    void lock() {
        // 原子地将flag设置为true，并返回旧值。如果旧值为true，说明锁已被占用，继续自旋。
        while (flag.test_and_set(std::memory_order_acquire));
    }

    void unlock() {
        // 原子地将flag设置为false，释放锁。
        flag.clear(std::memory_order_release);
    }

private:
    std::atomic_flag flag = ATOMIC_FLAG_INIT; // 初始化为false
};

SpinLock my_spinlock;
int shared_data = 0;

void worker_with_spinlock() {
    for (int i = 0; i < 100000; ++i) {
        my_spinlock.lock();
        shared_data++;
        my_spinlock.unlock();
    }
}

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 10; ++i) {
        threads.emplace_back(worker_with_spinlock);
    }

    for (auto& t : threads) {
        t.join();
    }

    std::cout << "Final shared_data value: " << shared_data << std::endl;
    return 0;
}

这个自旋锁是“无障碍”的，因为它不涉及操作系统调度，但它仍然是“阻塞”的，因为一个线程在 test_and_set 失败时会忙等待。

3.3 内存顺序（Memory Order）

std::memory_order 是原子操作中一个非常关键且复杂的概念，它定义了原子操作与非原子操作以及其他原子操作之间的可见性规则。理解内存顺序对于编写正确且高效的无锁/等待无关算法至关重要。

• std::memory_order_relaxed: 最宽松的顺序，只保证原子性，不保证任何同步或顺序。
• std::memory_order_consume: 用于依赖链的消费操作，不常用。
• std::memory_order_acquire: 读操作，确保此操作之后的所有内存访问都在此操作之后发生。
• std::memory_order_release: 写操作，确保此操作之前的所有内存访问都在此操作之前发生。
• std::memory_order_acq_rel: 读-改-写操作，兼具 acquire 和 release 的语义。
• std::memory_order_seq_cst: 最严格的顺序，保证所有线程看到的所有 seq_cst 操作都以相同的全局顺序发生。这是默认的内存顺序，最安全但可能开销最大。

在上面的自旋锁示例中，std::memory_order_acquire 确保了在锁被获取后，当前线程能看到所有其他线程在释放锁前所做的修改。std::memory_order_release 确保了在锁被释放前，当前线程所做的所有修改对其他线程可见。

四、构建等待无关原语：挑战与技术

实现等待无关算法的复杂性远超普通无锁算法。它需要解决一系列棘手的问题。

4.1 ABA 问题

ABA 问题是无锁编程中的一个经典陷阱，特别是在使用 compare_exchange 操作时。

问题描述：
假设一个共享变量的值从 A 变为 B，然后又变回 A。一个线程读取了 A，然后被挂起。在它被挂起期间，另一个线程将变量从 A 变为 B，再从 B 变回 A。当第一个线程恢复执行时，它执行 compare_exchange 操作，发现变量的当前值仍然是 A，于是认为变量没有被其他线程修改过，从而成功地将变量从 A 替换为 C。然而，这个“成功”是基于一个错误的前提，即变量在期间没有发生任何改变，这可能导致数据结构损坏。

解决方案：
最常见的解决方案是使用带标签的指针（Tagged Pointers）。这通常通过将一个版本号或计数器与指针捆绑在一起实现。每次指针被修改时，版本号也随之递增。这样，即使指针值变回 A，版本号也会不同，compare_exchange 操作就能检测到中间的修改。

• 64位系统上的实现：在64位系统上，指针通常只使用低48位或56位地址空间，高位字节通常是空闲的。我们可以将一个小的版本号存储在指针的高位字节中（如果指针支持），或者使用一个单独的 std::atomic<std::pair<T*, unsigned int>> 结构来封装指针和版本号。
• 示例（概念性）：

template<typename T>
struct TaggedPointer {
    T* ptr;
    unsigned int tag;

    bool operator==(const TaggedPointer& other) const {
        return ptr == other.ptr && tag == other.tag;
    }
};

// ... 在实际使用时，std::atomic<TaggedPointer<Node>> head;
// CAS操作会比较ptr和tag两个字段。
// 每次修改head时，都会递增tag。

4.2 内存管理与安全回收

在无锁/等待无关算法中，当一个节点从数据结构中被移除时，不能立即释放其内存。因为其他线程可能仍然持有指向该节点的指针，并在对其进行操作。如果此时内存被释放并重用，就会导致悬空指针（Dangling Pointer）问题。

常见的内存回收机制：

• 哈扎德指针（Hazard Pointers）：

  • 每个线程维护一组哈扎德指针，用于指向它当前正在访问的共享节点。
  • 当一个线程要删除一个节点时，它将其放入一个待回收列表，而不是立即释放。
  • 定期地，一个线程会扫描待回收列表。对于列表中的每个节点，它检查所有其他线程的哈扎德指针，看是否有任何线程正在访问该节点。
  • 如果没有任何哈扎德指针指向该节点，则可以安全地回收其内存。
  • 这是一种复杂的机制，需要仔细实现。

• 读-复制-更新（Read-Copy-Update, RCU）：

  • RCU 是一种针对读多写少场景优化的并发机制。
  • 读操作可以无锁地进行，直接访问数据。
  • 写操作通过创建数据的副本，修改副本，然后原子地将指针指向新副本。
  • 旧副本不能立即删除，必须等待所有正在进行的读操作完成后才能回收。这通常通过“宽限期”（Grace Period）机制实现。

• 基于纪元（Epoch-based Reclamation, EBR）：

  • 类似于 RCU，线程进入或退出一个“纪元”。
  • 被删除的节点被标记为属于某个纪元，只有当所有线程都进入了比该纪元更新的纪元后，这些节点才能被安全回收。

• 引用计数（Reference Counting）：

  • 使用原子引用计数（std::shared_ptr）可以保证对象在没有引用时才被删除。
  • 然而，在高并发下，std::shared_ptr 的引用计数更新本身可能成为瓶颈，且不能完全解决循环引用问题。对于复杂的无锁数据结构，直接使用 std::shared_ptr 可能不切实际。

在许多等待无关的场景中，为了简化，经常使用预分配的固定大小内存池，或者假设对象永远不被删除，只被逻辑移除。

4.3 构建等待无关队列 (Bounded Wait-Free Queue)

现在，让我们通过一个具体的例子来理解等待无关算法的复杂性：一个有界的等待无关队列。我们将实现一个多生产者、多消费者（MPMC）的等待无关队列。

核心思想：

• 使用一个固定大小的数组作为底层存储。
• 维护两个原子索引：head（队头，指向下一个可读位置）和 tail（队尾，指向下一个可写位置）。
• 生产者通过原子递增 tail 索引来找到写入位置。
• 消费者通过原子递增 head 索引来找到读取位置。
• 为了实现等待无关，每个操作必须在有限步内完成，即使遇到争用。这意味着不能有忙等待循环（spin loop）等待其他线程释放资源，但可以有忙等待循环等待 compare_exchange 成功。

设计挑战：

1. 生产者写冲突：多个生产者可能同时尝试在同一个位置写入，或同时尝试更新 tail 索引。
2. 消费者读冲突：多个消费者可能同时尝试在同一个位置读取，或同时尝试更新 head 索引。
3. 读写协调：生产者不能写入已被占用的位置，消费者不能读取空位置。
4. 队列满/空判断：需要在原子操作中正确判断队列是否满或空。
5. ABA 问题：在更新 head 和 tail 时需要防止 ABA 问题（尽管对于简单的索引递增，通常不会出现，因为索引是单调递增的，不会回到旧值）。

为了简化并实现等待无关，我们通常会采用一种叫做数组槽状态的方法。每个槽位不仅存储数据，还存储一个状态或版本号，以协调生产者和消费者。

Wait-Free Bounded MPMC Queue 实现思路：

我们使用一个环形缓冲区。每个槽位 std::atomic<T> 存储实际数据。为了协调生产者和消费者，我们可以在每个槽位旁边存储一个版本号或一个状态，或者更巧妙地，将版本号编码在数据中（如果 T 允许），或者利用队列索引的特点。

一个经典的 Wait-Free Queue 算法是Lamport’s Bakery Algorithm的变体，或者更实用的是基于 Herlihy’s Wait-Free Universal Construction 思想的简化版。对于 MPMC 队列，通常会使用更复杂的原子操作和版本号。

这里我们实现一个基于 CAS 的 MPMC 队列，它通过确保每个线程都在有限步内尝试其 CAS 操作，并最终通过一个辅助数组来保证 Wait-Free 属性，虽然其复杂性非常高。

一个相对简单且能达到 Wait-Free 的 MPMC 队列实现，通常会使用一个辅助数组来帮助线程“预留”槽位，或者使用一种叫做“双数组”的方法。

让我们先从一个较为直观，但可能在极端情况下不是严格 Wait-Free 的 Lock-Free MPMC Queue 开始，然后讨论如何使其 Wait-Free。

#include <atomic>
#include <vector>
#include <thread>
#include <iostream>
#include <optional> // C++17 for optional return value

// 一个经典的 Lock-Free MPMC 队列，通常不是严格 Wait-Free
// 但能很好地说明原子操作的用法。
// 要实现严格的 Wait-Free MPMC 队列，通常需要更复杂的结构，
// 例如 Herlihy 的通用构造，或基于线程ID和辅助数组的特定设计。
// 这里展示的是一个常见的 Lock-Free MPMC 队列模式，
// 它的每个操作在有限步内完成，但单个线程可能饥饿。
// 我们可以通过额外的机制（例如帮助其他线程完成其操作）来使其 Wait-Free。

template <typename T>
class LockFreeQueue {
public:
    explicit LockFreeQueue(size_t capacity) :
        capacity_(capacity),
        buffer_(new Node[capacity]), // Note: Node contains std::atomic<T> for elements.
        head_(0),
        tail_(0)
    {}

    ~LockFreeQueue() {
        delete[] buffer_;
    }

    bool push(const T& value) {
        // 原子地获取当前 tail 索引
        size_t current_tail = tail_.load(std::memory_order_relaxed);
        size_t next_tail;

        while (true) {
            next_tail = current_tail + 1;
            if (next_tail == capacity_) {
                next_tail = 0; // Wrap around
            }

            // 检查队列是否已满
            if (next_tail == head_.load(std::memory_order_acquire)) { // Full
                return false;
            }

            // 尝试原子地更新 tail 索引
            if (tail_.compare_exchange_weak(current_tail, next_tail,
                                            std::memory_order_release,
                                            std::memory_order_relaxed)) {
                break; // CAS successful, we've claimed the slot
            }
            // CAS failed, another producer updated tail. Retry with new current_tail.
        }

        buffer_[current_tail].data = value; // Write data to the claimed slot
        // Update a "version" or "state" associated with the slot to signal it's ready for consumption.
        // For simplicity, we assume `data` itself being written is enough, or we use a separate flag.
        // In a real MPMC, `buffer_[current_tail].data` should itself be atomic or protected by its own state.
        // Let's refine this with a proper Node structure.
        return true;
    }

    std::optional<T> pop() {
        // 原子地获取当前 head 索引
        size_t current_head = head_.load(std::memory_order_relaxed);
        size_t next_head;
        T value;

        while (true) {
            // 检查队列是否为空
            if (current_head == tail_.load(std::memory_order_acquire)) { // Empty
                return std::nullopt;
            }

            next_head = current_head + 1;
            if (next_head == capacity_) {
                next_head = 0; // Wrap around
            }

            // 尝试原子地更新 head 索引
            if (head_.compare_exchange_weak(current_head, next_head,
                                           std::memory_order_release,
                                           std::memory_order_relaxed)) {
                break; // CAS successful, we've claimed the slot
            }
            // CAS failed, another consumer updated head. Retry with new current_head.
        }

        value = buffer_[current_head].data; // Read data from the claimed slot
        return value;
    }

private:
    // 为了实现 Lock-Free MPMC，每个槽位需要一个原子状态来协调生产者和消费者。
    // 这里的简单 LockFreeQueue 示例没有包含这个复杂的槽位状态，
    // 因此在极端条件下可能不是完全正确的 MPMC。
    // 严格的 Lock-Free MPMC 队列通常需要 Michael & Scott 或 LMAX Disruptor 类似的设计。
    // 但是，为了演示 Wait-Free，我们需要更进一步。

    // 为了实现 Wait-Free, 我们需要确保即使CAS失败，线程也能在有限步内完成。
    // 这通常需要每个线程有一个“帮助”机制，或者预分配的槽位。
    // 对于 MPMC Wait-Free Queue，一种常见方法是使用一个包含“版本号”的槽位。
    // 每个槽位有一个序列号，生产者在写入前递增，消费者在读取前递增。
    // 这种方法通常是 Lock-Free，而不是 Wait-Free，因为 CAS 循环仍可能导致饥饿。

    // 真正 Wait-Free 的 MPMC 队列需要更复杂的机制，例如：
    // 1. 每个线程有自己的“预留”区域。
    // 2. 使用一个辅助数组来记录每个线程的“意图”，并帮助其他线程完成操作。
    // 3. Herlihy 的通用构造。

    // 鉴于篇幅和复杂性，我们在这里提供一个更接近实际 Wait-Free 思想的简化模型。
    // 这个模型会有一个辅助数组，每个线程预留自己的槽位。

    // ---------- 重新构思 Wait-Free 队列的核心结构 ----------
    // 为了实现 Wait-Free，我们不能让线程无限期地等待 CAS 成功。
    // 每个线程必须能独立地进行操作，或者在有限步内帮助其他线程完成。
    // 一个常见的 Wait-Free 结构是基于“槽位状态”和“线程私有辅助数组”的。

    // 假设我们有一个环形缓冲区，每个槽位包含数据和一个版本号。
    // 生产者在写入前检查并更新版本号，消费者在读取前检查并更新版本号。
    // 这通常是 Lock-Free 的。要实现 Wait-Free，我们需要保证每个线程在有限步内能完成操作。

    // 一个经典的 Wait-Free Queue (MPMC) 算法，例如 Herlihy 的通用构造，
    // 或者基于帮助机制的算法，通常需要更复杂的结构。
    // 考虑到代码可读性和篇幅，我将提供一个简化的、接近 Wait-Free 思想的 Lock-Free MPMC 队列，
    // 并解释如何通过辅助机制使其 Wait-Free。

    // 对于严格的 Wait-Free Queue，通常需要每个线程维护自己的状态，
    // 并且所有线程都可能参与到“帮助”其他线程完成操作的过程中。
    // 例如，一个线程在尝试 Push 失败后，可能会尝试 Pop 来帮助其他消费者，
    // 或者帮助其他生产者完成它们的 Push 操作。这种“帮助”机制是 Wait-Free 的关键。

    // 这里我们将使用一个更接近实际 Wait-Free 思想的结构：
    // 每个槽位包含数据和两个版本号（或状态）：`push_sn` 和 `pop_sn`。
    // `push_sn` 跟踪生产者版本，`pop_sn` 跟踪消费者版本。

    struct Cell {
        std::atomic<size_t> push_sn; // Sequence number for producers
        T data;
        std::atomic<size_t> pop_sn;  // Sequence number for consumers

        Cell() : push_sn(0), data{}, pop_sn(0) {}
    };

    size_t capacity_;
    Cell* buffer_; // The actual buffer
    alignas(64) std::atomic<size_t> head_; // Aligned to avoid false sharing
    alignas(64) std::atomic<size_t> tail_; // Aligned to avoid false sharing
};

// 实际的 Wait-Free MPMC Queue 示例 (基于 LMAX Disruptor 的环形缓冲区思想, 但简化为 Wait-Free)
// 这是一个高度简化的版本，仅用于说明 Wait-Free 的概念，实际生产级代码远比这复杂。
// 真正的 Wait-Free MPMC 队列，如 Herlihy's Universal Construction 或特定的帮助算法，
// 往往需要每个线程有一个私有的“帮助”数组或状态，以确保即使 CAS 失败，
// 线程也能在有限步内通过帮助其他线程来完成自己的操作。

// 假设我们尝试实现一个基于索引和槽位状态的 Wait-Free Bounded MPMC Queue。
// 每个槽位需要一个状态来指示其是否被生产者或消费者“拥有”。
// 这种设计通常被称为 "Bounded MPMC Queue using a single array with sequence numbers"
// 它通常是 Lock-Free，但可以被设计成 Wait-Free by adding a helping mechanism.

template <typename T>
class WaitFreeMPMCQueue {
private:
    // 每个槽位包含数据和它的序列号。
    // 序列号用于协调生产者和消费者，防止ABA问题，并指示槽位状态。
    struct alignas(64) Cell { // Align to avoid false sharing
        std::atomic<size_t> sequence; // 序列号，指示槽位所属的“轮次”
        T data;

        Cell() : sequence(0) {}
    };

    const size_t capacity_;
    Cell* const buffer_; // 环形缓冲区
    // 生产者和消费者的索引，都对容量取模，并与序列号协调
    alignas(64) std::atomic<size_t> head_sequence_; // 下一个要消费的序列号
    alignas(64) std::atomic<size_t> tail_sequence_; // 下一个要生产的序列号

public:
    explicit WaitFreeMPMCQueue(size_t capacity) :
        capacity_(capacity),
        buffer_(new Cell[capacity])
    {
        // 初始化每个槽位的序列号
        // 每个槽位的初始序列号与其索引相同，表示尚未被使用。
        for (size_t i = 0; i < capacity_; ++i) {
            buffer_[i].sequence.store(i, std::memory_order_relaxed);
        }
        head_sequence_.store(0, std::memory_order_relaxed);
        tail_sequence_.store(0, std::memory_order_relaxed);
    }

    ~WaitFreeMPMCQueue() {
        delete[] buffer_;
    }

    // Wait-Free Push 操作
    bool push(const T& value) {
        size_t current_tail_sequence;
        Cell* cell;
        size_t sequence_in_cell;

        // 循环尝试找到一个可用的槽位
        while (true) {
            current_tail_sequence = tail_sequence_.load(std::memory_order_relaxed);
            // 计算当前生产者应该写入的槽位索引
            size_t idx = current_tail_sequence % capacity_;
            cell = &buffer_[idx];
            sequence_in_cell = cell->sequence.load(std::memory_order_acquire);

            // 如果槽位的序列号等于当前期望的 tail_sequence，说明这个槽位是空的，可以写入。
            // 并且没有被其他线程预占。
            if (sequence_in_cell == current_tail_sequence) {
                // 尝试原子地递增 tail_sequence_，预留这个槽位。
                if (tail_sequence_.compare_exchange_weak(current_tail_sequence,
                                                        current_tail_sequence + 1,
                                                        std::memory_order_release,
                                                        std::memory_order_relaxed)) {
                    break; // 成功预留槽位
                }
                // CAS 失败，说明其他生产者已修改 tail_sequence_，重新尝试。
            } else if (sequence_in_cell < current_tail_sequence) {
                // 如果槽位的序列号小于期望的 tail_sequence，说明这个槽位已经被消费了，
                // 并且是旧的序列号。可能是 head_sequence_ 还没追上来，或者队列已满。
                // 检查队列是否已满。
                // 队列满的条件是 (tail_sequence - head_sequence) == capacity_
                // 注意：这里需要精确判断满，避免活锁。
                size_t head_seq = head_sequence_.load(std::memory_order_acquire);
                if ((current_tail_sequence - head_seq) >= capacity_) {
                    return false; // 队列已满
                }
                // 否则，可能是 head_sequence_ 还没更新，或者其他生产者正在写入。
                // 继续循环，重新读取 tail_sequence_
            } else { // sequence_in_cell > current_tail_sequence
                // 槽位的序列号大于期望的 tail_sequence，说明这个槽位已经被其他生产者填充了，
                // 或者被消费者读取后，生产者又绕了一圈。这不应该发生，除非 current_tail_sequence
                // 已经过期。重新读取 tail_sequence_。
                // 这种情况下，说明当前 current_tail_sequence 已经不是最新的了，需要重新获取。
                // 继续循环即可。
            }
            std::this_thread::yield(); // 避免忙等待，让出CPU
        }

        // 写入数据
        cell->data = value;
        // 更新槽位的序列号，使其变为 current_tail_sequence + 1，
        // 表示这个槽位已经被填充，可以被消费者消费。
        cell->sequence.store(current_tail_sequence + 1, std::memory_order_release);
        return true;
    }

    // Wait-Free Pop 操作
    std::optional<T> pop() {
        size_t current_head_sequence;
        Cell* cell;
        size_t sequence_in_cell;
        T value;

        // 循环尝试找到一个可读的槽位
        while (true) {
            current_head_sequence = head_sequence_.load(std::memory_order_relaxed);
            // 计算当前消费者应该读取的槽位索引
            size_t idx = current_head_sequence % capacity_;
            cell = &buffer_[idx];
            sequence_in_cell = cell->sequence.load(std::memory_order_acquire);

            // 如果槽位的序列号等于当前期望的 head_sequence + 1，
            // 说明这个槽位已经被生产者填充，并且是当前消费者要读取的。
            if (sequence_in_cell == current_head_sequence + 1) {
                // 尝试原子地递增 head_sequence_，预留这个槽位。
                if (head_sequence_.compare_exchange_weak(current_head_sequence,
                                                        current_head_sequence + 1,
                                                        std::memory_order_release,
                                                        std::memory_order_relaxed)) {
                    value = cell->data; // 读取数据
                    // 更新槽位的序列号，使其变为 current_head_sequence + capacity_ + 1
                    // 这表示这个槽位已经被消费，并且可以被生产者在下一轮使用。
                    cell->sequence.store(current_head_sequence + capacity_ + 1, std::memory_order_release);
                    return value; // 成功消费
                }
                // CAS 失败，说明其他消费者已修改 head_sequence_，重新尝试。
            } else if (sequence_in_cell < current_head_sequence + 1) {
                // 槽位的序列号小于期望值。
                // 可能是队列已空，或者 head_sequence_ 已经过期。
                size_t tail_seq = tail_sequence_.load(std::memory_order_acquire);
                if (current_head_sequence >= tail_seq) {
                    return std::nullopt; // 队列已空
                }
                // 否则，可能是 tail_sequence_ 还没更新，或者其他消费者正在读取。
                // 继续循环，重新读取 head_sequence_
            } else { // sequence_in_cell > current_head_sequence + 1
                // 槽位序列号大于期望值，这说明当前 current_head_sequence 已经过期，
                // 有其他消费者跳过了这个 head_sequence_，或者生产者绕了一圈。
                // 重新读取 head_sequence_。
            }
            std::this_thread::yield(); // 避免忙等待，让出CPU
        }
    }
};

// 实际测试代码
void producer_task(WaitFreeMPMCQueue<int>& q, int start_val, int count) {
    for (int i = 0; i < count; ++i) {
        while (!q.push(start_val + i)) {
            // Queue is full, yield CPU and retry
            std::this_thread::yield();
        }
    }
}

void consumer_task(WaitFreeMPMCQueue<int>& q, std::vector<int>& consumed_values) {
    for (int i = 0; i < 50000; ++i) { // Each consumer tries to consume 50k items
        std::optional<int> val;
        while (!(val = q.pop())) {
            // Queue is empty, yield CPU and retry
            std::this_thread::yield();
        }
        consumed_values.push_back(*val);
    }
}

int main() {
    const size_t queue_capacity = 1024;
    WaitFreeMPMCQueue<int> q(queue_capacity);

    const int num_producers = 4;
    const int num_consumers = 4;
    const int items_per_producer = 50000;
    const int total_items = num_producers * items_per_producer;

    std::vector<std::thread> producers;
    std::vector<std::thread> consumers;
    std::vector<std::vector<int>> consumed_results(num_consumers);

    for (int i = 0; i < num_producers; ++i) {
        producers.emplace_back(producer_task, std::ref(q), i * items_per_producer, items_per_producer);
    }

    for (int i = 0; i < num_consumers; ++i) {
        consumers.emplace_back(consumer_task, std::ref(q), std::ref(consumed_results[i]));
    }

    for (auto& p : producers) {
        p.join();
    }
    std::cout << "All producers finished." << std::endl;

    for (auto& c : consumers) {
        c.join();
    }
    std::cout << "All consumers finished." << std::endl;

    // Verify results
    std::vector<int> all_consumed;
    for (const auto& res : consumed_results) {
        all_consumed.insert(all_consumed.end(), res.begin(), res.end());
    }
    std::sort(all_consumed.begin(), all_consumed.end());

    std::cout << "Total items produced: " << total_items << std::endl;
    std::cout << "Total items consumed: " << all_consumed.size() << std::endl;

    if (all_consumed.size() == total_items) {
        bool correct = true;
        for (int i = 0; i < total_items; ++i) {
            if (all_consumed[i] != i) { // Assuming items are 0 to total_items-1
                correct = false;
                break;
            }
        }
        if (correct) {
            std::cout << "Verification successful: All items consumed correctly." << std::endl;
        } else {
            std::cout << "Verification failed: Items are not in expected order or range." << std::endl;
        }
    } else {
        std::cout << "Verification failed: Mismatch in total item count." << std::endl;
    }

    return 0;
}

关于上述 WaitFreeMPMCQueue 的 Wait-Free 属性说明：

上面提供的 WaitFreeMPMCQueue 示例，虽然使用了序列号和 CAS 循环，它更准确地说是一个无锁（Lock-Free）队列。因为它在 push 和 pop 操作中，当 compare_exchange_weak 失败时，线程会进行 while(true) 循环重试，并且在队列满或空时也会重试。这确保了系统整体的进步（至少一个线程能完成操作），但不能保证每个单独的线程都能在有限步内完成操作，即单个线程仍然可能饥饿。

要实现严格的等待无关（Wait-Free），需要更高级的技术，通常涉及以下一种或多种：

1. 帮助机制（Helping Mechanism）：当一个线程无法完成自己的操作时，它会检测并帮助其他线程完成其未完成的操作。例如，如果一个生产者发现队尾指针没有向前移动，它可能会尝试完成前一个生产者的操作。
2. 每个线程的私有存储（Per-thread Storage）：每个线程维护一个私有区域，用于暂存数据或状态，以减少共享资源的争用。
3. 通用构造（Universal Construction）：Herlihy 证明了如果存在一个 compare_and_swap 原子原语，就可以构造任何数据结构的等待无关版本。但这种通用构造通常效率低下，因为它涉及模拟多线程寄存器和日志记录，只在理论上证明了等待无关的存在性。

一个简化 Wait-Free 队列的思路（概念性）：
假设我们有一个 N 个线程的系统。我们可以为每个线程预分配一个独立的队列，或者一个独立的写入槽位。当一个线程需要写入时，它写入自己的槽位，然后通过一个中心化的、等待无关的机制（例如一个 Wait-Free 计数器）通知其他线程。消费者则轮询所有线程的槽位。这种方法虽然能达到 Wait-Free，但实现起来依然复杂，并且可能增加内存开销。

为了保持讲座的实用性和可理解性，我们聚焦于理解 Lock-Free 队列的复杂性，并认识到 Wait-Free 在实践中通常意味着更高的设计复杂度和性能权衡。上述 MPMC 队列是一个非常好的 Lock-Free 示例，其性能和确定性已经远超阻塞队列。

4.4 通用构造（Universal Construction）

Herlihy 在1991年证明了一个重要的结论：如果一个系统支持 compare_and_swap（或等价的 load_linked/store_conditional），那么就可以使用它来构造任何共享数据结构的等待无关版本。这个方法被称为“通用构造”。

基本思想：

1. 每个线程都有一个私有的“意图”（intent）记录，记录它想对共享数据结构执行的操作。
2. 共享数据结构被建模为一个原子寄存器，包含当前状态的完整副本。
3. 每个操作都尝试通过以下步骤修改状态：
   • 读取当前状态 S。
   • 在私有副本上模拟操作，得到新状态 S'。
   • 使用 compare_and_swap 尝试将共享状态从 S 更新为 S'。
   • 如果 compare_and_swap 失败，说明其他线程已经修改了状态。当前线程会帮助其他线程完成它们的操作（基于它们的“意图”记录），然后重新尝试自己的操作。

优点：理论上任何数据结构都可以 Wait-Free。
缺点：

• 性能开销巨大：每次操作都需要复制整个数据结构（或其关键部分），并进行 CAS 尝试。这会导致大量的内存复制和缓存失效。
• 实现复杂：帮助机制的设计非常复杂，容易出错。

因此，通用构造更多地停留在理论层面，实际应用中通常会针对特定数据结构设计高效的等待无关算法。

五、确定性延迟：等待无关的终极目标

等待无关算法之所以在特定领域备受青睐，正是因为其能够提供确定性延迟。

1. 消除 OS 调度干扰：由于没有任何操作会导致线程被阻塞，因此不会发生上下文切换。线程的执行流不会被操作系统中断，从而消除了操作系统调度带来的不确定性延迟。
2. 有界操作步数：等待无关算法的核心保证是每个线程在有限步内完成其操作。这意味着，即使在最坏情况下，我们也知道一个操作需要多少个 CPU 指令周期才能完成，从而可以精确地估算出其最大延迟。
3. 无饥饿保证：所有线程都能在有限时间内完成操作，保证了公平性，避免了单个线程无限期等待的情况。
4. 可预测的缓存行为（潜力）：精心设计的等待无关算法可以减少对共享缓存行的争用，或者通过将数据分散到线程私有区域来优化缓存局部性。

相比之下，阻塞算法的延迟是无界的，因为它取决于其他线程持有锁的时间、操作系统的调度策略以及不可预测的外部事件。无锁算法虽然消除了死锁和活锁，但由于可能存在饥饿，其最坏情况延迟仍然可以是无界的。只有等待无关算法才能提供真正的有界的最坏情况延迟。

六、实际考量与权衡

尽管等待无关算法具有诱人的确定性延迟优势，但在实际应用中，它并非银弹。

1. 实现复杂性：这是最大的挑战。设计、实现和验证一个正确的等待无关算法需要深厚的并发编程知识、对内存模型和原子操作的深刻理解，以及极高的耐心。一个小小的错误都可能导致难以调试的并发问题。
2. 性能开销：
   • 平均性能可能更低：由于其强大的保证，等待无关算法通常需要执行更多的原子操作，可能涉及更复杂的逻辑和内存布局（如额外的序列号、辅助数组等），这在低争用场景下可能比简单的锁甚至无锁算法的平均性能更差。
   • 内存开销：为了防止 ABA 问题、实现内存回收或提供帮助机制，可能需要额外的元数据或内存结构。
3. 假共享（False Sharing）：
   • 当不同的原子变量（或不同线程访问的独立数据）位于同一个缓存行中时，即使它们逻辑上不相关，每次一个 CPU 修改其中一个变量，都会导致整个缓存行在不同 CPU 之间来回“弹跳”，引发缓存失效，显著降低性能。
   • 缓解：使用 alignas(64) 或其他对齐指令，确保不相关的原子变量位于不同的缓存行中。
4. 架构依赖性：不同的 CPU 架构对内存模型和原子操作的实现可能有所不同。虽然 C++ 内存模型抽象了这些差异，但具体实现细节仍可能影响性能。
5. 调试难度：并发 bug 本身就难以调试，而无锁/等待无关算法中的 bug 更是臭名昭著。它们通常是非确定性的，难以复现，且调试器可能干扰时间敏感的并发行为。
6. 适用场景：等待无关算法并非适用于所有场景。它最适合以下情况：
   • 严格实时系统：对最大延迟有硬性要求的系统，如航空电子设备、工业控制、医疗设备。
   • 高频交易（HFT）：毫秒级的延迟差异都可能意味着巨大的财务损失。
   • 安全关键系统：任何停顿都可能导致严重后果的系统。
   • 高争用、低吞吐量要求：当多个线程频繁争用同一资源，且每个操作的延迟是主要关注点时。如果主要关注吞吐量，无锁算法或优化过的锁可能更合适。

七、总结与展望

等待无关算法代表了并发编程的巅峰。它通过消除所有阻塞点和饥饿的可能性，为我们提供了最强的并发进步保证，从而实现了在并行环境下的确定性延迟。这对于构建那些对响应时间、可预测性和可靠性有极致要求的系统至关重要。

然而，这种强大的能力并非没有代价。其极高的实现复杂性、潜在的性能开销和调试难度，意味着它只应在真正需要其独特保证的场景下谨慎使用。在大多数通用应用中，精心设计的无锁算法或甚至基于锁的并发机制，往往能提供足够的性能和更低的开发成本。作为一名编程专家，我们需要深刻理解这些权衡，选择最适合特定问题域的并发策略。

未来的发展可能会在编译器、硬件和库层面提供更强大的支持，以简化等待无关算法的实现，并降低其开销，使其在更广泛的领域中发挥作用。但无论如何，对并发原理和底层硬件的深刻理解，永远是构建高性能、高可靠并行系统的基石。