给 SPSC 队列加 loom

主线位置：主线 3「一次数据竞争怎么被抓到」在一个真实小库上的样子：队列的槽位放在 loom 的 UnsafeCell 里，两侧靠 head / tail 的 Release / Acquire 交接所有权，少一个 Acquire，槽位的读写就没有 happens-before。用法对应 06 用法与调试。代码在 exp-loom/hands-on/src/{sync,spsc}.rs。

sequenceDiagram
  participant P as 生产者
  participant T as tail / head
  participant C as 消费者
  P->>T: head.load(Acquire)：消费者已读完 head 以下的槽
  P->>P: buf[tail % cap].with_mut：写入
  P->>T: tail.store(tail + 1, Release)
  C->>T: tail.load(Acquire)：生产者已写好 tail 以下的槽
  C->>C: buf[head % cap].with：读出
  C->>T: head.store(head + 1, Release)
  Note over P,C: 任一侧的 Acquire 改成 Relaxed，槽位的写和读之间就没有 happens-before

切换层

src/sync.rs，照 loom 文档推荐的 mod sync 写法（cfg(loom) 切换层）：

#[cfg(loom)]
pub(crate) use loom::{
  cell::UnsafeCell,
  sync::atomic::{AtomicUsize, Ordering},
  sync::Arc,
};

#[cfg(not(loom))]
pub(crate) use std::sync::{
  atomic::{AtomicUsize, Ordering},
  Arc,
};

#[cfg(not(loom))]
#[derive(Debug)]
pub(crate) struct UnsafeCell<T>(std::cell::UnsafeCell<T>);

#[cfg(not(loom))]
impl<T> UnsafeCell<T> {
  pub(crate) const fn new(v: T) -> Self {
    UnsafeCell(std::cell::UnsafeCell::new(v))
  }
  pub(crate) fn with<R>(&self, f: impl FnOnce(*const T) -> R) -> R {
    f(self.0.get())
  }
  pub(crate) fn with_mut<R>(&self, f: impl FnOnce(*mut T) -> R) -> R {
    f(self.0.get())
  }
}

API 差别落在两处：

• UnsafeCell 只能用闭包访问。loom 要知道访问从哪里开始、到哪里结束，才能在开始时和结束时各查一次时钟（UnsafeCell 与原子的非同步访问）。所以 std 一侧反过来补一个同样形状的包装，队列代码只写一份。tokio 的 std/unsafe_cell.rs 就是这个样子。
• 构造不能是 const。std 一侧的 new 可以是 const fn，loom 的不行；队列的 channel(cap) 本来就在运行时建，不受影响。要是写一个 static 队列，就得给 loom 另写一份（坑：在模型外用 loom 类型）。

Cargo.toml 里 loom 只出现在 [target.'cfg(loom)'.dependencies]，并在 [lints.rust] 里声明 check-cfg = ["cfg(loom)"]，平时构建不依赖 loom、也没有 unexpected_cfgs 警告。这里用的是裸 cfg(loom)：下游开 --cfg loom 时这个库也会被切换，10 在本站中 讲了这在依赖树里意味着什么；做成要发布的库时，更稳妥的是 cfg(all(test, loom)) 或自己的 cfg 名。

队列

head、tail 只增不减，槽位 i % cap 在 i >= tail 时归生产者，在 head <= i < tail 时归消费者：

pub fn push(&self, v: T) -> Result<(), T> {
  let r = &*self.0;
  let tail = r.tail.load(Ordering::Relaxed); // only we write tail
  let head = r.head.load(PUSH_SEES_HEAD); // the consumer is done with slots below head
  if tail - head == r.buf.len() {
    return Err(v);
  }
  // SAFETY: tail - head < cap, so slot `tail % cap` is free and owned by us.
  r.buf[tail % r.buf.len()].with_mut(|p| unsafe { (*p).write(v) });
  r.tail.store(tail + 1, Ordering::Release);
  Ok(())
}

pop 对称：Relaxed 读自己的 head、Acquire 读 tail、with 读槽、Release 写 head。PUSH_SEES_HEAD、POP_SEES_TAIL 两个常量平时是 Acquire，打开 broken_push / broken_pop feature 时变成 Relaxed，用来做反向验证。Ring 的 Drop 把 head..tail 里剩下的值 drop 掉。

模型

tests/spsc_loom.rs：子线程做生产者，推 n 个 String，满了就 yield_now；主线程做消费者，空了就 yield_now，断言按顺序收到。自旋里的 yield_now 是必须的，否则撞分支上限（坑）。

容量取 1 和 2，是为了让「绕回去复用槽位」在两三步之内就发生：容量 1、两个元素，第二次 push 一定写在消费者刚读完的那个槽上，正好是 head 那个 Acquire 要保护的地方（写一个好模型 的「把常数调小」）。另有一个模型检查 drop：推两个 loom Arc 进去不取，drop 两端之后引用计数回到 1。

反向验证

同一组模型，每次只把一处 Acquire 改成 Relaxed。真实线程那一列是 tests/spsc_std.rs：不开 --cfg loom，生产者和消费者各一个 OS 线程，容量 64，传 100 万个数，断言顺序；每种改动连跑 300 次（本机 arm64，跑时负载约 3.4）。

两种错 loom 都在第一个交错就报出，而且报的正是那个槽位：前者是「读时发现并发的写」，后者是「写时发现并发的读」（04 检测器 的报错一览）。它不需要碰巧撞上坏的时序：只要第 1 轮里那次写和读之间没有同步，向量时钟就判得出来。

压测那一列说明为什么需要 loom：broken_pop 在 ARM 上 50 次里出一次，broken_push 300 次一次都没有。后者不是没有错，而是要求消费者对槽位的读被推迟到生产者覆盖之后，在这个压测形状下很难碰到。只看压测，它会被当成正确的代码合进去。

小结

• 切换层：loom 类型与 std 类型二选一；std 一侧补 with / with_mut，队列只写一份；loom 只是 cfg(loom) 下的依赖，并声明 check-cfg。
• 模型：容量 1–2、两三个元素，自旋里 yield_now；容量 2、3 个元素在上限 2 / 3 / 不设时分别是 157 / 473 / 6 096 个交错。
• 反向验证：任一侧 Acquire 改成 Relaxed，loom 都在第 1 个交错报竞争；同样的改动真实线程压测 300 次，一个失败 6 次，一个 0 次。
• 做成要发布的库时，切换条件优先用 cfg(all(test, loom)) 或自己的 cfg 名，免得下游开 --cfg loom 时被一起换掉。