第14章 Pool 内部实现：空闲队列、公平性、驱逐策略

空闲队列用的是无锁环形队列，信号量是对 tokio 与 async-std 两套实现的统一包装。PoolInner 的十个字段，每一个都在回答同一个问题：什么时候可以信任自己记下的状态。

14.1 问题引入：Pool 内部的五件事

第 13 章讲完 Pool 的外部 API——用户能调的所有方法。但每次 pool.acquire().await? 底下究竟发生了什么？Pool 内部至少在做五件事：

1. 管理一组连接：有空闲的（idle）、有借出去的（in-use），两类加起来不能超过 max_connections。
2. 协调并发：100 个 async task 同时 acquire，谁先拿到、谁等、等多久——信号量（Semaphore）解决。
3. 跟踪连接状态：一条连接从"空闲队列里"到"被借出用"到"归还回来"到"过期被关"——状态机。
4. 后台维护：空闲太久的连接要关、太老的连接要退休、min_connections 要补齐——后台 task 做。
5. 安全兜底：网络断、panic、cancel——每种异常路径都不能让 size 计数泄漏、不能让连接凭空消失。

sqlx Pool 内部用几个精心选择的并发原语把这五件事串起来：ArrayQueue（无锁队列）、AsyncSemaphore（公平异步信号量）、RAII guard（DecrementSizeGuard）、原子计数（AtomicU32 / AtomicUsize / AtomicBool）、event_listener（close 事件）。本章把它们逐一拆开。

14.2 PoolInner<DB> 的 10 个字段

sqlx-core/src/pool/inner.rs:26-37：

pub(crate) struct PoolInner<DB: Database> {
    pub(super) connect_options: RwLock<Arc<<DB::Connection as Connection>::Options>>,
    pub(super) idle_conns: ArrayQueue<Idle<DB>>,
    pub(super) semaphore: AsyncSemaphore,
    pub(super) size: AtomicU32,
    pub(super) num_idle: AtomicUsize,
    is_closed: AtomicBool,
    pub(super) on_closed: event_listener::Event,
    pub(super) options: PoolOptions<DB>,
    pub(crate) acquire_time_level: Option<Level>,
    pub(crate) acquire_slow_level: Option<Level>,
}

10 个字段（最后两个 Level 是同一类日志配置的两面）：

状态存储：

• idle_conns：空闲连接的队列。
• size：当前 pool 持有的连接总数（atomic）。
• num_idle：空闲连接数（atomic）。

并发原语：

• semaphore：控制 max_connections 并发。
• on_closed：close 事件的广播器。

配置：

• connect_options：可热更新的连接参数。
• options：Pool 自身的配置（max_connections 等）。
• 两个 Level：日志粒度。

标志：

• is_closed：atomic bool，标记 pool 已关闭。

PoolInner 这个结构体本身很小——真正随 max_connections 增长的内容都在堆上（ArrayQueue 一次性分配 capacity 个槽、信号量维护自己的等待者链表），结构体里只有定长的指针、原子量、锁和配置。

14.3 idle_conns: ArrayQueue 的无锁队列

sqlx-core/src/pool/inner.rs:28 的 idle_conns: ArrayQueue<Idle<DB>>——来自 crossbeam_queue crate的无锁环形队列。

为什么选 ArrayQueue 而不是 Mutex<VecDeque>？

ArrayQueue 是 bounded MPMC queue（多生产者多消费者、有界）：

• 无锁：push 和 pop 都通过 atomic CAS 完成，不走 Mutex。
• 固定容量：ArrayQueue::new(capacity) 分配一次，之后不扩容。
• 零分配：push / pop 在已分配 buffer 里操作。

Pool 的 idle_conns 容量设为 max_connections（sqlx-core/src/pool/inner.rs:44 算出 capacity、57 行 ArrayQueue::new(capacity)）——最多同时有这么多空闲连接。

为什么这个选择对 Pool 重要——Mutex<VecDeque> 下多线程同时 push/pop 会串行化到 Mutex 临界区；ArrayQueue 基于 CAS 原子操作，不存在"持锁的线程被挂起、其他线程全部干等"的情况。本专栏没有找到 sqlx 针对这两种方案的对比测量，所以只做定性判断。从 git 历史看，ArrayQueue 是 2020-01-11 的提交 95fac72a「Use crossbeam queues to achieve fairness in the pool」引入的（首个包含它的版本是 v0.1.4），替换掉的是原先的 async-std channel，而不是 Mutex<VecDeque>；那次提交把公平排队交给另一条等待者队列（SegQueue），后来 v0.5.6 的 #1320（「reimplement pool internals with futures-intrusive」）才改成由信号量负责排队。代价是"有界 + 满了直接 panic"这个约束（Pool 通过 size counter 在外部已保证不会满、约束天然满足）。

ArrayQueue 的代价：有界（但 Pool 恰好天然有界、不是问题）、push 满了返回 Err（release 源码 sqlx-core/src/pool/inner.rs:205-207 对应处理，满了直接 panic!("BUG: connection queue overflow in release()")——因为 size counter 理论上保证不会满）。

14.3.1 Idle<DB> 数据

sqlx-core/src/pool/connection.rs:27-35：

pub(super) struct Live<DB: Database> {
    pub(super) raw: DB::Connection,
    pub(super) created_at: Instant,
}

pub(super) struct Idle<DB: Database> {
    pub(super) live: Live<DB>,
    pub(super) idle_since: Instant,
}

Live：借出中的连接 + 建立时间。 Idle：Live + 进入 idle queue 的时间戳。

created_at 用于 max_lifetime 检查（连接活太久就退休）；idle_since 用于 idle_timeout 检查（空闲太久就关掉）。两个 Instant 一起让 maintenance task 能做精确的年龄判断。

14.4 AsyncSemaphore: 公平异步信号量

sqlx-core/src/pool/inner.rs:29 的 semaphore: AsyncSemaphore——来自 sqlx-core/src/sync.rs。它是对 tokio::sync::Semaphore 或 futures_intrusive::sync::Semaphore（async-std 侧）的统一包装（runtime 多后端支持，第 2 章 §2.5.1 讨论过）。

Semaphore 的作用：限制 acquire 的并发——最多 max_connections 个 permit。超过的 acquire 排队等。

AsyncSemaphore::new(fair, capacity) 的两个参数：

• capacity: usize——permit 总数 = max_connections。
• fair: bool——排队语义：

14.4.1 fair 语义：先来先得 vs 允许插队

PoolOptions::__fair 的文档注释（sqlx-core/src/pool/options.rs:324-336）把两种语义说得很清楚：

If set to true, calls to acquire() are fair and connections are issued in first-come-first-serve order. If false, "drive-by" tasks may steal idle connections ahead of tasks that have been waiting.

• fair = true（默认）：FIFO——先 acquire 的先拿到 permit。
• fair = false：路过的任务可以"顺手牵羊"，抢在排队者前面拿走空闲连接。

哪个更好？源码注释直接给了结论，而且标明了依据——sqlx-bench/benches/pg_pool：

According to sqlx-bench/benches/pg_pool this may slightly increase time to acquire() at low pool contention but at very high contention it helps avoid tasks at the head of the waiter queue getting repeatedly preempted by these "drive-by" tasks and tasks further back in the queue timing out because the queue isn't moving.

也就是说：低争抢时公平排队反而略慢一点；但高争抢时它能避免队头被反复插队、队尾大面积超时。所以默认取 true。

在 Tokio 下这一点可以直接验证：SQLite 内存库、max_connections(1)，先占住唯一的连接，再每隔 10ms spawn 一个任务去 acquire，共 10 个；放开连接后，它们拿到连接的顺序实测正是 [0, 1, 2, …, 9]，而且排队期间主任务反复调用 try_acquire() 一次也没抢到——Tokio 信号量把释放出来的 permit 直接分给队头的等待者，"路过"的任务没有机会插队。

两条重要的限定：

1. 这个 setter 是 #[doc(hidden)] 的 __fair，注释写明"目前只为 benchmark 暴露；多数情况下 fair = true 是更优选项"——不要在业务代码里调它。
2. 在 Tokio 运行时下它不起作用——AsyncSemaphore::new 里只有一句 debug_assert!(fair, "Tokio only has fair permits")，然后照常构造 tokio::sync::Semaphore。也就是说 release 构建下 fair = false 被静默忽略，debug 构建下则直接 panic（实测 SqlitePoolOptions::new().__fair(false).connect_with(..) 报 panicked at .../sqlx-core-0.8.6/src/pool/inner.rs:58:24: Tokio only has fair permits，#[track_caller] 把位置指回了调用方 new_arc）。fair = false 只在 async-std 那条路（futures_intrusive::sync::Semaphore::new(fair, permits)）才真正传下去。

14.4.2 AsyncSemaphore 对 Tokio Semaphore 的包装

sqlx-core/src/sync.rs 全文只有 137 行，做的事就是把两家的信号量塞进同一个壳里——注意 inner 是两个互斥的 cfg 字段，不是 Arc：

pub struct AsyncSemaphore {
    #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
    inner: futures_intrusive::sync::Semaphore,

    #[cfg(feature = "_rt-tokio")]
    inner: tokio::sync::Semaphore,
}

impl AsyncSemaphore {
    #[track_caller]
    pub fn new(fair: bool, permits: usize) -> Self {
        if cfg!(not(any(feature = "_rt-async-std", feature = "_rt-tokio"))) {
            crate::rt::missing_rt((fair, permits));
        }

        AsyncSemaphore {
            #[cfg(all(feature = "_rt-async-std", not(feature = "_rt-tokio")))]
            inner: futures_intrusive::sync::Semaphore::new(fair, permits),
            #[cfg(feature = "_rt-tokio")]
            inner: {
                debug_assert!(fair, "Tokio only has fair permits");
                tokio::sync::Semaphore::new(permits)
            },
        }
    }
    // acquire / try_acquire / release / permits 同样按 cfg 分派
}

几处值得留意的细节：

• 两个 runtime 都没启用时不是编译错误，而是 crate::rt::missing_rt 的运行时 panic——配合 #[track_caller] 把出错位置指回调用方。
• release(n) 在 tokio 侧映射成 add_permits(n)、在 async-std 侧映射成 release(n)——名字不同、语义相同。
• acquire 在 tokio 侧用 acquire_many(permits)，源码注释还吐槽了一句 Tokio 的怪癖："tokio::sync::Semaphore 大部分地方用 usize 计 permit，唯独这个和 try_acquire_many() 用 u32"。
• 选 futures-intrusive 而不是 async-lock 的理由也写在字段注释里：async-lock 的信号量不能加任意数量的 permit、也不保证公平（issue #22、#23）；同时注释坦承 futures-intrusive 已无人维护、还有 MIRI 下的 unsound 疑虑（issue #1668），"我们在找替代品"。

14.5 连接状态机：Live / Idle / Floating

三个状态类型（sqlx-core/src/pool/connection.rs:27-41）：

pub(super) struct Live<DB: Database> {
    pub(super) raw: DB::Connection,
    pub(super) created_at: Instant,
}

pub(super) struct Idle<DB: Database> {
    pub(super) live: Live<DB>,
    pub(super) idle_since: Instant,
}

pub(super) struct Floating<DB: Database, C> {
    pub(super) inner: C,
    pub(super) guard: DecrementSizeGuard<DB>,
}

三种状态的含义：

• Live<DB>：连接在用中——raw connection + 建立时间。
• Idle<DB>：连接在 idle queue 里——Live + 入队时间。
• Floating<DB, C>：过渡态——连接被"捞出"idle queue 但还没成正式的 PoolConnection、或者 PoolConnection drop 后还没归还——内带 DecrementSizeGuard 做 RAII 兜底。

状态转换图：

stateDiagram-v2
    [*] --> Connecting: acquire 新建
    Connecting --> Floating_Live: connect 成功
    Floating_Live --> PoolConnection: reattach
    PoolConnection --> Floating_Live: drop → return_to_pool
    Floating_Live --> Idle: release pushes into idle queue
    Idle --> Floating_Idle: acquire pops from idle queue
    Floating_Idle --> Floating_Live: into_live
    Idle --> [*]: close beyond timeout
    Floating_Live --> [*]: close_on_drop

五个转换：

1. Connecting → Floating_Live：新建连接成功。
2. Floating_Live → PoolConnection：包装成用户可见的 PoolConnection。
3. PoolConnection → Floating_Live：drop 开始归还。
4. Floating_Live → Idle：push 进 idle queue。
5. Idle → Floating_Idle：acquire 弹出。

14.5.1 Floating 的 RAII 保证

Floating 的类型参数 C 是 Live 或 Idle 之一——它自己是"我正在处理这个连接，还没决定归宿"的过渡态。

关键在 guard: DecrementSizeGuard<DB>——只要 Floating 没有经 release（放回空闲队列）或 reattach（交给用户）把 guard 取消掉，它被丢弃时（包括 close() 之后，这正是关连接的正常收尾）DecrementSizeGuard 的 Drop 就会把 pool size 减 1、释放信号量 permit——保证连接计数不泄漏。

这是 Rust RAII 的经典用法——状态变迁本来可能出错（网络断、panic、await 取消），用 RAII guard 兜底让计数器自动修正。

14.6 acquire 的完整实现

sqlx-core/src/pool/inner.rs:242-318 的 acquire 主体（简化后）：

pub(super) async fn acquire(self: &Arc<Self>) -> Result<Floating<DB, Live<DB>>, Error> {
    if self.is_closed() { return Err(Error::PoolClosed); }

    let acquire_started_at = Instant::now();
    let deadline = acquire_started_at + self.options.acquire_timeout;

    let acquired = crate::rt::timeout(self.options.acquire_timeout, async {
        loop {
            let permit = self.acquire_permit().await?;

            // 1. 先尝试 idle queue
            let guard = match self.pop_idle(permit) {
                Ok(conn) => match check_idle_conn(conn, &self.options).await {
                    Ok(live) => return Ok(live),      // ✓ 空闲连接可用
                    Err(guard) => guard,                // ✗ 连接有问题，guard 待重建
                },
                Err(permit) => {
                    // idle queue 空，尝试建新连接
                    if let Ok(guard) = self.try_increment_size(permit) {
                        guard
                    } else {
                        // 既没空闲也不能增 size——重试
                        crate::rt::yield_now().await;
                        continue;
                    }
                }
            };

            // 2. 建新连接
            return self.connect(deadline, guard).await;
        }
    }).await.map_err(|_| Error::PoolTimedOut)??;

    Ok(acquired)
}

三阶段：

1. 获取 permit（受 semaphore 限制）。
2. 尝试 pop idle——有就用（先 check_idle_conn 验证）、没有就尝试 try_increment_size 决定是否能建新。
3. 建连接——调 connect(deadline, guard)，失败 backoff 重试。

整段包在 rt::timeout(acquire_timeout, ...) 里——超总时限返回 PoolTimedOut（实测：acquire_timeout(300ms)、唯一的连接被占住时，第二次 acquire 约 302ms 后返回 Err(PoolTimedOut)）。排队、ping、建连的耗时都算在这一个总时限里。

14.6.1 check_idle_conn：acquire 时的健康检查

sqlx-core/src/pool/inner.rs:465-501（省略了原注释）：

async fn check_idle_conn<DB: Database>(
    mut conn: Floating<DB, Idle<DB>>,
    options: &PoolOptions<DB>,
) -> Result<Floating<DB, Live<DB>>, DecrementSizeGuard<DB>> {
    if options.test_before_acquire {
        if let Err(error) = conn.ping().await {
            tracing::info!(%error, "ping on idle connection returned error");
            return Err(conn.close_hard().await);
        }
    }

    if let Some(test) = &options.before_acquire {
        let meta = conn.metadata();
        match test(&mut conn.live.raw, meta).await {
            Ok(false) => return Err(conn.close().await),
            Err(error) => {
                tracing::warn!(%error, "error from `before_acquire`");
                return Err(conn.close_hard().await);
            }
            Ok(true) => {}
        }
    }

    Ok(conn.into_live())
}

两道检查依次过：

1. test_before_acquire：ping 一次（默认开）。ping 不通说明对端已经挂断或链路断了，走 close_hard——不做优雅关闭，因为连接本来就已经坏了。
2. before_acquire 回调：用户自定义校验（第 13 章 §13.11.2）。回调返回 Ok(false) 是「这条连接我不要」，属于正常拒绝，走 close() 优雅关闭；回调自己报错才走 close_hard。

任一失败返回 Err(DecrementSizeGuard)——让上层拿 guard 去建新连接（guard 已经持有"size 已算"的承诺，不用再 try_increment）。

值得留意这里没有 max_lifetime 检查。连接寿命和空闲超时不由 acquire 路径管，而是交给后台的回收任务：spawn_maintenance_tasks 每隔 min(max_lifetime, idle_timeout) 扫一遍空闲队列，is_beyond_idle_timeout(&conn, ..) || is_beyond_max_lifetime(&conn, ..) 命中就关掉并补足 min_connections（sqlx-core/src/pool/inner.rs:545-548）。分工是有道理的：acquire 在请求的关键路径上，只做"这条连接现在还能用吗"这一件事；"这条连接是不是该退休了"是后台的活，多花几微秒也没人等。

14.7 PoolInner::release 归还

sqlx-core/src/pool/inner.rs:200-214：

pub(super) fn release(&self, floating: Floating<DB, Live<DB>>) {
    // `options.after_release` and other checks are in `PoolConnection::return_to_pool()`.

    let Floating { inner: idle, guard } = floating.into_idle();

    if self.idle_conns.push(idle).is_err() {
        panic!("BUG: connection queue overflow in release()");
    }

    // NOTE: we need to make sure we drop the permit *after* we push to the idle queue
    // don't decrease the size
    guard.release_permit();

    self.num_idle.fetch_add(1, Ordering::AcqRel);
}

注意源码里那两条注释：after_release 之类的检查不在这里、在 PoolConnection::return_to_pool()（§14.9）；以及 permit 必须在 push 进空闲队列之后才释放——否则被唤醒的 acquire 可能先跑、看到一个空队列。

四步：

1. floating.into_idle() 把 Floating<Live> 转成 Idle——记录 idle_since = Instant::now()。
2. push 进 idle queue——成功是不变量（size counter 保证不会满）。
3. guard.release_permit()——释放信号量 permit（让下一个 acquire 能拿）但不减 size（连接还在 pool 里、只是状态变 idle）。
4. num_idle += 1。

注意 release_permit 和 decrement_size 的区别：

• release_permit：归还信号量名额——让并发 limit 恢复。
• decrement_size：实际减少连接总数——只在 connection 真的 close 时发生。

这两者分离是 Pool 设计的精妙之处——信号量限制"并发"、size 限制"总数"。一条连接 idle 时占 size 不占信号量（等它被拿起时信号量才被占）。

14.8 try_increment_size + DecrementSizeGuard

sqlx-core/src/pool/inner.rs:219-240：

pub(super) fn try_increment_size<'a>(
    self: &'a Arc<Self>,
    permit: AsyncSemaphoreReleaser<'a>,
) -> Result<DecrementSizeGuard<DB>, AsyncSemaphoreReleaser<'a>> {
    let result = self.size.fetch_update(Ordering::AcqRel, Ordering::Acquire, |size| {
        if self.is_closed() { return None; }
        size.checked_add(1).filter(|size| size <= &self.options.max_connections)
    });

    match result {
        Ok(_) => Ok(DecrementSizeGuard::from_permit((*self).clone(), permit)),
        Err(_) => Err(permit),
    }
}

原子 CAS 循环（fetch_update）—— size 已达 max 或 pool 已关返回 None（CAS 失败）、否则成功 +1、返回 DecrementSizeGuard。

DecrementSizeGuard 只有两个字段（sqlx-core/src/pool/inner.rs:577-580：pool: Arc<PoolInner<DB>> 和 cancelled: bool），Drop 实现在 sqlx-core/src/pool/inner.rs:610-618：

impl<DB: Database> Drop for DecrementSizeGuard<DB> {
    fn drop(&mut self) {
        if !self.cancelled {
            self.pool.size.fetch_sub(1, Ordering::AcqRel);

            // and here we release the permit we got on construction
            self.pool.semaphore.release(1);
        }
    }
}

配套的两个方法也很短（sqlx-core/src/pool/inner.rs:597-607）：release_permit(self) 是"释放 permit 但不减 size"——它先 semaphore.release(1) 再 cancel()；cancel(mut self) 只是把 cancelled 置 true。构造侧的 from_permit 则调用 permit.disarm()——把 permit 的所有权从 releaser 手里"接管"过来，之后由 guard 负责释放。

如果 guard 没被 cancel()——drop 时自动减 size + 释放 permit。

谁会调 cancel？——成功把连接变成 PoolConnection 时（DecrementSizeGuard::cancel）——这时 guard 不需要兜底、pool 确实多了一条连接。

没调 cancel 的路径：connect 失败、after_connect 失败、任何"已增 size 但连接没成型"场景。自动减 size 让 size counter 不泄漏——Pool 的核心不变量。

这条 RAII 设计让建连接失败的各种错误路径都自动正确——不用程序员在每个 ? 后面手动 decrement。

14.9 PoolConnection::Drop 归还机制

用户代码里 let conn = pool.acquire().await?; 的 conn 最终被 drop 时会怎样？

sqlx-core/src/pool/connection.rs:199-211：

impl<DB: Database> Drop for PoolConnection<DB> {
    fn drop(&mut self) {
        if self.close_on_drop {
            crate::rt::spawn(self.take_and_close());
            return;
        }

        // 还需要 spawn task 来维持 min_connections
        if self.live.is_some() || self.pool.options.min_connections > 0 {
            crate::rt::spawn(self.return_to_pool());
        }
    }
}

Drop 里 spawn 一个异步 task ——return_to_pool。PoolConnection::return_to_pool（sqlx-core/src/pool/connection.rs:134-155）先在同步部分就把 live take 出来 float 好（源码注释指向 issue #1396：万一返回的 Future 被扔进一个正在死掉的 runtime、根本没被执行，Future 被丢弃时里面的 DecrementSizeGuard 照样会减 size、还 permit，池子的名额不会因此泄漏），然后异步部分调 Floating::return_to_pool（sqlx-core/src/pool/connection.rs:275-328），依次五道关：

1. pool 已关？→ close()，返回 false。
2. 超过 max_lifetime？→ close()，返回 false。
3. after_release 回调（如果配置了）：返回 Ok(false) 走 close()；回调自己报错走 close_hard()。
4. 无条件 ping() 一次——注意这一步和 test_before_acquire 无关，每次归还都做。源码注释说明了原因：如果某个 Executor 的 future/stream 在 .await 中途被 drop，连接很可能处在不一致状态，这时不应该放回池子；顺便也把事务回滚这类"时效性"操作 flush 掉。ping 失败就 close_hard()。
5. 全过关 → self.release()，也就是 §14.7 的 PoolInner::release：push 进 idle queue、释放 permit、num_idle += 1。

如果 return_to_pool 返回 false（连接没回池子），外层还会顺手调一次 pool.min_connections_maintenance(None) 把缺口补上。

注意是 spawn 而不是 block——drop 是同步函数，不能 await。spawn 出一个 task 让异步清理异步进行（rt::spawn 返回的 JoinHandle 被直接丢弃，只是分离、不会取消任务）。代价是drop 返回时清理还没完成：实测 SQLite 内存库上 drop 之后立刻读，size=1, num_idle=0，等 50ms 再读才是 num_idle=1——但用户一般不关心（pool 内部最终一致）。

14.9.1 close_on_drop 的特殊路径

PoolConnection 有一个 close_on_drop: bool 字段，配套一个 pub fn close_on_drop(&mut self)（sqlx-core/src/pool/connection.rs:97-99）——让连接drop 时直接关而不是归还：

let mut conn = pool.acquire().await?;
conn.close_on_drop();  // 标记：用完关
// ... 用 conn ...
// drop → 走 take_and_close()，而不是 return_to_pool()

文档注释说明了它和 PoolConnection::close().await 的分工：close() 会等关闭完成，close_on_drop() 适用于"等不起那个 await、但仍希望优雅关闭让服务端回收资源"的场景。take_and_close（sqlx-core/src/pool/connection.rs:157-176）里还包了一层 crate::rt::timeout(CLOSE_ON_DROP_TIMEOUT, ...)——CLOSE_ON_DROP_TIMEOUT 是文件顶部的常量 5 秒，防止关闭动作卡住时永远占着这条连接。

典型用途：你做了一些可能破坏连接状态的操作（临时改了 session 参数、触发了某种错误）——不想让这条连接污染其他请求——直接关掉让 pool 建新的。

同一个 impl 块里还有两个"脱离池子"的方法值得知道：detach() 把裸连接拿走、允许池子再开一条（等于应用可以超过 max_connections）；leak() 也拿走裸连接但把这条名额永久扣掉——每用一次池子的实际上限就少 1。

14.10 spawn_maintenance_tasks：后台驱逐循环

sqlx-core/src/pool/inner.rs:503-571 是 Pool 的"后台维护任务"入口（简化）：

fn spawn_maintenance_tasks<DB: Database>(pool: &Arc<PoolInner<DB>>) {
    let pool_weak = Arc::downgrade(pool);

    let period = match (pool.options.max_lifetime, pool.options.idle_timeout) {
        (Some(it), None) | (None, Some(it)) => it,
        (Some(a), Some(b)) => cmp::min(a, b),
        (None, None) => {
            // 都没设——只维持 min_connections
            if pool.options.min_connections > 0 {
                crate::rt::spawn(async move { ... pool.min_connections_maintenance(None).await; ... });
            }
            return;
        }
    };

    let mut close_event = pool.close_event();

    crate::rt::spawn(async move {
        let _ = close_event.do_until(async {
            while let Some(pool) = pool_weak.upgrade() {
                if pool.is_closed() { return; }

                let next_run = Instant::now() + period;

                // 扫描 idle queue，关过期连接
                for _ in 0..pool.num_idle() {
                    if let Some(conn) = pool.try_acquire() {
                        if is_beyond_idle_timeout(&conn, &pool.options)
                            || is_beyond_max_lifetime(&conn, &pool.options)
                        {
                            let _ = conn.close().await;
                            pool.min_connections_maintenance(Some(next_run)).await;
                        } else {
                            pool.release(conn.into_live());
                        }
                    }
                }

                // 睡眠期间不持有强引用
                drop(pool);

                // 等到 next_run（已过期则只 yield_now）
                crate::rt::sleep(/* next_run - now */).await;
            }
        }).await;
    });
}

四个关键点：

1. period 由 min(idle_timeout, max_lifetime) 决定——默认 10 分钟。意味着每 10 分钟扫描一次 idle queue。
2. Arc::downgrade 成 Weak——后台任务持有 Weak 引用、每轮扫描完还会 drop(pool) 再睡，不阻止 PoolInner drop。Pool 最后一个 handle drop 时 PoolInner::drop 会 mark_closed()，close_event 触发，任务随即结束；即便错过，下一轮 Weak 升级也会失败、循环退出。
3. close_event 可取消——Pool close 时 close_event 触发、do_until 退出、task 结束。
4. 扫描过程自身用 try_acquire + release——连接先拿出来检查、不过期就放回去、过期就 close。

为什么用 try_acquire 而不是直接操作 idle_conns？——因为 try_acquire 走正常路径（抢 permit + 弹 idle），检查中的连接和借出的连接一样占着一个 permit，记账才对得上：不过期时要用 release 放回，而 release 会释放一个 permit；若当初没拿 permit 就直接 pop，这一放就凭空多出一个 permit。反过来，permit 全被占满时 try_acquire 返回 None，这一轮就跳过那条连接，不和排队的用户请求抢名额。

还要注意 min_connections 的补齐只挂在"关掉一条过期连接之后"。按 sqlx-core/src/pool/inner.rs:503-571，只要 idle_timeout 或 max_lifetime 有一个是 Some（默认两者都是），就不会走 (None, None) 分支里那次一次性补齐；而 connect_lazy* 建出的池子在循环第一轮时空闲队列为空、什么也不关，接着就睡满一个周期。实测（SQLite 内存库）：min_connections(3).connect_lazy_with(..) 等 500ms 后 size() 仍是 0；把两个超时都设成 None 后是 3；用 connect_with 建池则是 3（它在返回前直接调 try_min_connections）。

14.10.1 min_connections_maintenance

实际干活的是 try_min_connections（sqlx-core/src/pool/inner.rs:396-417），min_connections_maintenance（sqlx-core/src/pool/inner.rs:420-434）只是它外面一层"补上默认 deadline + 吞掉错误只打日志"的包装：

pub async fn try_min_connections(self: &Arc<Self>, deadline: Instant) -> Result<(), Error> {
    while self.size() < self.options.min_connections {
        // Don't wait for a semaphore permit.
        let Some(permit) = self.semaphore.try_acquire(1) else {
            return Ok(());
        };

        // We must always obey `max_connections`.
        let Some(guard) = self.try_increment_size(permit).ok() else {
            return Ok(());
        };

        // We skip `after_release` since the connection was never provided to user code
        self.release(self.connect(deadline, guard).await?);
    }

    Ok(())
}

pub async fn min_connections_maintenance(self: &Arc<Self>, deadline: Option<Instant>) {
    let deadline = deadline.unwrap_or_else(|| Instant::now() + Duration::from_secs(300));

    match self.try_min_connections(deadline).await {
        Ok(()) => (),
        Err(Error::PoolClosed) => (),
        Err(Error::PoolTimedOut) => tracing::debug!("unable to complete `min_connections` maintenance before deadline"),
        Err(error) => tracing::debug!(%error, "error while maintaining min_connections"),
    }
}

几个关键点：

• 用 try_acquire 而不是 acquire——源码注释解释："如果没有多余的 permit，本来就不该再去开连接。"补齐是尽力而为，绝不和真正的用户请求抢名额。
• 仍然走 try_increment_size——注释写着 "We must always obey max_connections"。
• 补出来的连接直接 release 进空闲队列，跳过 after_release（这条连接从没交给用户代码）。
• 默认 deadline 是 5 分钟（Instant::now() + Duration::from_secs(300)）——调用方不关心时给的兜底值。
• 所有错误只打 tracing::debug!——min_connections 补不齐不会让任何用户请求失败，下一轮再试。

调用时机：建池时走的是 try_min_connections 本体（sqlx-core/src/pool/options.rs:546-549，min_connections > 0 才调，并且这里的错误会直接冒泡出去让 connect_with 失败）；min_connections_maintenance 这层包装则用在两个超时都为 None 时建池就 spawn 的那次一次性任务（sqlx-core/src/pool/inner.rs:513-520）、maintenance 循环每关掉一条过期连接之后（sqlx-core/src/pool/inner.rs:550 的 min_connections_maintenance(Some(next_run))）、连接归还失败时（return_to_pool 在 sqlx-core/src/pool/connection.rs:134 声明，调用点在 :151-153）、以及标了 close_on_drop 的连接关闭之后（take_and_close 在 :157 声明，调用点在 :174）。

14.11 连接建立的指数退避

sqlx-core/src/pool/inner.rs:321-393 的 connect 方法有一个精彩的 backoff 机制：

let mut backoff = Duration::from_millis(10);
let max_backoff = deadline_as_timeout(deadline)? / 5;

loop {
    let timeout = deadline_as_timeout(deadline)?;

    match crate::rt::timeout(timeout, connect_options.connect()).await {
        Ok(Ok(mut raw)) => { /* 成功路径 */ }

        // IO 错误（连接被拒）——可能是 DB 刚启动
        Ok(Err(Error::Io(e))) if e.kind() == ConnectionRefused => (),

        // 瞬时 DB 错误
        Ok(Err(Error::Database(error))) if error.is_transient_in_connect_phase() => (),

        Ok(Err(e)) => return Err(e),
        Err(_) => return Err(Error::PoolTimedOut),
    }

    crate::rt::sleep(backoff).await;
    backoff = cmp::min(backoff * 2, max_backoff);
}

backoff 策略：

• 初始 10ms。
• 每次失败 ×2：10 → 20 → 40 → 80 → ...
• 上限是 deadline_as_timeout(deadline)? / 5——注意这是首次进入 connect 时到 deadline 的剩余时间的五分之一，不是 acquire_timeout 的五分之一（虽然在 acquire 路径上两者接近）。

三类会退避重试的情况：

• ConnectionRefused——源码注释："连接时的 IO 错误被假定为系统正在启动中"。
• is_transient_in_connect_phase()——DB 报的瞬时错误。0.8.6 里只有 Postgres 驱动覆写了它，认 53300（too_many_connections）和 57P03（cannot_connect_now）两个错误码（sqlx-postgres/src/error.rs:189-201）；trait 默认实现返回 false（sqlx-core/src/error.rs:224-226）。
• after_connect 回调失败——这条容易被忽略：连接建起来了但回调报错，sqlx 会 close_hard() 掉它（注释："连接已经坏了，别试着优雅关闭"），然后落进同一条退避重试路径。

其他错误直接返回——认证失败、DNS 不可解等，重试没意义。超时则返回 Error::PoolTimedOut。

这条退避机制对容器化部署特别重要——app 和 DB 可能同时启动、app 先 up 时 DB 还没 ready——backoff 让 app 能平滑等 DB 就绪。

14.12 close 的关闭协议

sqlx-core/src/pool/inner.rs:97-115：

pub(super) fn close<'a>(self: &'a Arc<Self>) -> impl Future<Output = ()> + 'a {
    self.mark_closed();

    async move {
        for permits in 1..=self.options.max_connections {
            // 关所有 idle
            while let Some(idle) = self.idle_conns.pop() {
                let _ = idle.live.float((*self).clone()).close().await;
            }

            if self.size() == 0 { break; }

            // 等所有 permit 都 release
            let _permits = self.semaphore.acquire(permits).await;
        }
    }
}

三阶段：

1. mark_closed——atomic bool 置 true，on_closed 事件广播。
2. 关所有 idle——循环 pop idle queue、close。
3. 等所有借出的连接归还——每轮 semaphore.acquire(permits) 要的 permit 数加 1，拿到后（循环体结束时释放）再回头清一遍空闲队列，直到 size() 归零或走完 max_connections 轮；最后一轮要一次拿满 max_connections 个 permit，只有借出的连接全部交回 permit 才能完成。关闭后归还的连接会在 return_to_pool 第一关看到 is_closed，直接关掉、size 减 1。

这是一个精妙的用法——用信号量的 acquire 作为"等所有 release"的同步——不需要显式 channel 或 condvar。

14.13 Pool 内部关键流程的整体图

把本章讨论的几条主路径合一张总图：

flowchart TD
    subgraph 用户线程
        A1["pool.acquire await"]
        A2["conn.execute ..."]
        A3["drop conn"]
    end
    subgraph PoolInner
        S["semaphore.acquire"]
        IQ["idle_conns.pop"]
        CC["check_idle_conn"]
        TI["try_increment_size"]
        CN["connect backoff loop"]
        RT["return_to_pool"]
        REL["release: idle_conns.push"]
    end
    subgraph 后台任务
        M1["maintenance loop<br/>每 min(idle_timeout,max_lifetime)"]
        M2["扫描 idle queue"]
        M3["关过期连接"]
        M4["min_connections_maintenance"]
    end
    A1 --> S
    S --> IQ
    IQ -->|有空闲| CC
    CC -->|OK| A2
    CC -->|fail| TI
    IQ -->|无空闲| TI
    TI --> CN
    CN --> A2
    A3 --> RT
    RT --> REL
    REL --> IQ
    M1 --> M2 --> M3 --> M4
    M4 --> CN

这张图包含四条主路径：

1. 用户 acquire + 已有 idle：acquire → semaphore → pop idle → check → 返回。
2. 用户 acquire + 无 idle：acquire → semaphore → try_increment → connect → 返回。
3. 用户 drop conn：drop → spawn return_to_pool → release → 入 idle queue。
4. 后台 maintenance：扫描 → 关过期 → 触发 min_connections 补齐。

理解这四条路径你就理解了 Pool 整体的动态行为。

14.14 阶段小结

前面几节把 PoolInner 的每一个机制拆开：

1. PoolInner 10 字段（§14.2）—— idle_conns / semaphore / size / num_idle / is_closed / on_closed + 配置和日志。
2. ArrayQueue 无锁队列（§14.3）—— crossbeam-queue 的 MPMC 有界队列，基于 CAS 原子操作、高并发下不被 Mutex contention 拖长尾。
3. AsyncSemaphore fair 语义（§14.4）—— 默认 FIFO 公平。抽象 Tokio/async-std 差异。
4. 三状态 Live/Idle/Floating（§14.5）—— Floating 是过渡态、带 DecrementSizeGuard RAII。
5. acquire 的 loop（§14.6）—— 获取 permit → pop idle 或 try_increment_size → 建连接。check_idle_conn 两道检查。
6. release 的双计数分离（§14.7）—— permit 管并发、size 管总数——idle 时占 size 不占 permit。
7. DecrementSizeGuard RAII（§14.8）—— fetch_update 原子 +size、Drop 自动 -size 除非 cancel——错误路径的自动修正。
8. PoolConnection::Drop spawn 归还（§14.9）—— 异步清理 + after_release 回调 + close_on_drop 特殊路径。
9. maintenance task（§14.10）—— 后台扫描 idle queue、关过期连接、维持 min_connections。用 Weak 引用不阻止 PoolInner drop。
10. 指数退避 backoff（§14.11）—— 建连接被拒或遇瞬时错误时 10ms→20ms→40ms...，上限为进入 connect 时剩余时限的 1/5。
11. close 的三阶段（§14.12）—— mark_closed → 关 idle → 等 in-use 归还。用 semaphore.acquire 作为"等归还"的同步。

14.15 Pool 内部的并发正确性

Pool 内部是高并发代码——多个 async task 同时 acquire、maintenance task 后台扫描、close 可能并发——如何保证正确性？

sqlx 用了几条并发技巧：

1. 原子计数 + CAS 循环。size.fetch_update 用乐观 CAS——读当前值、算新值、原子替换失败就重试，全程不加锁：

let result = self.size.fetch_update(..., |size| {
    if self.is_closed() { return None; }
    size.checked_add(1).filter(|size| size <= &max)
});

checked_add + filter 一起表达"没满才能加"。成功时返回 Ok(旧值)；只有闭包返回 None 时才返回 Err(旧值)——CAS 竞争失败不会变成 Err，fetch_update 内部会拿新读到的值重新调用闭包（实测 AtomicU32::new(5) 上闭包返回 Some(x + 1) 得 Ok(5)，接着返回 None 得 Err(6)）。

2. 分离 permit 和 size。permit 管"允许多少并发 acquire"、size 管"实际多少连接"。两者独立——idle 连接占 size 不占 permit。通过 DecrementSizeGuard::release_permit（不减 size）vs Drop（减 size）实现精细控制。

3. Weak 引用防止循环。maintenance task 持 Weak<PoolInner>——不阻止 PoolInner drop。Arc 计数为 0 时 Weak upgrade 返回 None、task 退出。如果任务持的是 Arc，它自己就一直撑着 PoolInner 的强引用计数：Pool 永远不被回收，任务也永远不退出。

4. event_listener 用于 close 信号。on_closed: event_listener::Event——多个 waiter 同时 listen、一次 notify(usize::MAX) 唤醒所有。

5. crate::rt::yield_now() 在冲突时让步。acquire 循环里如果 pop idle + try_increment 都失败（源码注释说这只会发生在子池到达上限、或池子在两步之间被关闭时），先 yield 再重试——注释给的理由是：这种情况多半在 Tokio 的 current-thread 运行时里，要让已 spawn 的 return_to_pool 任务有机会跑。

6. do_until 用于可取消操作。close_event().do_until(future) 让长任务能在 Pool close 时取消——优雅 shutdown 的基础。

14.16 性能特征和瓶颈分析

Pool 在不同负载下的定性性能特征（具体数值需自行 benchmark）：

场景 A：低并发（< 10 acquire/s）

• 几乎没 semaphore 争抢——permit 永远有。
• 大多数 acquire 走 idle queue pop 快路径（纳秒级原子操作）。
• 主要开销是两次 ping（acquire 前的 test_before_acquire 和归还时的无条件 ping，见 §14.24），其次才是 PoolConnection::Drop 里的一次 spawn。

场景 B：中等并发（100-1000 acquire/s）

• idle queue pop / push 频繁——但都是进程内的原子操作。
• semaphore 可能有短暂等待。
• acquire 本身不会成为瓶颈——真正的耗时在 §14.24 讲的那两次 ping 和可能的建连上。

场景 C：高并发 + 连接饱和（acquire > max_connections）

• 排队成为主要开销——fair 模式 FIFO。
• P99 可能触及 acquire_timeout（默认 30s）。
• 症状：P99 延迟剧烈震荡，偶发 PoolTimedOut。

场景 D：连接抖动

• 网络不稳、连接频繁断开重建。
• 归还时的 ping 频繁失败、连接被 close_hard 掉，acquire 频繁走建新连接；connect backoff 延长。
• 症状：num_idle 长期 0，size 剧烈波动。

瓶颈定位：

• 低并发慢：检查业务 query 本身（不是 Pool 问题）。
• 中等并发偶发慢：看 acquire_slow_level 的日志（默认就是 Warn 级、阈值 2s）——看哪些 acquire 变慢。
• 高并发排队：增大 max_connections 或加实例水平扩展。
• 抖动：排查网络 / DB 健康 / PgBouncer 状态。

这条性能特征分析让你在生产排查时对症下药——不会"P99 高了就盲目调 max_connections"。

14.17 Pool 作为 Rust 并发编程的教材

读完本章你会发现 sqlx 的 pool/ 模块是一份Rust 高并发代码的教材——浓缩了许多值得学习的模式：

技术	在 Pool 里的用法	通用用途
无锁 MPMC 队列	idle_conns: ArrayQueue	任何"多生产者多消费者 + 有界"场景
异步信号量	AsyncSemaphore 控制 max_connections	任何"并发数限制"场景
原子 CAS 循环	size.fetch_update	无锁计数器
RAII 兜底 guard	DecrementSizeGuard	错误路径自动修正计数
Weak 引用不延长寿命	maintenance task 持 Weak<PoolInner>	后台任务不阻止主对象回收
event_listener	on_closed 事件广播	多 waiter 的单次通知
异步超时 + 退避	connect backoff 10→max	网络操作的重试
Drop 里 spawn	PoolConnection::Drop → spawn return_to_pool	异步清理资源

每一条都是 Rust 生态里反复出现的技术——掌握它们的具体用法（sqlx 展示的）可以直接迁移到你自己的项目。如果你在写 HTTP client pool、gRPC channel pool、任何"管理一组异步资源"的代码——sqlx 的 Pool 就是你的参考实现。

14.18 三点必记的 Pool 内部常识

本章最后，精炼成三点生产工程师必记的 Pool 内部常识：

1. acquire 成功的条件是 "拿到 permit AND （pop idle 且通过检查 OR try_increment 成功后连上）"——一个 AND 和一个 OR。排查"为什么我的 acquire 超时"，就看时间耗在哪一段：permit 排队，还是拿到 permit 之后的 ping / 建连（含退避重试）。不必怀疑 size 撞上限——对普通（非子）池，拿到 permit 却 try_increment_size 失败只会发生在池子已关闭时（sqlx-core/src/pool/inner.rs:275-277 的注释），因为空闲连接不占 permit，手里有 permit 而空闲队列为空就说明 size 还没到顶。

2. size 泄漏永远由 DecrementSizeGuard 兜底——哪怕 connect 过程 panic、task 被取消，只要 guard 没 cancel，drop 时自动减 size。这是 Pool 永远不会"size 单调上升超过实际连接数"的保证。

3. maintenance task 是 Weak 引用——Pool 的最后一个 handle drop 后、maintenance 会退出。你不需要手动关 pool——drop 所有 handle 就行。但生产建议显式 close().await 让 idle 连接优雅关——不然 PoolInner 析构时空闲连接只是随 ArrayQueue 一起被丢弃，不走 Connection::close()：Postgres / MySQL 驱动不会发送协议层的退出消息，服务端看到的是 socket 直接断开。

14.19 源码阅读推荐路径

如果读完本章你想自己直接读 sqlx 的 Pool 源码——推荐按以下顺序：

1. sqlx-core/src/pool/mod.rs（668 行）——从 Pool struct 和公共方法开始，前半是大段文档注释。
2. sqlx-core/src/pool/options.rs（596 行）——PoolOptions 的 13 个配置项，大部分是 builder 方法。
3. sqlx-core/src/pool/inner.rs（619 行）——核心实现，最硬的一部分。
4. sqlx-core/src/pool/connection.rs（421 行）——PoolConnection + Live/Idle/Floating 状态类型。
5. sqlx-core/src/sync.rs（137 行）——AsyncSemaphore 的 runtime 抽象。

再加上 executor.rs（142 行）和 maybe.rs（43 行），整个 pool/ 模块 2489 行。哪些该 Mutex、哪些该 atomic、哪些该 Weak、哪些该 spawn、哪些该 RAII guard——这几个文件里都有具体示范。

重点关注源码中的注释——不少关键决策旁边有英文解释。比如 release 里"permit 必须在 push 之后释放"、acquire 里 yield_now() 前那段关于 current-thread 运行时的说明、num_idle() 为什么不用 idle_conns.len()（sqlx-core/src/pool/inner.rs:80-84：高频进出时 len() 要等头尾指针稳定下来，可能很慢）——作者在和你对话"为什么这么做"。

14.20 Pool 内部设计的通用启示

最后升华到 API 设计层面，sqlx Pool 内部代码给我们的几条启示：

1. 分离语义层次的计数器。permit（并发控制）和 size（资源总数）分开——让"一条 idle 连接占 size 但不占 permit" 这种细粒度控制成为可能。

2. 用 RAII guard 处理错误路径。DecrementSizeGuard 的 Drop 让错误路径"默认正确"——程序员不用手写 "if err { decrement }"。这条设计在 Rust 里极其常见——Mutex 的 MutexGuard、RefCell 的 RefMut、tokio 的 tokio::sync::SemaphorePermit——都是同一套思路。遇到"需要成对做的操作"永远考虑 RAII。

3. 后台任务用 Weak 引用。maintenance task 持 Weak<PoolInner> 不阻止 PoolInner drop——这让 "Pool 最后一个 handle drop 后整个资源链自动回收" 成立。如果后台任务持 Arc，任务自己就撑着 PoolInner 的强引用，Pool 就永远不会被回收。这条"引用类型的方向性"是 Rust 异步代码里最容易踩坑的地方——但学会后能避免很多生产事故。

4. 多状态对象用类型系统表达状态。Live / Idle / Floating 三个类型就是状态机的三个状态——转换通过 into_xxx 方法、编译器保证状态迁移合法（你不能把 Live 直接推回 Idle queue、必须先 into_idle）。比用 bool flag 区分状态类型安全得多。

5. 资源上限由数据结构本身保证。ArrayQueue 是有界的、Pool 的 semaphore 也是有界的（max_connections）——无界队列 / 无界信号量听起来灵活，放到生产里容易把内存或下游撑爆。

14.21 parent_pool 的特殊机制

sqlx-core/src/pool/inner.rs:127-171 的 acquire_permit 里有一段关于 parent_pool 的处理——这是 sqlx 一个少用但精彩的机制。

PoolOptions::parent(pool) 让一个子 Pool 从父 Pool 偷 permit。先说清楚：这个方法带 #[doc(hidden)]，文档注释写明 "This is currently an internal-only API"（sqlx-core/src/pool/options.rs:505），并且如果子 Pool 的 max_connections 大于父的、或者 fair 和父不一致就会 panic（sqlx-core/src/pool/inner.rs:47-48 的两条 assert）。所以下面这段代码是为了讲清机制，不是推荐你在业务里这么写。

let main_pool = PgPoolOptions::new().max_connections(100).connect(url).await?;

// 子 Pool 共享父的连接预算
let subtask_pool = PgPoolOptions::new()
    .max_connections(20)
    .parent(main_pool.clone())
    .connect_lazy_with(opts);

语义：

• 子 Pool 的 max_connections = 20 —— 子池 size 的上限（由 try_increment_size 保证）。
• 父 Pool 的 max_connections = 100 —— 子池建池时信号量容量是 0，没有自己的 permit，每条连接用的 permit 都是从父池偷来的，所以子池的连接也计在父池这 100 个名额里。
• 子 Pool 的 acquire 先 poll 自己的信号量（子池连接归还时 permit 回到的是子池的信号量），拿不到才去父池偷；子池 size 已到 20 时根本不碰父池。
• 效果：子池最多 20 条连接，最多占掉父池 100 个名额中的 20 个；偷来的 permit 在子池销毁前不会还给父池。

典型用例：一个服务里有主路径 pool（业务查询）和后台任务 pool（ETL / 统计）。后台 pool 不能占满主路径的连接——设成子 Pool、max_connections 给个适量（20 条），让后台跑慢但不撞主路径。

实现上 acquire_permit（sqlx-core/src/pool/inner.rs:127-171）的 future::poll_fn 里同时 poll 子和父的 semaphore——先 poll 子（优先用自己的），子那边拿不到才 poll 父（借）。子池手里的 permit 全都来自父池（子的 semaphore capacity 是 0）：偷来之后在子池内部循环使用，直到子池销毁才归还。

这条机制目前是内部 API（sqlx 自己在 #[sqlx::test] 的隔离场景里用得上），日常业务代码碰不到——但它展示了"多个逻辑池共享同一份连接预算"可以怎么实现。至于 PoolInner::drop（sqlx-core/src/pool/inner.rs:437-445）里那句 parent.0.semaphore.release(self.semaphore.permits())——子池销毁时把偷来的 permit 还给父池，正是这套机制的收尾。

14.22 Pool 并发测试的挑战

值得说明的是：sqlx 仓库里并没有针对 Pool 的专门并发测试套件——没有 tests/pool/ 目录，也没有引入 loom 做穷尽并发状态检查。Pool 相关的测试散落在各驱动的集成测试里（例如 tests/postgres/postgres.rs 的 pool_smoke_test、tests/any/pool.rs）。其中 pool_smoke_test 是最接近压力测试的一个：它标了 #[ignore]，用 200 个任务抢一条连接、另有 50 个任务反复发起 acquire 又立刻放弃，跑 30 秒后检查 close() 能在 30 秒内完成——验证的是"不死锁、取消处理正确"，而不是穷举 race。

那它的正确性靠什么？主要靠代码结构本身压缩竞争空间：

• atomic CAS 让 size 更新无 race。
• RAII guard 让错误路径自动修正。
• Weak 引用让后台任务不阻碍 drop。

但这不等于没有 bug。git 历史里 pool 模块经历过多轮修复：v0.5.6 的 #1320 用 futures-intrusive 重写了内部实现，v0.5.12 的 #1799 修复了"drop 任务没跑时连接泄漏"（fixes #1396，也就是 §14.9 里"先 float 再 spawn"那段注释的来历），v0.6.0 的 #1901 又是一轮「pool fixes and breaking changes」。结构上的 RAII 与原子操作能挡住一类错误，剩下的仍要靠问题暴露后逐个修。

14.23 acquire_permit 函数的 poll_fn 精读

前面 §14.6 一笔带过了 acquire_permit（sqlx-core/src/pool/inner.rs:127-171）。先说清楚它的两条分支：没有父池时（也就是你在业务代码里几乎总会遇到的情况）它只有一行——

close_event.do_until(acquire_self).await

即"等自己的信号量，池子一关就放弃"。下面这段 poll_fn（sqlx-core/src/pool/inner.rs:143-166）只在配了 parent_pool 时才走，但它的写法值得精读：

future::poll_fn(|cx| {
    if close_event.as_mut().poll(cx).is_ready() {
        return Poll::Ready(Err(Error::PoolClosed));
    }

    if parent_close_event.as_mut().poll(cx).is_ready() {
        self.mark_closed();
        return Poll::Ready(Err(Error::PoolClosed));
    }

    if let Poll::Ready(permit) = acquire_self.as_mut().poll(cx) {
        return Poll::Ready(Ok(permit));
    }

    // 父 Pool 的 poll 有延迟
    if poll_parent {
        acquire_parent.as_mut().poll(cx).map(Ok)
    } else {
        poll_parent = true;
        cx.waker().wake_by_ref();
        Poll::Pending
    }
}).await

四个 future 手工按优先级 poll——close_event / parent_close_event / acquire_self / acquire_parent：

1. close_event：Pool 已关，立即返回 PoolClosed。
2. parent_close_event：父 Pool 关，propagate 到子、同样 PoolClosed。
3. acquire_self：自己的 semaphore 有 permit，立即拿。
4. acquire_parent：从父借 permit——但第一次 poll 故意返回 Pending + wake，给子一次"不借父"的机会。

第四步的延迟 poll 很精彩——poll_parent = false 时调用者第一次 poll 返回 Pending + 立刻 wake——让 executor 再 poll 一次（这次 poll_parent = true）。中间可能有子的 semaphore 释放（其他 task 归还）——让子 Pool 优先用自己的配额、只有真找不到时才借父。

源码里对第四步的注释只有一句 "Don't try the parent right away."——但意图很清楚：这是"优先本地、不得已才借外部"的语义，用 poll_fn 里的一个 bool 状态精确表达。如果同时 poll 父和子，父可能抢先返回、白白占掉父池的名额；延迟一个 poll cycle 就把这个 race 规避掉了。另外 acquire_permit 开头还有一层过滤（sqlx-core/src/pool/inner.rs:128-132）：子池自己已经到 max_connections 时压根不去碰父池，注释说这"只会在 acquire() 里造成不必要的抖动"。

14.24 Pool 热路径上到底做了多少事

本专栏不给没有出处的耗时数字——但可以精确地数清楚"命中空闲连接的那条最快路径上做了哪些操作"，这比拍脑袋的纳秒数有用得多：

步骤	实际动作	涉及的同步原语
acquire_permit	semaphore.acquire(1)（无争抢时立即就绪）	信号量
pop_idle	idle_conns.pop() + num_idle.fetch_sub(1)	无锁队列 + 一次原子
check_idle_conn	默认还要 ping() 一次（test_before_acquire 默认 true）	一个网络往返
reattach	guard.cancel() + 构造 PoolConnection	无（只改一个 bool）
drop 归还	rt::spawn 一个 task → ping() → push → release_permit	又一个网络往返

结论比任何数字都清楚：这条路径上唯一数量级较大的开销是两次 ping——一次在 acquire 前（可以用 test_before_acquire(false) 关掉），一次在归还时（§14.9，无条件执行、关不掉）。其余全是进程内的原子操作和一次 task spawn。

所以想让 acquire 更快，能动的只有：关掉 test_before_acquire、或者用 before_acquire 回调按 idle_for 有选择地 ping。至于队列和信号量那一层，都是进程内的原子操作，不是这条路径上的大头。

14.25 一条实战排查案例

用一个典型的生产场景串起本章内容——假设你的服务突然报大量 PoolTimedOut：

Step 1：看监控仪表盘——pool.size() 恒定在 max_connections、pool.num_idle() 恒 0——Pool 饱和。

Step 2：看 acquire_slow_level 的日志（默认就是 Warn 级、阈值 2s，不用额外打开）——大量出现 acquired connection, but time to acquire exceeded slow threshold，每次 acquire 等 2s+。

Step 3：查 pg_stat_activity——发现大量连接在同一条长查询上。

Step 4：定位那条查询——SELECT ... FROM huge_table WHERE ... ORDER BY col——没索引、全表扫。

Step 5：根因——这条慢查询把 pool 的所有连接占满、新请求全排队超时。

Step 6：修复——加索引 / 加 statement_timeout（从 after_connect 里设）——让慢查询自动被 kill。

这条排查路径利用了本章讨论的几个机制：

• pool.size() / num_idle() 做饱和判断（§13.8）。
• acquire_slow_level 定位慢 acquire（§13.10）。
• after_connect 设 statement_timeout（§13.11）。
• Pool 的 size 一直等于 max 而不增不减——背后是 §14.8 的 size counter 稳定语义。
• 理解 idle 队列为空意味着所有连接在 in-use——这来自本章的 idle vs in-use 分离概念（§14.7）。

不理解 Pool 内部的工程师可能会乱改配置（增大 max_connections、改 acquire_timeout）——治标不治本。理解本章内容后你知道根因不在 Pool——而在业务查询——从这个方向调才是对的。

这个案例说明：读懂 Pool 内部不是学术锻炼、是生产武器。

14.26 Pool 源码的代码品质观察

作为本章结尾，回看 sqlx pool 源码的代码品质几点观察：

1. 注释密度恰当。关键决策点大多有英文注释解释"为什么这么做"——长的如 release 前后的 permit 顺序、acquire 里 yield_now() 的理由、num_idle() 不用 len() 的原因，短的如 poll_fn 延迟 poll 父池只有一句 "Don't try the parent right away."。其他显然的代码不加废话注释。这是良好开源代码的标志——新手能读懂、老手不被冗余。

2. 函数长度控制。多数函数一屏就能看完，没有超过 100 行的。最长的是 PoolInner::acquire（约 78 行，核心循环 + 超时 + 慢日志），其次是 connect（约 73 行）和 spawn_maintenance_tasks（约 69 行）——这种"必要的长"可以接受。

3. panic! 使用谨慎。整个 pool 模块显式的 panic! 只有一处：release 里的 "BUG: connection queue overflow"（另有 new_arc 里校验父池配置的两条 assert! 和几处 expect）——因为逻辑上 size counter 保证 idle_conns 不会满、如果满了一定是 bug。这种不变量 violation 用 panic比用 Result 合理——把"不可能发生"的代码路径标出来、避免用户处理不存在的错误。

4. 测试友好的设计。parent_pool 机制（§14.21）服务于 #[sqlx::test]：Postgres / MySQL 的测试支持代码（sqlx-postgres/src/testing/mod.rs:182、sqlx-mysql/src/testing/mod.rs:182）给每个测试建一个 max_connections(5) 的子池、挂在主池下面，单个测试就占不满全部连接。它没有用 #[cfg(test)] 隔离，而是一个 #[doc(hidden)] 的普通方法。

5. #[allow(dead_code)] 使用克制。pool/ 模块里只有两处：sqlx-core/src/pool/mod.rs:659 那个只做编译期 trait 断言的测试函数，和 sqlx-core/src/pool/maybe.rs:6 的 MaybePoolConnection::Connection 变体。

学 sqlx 不只是学库怎么用、也是学怎么写一份能长期维护的开源 Rust 库。

14.27 和《Tokio 源码深度解析》专栏第 12 章的关联

sqlx Pool 的核心门禁是 AsyncSemaphore——本章看到 acquire_permit、DecrementSizeGuard、fair 队列这些概念——它们背后就是《Tokio 源码深度解析》专栏第 12 章 §12.4 Semaphore：基础 building block 和 §12.5 公平锁 vs 非公平锁的 trade-off 里讨论的数据结构。

具体对应：

• sqlx PoolOptions::fair = true 默认（sqlx-core/src/pool/options.rs:163）↔ Tokio 专栏 §12.5 讲 tokio::sync::Semaphore 的公平队列——底层是侵入式链表（tokio/src/sync/batch_semaphore.rs 里的 Waitlist.queue），每个 waiter 是链表上的一个节点、按到达顺序排队。
• sqlx Pool 的 acquire_permit().await ↔ Tokio 专栏 §12.4 讲 Semaphore::acquire(1) 返回 SemaphorePermit 的 RAII guard——drop 时自动 release（sqlx 在 tokio 侧实际调用的是 acquire_many(1)，交给 DecrementSizeGuard::from_permit 时就 disarm()（即 forget()），改由 guard 负责归还，见 §14.8）。
• sqlx DecrementSizeGuard 的"资源获取和释放绑定同一 struct" ↔ Tokio 专栏 §12.4 讲的 OwnedSemaphorePermit。

为什么要看 Tokio 那章——sqlx 池是在 Semaphore 之上封装业务语义（最大连接数、idle queue、min_connections 维护）——底层 Semaphore 的 waker 管理、公平性实现、poll 协议都在 Tokio crate 里。不看 Tokio 那一层、sqlx 的 acquire 为什么能被唤醒就只能靠直觉理解——看了就知道 Tokio Semaphore 内部 poll_acquire 是怎么挂 waker、怎么在 add_permits 时按序唤醒。

一条建议——如果本章读得有点吃力、先看《Tokio 源码深度解析》专栏第 12 章的 §12.4-§12.5 半小时——再回来读本章会顺得多。