用 parking_lot_core 写计数信号量

主线位置：主线 1 的骨架换一种状态。RawMutex 的「锁 + 有人停车」两位，换成「许可数 + 有人等待」；park / unpark_one 的用法与回调约束不变（park 与四种 unpark）。代码在 exp-parking_lot/hands-on/src/sema.rs。

stateDiagram-v2
  direction LR
  free: n 个许可，无人等（n×2）
  empty: 0 个许可，无人等（0）
  waiting: 0 个许可，有人等或将要等（1）
  mixed: n 个许可，WAITERS 还没清（n×2+1）
  free --> empty: acquire 取走最后一个
  empty --> waiting: acquire 置 WAITERS
  waiting --> waiting: park，validate 要求状态恰好是 1
  waiting --> mixed: release fetch_add(2)
  mixed --> free: unpark_one 回调里，没有更多等待者，清 WAITERS
  mixed --> waiting: 被叫醒者或插队者取走许可
  empty --> free: release

状态

const WAITERS: usize = 1;
const ONE: usize = 2;

pub struct Semaphore {
  state: AtomicUsize, // permits * ONE | WAITERS
}

把「有人等」放进同一个字，release 一次 fetch_add 就能同时知道「要不要去叫人」，和 RawMutex::unlock 快路径失败就说明有 PARKED_BIT 是一个道理。

acquire：抢、转、置位、停车

fn acquire_slow(&self) {
  let addr = self as *const _ as usize;
  let mut spin = SpinWait::new();
  let mut s = self.state.load(Ordering::Relaxed);
  loop {
    if s >= ONE {
      match self.state.compare_exchange_weak(s, s - ONE, Ordering::Acquire, Ordering::Relaxed) {
        Ok(_) => return,
        Err(x) => s = x,
      }
      continue;
    }
    if s & WAITERS == 0 && spin.spin() {
      s = self.state.load(Ordering::Relaxed);
      continue;
    }
    if s & WAITERS == 0 {
      if let Err(x) = self.state.compare_exchange_weak(s, s | WAITERS, Ordering::Relaxed, Ordering::Relaxed) {
        s = x;
        continue;
      }
    }
    // validate runs under the bucket lock: no permits, and WAITERS still set.
    let validate = || self.state.load(Ordering::Relaxed) == WAITERS;
    // SAFETY: addr is our own address; the callbacks only do atomic loads and
    // never panic or call back into parking_lot_core.
    let r = unsafe { park(addr, validate, || {}, |_, _| unreachable!("no timeout"), DEFAULT_PARK_TOKEN, None) };
    debug_assert!(matches!(r, ParkResult::Unparked(_) | ParkResult::Invalid));
    spin.reset();
    s = self.state.load(Ordering::Relaxed);
  }
}

逐行对照 RawMutex::lock_slow：有许可就抢（插队，哪怕有人在等）；没人等才自旋；置 WAITERS；validate 要求状态恰好是 WAITERS（0 个许可、等待位还在），否则不睡、回到开头。timed_out 写成 unreachable!：不传超时就不会被调用；如果真被调用，它在桶锁里 panic，违反了 park 的契约，所以这一行是「逻辑上不可达」的声明，不是错误处理。Once 的源码里也是这样写的（Once）。

许可的获取用 Acquire，与 release 的 Release 配对，上一个持有者在许可保护下的写入对下一个可见；WAITERS 位只用来决定「要不要走慢路径」，用 Relaxed。

release：加许可，必要时叫一个

pub fn release(&self) {
  let old = self.state.fetch_add(ONE, Ordering::Release);
  if old & WAITERS != 0 {
    self.release_slow();
  }
}

fn release_slow(&self) {
  let addr = self as *const _ as usize;
  let callback = |r: parking_lot_core::UnparkResult| {
    if !r.have_more_threads {
      self.state.fetch_and(!WAITERS, Ordering::Relaxed);
    }
    DEFAULT_UNPARK_TOKEN
  };
  // SAFETY: same key as acquire; the callback is one atomic op and cannot panic.
  unsafe { unpark_one(addr, callback) };
}

have_more_threads 是 UnparkResult 里「摘下这个之后队列里还有同 key 的线程」（park 与四种 unpark）。清 WAITERS 只在回调里做，也就是只在桶锁里做。三种交错都安全：

• 回调清位之前，另一个线程已经在桶锁里验证并入队：那时 have_more_threads 为真，不清。
• 回调清位之后，一个已经置过位、正要停车的线程来验证：状态里没有 WAITERS，验证失败，回去重新置位。
• 没找到任何停车者（置位的线程验证失败走了）：have_more_threads 为假，清位，下次 release 不再走慢路径。

被叫醒的线程不直接拿到许可，回到循环开头去抢，可能被插队者抢先，那就再停车。这里没有做「交接」（TOKEN_HANDOFF），所以和 Mutex 一样不公平。

测试

tests/hands_on.rs，看门狗下运行，连续 3 遍通过：

• semaphore_never_exceeds_permits：许可 1 / 3 / 8，线程 8 / 16 / 32，每个线程 2 万次「acquire → 在用计数加一并记最大值 → 偶尔 yield_now → 在用计数减一 → release」。断言：最大同时持有数 ≤ 许可数、结束后许可数复原、总次数精确。输出：

  Copy

  permits=1 threads=8: max in use 1, permits left 1, acquisitions 160000
  permits=3 threads=16: max in use 3, permits left 3, acquisitions 320000
  permits=8 threads=32: max in use 8, permits left 8, acquisitions 640000
  

• semaphore_wakes_every_parked_thread：许可 0，32 个线程 acquire 后停车；50 ms 后逐个 release 32 次，32 个线程全部返回，许可数回到 0。这一项专测「等待位清早了、有人永远睡着」。

反向验证（feature broken_sema）：把 acquire 换成「先读、看到有许可、再写回减一后的值」两步，中间放一次 spin_loop。3 次运行都是同样两项失败：许可 1 时最大同时持有数 8（max in use 8 > permits 1）；32 个线程那项许可数对不上（left: 2, right: 0）或看门狗超时。说明这组断言能发现「超发许可」和「许可数被写乱」。

局限

• 没有用 loom：parking_lot_core 用真实的原子、线程局部变量和系统调用，不能放进 loom 的模型里跑；也就无法穷举交错。上面的证据是压力测试加反向验证：能说明测法有效，不能证明没有更罕见的交错。
• 没有超时（try_acquire_for 要处理 ParkResult::TimedOut 和 timed_out 回调里「最后一个等待者清位」，照 RawMutex 的写法即可）；release 的 fetch_add 不检查溢出。
• 和 tokio 的 Semaphore（异步、FIFO 公平、可一次取多个许可，tokio 卷 batch-semaphore）不是一类东西：这里是阻塞线程的同步信号量。

小结

• 计数信号量的状态就是 RawMutex 的两位换成「许可数 + 等待位」，慢路径照抄 lock_slow 的四段。
• validate 要求状态恰好是「0 个许可 + 等待位」；等待位只在 unpark_one 的回调里、即桶锁里清。
• 压测：3 组配置同时持有数都没超过许可数，许可数复原；32 个停车者全部被叫醒。
• 反向：「先看再减」的写法让 1 个许可同时被 8 个线程持有。
• 没有 loom，证据只有压测加反向验证。