Condvar

主线位置：主线 2「一次 notify_all」全部。wait → wait_until_internal → park(condvar 地址, validate = 登记 mutex 地址, before_sleep = 放开 mutex)；notify_all → unpark_requeue(condvar, mutex, validate, callback)：mutex 被锁着就 RequeueAll（一个都不叫醒，全部挪到 mutex 的队列），否则 UnparkOneRequeueRest；被挪走的线程此后由 mutex 的 unlock 逐个叫醒。

sequenceDiagram
  participant W as 等待者 W1..Wn
  participant C as Condvar 桶（key = condvar）
  participant M as Mutex 桶（key = mutex）
  participant N as 通知者（持有 mutex）
  W->>C: wait：桶锁里 validate，state = mutex 地址，入队
  W->>W: before_sleep：mutex.unlock()，睡
  N->>C: notify_all：lock_bucket_pair 锁两个桶
  N->>N: validate：state 清空；mutex 被锁，mark_parked_if_locked，RequeueAll
  C->>M: 全部等待者改 key，接到 mutex 桶队尾
  N->>M: 放开 mutex：unlock_slow，unpark_one(mutex)
  M->>W: 叫醒 W1（TOKEN_NORMAL 或 TOKEN_HANDOFF）
  W->>W: W1 拿到锁，wait 返回；之后每次 unlock 叫醒下一个

一个指针

condvar.rs:90-102：

pub struct Condvar {
    state: AtomicPtr<RawMutex>,
}

impl Condvar {
    #[inline]
    pub const fn new() -> Condvar {
        Condvar {
            state: AtomicPtr::new(ptr::null_mut()),
        }
    }
}

state 只记一件事：当前有等待者时，它们配的是哪一把 mutex；没有等待者时为 null。等待队列在停车场里，key 是 condvar 自己的地址。所以 Condvar 是 8 字节、const fn new、没有 Drop（E1：std 在 macOS 上是 16 字节，Linux 上 4 字节）。

文档列了与 std 的差异（48-59 行），逐条对照代码：

wait：先登记，再放锁，再睡

wait_until_internal：

fn wait_until_internal(&self, mutex: &RawMutex, timeout: Option<Instant>) -> WaitTimeoutResult {
    let result;
    let mut bad_mutex = false;
    let mut requeued = false;
    {
        let addr = self as *const _ as usize;
        let lock_addr = mutex as *const _ as *mut _;
        let validate = || {
            // Ensure we don't use two different mutexes with the same
            // Condvar at the same time. This is done while locked to
            // avoid races with notify_one
            let state = self.state.load(Ordering::Relaxed);
            if state.is_null() {
                self.state.store(lock_addr, Ordering::Relaxed);
            } else if state != lock_addr {
                bad_mutex = true;
                return false;
            }
            true
        };
        let before_sleep = || {
            // Unlock the mutex before sleeping...
            unsafe { mutex.unlock() };
        };
        let timed_out = |k, was_last_thread| {
            // If we were requeued to a mutex, then we did not time out.
            // We'll just park ourselves on the mutex again when we try
            // to lock it later.
            requeued = k != addr;

            // If we were the last thread on the queue then we need to
            // clear our state. This is normally done by the
            // notify_{one,all} functions when not timing out.
            if !requeued && was_last_thread {
                self.state.store(ptr::null_mut(), Ordering::Relaxed);
            }
        };
        result = unsafe {
            parking_lot_core::park(addr, validate, before_sleep, timed_out, DEFAULT_PARK_TOKEN, timeout)
        };
    }

    // Panic if we tried to use multiple mutexes with a Condvar. Note
    // that at this point the MutexGuard is still locked. It will be
    // unlocked by the unwinding logic.
    if bad_mutex {
        panic!("attempted to use a condition variable with more than one mutex");
    }

    // ... and re-lock it once we are done sleeping
    if result == ParkResult::Unparked(TOKEN_HANDOFF) {
        unsafe { deadlock::acquire_resource(mutex as *const _ as usize) };
    } else {
        mutex.lock();
    }

    WaitTimeoutResult(!(result.is_unparked() || requeued))
}

顺序是这个设计的核心：

1. validate 在 condvar 的桶锁里登记 mutex 地址。state 的读写都是 Relaxed，因为改它的地方（这里、notify_* 的 validate 和回调、超时回调）全都在 condvar 所在的桶锁里，桶锁本身负责先后顺序。
2. 入队之后、睡之前，before_sleep 放开 mutex。这一刻起通知者可以拿到 mutex 并 notify，而等待者已经在队列里，通知不会丢。std 的 pthread 条件变量靠 pthread_cond_wait 在内核或 libc 里做同样的原子「放锁并睡」；parking_lot 用「先入队、后放锁」达到同样效果。before_sleep 在桶锁之外执行，里面调 mutex.unlock() 可能走 unlock_slow 去 unpark_one，park 的契约允许 before_sleep 叫醒别人，但不允许它停车（park 与四种 unpark）。
3. 醒来后看令牌：被 requeue 到 mutex 上的线程，是被 mutex 的 unlock_slow 叫醒的，令牌也由它给。如果那次解锁是公平交接（TOKEN_HANDOFF），锁已经在手里，只补一次死锁检测记账；否则 mutex.lock() 自己去抢（这时可能又被插队）。

返回值 timed_out() 为真的条件是「没被叫醒，也没被挪走」：被挪到 mutex 队列之后才超时，不算超时（timed_out 回调里 requeued = k != addr，挪走后线程的 key 已经改成了 mutex 地址）。

notify_all：挪，不叫

notify_all_slow：

let validate = || {
    if self.state.load(Ordering::Relaxed) != mutex {
        return RequeueOp::Abort;
    }
    // Clear our state since we are going to unpark or requeue all
    // threads.
    self.state.store(ptr::null_mut(), Ordering::Relaxed);

    if unsafe { (*mutex).mark_parked_if_locked() } {
        RequeueOp::RequeueAll
    } else {
        RequeueOp::UnparkOneRequeueRest
    }
};
let callback = |op, result: UnparkResult| {
    // If we requeued threads to the mutex, mark it as having
    // parked threads. The RequeueAll case is already handled above.
    if op == RequeueOp::UnparkOneRequeueRest && result.requeued_threads != 0 {
        unsafe { (*mutex).mark_parked() };
    }
    TOKEN_NORMAL
};
let res = unsafe { parking_lot_core::unpark_requeue(from, to, validate, callback) };

res.unparked_threads + res.requeued_threads

unpark_requeue 用 lock_bucket_pair 同时锁住 condvar 和 mutex 两个桶，在锁里跑 validate，把 condvar 的等待者摘下来，改 key、接到 mutex 桶的队尾，再跑回调，最后（如果有）叫醒一个。

• mutex 正被锁着（通常如此：惯用法是持锁改条件再 notify）：mark_parked_if_locked 给 mutex 补上 PARKED_BIT，返回 RequeueAll，一个线程也不叫醒。通知者放开 mutex 时走 unlock_slow，从 mutex 队列叫醒第一个；那个线程用完锁再放开，叫醒下一个。每次只有一个等待者醒来，每个都能拿到锁。
• mutex 没被锁（先放锁、后 notify）：叫醒第一个，其余挪走，回调里给 mutex 补 PARKED_BIT。注释说明了 validate 里检查完「没被锁」之后 mutex 又被别人锁上的竞态「doesn't matter」：被叫醒的那个线程去 lock() 时会正常排队。

两种情况下 PARKED_BIT 都是在桶锁里补上的，这是 RawMutex 那条不变量的要求：被挪进 mutex 队列的线程，必须有 PARKED_BIT 指着，才会有人去叫（RawMutex 的一字节状态机）。这也是 Condvar 只能配 parking_lot 自己的 RawMutex 的原因：它要在停车场的桶锁里直接改 mutex 的状态字节。

返回值 unparked_threads + requeued_threads 是「受影响的线程数」：持锁 notify_all 时恒等于等待者数，而真正醒来的是 0 个。文档写的是「Returns the number of threads woken up」，和代码不符，以代码为准（坑 notify-return-value）。

notify_one：也可能一个都不叫醒

notify_one_slow 的 validate 结构相同：mutex 被锁着就 RequeueOne（把一个等待者挪到 mutex 队列），否则 UnparkOne。回调里如果 condvar 上没有更多等待者就把 state 清空。返回 unparked + requeued != 0。

所以 notify_one() 返回 true 只表示「有一个等待者被处理了」，不表示有线程马上会运行：持锁调用时，那个线程只是换了个队列，要等通知者放锁才会醒。

只能配一把 mutex

validate 里发现 state 已经是别的 mutex，就设 bad_mutex 并返回 false，park 立即返回 Invalid——不入队、不调 before_sleep，所以 mutex 仍然被这个线程持有。然后 panic!，守卫在 unwind 时把锁放开。文档说明：同一时刻只能配一把 mutex；没有等待者之后可以换一把（41-46 行）。

std 在 macOS 上的 pthread 条件变量也检查，但更严：condvar/pthread.rs:31-41 用 compare_exchange(0, addr) 记下第一次用到的 mutex，之后从不清零。

实验 E15：一个 Condvar，两把 mutex

方法（exp-parking_lot/condvar-two-mutexes，功能实验，不计时）：线程 A 用 m1 在 cv 上等；50 ms 后线程 B 用 m2 调 cv.wait，记录 B 的结果、从调用 wait 到返回或 panic 的耗时，以及 panic 时 m2 是否仍被持有。「顺序」模式：先通知 A、等 A 退出，再让 B 用 m2 等，之后通知 B。两种实现、两种模式各跑 3 次：

• panic 发生在第二个等待者调用 wait 时，不是「醒来之后」：几十到几百微秒，说明根本没有睡下。
• parking_lot 只在「同时」有两把 mutex 时报错，换锁是允许的；std（pthread 版）只要换过 mutex 就报错。Linux 上 std 用 futex 版条件变量，本卷没有核实它是否检查。
• std 的 m2 被毒化：守卫在 panic 的 unwind 中被 drop。

能传给 wait 的只有 MutexGuard

wait 的签名是 wait<T: ?Sized>(&self, mutex_guard: &mut MutexGuard<'_, T>)（254-257 行），MutexGuard<'_, T> 是 lock_api::MutexGuard<'_, RawMutex, T>。实测（exp-parking_lot/mapped-guard-condvar，cargo build）：

• MutexGuard::map 之后的 MappedMutexGuard：error[E0308]: mismatched types，「expected &mut MutexGuard<'_, RawMutex, _>, found &mut MappedMutexGuard<'_, _, bool>」。
• FairMutex 的守卫：同样 E0308，「found &mut MutexGuard<'_, RawFairMutex, bool>」。FairMutex 也不能配 Condvar。
• 反向：先在 MutexGuard 上 wait_while，返回后再 MutexGuard::map，编译运行正常。

映射守卫被拒绝是合理的：wait 要临时放开再重新加锁，而映射守卫拿不回原来的 Mutex（lock_api 的设计，外壳与守卫）。

wait_while 与谓词

parking_lot 承诺没有虚假唤醒，文档示例因此用 if 而不是 while（82-88 行）。这在示例里成立（只有一个通知者、只通知一次），一般情况下仍然要在锁里重新检查条件：

• notify_all 之后，等待者是一个接一个拿到锁的，排在前面的可能已经把条件改回去了；
• 被叫醒和拿到锁之间有插队窗口（TOKEN_NORMAL 时 mutex.lock() 要重新抢），别的线程可能先进来改掉条件。

所以照样用 wait_while(&mut guard, |s| !s.ready)，或者自己写 while。和 rayon 的 LockLatch（std Mutex<bool> + Condvar，while !*guard { guard = v.wait(guard) }）写法一样。

wait_while_until_internal：

let mut result = WaitTimeoutResult(false);

while !result.timed_out() && condition(mutex_guard.deref_mut()) {
    result =
        self.wait_until_internal(unsafe { MutexGuard::mutex(mutex_guard).raw() }, timeout);
}

result

逐行看超时分支：某次 wait_until_internal 返回 timed_out() == true 时，循环条件的第一项为假，不再调用 condition 就退出，返回「超时」。但从超时到重新拿到锁之间，条件可能已经被别人满足了。调用方看到「超时」，却不知道此刻条件其实成立。0.13 的 CHANGELOG 有一条「Fixed conditional Condvar waits reporting a timeout without rechecking the predicate. (#533)」，见 去掉的与修正的。在 0.12.5 上，wait_while_for 返回超时后，自己再检查一次条件。

实验 E6：notify_all 的惊群

断言：parking_lot 的 requeue 让 notify_all 之后同一时刻只有一个等待者醒来，避免「全醒、抢锁、大多数再睡」；std（macOS pthread）会把所有等待者都叫醒。

方法（exp-parking_lot/condvar-herd，不开 deadlock_detection）：N 个线程在同一个 condvar 上等，每个醒来拿到锁后持锁忙等 20 µs 再放开。主线程两种通知方式：持锁时 notify_all（hold）、放锁后 notify_all（release）。记录 notify_all 的返回值、从通知到最后一个等待者做完的总时间（total）、从通知到第一个等待者拿到锁（first）、相邻两个持锁者之间的交接间隔中位数（gap）、整个进程在这段时间里的上下文切换次数（Mach task_info(TASK_EVENTS_INFO).csw；macOS 上 getrusage 的 ru_nvcsw 恒为 0）。中间对照：parking_lot 持锁时循环 N 次 notify_one（ploop）。N 取 1、4、16、64；每个配置程序内部重复 11 次取中位数，整组跑 3 遍。M2 Max（8P + 4E）。

结果（22:36–22:37，开始时 1 / 5 分钟负载 2.11 / 2.92，结束时 2.67 / 2.98）。每格是 3 遍的中位数（每遍本身是 11 次的中位数）；total 单位 µs（N=64 时为 ms），first、gap 单位 µs：

• 返回值：parking_lot 的 notify_all 在所有配置、所有重复里都等于 N，持锁时（RequeueAll，0 个叫醒 + N 个挪走）和放锁时（1 + N − 1）没有区别；ploop 的 N 次 notify_one 每次都返回 true。
• 总时间由 N × 20 µs 的持锁工作串行决定，两边都约等于 N × 31 µs，差别在 10% 以内。
• 差别在开头：从 notify_all 到第一个等待者拿到锁，N = 64 时 std 要 95–132 µs，parking_lot 18–29 µs；N = 16 时 std 23–53 µs，parking_lot 11–21 µs。std 一次叫醒全部线程，第一个拿到锁的线程要和其余线程一起被调度、一起抢 pthread mutex；parking_lot 只叫醒一个（持锁时要等通知者放锁才叫醒）。
• 惊群在本机并不明显：上下文切换次数 parking_lot 恰好约 2N（每个等待者醒一次、退出一次），std 只多 2N 的 5%–15%（N = 64 时 136 对 128），交接间隔也只比 parking_lot 多 1–2 µs。如果 64 个线程真的全部醒来、抢锁失败、再睡，切换次数应当接近翻倍。推测 macOS 的 pthread 条件变量在内核里对 broadcast 有类似 requeue 的处理，本卷没有 libpthread / 内核源码可查，未验证；Linux 上 std 的 futex 条件变量 notify_all 是计数器加一后 futex_wake_all（condvar/futex.rs:27-30），情况可能不同，本卷不在 Linux 上实验。
• 反向验证：N = 1 时两边的切换次数都是 2，总时间同量级（27–35 µs），说明这个测法本身不偏向任何一边。ploop（N 次 notify_one）与一次 notify_all 在所有指标上都分不开：持锁时每次 notify_one 都是 RequeueOne，效果等于逐个挪。

E6 的结论比规划时的预期弱：parking_lot 的 requeue 在本机带来的是「第一个等待者更早拿到锁」和略少的切换，总吞吐几乎不变。

小结

• Condvar 是一个 AtomicPtr<RawMutex>，记当前配对的 mutex；队列在停车场，key 是 condvar 地址。8 字节、const fn new、无 Drop。
• wait：在桶锁里登记 mutex（另一把就 panic）→ 入队 → before_sleep 放开 mutex → 睡；醒来拿到 TOKEN_HANDOFF 就已持有 mutex，否则自己 lock()。
• notify_all 用 unpark_requeue 把等待者整体挪到 mutex 队列：mutex 被锁时 RequeueAll，一个都不叫醒；否则叫醒一个、挪走其余。返回值是受影响的线程数，不是醒来的线程数；notify_one 的 true 也可能只是挪了一个。
• 只接受 parking_lot::MutexGuard：映射守卫、FairMutex 的守卫都是编译错误。
• 两把 mutex 同时用会在第二个 wait 调用时立即 panic；parking_lot 允许在没有等待者后换锁，本机 std 不允许（E15）。
• 没有虚假唤醒，但谓词仍要在锁里检查；0.12.5 的 wait_while_* 超时时不复查谓词。
• E6（本机）：notify_all 返回值恒为 N；总时间由持锁工作串行决定、两边相近；N = 64 时第一个等待者拿到锁 parking_lot 约 19–27 µs、std 约 103–132 µs；上下文切换 parking_lot 约 2N、std 多 5%–15%。