async Mutex 与 Semaphore

主线位置：跨任务主线的另一种形态。任务之间传的不是数据，而是「使用权」：拿不到就登记 waker、返回 Pending，持有者放手时叫醒等待者。和 Channel 一样，内部状态由一把 RawMutex 在很短的时间里保护，等待靠 等待队列 里的那几种槽。

flowchart TB
  lock["lock poll"] --> held{"locked"}
  held -->|否| take["locked = true"]
  take --> ready["Ready Guard"]
  held -->|是| reg["waker.register"]
  reg --> pend["Pending"]
  guard["Guard drop"] --> unlock["locked = false"]
  unlock --> wake["waker.wake"]
  wake -.-> lock

文件：mutex.rs（392 行）、rwlock.rs（386 行）、semaphore.rs（778 行，第 493 行以后是测试）、pipe.rs（927 行）、once_lock.rs（245 行）、lazy_lock.rs（174 行）。实验在 exp-embassy/sync-std（platform-std 执行器，release，M2 Max、macOS 27、rustc 1.99.0；同机负载 20 以上）。

Mutex：一个布尔，一个 waker 槽

Mutex<M, T> 是一把 RawMutex 护着的小状态，加一个放数据的 UnsafeCell<T>：

#[derive(Debug)]
struct State {
    locked: bool,
    waker: WakerRegistration,
}

// ...
pub struct Mutex<M, T>
where
    M: RawMutex,
    T: ?Sized,
{
    state: BlockingMutex<M, RefCell<State>>,
    inner: UnsafeCell<T>,
}

文档特意说明 RawMutex「is held for very short periods only, while locking and unlocking. It is not held for the entire time the async Mutex is locked.」。持锁期间可以 .await，中断也照常进来；RawMutex 只在改 locked 和 waker 的那几条指令里关中断（CriticalSectionRawMutex 时）。

lock 是一个 poll_fn：

    pub fn lock(&self) -> impl Future<Output = MutexGuard<'_, M, T>> {
        poll_fn(|cx| {
            let ready = self.state.lock(|s| {
                let mut s = s.borrow_mut();
                if s.locked {
                    s.waker.register(cx.waker());
                    false
                } else {
                    s.locked = true;
                    true
                }
            });

            if ready {
                Poll::Ready(MutexGuard { mutex: self })
            } else {
                Poll::Pending
            }
        })
    }

MutexGuard 的 Drop 把 locked 置回 false，再 waker.wake()。被叫醒的任务要再 poll 一次 lock 才能真正拿到锁，中间如果有别的任务先 poll 到 lock，锁就归它了：没有「交接」，只有「放开，叫一个人来抢」。lock 返回的 future 被 drop 时什么也不做，槽里可能留着它的 waker，下次解锁时白叫醒一次。

只有一个 WakerRegistration，所以三个以上的任务等同一把锁时，后来的把先来的换出来叫醒，几个等待者在锁被占着的整段时间里互相 poll，和 Channel 满了时的发送方一样。

sync-std/src/bin/mutex_fair.rs：按 0、1、2、3 的顺序 spawn 四个任务，每个 M.lock().await 一次，拿到后记下自己的编号、持锁 50 ms（中间 Timer::after_millis(50).await）再放开：

  mutex: task 3 locked at 0 ms after 1 polls
  mutex: task 1 locked at 55 ms after 991907 polls
  mutex: task 2 locked at 110 ms after 1143834 polls
  mutex: task 0 locked at 165 ms after 1143835 polls
mutex order [3, 1, 2, 0]

再跑两次是 [3, 1, 2, 0] 和 [3, 1, 0, 2]。

• 第一个拿到锁的是最后 spawn 的任务 3：执行器的运行队列是一个栈，一批里的任务按入队的反序被 poll（run_queue.rs 的注释：「batches will be iterated in the reverse order as they were enqueued」，见 状态与运行队列）。
• 之后的顺序不是 spawn 顺序，也不是第一次 poll 的顺序，每次运行还可能不同。
• 每个等待者在等的 50 到 150 ms 里被 poll 一百万次左右。

Mutex<M, T> 是 Sync 的条件是 M: Sync 且 T: Send，和 blocking_mutex::Mutex 一样；MutexGuard::map 可以把 guard 缩小到数据的一个字段（0.8.0 去掉了 U: Sized 的限制）。tokio 的 Mutex 建在 FIFO 的 batch_semaphore 上，等待者按到达顺序拿锁；embassy 这里换来的是一个 bool 加 8 字节的 waker 槽。

RwLock：读者计数，同一个槽

RwLock<M, T>（0.7.0 加入）的状态是 readers: usize、writer: bool 和一个 WakerRegistration。read 只看 writer：没有写者持锁就 readers += 1 进去；write 要 writer == false && readers == 0。读者 guard drop 时只有 readers 归零才叫醒，写者 guard drop 时总是叫醒。

read 不看「有没有写者在等」，所以读者一个接一个来时，写者一直进不去（sync-std/src/bin/rwlock_order.rs：读者 1 在 0 ms 拿读锁、各持 30 ms，写者在 1 ms 开始等，读者 2、3、4 分别在 20、40、60 ms 到达）：

  reader 1 got the lock at 0 ms
  reader 2 got the lock at 26 ms
  reader 3 got the lock at 43 ms
  reader 4 got the lock at 61 ms
  writer got the lock at 95 ms

写者等到最后一个读者放手才进去。读者之间的持锁时间只要有重叠，写者就会一直饿着。

Semaphore：Greedy 和 Fair

Semaphore 是一个 trait：acquire(n)、try_acquire(n)、acquire_all(min)（至少 min 个，有多少拿多少）、release(n)、set(n)。拿到的许可装在 SemaphoreReleaser 里，drop 时还回去，disarm() 可以把许可「拿走不还」。

GreedySemaphore<M> 是 Cell<SemaphoreState { permits, waker: WakerRegistration }>，等待规则和 Mutex 完全一样。「Greedy」指的是：一个要 3 个许可的任务在等，另一个只要 1 个的任务来了，有 1 个就先给它，大请求可能一直等下去。

FairSemaphore<M, N> 是 embassy-sync 里唯一按到达顺序排队的原语。它的状态 FairSemaphoreState：

struct FairSemaphoreState<const N: usize> {
    permits: usize,
    next_ticket: usize,
    wakers: Deque<Option<Waker>, N>,
}

• 第一次拿不到时 register 在队尾放一个 waker，票号是 next_ticket + 队列长度，记在 future 自己的 ticket 字段里。队列满了（超过 N 个等待者）返回 Err(WaitQueueFull)，acquire 直接以错误完成。
• take 只给票号等于 next_ticket 的那个（排头）；没有票号的新来者（包括 try_acquire）在队列非空时一律拿不到。排头拿到后出队，剩下还有许可就叫醒新的排头。
• release 只叫醒排头一个（wake_by_ref，槽里的 waker 保留）。
• future 被 drop 时 FairAcquire 的 Drop 调 cancel，把自己那一格置为 None，之后 pop_canceled 会把队首的 None 清掉。

同样四个任务换成 FairSemaphore<_, 4>::new(1) 的 acquire(1)：

  fair semaphore: task 3 acquired at 305 ms after 1 polls
  fair semaphore: task 2 acquired at 360 ms after 2 polls
  fair semaphore: task 1 acquired at 410 ms after 2 polls
  fair semaphore: task 0 acquired at 465 ms after 2 polls

顺序是 3、2、1、0，正是它们第一次被 poll 的顺序（运行队列的反序），三次运行完全一样；每个等待者只被 poll 2 次：第一次排队，第二次被叫醒时拿到。

排头被取消

上面那段「叫醒新排头」只发生在 take 里。如果排头已经被 release 叫醒，却在下一次 poll 之前被 drop 了呢？

static S: FairSemaphore<CriticalSectionRawMutex, 4> = FairSemaphore::new(1);
static STOP: Signal<CriticalSectionRawMutex, ()> = Signal::new();

#[embassy_executor::task]
async fn a() {
  match select(STOP.wait(), S.acquire(1)).await {
    Either::First(()) => println!("  A: stopped, acquire future dropped at {} ms", Instant::now().as_millis()),
    Either::Second(_) => println!("  A: acquired"),
  }
}

#[embassy_executor::task]
async fn b() {
  let _p = S.acquire(1).await.unwrap();
  println!("  B: acquired at {} ms", Instant::now().as_millis());
}

// main:
let holder = S.try_acquire(1).unwrap();
spawner.spawn(a().unwrap());
Timer::after_millis(1).await;
spawner.spawn(b().unwrap());
Timer::after_millis(1).await;
// same synchronous step: give the permit back and tell A to stop
drop(holder);
STOP.signal(());
Timer::after_millis(200).await;
println!("after 200 ms: try_acquire from main -> {}", S.try_acquire(1).is_some());
println!("main: set(1) (permits already 1)");
S.set(1);
Timer::after_millis(10).await;
println!("main: release(1)");
S.release(1);

Pipe：字节流

Pipe<M, N> 是 N 字节的环形缓冲区（crate 内部的 RingBuffer<N>：start、end、full）加读、写两个 WakerRegistration，实现了 embedded_io_async 的 Read / Write。和 Channel 的差别在粒度：write 写进「一段连续的空闲区」能放下的那么多字节就返回，所以缓冲区回绕时即使总空间够，一次 write 也可能只写一部分，文档写明了这一点。split 出来的 Reader 还实现了 BufRead（直接借出缓冲区里的连续一段），因此只能有一个。适合在任务之间转发串口、USB CDC 这类字节流。

OnceLock 与 LazyLock

OnceLock<T> 是 AtomicBool 加 Cell<MaybeUninit<T>>，没有 RawMutex 参数，初始化固定用 critical_section::with。它的 get 不登记任何 waker：

    pub fn get(&self) -> impl Future<Output = &T> {
        poll_fn(|cx| match self.try_get() {
            Some(data) => Poll::Ready(data),
            None => {
                cx.waker().wake_by_ref();
                Poll::Pending
            }
        })
    }

没初始化就叫醒自己再返回 Pending，等于 yield_now 循环，在等的期间执行器不会睡。类型的文档写明了「this implementation uses a busy loop to poll the value」，并说用于早期初始化的 static 时可以接受。

0.8.0 的 try_get 用 Ordering::Relaxed 读标志，init 用 Relaxed 写，单核上没问题（同一个核看自己的写入总是有序的），多核或 std 多线程上另一个核可能先看到 true、后看到数据。main 上的 #6000（提交「embassy-sync: fix OnceLock publication memory ordering」）改成了 Acquire / Release，尚未发版。

LazyLock<T, F> 用一个 union 先放初始化函数、后放值；ensure_init 在 critical_section::with 里调 f()，标志用的是 Acquire / Release。get() 是同步的，没有 async 版本。注意 f() 在临界区里执行：在 Cortex-M 单核实现下，初始化函数跑多久，中断就被关多久。OnceLock::get_or_init 也一样。

动脑筋答案

  A: stopped, acquire future dropped at 2 ms
after 200 ms: try_acquire from main -> false
main: set(1) (permits already 1)
main: release(1)
  B: acquired at 214 ms

• drop(holder) 让 permits 变成 1，release 只叫醒排头 A。A 的 acquire 被 drop，cancel 把它那一格置为 None，但没有叫醒排在后面的 B。
• 200 ms 后许可还是 1 个，B 还在等。主任务的 try_acquire 没有票号，队列里有 B，take 返回 None：许可闲着，谁也拿不到。
• set(1) 只在新值大于当前值时叫醒，1 不大于 1，没用。直到 release(1) 让 permits 变成 2、叫醒排头 B，B 才在 214 ms 拿到。没有这次额外的 release，B 会一直等下去。
• 这是 0.8.0 的 bug：被叫醒的排头没来得及 poll 就被取消，唤醒丢了。main 上 #6964（提交「embassy-sync: wake the next waiter when a FairSemaphore acquire is canceled」）在 cancel 里加了「被取消的是排头就叫醒新排头」，并加了单元测试 cancel_wakes_next_waiter，尚未发版。把 acquire 放进 select 或 with_timeout 时最容易撞上。

小结

• Mutex：RawMutex 只护一个 bool 和一个 WakerRegistration，持锁期间不关中断；解锁是「放开再叫一个人来抢」。多个等待者互相叫醒，拿锁顺序不固定，实测每个等待者被 poll 约百万次。
• RwLock：读者计数加同一个槽，read 不看等待中的写者，写者会饿。
• GreedySemaphore 和 Mutex 同构；FairSemaphore<M, N> 用票号和 Deque<Option<Waker>, N> 做 FIFO，实测按到达顺序、每个等待者 poll 2 次，超过 N 个等待者返回 WaitQueueFull。
• 0.8.0 的 FairSemaphore 在被叫醒的排头被取消时不叫醒下一个，许可会闲置到下一次 release；main 上已修（#6964）。
• Pipe 是带两个 waker 的字节环形缓冲区；OnceLock::get 忙等，0.8.0 的内存序只在单核上成立；LazyLock 在临界区里跑初始化函数。