一、什么是 Sticky Counter（粘性计数器）

1⃣ 基本含义

所谓 sticky counter，直译是“会粘在 0 上的计数器”：

一旦计数器变成 0，就永远不能再被加回去
换句话说：

• 计数器 只能在 >0 时递增
• 计数器 可以递减到 0
• 0 是终态（absorbing state）
  用数学语言描述：
• 状态空间：$c\in \mathbb{N}$
• 转移规则：
  • $c>0\Rightarrow c\leftarrow c+1$
  • $c>0\Rightarrow c\leftarrow c-1$
  • $c=0\Rightarrow \text{不可再增}$
    $$0\text{ 是吸收态（absorbing state）}$$

二、为什么它这么重要？

1⃣ std::weak_ptr::lock 的核心需求

std::weak_ptr<T>::lock() 的语义是：

如果对象还活着，安全地增加引用计数并返回 shared_ptr；否则返回空
这里有一个竞争条件（race condition）：

• 线程 A：尝试 lock()（需要把计数从 1 → 2）
• 线程 B：正在释放最后一个 shared_ptr（1 → 0）
  如果设计不当，可能发生：

对象已经析构了，但引用计数又被加回来了
这是灾难级 bug 。

2⃣ Sticky counter 解决了什么？

Sticky counter 保证：

只要引用计数一旦变为 0，就再也不可能被加回
这正是 weak_ptr::lock() 所需要的原子语义。

三、接口语义逐条解释（非常关键）

struct Counter {
    // 如果计数器 > 0，则加一并返回 true
    // 否则（==0）什么都不做，返回 false
    bool increment_if_not_zero();
    // 将计数器减一
    // 如果减完后等于 0，返回 true
    // 否则返回 false
    // 前置条件：计数器 != 0
    bool decrement();
    // 返回当前计数值
    uint64_t read();
};

increment_if_not_zero()

语义：
$$\text{if }c>0\Rightarrow c\leftarrow c+1,\text{return true}$$
$$\text{if }c=0\Rightarrow \text{do nothing},\text{return false}$$
关键点：

检查和加一必须是 一个不可分割的原子操作
否则会出现 ABA 问题。

decrement()

语义：
$$c\leftarrow c-1$$
返回值表示：
$$\text{return }(c==0)$$
前置条件：
$$c\ne 0$$
在 shared_ptr 中，这通常意味着：

• 返回 true → 需要析构对象
• 返回 false → 还有人持有对象

四、为什么不能用普通 atomic++ / atomic–？

✗ 错误做法（示例）

if (counter.load() > 0) {
    counter.fetch_add(1);
}

这是错误的，因为：

1. load() 和 fetch_add() 之间不是原子
2. 可能发生：
   • 线程 A：load() == 1
   • 线程 B：decrement() → 0，析构对象
   • 线程 A：fetch_add() → 1（复活对象）
     对象“死而复生”

五、正确实现：CAS（Compare-And-Swap）

下面是一个标准且正确的 sticky counter 实现。

六、带完整注释的代码实现

#include <atomic>
#include <cstdint>
struct Counter {
    // 使用无符号 64 位原子变量保存计数值
    std::atomic<uint64_t> value{0};
    // 如果当前值不为 0，则原子地加 1 并返回 true
    // 如果当前值为 0，则什么都不做，返回 false
    bool increment_if_not_zero() {
        uint64_t old = value.load(std::memory_order_acquire);
        while (old != 0) {
            // 尝试把 old -> old + 1
            // 如果期间 value 没被别人修改，CAS 成功
            if (value.compare_exchange_weak(
                    old,                    // 期望值（失败时会被更新）
                    old + 1,                // 新值
                    std::memory_order_acq_rel,
                    std::memory_order_acquire)) {
                return true;
            }
            // 如果 CAS 失败，old 会被更新为当前 value
            // 然后继续循环判断 old 是否为 0
        }
        // old == 0，说明计数器已经“粘”在 0 上
        return false;
    }
    // 将计数器减 1
    // 返回是否减完后变成 0
    // 前置条件：value != 0
    bool decrement() {
        // fetch_sub 返回的是“减之前”的值
        uint64_t old = value.fetch_sub(1, std::memory_order_acq_rel);
        // 如果原来的值是 1，减完后就是 0
        return old == 1;
    }
    // 读取当前值（只读，不修改）
    uint64_t read() const {
        return value.load(std::memory_order_acquire);
    }
};

七、与 shared_ptr / weak_ptr 的对应关系

---

Counter 操作	shared_ptr / weak_ptr 含义
increment_if_not_zero	weak_ptr::lock()
decrement	shared_ptr 析构
变为 0	对象析构、控制块释放

---

八、核心不变量（Invariant）

整个设计依赖一个极其重要的不变量：
$$\text{一旦 }c=0,\forall t,c(t)=0$$

0 之后永远是 0
这是 sticky counter 的灵魂。

九、一句话总结（面试/论文级）

Sticky counter 是一种 以 0 为吸收态的原子计数器，
它通过 CAS 保证“检查 + 修改”的不可分割性，
是 std::weak_ptr::lock() 和安全并发内存回收的基础构件。

一、先看这份「第一次实现」

struct Counter {
    // 如果 counter > 0，就加 1 并返回 true
    bool increment_if_not_zero() {
        if (counter > 0) {
            counter++;          // ✗ 非原子
            return true;
        }
        return false;
    }
    // 递减计数器，如果减完后为 0，返回 true
    bool decrement() {
        return (--counter == 0); // ✗ 非原子
    }
    // 读取当前值
    uint64_t read() { return counter; }
    // 初始值为 1，表示“对象还活着”
    uint64_t counter{1};
};

乍一看非常“合理”，但这是一个经典的并发陷阱。

二、设计目标回顾（必须牢记）

我们想要的语义是：

如果计数器已经变成 0，就永远不能再被加回
用数学不变量表示：
$$\forall t,c(t_0)=0\Rightarrow c(t)=0$$
也就是说：
0 是吸收态（absorbing state）

三、关键问题：increment_if_not_zero 不是原子操作

这一段代码：

if (counter > 0) {
    counter++;
    return true;
}

在并发环境中，实际上被拆成了两个独立步骤：

1. 读 counter
2. 写 counter
   中间可以被任意线程打断。

四、致命竞态：两个线程交错执行

我们来看你给出的线程交错场景。

初始状态

counter = 1

五、逐步时间线分析（非常重要）

Thread 1：increment_if_not_zero()

if (counter > 0) {

• 读取 counter == 1
• 条件成立
• 此时还没执行 counter++
• 被调度器打断

Thread 2：decrement()

return (--counter == 0);

执行步骤：

• --counter：
  $$counter:1\to 0$$
• 返回 true
• 对象被认为可以析构
  现在系统状态是：

counter = 0   ← 对象已“死亡”

Thread 1：继续执行

counter++;
return true;

执行：

• counter++：
  $$counter:0\to 1$$
  结果：

counter = 1   ← 对象“复活”了

灾难发生

六、发生了什么？（核心问题）

✗ 违反了 sticky counter 的根本不变量

我们刚刚证明：
$$c(t_1)=0\text{但}c(t_2)=1$$
这直接违反了：
$$0\text{ 必须是终态}$$

七、从语义角度理解这个 bug

1⃣ Thread 2 的视角

• 它看到 counter 从 1 变成 0
• 认为：

  “我是最后一个引用，可以安全析构对象”

2⃣ Thread 1 的视角

• 它基于过期信息（counter 曾经 > 0）
• 在对象已经“死亡”后：

  又把引用计数加回来了

3⃣ 结果

• 引用计数 ≠ 实际对象生命周期
• 后果可能是：
  • use-after-free
  • double free
  • 内存破坏

八、为什么这是「必然会出错」的设计？

因为下面这个操作 不是原子的：
$$\text{check}(c>0)+\text{increment}(c)$$
在并发系统中，这两个步骤之间：

• 任何线程
• 任何状态变化
• 任何析构行为
  都可能发生。

九、形式化总结这个错误

我们期望的操作是：
$$\text{if }c>0\text{ then }c\leftarrow c+1$$
但代码实际实现的是：
$$\text{load}(c);\text{later}\text{store}(c+1)$$
这两个操作之间没有原子性保证。

十、为什么加 mutex 也能解决？

是的，加锁可以解决：

std::mutex m;
bool increment_if_not_zero() {
    std::lock_guard<std::mutex> lock(m);
    if (counter > 0) {
        counter++;
        return true;
    }
    return false;
}

但问题是：

• std::shared_ptr / weak_ptr 不能用 mutex
• 要求：
  • 无阻塞
  • 高性能
  • 可在析构路径使用
    所以必须使用 原子 + CAS。

十一、这段代码在讲什么「教学意义」？

这份 “first implementation” 的目的不是让你用它，而是让你理解：

✗ “看起来正确”的代码，在并发下几乎一定是错的
尤其是：

• check-then-act
• read-modify-write
• 生命周期管理

十二、一句话总结（非常重要）

这个实现失败的根本原因是：
increment_if_not_zero() 中的「检查 + 自增」不是一个原子操作，
导致计数器在变成 0 后仍然可能被加回，破坏了 sticky counter 的核心不变量。

一、先看“加锁后的实现”

struct Counter {
    // 如果 counter > 0，则加 1 并返回 true
    bool increment_if_not_zero() {
        std::lock_guard g_{m};   //  进入临界区
        if (counter > 0) {
            counter++;           // ✓ 在锁保护下修改
            return true;
        }
        return false;
    }
    // 递减计数器，如果减完后为 0 返回 true
    bool decrement() {
        std::lock_guard g_{m};   //  进入临界区
        return (--counter == 0); // ✓ 原子化于 mutex 级别
    }
    std::mutex m;               // 互斥锁，保护 counter
    uint64_t counter{1};        // 初始引用计数
};

二、这一版「修复了什么问题？」

1⃣ 关键修复点

现在下面这两个操作：

if (counter > 0) {
    counter++;
}

在 mutex 的保护下变成了：

一个不可被其他线程打断的临界区
从并发语义上等价于：
$$\text{atomically: }\{\begin{array}{l}\text{if }c>0\\ c\leftarrow c+1\end{array}$$

2⃣ sticky counter 的核心不变量恢复了

现在可以保证：
$$\forall t,c(t_0)=0\Rightarrow c(t)=0$$
原因是：

• 一旦 decrement() 把 counter 变成 0
• 它仍然持有 mutex
• 在释放锁之前，任何线程都无法执行 increment_if_not_zero()

三、用线程交错证明「不会再出错」

初始状态

counter = 1
mutex = unlocked

Thread 1：increment_if_not_zero()

std::lock_guard g_{m};

• 成功获得锁
• Thread 2 被阻塞

if (counter > 0) {

• 读到 counter == 1
• 条件成立

Thread 2：decrement()

std::lock_guard g_{m};

• ✗ 无法获得锁
• 被阻塞

Thread 1：继续执行

counter++;   // 1 → 2
return true;

• 离开作用域，释放 mutex

Thread 2：继续执行

--counter;   // 2 → 1
return false;

✓ 最终状态

counter = 1

没有“死而复生”的问题

四、从语义上看：这是“正确”的 sticky counter

我们需要的语义是：

$$\text{increment\_if\_not\_zero}\equiv \{\begin{array}{l}\text{if }c>0\Rightarrow c\leftarrow c+1\\ \text{else }\text{do nothing}\end{array}$$
mutex 保证了：

• 检查
• 修改
  这两个动作在并发意义下是 不可分割的

五、那为什么这仍然不是最终解？

既然正确，为什么 std::shared_ptr 不用 mutex？

答案是：用不了

六、为什么 mutex 在这里「设计上不合格」？

1⃣ 析构路径中不能安全加锁

在 shared_ptr / weak_ptr 的实现中：

• decrement() 可能发生在对象析构期间
• 析构期间：
  • allocator 可能已被销毁
  • mutex 本身可能已经被释放
  • 再加锁 = 未定义行为

2⃣ mutex 可能导致死锁

考虑：

• 线程 A：持有 mutex，正在 decrement → 析构对象
• 析构函数中：
  • 间接调用 weak_ptr::lock()
  • 再次尝试获取 mutex
    $$\text{deadlock}$$

3⃣ mutex 破坏无阻塞（lock-free）要求

C++ 标准对 shared_ptr 有重要隐含要求：

• weak_ptr::lock() 必须是 无阻塞的
• 至少不能：
  • 等待
  • sleep
  • 被操作系统调度
    mutex 不满足这一点。

4⃣ 性能不可接受

shared_ptr 是高频路径：

• 每次拷贝
• 每次析构
• 每次 lock()
  mutex 会导致：
• cache line 抖动
• syscalls（在高竞争时）
• 严重性能退化

七、数学视角下的差异

mutex 版本保证的是：

$$\text{mutual exclusion}$$

真正需要的是：

$$\text{atomic read–modify–write with state transition}$$
也就是：
$$c>0\stackrel{\text{CAS}}{\to }c+1$$

八、教学重点总结（非常重要）

这一版代码的意义不在于“最终方案”，
而在于让你意识到：

• ✗ 问题不是“逻辑写错”
• ✗ 问题也不是“没加锁”
• ✓ 问题是：需要一个更强的原子语义，而不是互斥

九、一句话总结

使用 mutex 可以让 sticky counter 在语义上正确，
但由于析构安全性、死锁风险和性能要求，
它不适合用于 std::shared_ptr / weak_ptr，
因此必须使用基于 CAS 的原子实现。

一、到目前为止我们学到了什么？

1⃣ Locks（锁）的本质

锁几乎可以解决所有并发正确性问题
原因很简单：

• 锁通过 互斥（mutual exclusion）
• 把并发问题 退化成单线程问题

但代价是什么？

锁的代价是：消灭并发
用数学直觉理解：

• 原本可以并行的 $N$ 个线程
• 被串行化为：
  $$T_1\to T_2\to \cdots \to T_N$$

锁的典型问题

• 阻塞（blocking）
• 死锁（deadlock）
• 优先级反转
• 析构路径不可用
• 性能抖动（cache + syscall）

重要免责声明（非常专业）

Always measure performance. Never guess.
永远用数据说话，不要凭感觉判断性能。

二、进展保证（Progress Guarantees）

这是理论上用来分类并发算法的标准。

1⃣ Blocking（阻塞）

✗ 没有任何进展保证

• 一个线程被阻塞
• 可能导致整个系统停住
  例如：
• mutex
• condition_variable

2⃣ Obstruction-free（无阻碍）

定义

单个线程在“完全孤立”执行时，一定能在有限步内完成操作
数学描述：
$$\exists N,\text{steps}\le N$$

特点

• ✓ 不会死锁
• ✗ 在高竞争下可能一直失败重试
• ✗ 不保证系统整体吞吐

3⃣ Lock-free（无锁）

定义（非常重要）

任意时刻，至少有一个线程在前进
形式化表达：
$$\forall t,\exists \text{thread }i\text{ makes progress}$$

含义

• 系统不会整体停住
• 总有操作完成
• ✗ 某个特定线程可能永远失败（饥饿）

这是 shared_ptr 的最低要求

系统级吞吐量保证

4⃣ Wait-free（无等待）

定义

每个线程的每个操作，都能在有限步内完成
$$\forall i,\exists N_i$$

特点

• ✓ 最强保证
• ✗ 实现极其复杂
• ✗ 性能通常不如 lock-free

三、为什么我们需要 Lock-free？

回到 sticky counter 的需求：

• 不能阻塞（析构路径）
• 必须高并发
• 必须保证：
  $$c=0\Rightarrow \text{永远不能再增加}$$
  mutex：
• ✓ 正确
• ✗ blocking
• ✗ 析构不安全
  lock-free 是最合适的折中

四、Lock-free Sticky Counter 代码详解

作者：Daniel Anderson
来源：danielanderson.net

1⃣ 完整代码

struct Counter {
    // 如果 counter > 0，则原子地加 1
    bool increment_if_not_zero() {
        // 先读取当前值
        auto current = counter.load();
        // 只要 current > 0 且 CAS 失败，就不断重试
        while (current > 0 &&
               !counter.compare_exchange_weak(current, current + 1)) {
            // CAS 失败时：
            // current 会被更新为 counter 的最新值
            // 然后重新判断 current > 0
        }
        // 如果 current > 0，说明成功加 1
        return current > 0;
    }
    // 原子递减
    bool decrement() {
        // fetch_sub 返回的是“减之前”的值
        // 如果之前是 1，说明现在变成 0
        return counter.fetch_sub(1) == 1;
    }
    // 读取当前值
    uint64_t read() { return counter.load(); }
    // 原子计数器，初始值为 1
    std::atomic<uint64_t> counter{1};
};

2⃣ increment_if_not_zero() 的核心思想

我们想要的语义是：
$$\text{if }c>0\Rightarrow c\leftarrow c+1$$
并且必须是 原子操作。

CAS 循环的逻辑

auto current = counter.load();

• 读一个“期望值”（expected）

while (current > 0 &&
       !counter.compare_exchange_weak(current, current + 1)) {
}

循环条件含义：

1. current > 0
   sticky 语义保证
2. CAS 失败就重试

compare_exchange_weak 的语义

compare_exchange(expected, desired)

等价于：

if (counter == expected) {
    counter = desired;
    return true;
} else {
    expected = counter; // 写回最新值
    return false;
}

为什么失败后要继续？

因为在并发环境中：

• 其他线程可能在你操作期间修改了 counter
• CAS 失败并不是错误，只是竞争失败

3⃣ 为什么 current > 0 的判断必须在循环里？

因为 CAS 失败后：

• current 会被更新为最新的 counter
• 这个值 可能已经变成 0
  数学上：
  $$c_{old}>0\nRightarrow c_{now}>0$$

4⃣ decrement() 为什么这么简单？

return counter.fetch_sub(1) == 1;

这是一个标准模式：

• fetch_sub(1) 返回的是旧值
• 如果旧值是 1：
  $$1\to 0$$
  说明这是 最后一个引用

5⃣ 为什么这是 lock-free？

increment_if_not_zero()

• CAS 可能失败
• 但失败说明 别的线程成功了
• 至少有一个线程在前进

decrement()

• 单条原子指令
• 无阻塞

满足 lock-free 定义

$$\forall t,\exists \text{thread making progress}$$

五、与 mutex 版本的本质区别

---

特性	mutex	lock-free
阻塞	是	否
死锁	可能	不可能
析构安全	否	是
吞吐保证	无	有
实现复杂度	低	高

---

六、最终总结（一句话）

锁通过消灭并发换取简单正确性，
而 lock-free 通过 CAS 在保持并发的同时保证系统级进展，
sticky counter 正是 lock-free 设计的经典最小示例。

一、什么是 “CAS loop”（比较-交换循环）

CAS loop 是 lock-free 算法与数据结构的“面包和黄油”（bread and butter）
意思是：
最基础、最常用、最核心的构件

1⃣ CAS loop 的标准流程

一个典型 CAS loop 包含以下步骤：

（1）读取当前状态

auto old = atomic_var.load();

表示当前状态：
$${\text{state}}_{old}$$

（2）基于当前状态计算新状态

auto desired = f(old);

数学上：
$$\text{state}\ast new=f(\text{state}\ast old)$$

（3）尝试提交修改（compare-exchange）

if (atomic_var.compare_exchange_weak(old, desired)) {
    // 成功
}

等价语义：
$$\text{if }state==old\Rightarrow state\leftarrow desired$$

（4）如果失败，说明“别人赢了”，那就重试

while (!CAS) {
    // old 会被更新为最新值
}

2⃣ 用伪代码总结 CAS loop

do {
    old = load(state)
    new = compute(old)
} while (!compare_exchange(state, old, new))

二、为什么 CAS loop 是 lock-free 的？

核心原因（非常重要）

如果某个线程 CAS 失败了，那只能说明：
在此期间，另一个线程成功完成了一次 CAS
也就是说：
$$\text{CAS 失败}\Rightarrow \exists \text{其他线程取得进展}$$

形式化描述（lock-free 定义）

$$\forall t,\exists \text{线程 }i\text{ 在 }t\text{ 附近完成操作}$$
因此：

• 系统整体在前进
• 不会发生整体停顿
• lock-free

三、为什么 CAS loop 不是 wait-free？

问题在于：循环次数不受限

一个线程可能：

• 每次 CAS 都失败
• 因为别的线程总是“抢先一步”
  数学上：
  $$\exists \text{线程 }i,\forall k,\text{第 }k\text{ 次 CAS 失败}$$
  也就是说：

✗ 不能保证单个操作在有限步内完成

一个形象比喻

• 多个人抢一扇门
• CAS = “先碰到门把手的人进去”
• lock-free：
  • 总有人进去
• wait-free：
  • 每个人都有保证能进去

四、Wait-free 的工具箱（Tools for wait freedom）

1⃣ 核心约束（非常重要）

Wait-free 算法不能包含“无限 CAS loop”
原因：
$$\text{Wait-free}\Rightarrow \exists N,\text{steps}\le N$$
而 unbounded loop 显然违反这个条件。

2⃣ 这并不意味着不能用 CAS

正确说法是：

✗ 不能在「无界循环」中用 CAS
✓ 可以在「有界步骤」中用 CAS

3⃣ Wait-free 常用的原子操作

compare_exchange

compare_exchange_weak(expected, desired)
compare_exchange_strong(expected, desired)

语义回顾：

if (current == expected) {
    current = desired;
    return true;
} else {
    expected = current;
    return false;
}

fetch_add / fetch_sub

old = x.fetch_add(delta);

语义：
$$x\leftarrow x+\delta ,\text{return old}$$
单条原子指令 → 天然 wait-free

exchange

old = x.exchange(new_value);

语义：
$$x\leftarrow new,\text{return old}$$

4⃣ 为什么这些操作是 wait-free 的？

因为：

• 没有循环
• 步数是固定的
• 不依赖其他线程“是否成功”

五、从 lock-free 走向 wait-free

1⃣ Lock-free 的本质问题

线程是“竞争”的（competitive）

• 多个线程
• 争抢同一个 CAS
• 失败的线程反复重试
  结果：
• 有人成功
• 有人可能永远失败（饥饿）

2⃣ Wait-free 的思想转变

从竞争 → 协作（collaborative）
这是一个非常重要的设计哲学转变。

六、Wait-free 算法设计的核心：Helping（帮助）

1⃣ 什么是 Helping？

当一个线程发现：

• 另一个线程正在执行一个操作
• 但尚未完成

它不会抢着自己完成操作，而是“帮对方完成”

2⃣ Helping 的直觉理解

• 每个操作都有一个“任务描述”
• 其他线程看到后：
  • 帮忙把这个任务完成
  • 再继续自己的操作

3⃣ Helping 带来的保证

如果每个操作：

• 都能被其他线程“顺手完成”
• 那么：
  $$\forall \text{operation},\exists N$$
  wait-free 成立

七、为什么 Helping 不容易？

1⃣ 首先要能“看见别人正在干什么”

必须有：

• 公共状态
• 标记某个操作“进行中”

2⃣ 其次要能“安全地帮忙”

• 多个线程可能同时帮同一个操作
• 必须避免重复提交
• 需要精巧的状态设计

3⃣ 设计复杂度陡增

---

特性	lock-free	wait-free
设计难度	中	极高
性能	高	未必
保证	系统级	每线程
常见程度	常见	罕见

---

八、一句话总结（核心思想）

CAS loop 是 lock-free 的核心构件，
但正是“无界 CAS loop”阻止了它成为 wait-free；
要实现 wait-free，必须从“竞争”走向“协作（helping）”。

九、学习路线建议

如果你继续深入，我建议：
1⃣ 完全吃透 CAS loop（你已经在这一步）
2⃣ 学 Michael–Scott queue（经典 lock-free）
3⃣ 再看 Herlihy 的 wait-free universal construction
4⃣ 最后再碰生产级 wait-free 算法

一、Wait-free 计数器的根本难点

我们要的语义仍然是 sticky counter：
$$c=0\Rightarrow \text{永远不能再增加}$$
同时还要满足 wait-free：
$$\forall \text{operation},\exists N,\text{steps}\le N$$
不能有无界 CAS loop

1⃣ 为什么需要“检测其他操作正在进行中”？

在 wait-free 设计中：

• 操作不能“等别人”
• 但又必须知道：
  • 是否已经有人在把计数器变成 0
  • 是否需要帮忙完成这个过程

2⃣ 大思路：偷高位当标志位（flags）

我们把一个 uint64_t 拆成两部分：

[ flags | counter ]
 63 62     0 ... 61

示意：

10 00000000000000000000011
^^
||
flags

两个标志位

static constexpr uint64_t is_zero = 1ull << 63;   // 最高位
static constexpr uint64_t helped  = 1ull << 62;   // 次高位

含义：

flag	含义
is_zero	对外宣告：计数器已经为 0
helped	有线程已经“帮忙”完成了置零

---

二、Step 1：没有 read 的 wait-free counter

代码

struct Counter {
    static constexpr uint64_t is_zero = 1ull << 63;
    // 如果 is_zero 没被置位，说明还没变成 0
    bool increment_if_not_zero() {
        // fetch_add 一定是 wait-free
        return (counter.fetch_add(1) & is_zero) == 0;
    }
    bool decrement() {
        // 如果原值是 1，说明递减后为 0
        if (counter.fetch_sub(1) == 1) { // 我是最后一个
            uint64_t e = 0;
            // 尝试把 0 → is_zero
            return counter.compare_exchange_strong(e, is_zero);
        }
        return false;
    }
    std::atomic<uint64_t> counter{1};
};

Step 1 的关键思想

• fetch_add / fetch_sub：
  固定步数 → wait-free
• compare_exchange：
  不是循环 → 仍然 wait-free

Step 1 的隐藏问题

counter.compare_exchange_strong(e, is_zero);

如果返回 false 呢？

说明：

• 有别的线程
• 在你 fetch_sub 之后
• 改动了 counter
  decrement 可能“失去线性化点”

三、Step 2：加上 read —— 问题爆炸

天真版 read

uint64_t read() {
    auto val = counter.load();
    return (val & is_zero) ? 0 : val;
}

致命问题：0 不一定是真的 0

线程交错：

counter = 1

Thread 1

counter.fetch_sub(1);  // counter = 0

Thread 2

counter.load();        // 读到 0
return 0;

Thread 3

counter.fetch_add(1); // counter = 1

结论

读到的 0 只是一个“中间态”
数学上：
$$\text{read}(c)=0;\nRightarrow ;c\text{ 已稳定为 0}$$

四、Step 2 的关键想法：Helping（帮助）

如果我读到 0，说明一定有线程“正在”把它变成 is_zero
那我可以：

• 帮它把 flag 设置好
• 保证对外一致性

Helping 版 read

uint64_t read() {
    auto val = counter.load();
    if (val == 0 &&
        counter.compare_exchange_strong(val, is_zero)) {
        return 0; // 我帮忙完成了置零
    }
    return (val & is_zero) ? 0 : val;
}

新问题出现了

如果 read 帮忙把 is_zero 设好了，
真正的 decrement 再也不会返回 true
因为：

• decrement 的 CAS 会失败
• 但 decrement 才是“应该负责析构”的那个操作

五、Step 3：引入 helped 标志位

新的 flag

static constexpr uint64_t helped = 1ull << 62;

表示：

“置零是被别人帮忙完成的”

read：帮忙时打标记

if (val == 0 &&
    counter.compare_exchange_strong(val, is_zero | helped))
    return 0;

decrement：接手“功劳”

bool decrement() {
    if (counter.fetch_sub(1) == 1) {
        uint64_t e = 0;
        // 自己成功置零
        if (counter.compare_exchange_strong(e, is_zero))
            return true;
        // read 帮忙了，我来“收尾”
        else if ((e & helped) &&
                 (counter.exchange(is_zero) & helped))
            return true;
    }
    return false;
}

关键语义解释

• helped = 有人已经完成 CAS
• exchange(is_zero)：
  • 把状态规范化
  • 同时拿回“置零的功劳”

六、Step 4：最终思想（而非代码）

最后的警告

可能有多个线程：

• 都在 decrement
• 都在等待 someone else 完成置零

解决原则

必须保证：

• 只有一个 decrement
• 对“counter 变成 0”负责
  数学上：
  $$\exists !\text{decrement op s.t. }c\to 0$$

七、这一整套设计的本质

从竞争到协作

---

Lock-free	Wait-free
CAS 竞争	Helping 协作
重试	帮别人完成
系统进展	每线程进展

---

为什么这么复杂？

因为 wait-free 要保证：
$$\forall \text{线程},\exists \text{有界完成}$$
这是 并发算法里最强的保证。

八、一句话总结（极其重要）

Wait-free 算法的关键不在于“我能不能成功”，
而在于“即使我失败了，也要保证别人能成功，
并且我可以帮别人成功”。

一、整体设计回顾（先给你一个“全局视角”）

我们用一个 uint64_t 原子变量同时表示：

[ flag 位 | 实际计数值 ]
 63  62     0 ... 61

bit 63 : is_zero  —— 对外宣布：计数器已经稳定为 0（sticky）
bit 62 : helped   —— 置零是“被别人帮忙完成”的
bit 0-61 : 实际引用计数

设计不变量（最重要）

$$\text{is\_zero}=1\Rightarrow \text{counter 语义上为 0}$$
且：
$$\exists !\text{一个 decrement 操作对“置零”负责}$$

二、完整代码 + 详细注释

struct Counter {
    // 最高位：表示“计数器已经被宣布为 0（sticky）”
    static constexpr uint64_t is_zero = 1ull << 63;
    // 次高位：表示“置零是由其他线程帮忙完成的”
    static constexpr uint64_t helped  = 1ull << 62;
    // ===============================
    // increment_if_not_zero
    // ===============================
    //
    // 语义：
    //   如果计数器还没被宣布为 0，则递增并返回 true
    //   如果已经 is_zero，则递增无意义，返回 false
    //
    // wait-free：
    //   - 单条 fetch_add
    //   - 固定步数完成
    //
    bool increment_if_not_zero() {
        // fetch_add 返回的是“加之前”的值
        // 如果旧值的 is_zero 位为 0，说明：
        //   在这次递增线性化时，计数器仍然是“活的”
        return (counter.fetch_add(1) & is_zero) == 0;
    }
    // ===============================
    // decrement
    // ===============================
    //
    // 语义：
    //   - 递减计数
    //   - 如果这是最后一个引用（1 → 0）
    //     必须由某一个 decrement 返回 true
    //
    bool decrement() {
        // fetch_sub 返回的是“减之前”的值
        // 只有旧值 == 1，才意味着：
        //   我是最后一个 decrement
        if (counter.fetch_sub(1) == 1) {
            // 期望值 e：我们希望看到一个“纯 0”
            // （没有 flag，没有计数）
            uint64_t e = 0;
            // 情况 1：
            // 我自己成功地把 0 → is_zero
            // 这是最理想的情况
            if (counter.compare_exchange_strong(e, is_zero))
                return true;
            // 情况 2：
            // compare_exchange 失败，说明：
            //   有其他线程（通常是 read）
            //   已经“帮忙”把 is_zero 设好了
            //
            // e 会被更新为当前 counter 的值
            //
            // 如果：
            //   - helped 位被置位
            //   - 且我能通过 exchange 把状态规范化为 is_zero
            // 那么：
            //   我“接管”置零的功劳
            else if ((e & helped) &&
                     (counter.exchange(is_zero) & helped))
                return true;
        }
        // 不是最后一个 decrement
        return false;
    }
    // ===============================
    // read
    // ===============================
    //
    // 语义：
    //   - 返回当前计数值
    //   - 如果已经 is_zero，返回 0
    //   - 如果读到“中间态 0”，主动帮忙完成置零
    //
    uint64_t read() {
        // 读取当前值（可能是中间态）
        auto val = counter.load();
        // 如果 val == 0：
        //   说明：
        //     - 有线程刚刚 fetch_sub(1)
        //     - 但还没来得及设置 is_zero
        //
        // read 不等待它，
        // 而是主动帮忙：
        if (val == 0 &&
            counter.compare_exchange_strong(
                val,
                is_zero | helped   // 同时设置 is_zero + helped
            ))
            return 0;  // 我帮你完成了置零
        // 如果 is_zero 已经被设置
        // 无论低位是什么，语义上都必须返回 0
        return (val & is_zero) ? 0 : val;
    }
    // 原子计数器
    // 初始值为 1，表示“对象活着，有一个引用”
    std::atomic<uint64_t> counter{1};
};

三、几个「极其容易忽略」但非常关键的点

1⃣ 为什么 read 可以“帮忙”？

因为 wait-free 的核心原则是：

不等待别人完成，而是自己帮忙完成
否则 read 可能永远卡在“0 的中间态”。

2⃣ 为什么 helped 位是必须的？

如果没有 helped：

• read 抢先完成 is_zero
• decrement 永远无法知道：
  • “是不是我该负责析构？”
    结果：

✗ 所有 decrement 都返回 false
✗ 析构永远不会发生

3⃣ 为什么 decrement 必须“拿回功劳”？

因为：

必须有且只有一个 decrement 返回 true
这决定了：

• 对象析构
• 控制块释放
• 内存回收
  数学表达：
  $$\exists !\text{decrement 返回 true}$$

4⃣ 为什么这是 wait-free？

• 没有 CAS loop
• 每个操作：
  • fetch_add / fetch_sub
  • 最多 1–2 次 CAS / exchange
• 步数有上界

四、最终一句话总结（请记住）

这段代码的真正精髓不是 bit hack，
而是：
“当我无法前进时，我帮助别人前进，并确保责任最终被正确地认领。”

一、Benchmark 在测什么？（先搞清楚问题）

测试对象

atomic<shared_ptr> 的 load 操作
区别只有一个：

• 版本 A：Lock-free counter（CAS loop）
• 版本 B：Wait-free counter（fetch_add / fetch_sub + helping）
  只影响引用计数，不影响指针本身

延迟指标说明

---

指标	含义
1%	最快的 1% 操作（best-case）
50%	中位数（median）
99%	最慢的 1% 操作（tail latency）

---

并发算法最重要的往往不是平均值，而是 99% 尾延迟

二、Benchmark #1：100% load（纯读）

𝒑 个线程，全都在 load

Lock-free Counter（CAS loop）

---

p	1%	50%	99%
1	21ns	24.8ns	32.7ns
28	217ns	1.75µs	13.1µs
56	535ns	5.36µs	31.1µs

---

Wait-free Counter

---

p	1%	50%	99%
1	21.7ns	24.8ns	31.7ns
28	158ns	1.31µs	8.40µs
56	146ns	2.43µs	13.3µs

---

关键观察

• 单线程（p=1）：几乎没有差别
• 多线程下：
  • wait-free 明显降低 99% 尾延迟
  • p=56 时：
    $$\text{99\% 延迟下降约 }\frac{31.1-13.3}{31.1}\approx 57$$

原因分析

Lock-free 的问题

• load 内部有 CAS loop
• 多线程同时 load：
  • 不断失败
  • cache line 来回 bouncing
• 导致：

某些线程可能一直在失败 → 尾延迟暴涨

Wait-free 的优势

• load：
  • 固定步数
  • 无自旋
• helping 消除了中间态等待

每个 load 都能在有限步骤内完成

三、Benchmark #2：50% load / 50% store（读写均衡）

Lock-free Counter

---

p	1%	50%	99%
28	212ns	1.19µs	8.67µs
56	215ns	2.41µs	26.7µs

---

Wait-free Counter

---

p	1%	50%	99%
28	164ns	1.21µs	7.65µs
56	153ns	2.05µs	26.4µs

---

观察

• 差距缩小
• wait-free 在：
  • 1%
  • 50%
  • 99%
    上仍然略优或持平

原因

• store（写）：
  • 本身就是 RMW
  • wait-free 带来的优势被写竞争部分抵消
• 但：
  • helping 仍然降低了 load 的 worst-case

四、Benchmark #3：10% load / 90% store（写密集）

Lock-free Counter

---

p	1%	50%	99%
28	217ns	1.73µs	14.1µs
56	240ns	6.40µs	82.6µs

---

Wait-free Counter

---

p	1%	50%	99%
28	188ns	1.41µs	13.3µs
56	244ns	7.46µs	86.4µs

---

关键结论

• 在 写为主 的场景下：
  • wait-free 不再占优
  • 甚至略差

为什么？

• wait-free：
  • 帮忙机制
  • 更多原子操作
• 写密集下：
  • 额外开销变成负担
• lock-free：
  • “竞争但简单”
  • 在高写压下反而更直接

五、总结页逐条解读

Always measure performance. Never guess.

并发算法的性能是反直觉的
不能仅凭：

• 理论复杂度
• “wait-free 听起来更高级”

哪个算法最好，取决于负载

关键维度：

1. 读 / 写比例
2. 线程数 $p$
3. 硬件拓扑（NUMA / cache）

Progress guarantees 的实际意义

---

类型	理论保证	性能启示
Blocking	无	简单但危险
Lock-free	系统吞吐有保证	尾延迟可能很差
Wait-free	每个操作有上界	延迟稳定

---

Wait-free 的“灵魂”：helping

不是等别人完成，也不是抢先完成，而是：
帮别人完成
这是从：

• 竞争（lock-free）
  → 合作（wait-free）

六、一句“带走就够”的总结

**Wait-free ≠ 永远更快
Lock-free ≠ 一定不稳定

Progress guarantee 是“性能假设工具”，
Benchmark 才是“裁判”。**