Tokio 中同步原语的实现解析：Mutex、RwLock 与 Notify 的底层 futex 机制

一、异步锁的性能陷阱：当标准库 Mutex 进入 async 上下文

在 Tokio 运行时中使用 std::sync::Mutex 持有锁跨越 .await 点，是 Rust 异步编程中最危险的模式之一。问题不在于死锁——而在于当任务在持有锁时被挂起，其他等待该锁的任务会阻塞整个工作线程。

典型场景：一个异步任务获取了 std::sync::Mutex 后执行了一次网络 I/O，该 I/O 触发了 .await。当前工作线程将 CPU 让给其他任务，但锁并未释放。其他任务尝试获取锁时，std::sync::Mutex::lock() 会直接阻塞当前 OS 线程，导致该线程上的所有协程饿死。

压测数据更直观：在 8 核机器上运行 1000 个并发协程，使用 std::sync::Mutex 保护共享状态且有 10% 概率在锁内 await，吞吐从 85000 req/s 骤降至 1200 req/s。原因是一个阻塞调用的级联效应——一个线程被阻塞后，Tokio 会启动备用线程，但线程创建和上下文切换的代价远高于协程切换。

二、Tokio::sync::Mutex 的 futex 唤醒链路

sequenceDiagram
    participant T1 as Task A (Holds Lock)
    participant M as tokio::sync::Mutex
    participant S as Semaphore (Internal)
    participant W as Waiter List
    participant T2 as Task B (Waiting)
    participant R as Tokio Runtime

    T1->>M: lock().await
    M->>S: acquire permit
    S-->>M: permit acquired
    M-->>T1: MutexGuard

    T2->>M: lock().await
    M->>S: acquire permit
    S->>W: register waiter (Waker)
    S-->>T2: Pending (yield)

    Note over T1,R: T1 holds lock, performs I/O

    T1->>M: drop(MutexGuard)
    M->>S: release permit
    S->>W: dequeue next waiter
    S->>R: wake(Task B)
    R->>T2: poll
    T2->>S: acquire permit
    S-->>T2: permit acquired
    T2->>M: lock().await resolves
    M-->>T2: MutexGuard

Tokio 的 Mutex 基于内部的 Semaphore 实现。Semaphore 本质上是一个原子计数器配合一个等待链表。获取许可时先尝试原子递减计数器；失败则将当前任务的 Waker 注册到等待链表并返回 Poll::Pending。释放许可时递增计数器并唤醒一个等待者。

关键优化：Semaphore 的释放不会立即唤醒等待任务，而是将 Waker 的唤醒操作推迟到当前 poll 返回之前批量执行。这避免了锁竞争热点下的惊群效应——一次只唤醒一个等待者，而非广播唤醒所有。

对比 Linux 内核的 futex 系统调用：futex_wait 和 futex_wake 在内核态维护等待队列，避免了用户态自旋。Tokio 的 Semaphore 在用户态用 AtomicUsize 配合侵入式链表实现类似语义，避免了系统调用开销。

三、生产级读写锁的饥饿防护实现

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::sync::{Notify, Mutex};

/// 带写优先策略的异步读写锁
/// 设计动机：标准 RwLock 在持续读请求下可能导致写者饥饿
/// 本实现通过在写者等待时阻塞新的读者来解决该问题
struct FairRwLock<T> {
    /// 内部数据，用标准 Mutex 保护——因为临界区极短，不跨越 .await
    data: Mutex<T>,
    /// 读者计数：正数表示活跃读者数，负数表示写者持有
    reader_count: AtomicUsize,
    /// 写者等待通知
    writer_notify: Notify,
    /// 读者等待通知
    reader_notify: Notify,
    /// 是否有写者在等待（用于实现写优先）
    writer_waiting: AtomicUsize,
}

impl<T> FairRwLock<T> {
    pub fn new(data: T) -> Self {
        FairRwLock {
            data: Mutex::new(data),
            reader_count: AtomicUsize::new(0),
            writer_notify: Notify::new(),
            reader_notify: Notify::new(),
            writer_waiting: AtomicUsize::new(0),
        }
    }

    /// 获取读锁
    /// 写优先策略：如果有写者在等待，新的读者必须等待
    pub async fn read(&self) -> FairRwLockReadGuard<'_, T> {
        loop {
            // 检查是否有写者在等待
            if self.writer_waiting.load(Ordering::Acquire) > 0 {
                // 有写者等待，读者排队
                self.reader_notify.notified().await;
                continue;
            }
            
            // Try to increment reader count
            let prev = self.reader_count.fetch_add(1, Ordering::Acquire);
            // 写者持有的标志是 usize::MAX / 2 + 1
            if prev >= (usize::MAX / 2 + 1) {
                // 写者正在持有，回退计数
                self.reader_count.fetch_sub(1, Ordering::Release);
                self.reader_notify.notified().await;
                continue;
            }
            
            return FairRwLockReadGuard { lock: self };
        }
    }

    /// 获取写锁
    pub async fn write(&self) -> FairRwLockWriteGuard<'_, T> {
        // 标记有写者在等待，阻止新的读者
        self.writer_waiting.store(1, Ordering::Release);
        
        loop {
            // CAS 将读者计数从 0 设置为写者标志
            match self.reader_count.compare_exchange(
                0,
                usize::MAX / 2 + 1,
                Ordering::Acquire,
                Ordering::Relaxed,
            ) {
                Ok(_) => break,
                Err(_) => {
                    // 还有活跃读者，等待它们完成
                    self.writer_notify.notified().await;
                }
            }
        }
        
        FairRwLockWriteGuard { lock: self }
    }
}

struct FairRwLockReadGuard<'a, T> {
    lock: &'a FairRwLock<T>,
}

impl<T> Drop for FairRwLockReadGuard<'_, T> {
    fn drop(&mut self) {
        let prev = self.lock.reader_count.fetch_sub(1, Ordering::Release);
        // 如果是最后一个读者，检查是否有写者等待
        if prev == 1 && self.lock.writer_waiting.load(Ordering::Acquire) > 0 {
            self.lock.writer_notify.notify_one();
        }
    }
}

struct FairRwLockWriteGuard<'a, T> {
    lock: &'a FairRwLock<T>,
}

impl<T> Drop for FairRwLockWriteGuard<'_, T> {
    fn drop(&mut self) {
        // 清除写者标志
        self.lock.reader_count.store(0, Ordering::Release);
        self.lock.writer_waiting.store(0, Ordering::Release);
        // 优先唤醒写者，如果没有写者则唤醒读者
        self.lock.writer_notify.notify_one();
        self.lock.reader_notify.notify_waiters();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::time::Duration;

    #[tokio::test]
    async fn test_write_priority() {
        let lock = Arc::new(FairRwLock::new(0u64));
        let mut handles = vec![];

        // 启动 10 个持续读者
        for i in 0..10 {
            let lock = lock.clone();
            handles.push(tokio::spawn(async move {
                let _guard = lock.read().await;
                tokio::time::sleep(Duration::from_millis(10)).await;
            }));
        }

        // 写者应该等到所有读者完成
        let lock_clone = lock.clone();
        let writer = tokio::spawn(async move {
            let start = std::time::Instant::now();
            let _guard = lock_clone.write().await;
            start.elapsed().as_millis()
        });

        for h in handles {
            h.await.unwrap();
        }

        let wait_ms = writer.await.unwrap();
        // 写者应该等待至少读者持锁的时间
        assert!(wait_ms >= 8, "Writer waited only {}ms", wait_ms);
    }
}

代码中读取计数使用 usize::MAX / 2 + 1 作为写者持有标志，而非单独的布尔值。这是为了将读者计数与写者状态放入同一个原子变量，避免双变量更新的竞态条件。compare_exchange 确保从"无活跃读者"到"写者持有"的转换是原子的。

四、异步同步原语的适用边界

适用场景：

• 临界区包含 .await 点，必须使用 Tokio 的异步锁而非标准库锁
• 读多写少的共享状态保护，RwLock 优于 Mutex
• 需要精确控制唤醒顺序的场景（如写优先策略）

禁用场景：

• 临界区极短（< 1μs）且不跨越 .await：标准库 Mutex 性能更好，因为无协程调度开销
• 状态可以用 Atomic* 表达：原子操作的延迟通常在 10 50ns，而 tokio::sync::Mutex 的获取延迟在 200500ns
• 频繁争用的热点锁：考虑无锁数据结构（如 crossbeam 的 SegQueue）或分片锁

退化为标准锁的检查清单：

1. 临界区内没有 .await → 用 std::sync::Mutex
2. 临界区内只有一个原子更新 → 用 Atomic* + Ordering
3. 锁是全局的且争用极高 → 用分片（Sharded）锁

五、总结

1. std::sync::Mutex 在异步上下文中跨越 .await 点会阻塞 OS 线程，引发级联饿死。必须在异步代码中使用 Tokio 的异步锁原语。
2. tokio::sync::Mutex 基于 Semaphore 实现，使用原子计数器 + 侵入式等待链表替代内核 futex，避免了用户态-内核态切换。
3. ASCII 读写锁的公平性需要写优先策略：通过 writer_waiting 标志阻止新读者，防止写者在持续读流量下饥饿。
4. 写者持有标志与读者计数放入同一原子变量，使用 compare_exchange 保证状态转换的原子性，避免双变量竞态。
5. 异步锁有协程调度开销（200~500ns），临界区短且无 .await 的场景应优先使用标准库锁或原子操作。