读、写与写者优先

主线位置：主线 1 的读写锁版本。写者 lock_exclusive_slow：第一步在 lock_common 里抢 WRITER_BIT（抢不到就在主队列停车），第二步 wait_for_readers（读者没走光就在 addr + 1 停车）；最后一个读者 unlock_shared 看到 WRITER_PARKED_BIT 就叫醒它。写者解锁时 wake_parked_threads 用 unpark_filter 从主队列挑出一批读者或一个写者。

sequenceDiagram
  participant R1 as 读者 R1（持有中）
  participant W as 写者 W
  participant R2 as 新读者 R2
  participant S as 状态字
  W->>S: lock_common：CAS 置 WRITER_BIT（R1 还在）
  W->>S: wait_for_readers：自旋 10 轮后置 WRITER_PARKED_BIT
  W->>W: park(addr+1)
  R2->>S: try_lock_shared_fast：看到 WRITER_BIT，失败
  R2->>S: lock_common：置 PARKED_BIT，park(addr)
  R1->>S: unlock_shared：fetch_sub，旧值 = ONE_READER + WRITER_PARKED_BIT
  R1->>W: unlock_shared_slow：unpark_one(addr+1)，清 WRITER_PARKED_BIT
  W->>W: 醒来，读者数 = 0，持有写锁
  W->>S: unlock_exclusive：快路径失败（PARKED_BIT）
  W->>R2: wake_parked_threads：unpark_filter 叫醒队头读者

共同的骨架：lock_common

读、写、可升级三种锁的慢路径都调 lock_common，只是传进去的 try_lock 闭包、ParkToken 和 validate_flags 不同：

fn lock_common(
    &self,
    timeout: Option<Instant>,
    token: ParkToken,
    mut try_lock: impl FnMut(&mut usize) -> bool,
    validate_flags: usize,
) -> bool {
    let mut spinwait = SpinWait::new();
    let mut state = self.state.load(Ordering::Relaxed);
    loop {
        // Attempt to grab the lock
        if try_lock(&mut state) {
            return true;
        }
        // If there are no parked threads, try spinning a few times.
        if state & (PARKED_BIT | WRITER_PARKED_BIT) == 0 && spinwait.spin() {
            state = self.state.load(Ordering::Relaxed);
            continue;
        }
        // Set the parked bit
        // ...（与 RawMutex::lock_slow 第 3 步相同）
        let validate = || {
            let state = self.state.load(Ordering::Relaxed);
            state & PARKED_BIT != 0 && (state & validate_flags != 0)
        };
        // park(addr, validate, .., token, timeout)：
        //   TOKEN_HANDOFF -> return true；其余 -> spinwait.reset()，重来
    }
}

结构和 RawMutex::lock_slow 一样：先抢，没人停车（两个停车位都是 0）才自旋，再置 PARKED_BIT、停车。validate 在桶锁里确认两件事：PARKED_BIT 还在，且挡住自己的那些位（validate_flags）还在；挡路的东西已经没了就不睡，回去重抢。

读锁和可升级读锁的 try_lock 闭包里还有一个内层循环：CAS 加读者计数失败（说明别的读者同时在改计数）就 spin_no_yield() 再试（714-718 行），注释说是为了在读者计数「高度争用」时给别的线程留出时间。所有读者都对同一个字做 CAS，这就是 crossbeam 卷 ShardedLock 页里 parking_lot 只读吞吐上不去的原因所在：那里的推测「只读时所有线程同时改同一个计数」在代码上成立——每次 read() 至少一次对共享状态字的 CAS，失败就退避重试。至于这是否就是「全写比只读快」的全部原因，代码只能说明读者之间存在这种争用，没有量过 CAS 失败次数。

写锁：两步

lock_exclusive_slow：

// Step 1: grab exclusive ownership of WRITER_BIT
let timed_out = !self.lock_common(timeout, TOKEN_EXCLUSIVE, try_lock, WRITER_BIT | UPGRADABLE_BIT);
if timed_out {
    return false;
}

// Step 2: wait for all remaining readers to exit the lock.
self.wait_for_readers(timeout, 0)

第一步的 try_lock 注释写着「Grab WRITER_BIT if it isn't set, even if there are parked threads」：和 Mutex 一样允许插队。

第二步 wait_for_readers：读者数不为 0 就先 SpinWait::spin，转完 10 轮还有读者，就 CAS 置 WRITER_PARKED_BIT，在 addr + 1 上停车，validate 要求「读者数不为 0 且 WRITER_PARKED_BIT 仍在」。醒来（或 validate 失败）后总是重读状态再判断，注释解释了为什么醒来不能直接认为读者已经走光：之前某个写者超时回滚时，可能有读者趁机溜了进来。

另一侧是最后一个读者。unlock_shared：

unsafe fn unlock_shared(&self) {
    self.deadlock_release();
    let state = if have_elision() {
        self.state.elision_fetch_sub_release(ONE_READER)
    } else {
        self.state.fetch_sub(ONE_READER, Ordering::Release)
    };
    if state & (READERS_MASK | WRITER_PARKED_BIT) == (ONE_READER | WRITER_PARKED_BIT) {
        self.unlock_shared_slow();
    }
}

一次 fetch_sub；只有「我是最后一个读者，且有写者在第二队列睡着」时才进慢路径，unpark_one(addr + 1)，在回调里清 WRITER_PARKED_BIT（724-742 行）。读锁的解锁从不碰主队列：主队列里等着的人由写者或可升级读者解锁时负责叫醒，wake_parked_threads 开头的注释写的就是这个约定（「unlock_shared does not call wake_parked_threads」）。

读锁：看见 WRITER_BIT 就排队

try_lock_shared_fast：

fn try_lock_shared_fast(&self, recursive: bool) -> bool {
    let state = self.state.load(Ordering::Relaxed);

    // We can't allow grabbing a shared lock if there is a writer, even if
    // the writer is still waiting for the remaining readers to exit.
    if state & WRITER_BIT != 0 {
        // To allow recursive locks, we make an exception and allow readers
        // to skip ahead of a pending writer to avoid deadlocking, at the
        // cost of breaking the fairness guarantees.
        if !recursive || state & READERS_MASK == 0 {
            return false;
        }
    }
    // ... checked_add(ONE_READER) 后一次 compare_exchange_weak(Acquire)
}

「even if the writer is still waiting for the remaining readers」：写者只要占住了 WRITER_BIT，新读者一律去排队，哪怕此刻锁里全是读者。这就是写者不饿死的机制：读者流再密，也只能把写者挡在第二步，挡不住第一步；第一步一完成，读者流就被截断，剩下的读者一走光写者就进去。

注意：只有已经抢到 WRITER_BIT 的写者才挡读者。在主队列里排队的写者（例如前面还有一个写者持有锁）不会改变读者的判断；但那种情况下 WRITER_BIT 本来就置着（被前一个写者持有），读者同样进不来。

try_lock_shared（不带 _fast）失败就返回 false，不停车，规则相同。

写者解锁：一批读者，或一个写者

unlock_exclusive_slow 把活交给 wake_parked_threads(0, callback)，后者调 unpark_filter，过滤函数（见 状态字与两条队列里的代码）从主队列头开始：

• 遇到读者：叫醒，累加 ONE_READER，继续往后看；
• 遇到写者或可升级读者：如果累加器里还没有可升级读者，就叫醒它、加上它的 token，并且——如果它是写者——下一个线程起返回 Stop；如果累加器里已经有一个可升级读者，就 Skip 它（留在队列里）。

结果是「从队头起所有读者 + 至多一个可升级读者」，或者「队头的一个写者」。注意 Stop 只在累加器里已经有 WRITER_BIT 时才返回：如果队头是读者、后面跟着写者，写者会被叫醒（累加器从 0 加到 ONE_READER，再遇到写者时 UPGRADABLE_BIT 为 0，不 Skip，于是 Unpark），然后才 Stop。所以「一批读者 + 紧随其后的那个写者」会被一起叫醒。不公平分支（TOKEN_NORMAL）里它们各自回到 lock_common 开头去抢，谁先抢到不确定：读者先进来，写者就在第二步等这批读者走光；写者先占了 WRITER_BIT，这批读者又得重新排队。两种情况下读阶段和写阶段都还是按排队顺序交替，只是边界不那么整齐。公平分支里累加器原样写进状态字（读者数 + WRITER_BIT），被叫醒的读者直接持有读锁，写者直接进入「已占 WRITER_BIT、等读者退出」的状态，并在 lock_common 返回后进入 wait_for_readers。

回调里决定公平与否的条件和 Mutex 一样：force_fair || be_fair（unlock_exclusive_fair 传 true，桶上的计时器给 be_fair）。读锁的 unlock_shared_fair 直接就是 unlock_shared（154-159 行，注释「Shared unlocking is always fair in this implementation」）。

超时的写者要回滚

带超时的写锁（try_write_for）可能在第二步超时：WRITER_BIT 已经占了，读者还没走光。这时不能简单返回 false，否则 WRITER_BIT 一直挡着新读者。1041-1063 行：

ParkResult::TimedOut => {
    // We need to release WRITER_BIT and revert back to
    // our previous value. We also wake up any threads that
    // might be waiting on WRITER_BIT.
    let state = self
        .state
        .fetch_add(prev_value.wrapping_sub(WRITER_BIT), Ordering::Relaxed);
    if state & PARKED_BIT != 0 {
        // ... wake_parked_threads(prev_value, ..)
    }
    return false;
}

prev_value 对普通写者是 0（直接去掉 WRITER_BIT），对正在 upgrade 的可升级读者是 ONE_READER | UPGRADABLE_BIT（变回可升级读）。回滚后叫醒主队列里被这个 WRITER_BIT 挡住的读者。0.12.4 的发布说明里有两条 RwLock 修复与这一带有关，原文是「Fix parked upgraders potentially not being woken up after a write lock」「Fix clearing PARKED_WRITER_BIT after a timeout」；0.12.5 还有一条「Fixed RawRwLock::bump_*() not releasing lock when there are multiple readers (#471)」。

read_recursive：故意打破的那条规矩

同一个线程拿着读锁再 read()，如果这期间恰好有写者占了 WRITER_BIT，第二次 read() 就排到写者后面，而写者在等这个线程的第一个读锁释放：死锁，没有报错。文档（rwlock.rs:17-21）明说了：「attempts to recursively acquire a read lock within a single thread may result in a deadlock」。

read_recursive() 传 recursive = true：WRITER_BIT 置着但读者数不为 0 时仍然放行（上面代码 517 行）。读者数不为 0 说明写者还在第二步等，此时多进一个读者不会让任何已持有的东西失效，只是让写者多等一会儿——注释的说法是「at the cost of breaking the fairness guarantees」。读者数为 0 时（写者真的持有写锁）read_recursive 也得排队，这种情况下当前线程不可能持有读锁，不会死锁。

lock_api 里对应的 trait 是 RawRwLockRecursive，文档说用它「can cause writers to starve since readers no longer block if a writer is waiting」（lock_api/src/rwlock.rs:187-193）。E4 的反向验证组量的就是这件事。0.13 删掉了 read_recursive，换成单独的 RecursiveRwLock 类型，见 去掉的与修正的。

实验 E5：同线程重复读

方法（exp-parking_lot/recursive-read，不开 deadlock_detection）：线程 A read()，睡 100 ms；期间线程 B（30 ms 时）申请写锁；A 醒来再 read()。2 秒看门狗，超时打印并以退出码 1 结束。四种模式各跑 3 次：

std 的文档对此的说法是优先策略「依赖底层操作系统的实现」，并给了同样形状的「Potential deadlock example」（sync/poison/rwlock.rs:22-38）；本机 std 用的 queue.rs 在模块文档里写明自己「writer-preferring」，所以也卡死（和 std RwLock 比）。tokio watch::Receiver::borrow 文档提醒的「同一线程上可能死锁」就是这个坑（tokio 卷 watch），详细改法见坑 recursive-read。

小结

• 三种锁的慢路径共用 lock_common：先抢，没人停车才自旋，置 PARKED_BIT，在主队列（addr）停车，validate 确认挡路的位还在。
• 写锁两步：抢 WRITER_BIT（有读者也能抢到）→ wait_for_readers，转完还有读者就在 addr + 1 停车；最后一个读者 unlock_shared 负责叫醒它。读锁解锁从不碰主队列。
• 读者看到 WRITER_BIT 一律排队，写者不会被读者流饿死。
• 写者解锁用 unpark_filter 从主队列头叫醒「所有读者 + 至多一个写者或可升级读者」；公平时把 token 累加值原样写进状态字。
• 超时的写者回滚 WRITER_BIT 并叫醒被它挡住的人。
• read_recursive 在读者数不为 0 时越过等待中的写者，避免同线程重复读死锁，代价是可能饿写者；E5：read() 3/3 死锁、read_recursive() 3/3 正常、本机 std 也死锁。