获取连接

    pub fn acquire(&self) -> impl Future<Output = Result<PoolConnection<DB>, Error>> + 'static {
        let shared = self.0.clone();
        async move { shared.acquire().await.map(|conn| conn.reattach()) }
    }

    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 {
                    // Handles the close-event internally
                    let permit = self.acquire_permit().await?;


                    // First attempt to pop a connection from the idle queue.
                    let guard = match self.pop_idle(permit) {

                        // Then, check that we can use it...
                        Ok(conn) => match check_idle_conn(conn, &self.options).await {

                            // All good!
                            Ok(live) => return Ok(live),

                            // if the connection isn't usable for one reason or another,
                            // we get the `DecrementSizeGuard` back to open a new one
                            Err(guard) => guard,
                        },
                        Err(permit) => if let Ok(guard) = self.try_increment_size(permit) {
                            // we can open a new connection
                            guard
                        } else {
                            // ...
                            tracing::debug!("woke but was unable to acquire idle connection or open new one; retrying");
                            // ...
                            crate::rt::yield_now().await;
                            continue;
                        }
                    };

                    // Attempt to connect...
                    return self.connect(deadline, guard).await;
                }
            }
        )
            .await
            .map_err(|_| Error::PoolTimedOut)??;
        // ... 慢获取的日志，见下文
        Ok(acquired)
    }

            close_event.do_until(acquire_self).await

let held = pool.acquire().await?;
let mut tasks = Vec::new();
for i in 0..5 {
  let (pool, order, clock) = (pool.clone(), order.clone(), clock.clone());
  tasks.push(tokio::spawn(async move {
    let conn = pool.acquire().await.unwrap();
    println!("{:>4}ms waiter {i} got it", clock.ms());
    order.lock().unwrap().push(i);
    tokio::time::sleep(Duration::from_millis(20)).await;
    drop(conn);
  }));
  // let each waiter reach the semaphore before the next one starts
  tokio::time::sleep(Duration::from_millis(10)).await;
}

println!("{:>4}ms try_acquire while 5 are queued: {:?}", clock.ms(), pool.try_acquire().map(|_| ()));
drop(held);
println!("{:>4}ms holder dropped", clock.ms());
// a latecomer arriving right after the release
let late = {
  let (pool, clock) = (pool.clone(), clock.clone());
  tokio::spawn(async move {
    let _conn = pool.acquire().await.unwrap();
    println!("{:>4}ms latecomer got it", clock.ms());
  })
};

  59ms try_acquire while 5 are queued: None
  59ms holder dropped
  60ms waiter 0 got it
  83ms waiter 1 got it
 106ms waiter 2 got it
 128ms waiter 3 got it
 151ms waiter 4 got it
 174ms latecomer got it
order: [0, 1, 2, 3, 4]

            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() {
                    // Propagate the parent's close event to the child.
                    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));
                }

                // Don't try the parent right away.
                if poll_parent {
                    acquire_parent.as_mut().poll(cx).map(Ok)
                } else {
                    poll_parent = true;
                    cx.waker().wake_by_ref();
                    Poll::Pending
                }
            })

        if let Some(idle) = self.idle_conns.pop() {
            self.num_idle.fetch_sub(1, Ordering::AcqRel);
            Ok(Floating::from_idle(idle, (*self).clone(), permit))
        } else {
            Err(permit)
        }

    if options.test_before_acquire {
        // Check that the connection is still live
        if let Err(error) = conn.ping().await {
            // an error here means the other end has hung up or we lost connectivity
            // either way we're fine to just discard the connection
            // the error itself here isn't necessarily unexpected so WARN is too strong
            tracing::info!(%error, "ping on idle connection returned error");
            // connection is broken so don't try to close nicely
            return Err(conn.close_hard().await);
        }
    }

    if let Some(test) = &options.before_acquire {
        // ...
    }

    // No need to re-connect; connection is alive or we don't care
    Ok(conn.into_live())

let pool = PgPoolOptions::new()
  .before_acquire(move |_conn, _meta| {
    let flag = flag.clone();
    Box::pin(async move { Ok(!flag.swap(false, Ordering::SeqCst)) })
  })
  .connect_with(lab::opts("pool-reject"))
  .await?;
let mut obs = lab::observer().await;
let (a, b) = (pool.acquire().await?, pool.acquire().await?);
drop((a, b));
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
println!("before: size={} idle={} pids={:?}", pool.size(), pool.num_idle(), lab::backends(&mut obs, "pool-reject").await);
reject_next.store(true, Ordering::SeqCst);
let mut conn = pool.acquire().await?;
let pid: i32 = sqlx::query_scalar("select pg_backend_pid()").fetch_one(&mut *conn).await?;
println!("got pid={pid}: size={} idle={} pids={:?}", pool.size(), pool.num_idle(), lab::backends(&mut obs, "pool-reject").await);

before: size=2 idle=2 pids=[499, 501]
got pid=502: size=2 idle=1 pids=[501, 502]

for test in [true, false] {
  p.quiet.store(true, Ordering::SeqCst);
  let pool = PgPoolOptions::new()
    .max_connections(1)
    .test_before_acquire(test)
    .connect_with(lab::via_proxy("pool-pings", p.port))
    .await?;
  tokio::time::sleep(Duration::from_millis(100)).await;
  p.quiet.store(false, Ordering::SeqCst);

  println!("--- test_before_acquire({test}) ---");
  let conn = pool.acquire().await?;
  println!("acquired: size={} num_idle={}", pool.size(), pool.num_idle());
  let mut conn = conn;
  let one: i32 = sqlx::query_scalar("select 1").fetch_one(&mut *conn).await?;
  println!("query -> {one}");
  drop(conn);
  println!("dropped:  size={} num_idle={}", pool.size(), pool.num_idle());
  tokio::time::sleep(Duration::from_millis(100)).await;
  println!("later:    size={} num_idle={}", pool.size(), pool.num_idle());

  if test {
    println!("--- try_acquire (test_before_acquire(true)) ---");
    let conn = pool.try_acquire();
    println!("try_acquire -> {:?}", conn.as_ref().map(|_| "PoolConnection"));
    println!("again while held -> {:?}", pool.try_acquire().map(|_| "PoolConnection"));
    drop(conn);
    tokio::time::sleep(Duration::from_millis(100)).await;
  }

  p.quiet.store(true, Ordering::SeqCst);
  pool.close().await;
}

--- test_before_acquire(true) ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
acquired: size=1 num_idle=0
  [conn 1] C->S Parse(sqlx_s_1, "select 1"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
query -> 1
dropped:  size=1 num_idle=0
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
later:    size=1 num_idle=1
--- try_acquire (test_before_acquire(true)) ---
try_acquire -> Some("PoolConnection")
again while held -> None
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
--- test_before_acquire(false) ---
acquired: size=1 num_idle=0
  [conn 2] C->S Parse(sqlx_s_1, "select 1"), Describe(S, sqlx_s_1), Sync
  [conn 2] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
query -> 1
dropped:  size=1 num_idle=0
  [conn 2] C->S Sync
  [conn 2] S->C ReadyForQuery(I)
later:    size=1 num_idle=1

        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 {
            // we successfully incremented the size
            Ok(_) => Ok(DecrementSizeGuard::from_permit((*self).clone(), permit)),
            // the pool is at max capacity or is closed
            Err(_) => Err(permit),
        }

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);
        }
    }
}

let parent = PgPoolOptions::new()
  .max_connections(2)
  .connect_with(lab::opts("pool-ct"))
  .await?;
let child = PgPoolOptions::new()
  .max_connections(1)
  .parent(parent.clone())
  .connect_lazy_with(lab::opts("pool-ct"));
let (c1, c2) = (child.clone(), child.clone());
let a = tokio::spawn(async move { drop(c1.acquire().await.unwrap()); "a done" });
let b = tokio::spawn(async move { drop(c2.acquire().await.unwrap()); "b done" });
println!("{} / {}", a.await?, b.await?);

...
DEBUG sqlx_core::pool::inner: woke but was unable to acquire idle connection or open new one; retrying
a done / b done
...

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

        loop {
            let timeout = deadline_as_timeout(deadline)?;

            // clone the connect options arc so it can be used without holding the RwLockReadGuard
            // across an async await point
            let connect_options = self
                .connect_options
                .read()
                .expect("write-lock holder panicked")
                .clone();

            // result here is `Result<Result<C, Error>, TimeoutError>`
            // if this block does not return, sleep for the backoff timeout and try again
            match crate::rt::timeout(timeout, connect_options.connect()).await {
                // successfully established connection
                Ok(Ok(mut raw)) => {
                    // ... 跑 after_connect，成功就 return Ok(Floating::new_live(raw, guard))，
                    // 出错就 close_hard 然后落到下面的退避
                }

                // an IO error while connecting is assumed to be the system starting up
                Ok(Err(Error::Io(e))) if e.kind() == std::io::ErrorKind::ConnectionRefused => (),

                // We got a transient database error, retry.
                Ok(Err(Error::Database(error))) if error.is_transient_in_connect_phase() => (),

                // Any other error while connection should immediately
                // terminate and bubble the error up
                Ok(Err(e)) => return Err(e),

                // timed out
                Err(_) => return Err(Error::PoolTimedOut),
            }

            // If the connection is refused, wait in exponentially
            // increasing steps for the server to come up,
            // capped by a factor of the remaining time until the deadline
            crate::rt::sleep(backoff).await;
            backoff = cmp::min(backoff * 2, max_backoff);
        }

async fn try_pool(name: &str, port: u16) {
  let started = Instant::now();
  let res = PgPoolOptions::new()
    .acquire_timeout(Duration::from_secs(1))
    .connect_with(opts_at(port))
    .await;
  println!("{name}: after {:?}: {}", started.elapsed(), res.err().map(|e| e.to_string()).unwrap_or_default());
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
  // port 1 on loopback: nothing listens, the kernel answers RST -> ConnectionRefused
  try_pool("closed port", 1).await;

  let t0 = Instant::now();
  let (port, hits) = fake_server("57P03", "the database system is starting up", t0).await;
  try_pool("57P03", port).await;
  println!("  attempts at ms: {:?}", hits.lock().unwrap());

  let t0 = Instant::now();
  let (port, hits) = fake_server("28P01", "password authentication failed for user \"lab\"", t0).await;
  try_pool("28P01", port).await;
  println!("  attempts at ms: {:?}", hits.lock().unwrap());
  Ok(())
}

closed port: after 1.005184208s: pool timed out while waiting for an open connection
...
28P01: after 425.542µs: error returned from database: password authentication failed for user "lab"
  attempts at ms: [0]

let pool = PgPoolOptions::new()
  .max_connections(1)
  .acquire_timeout(Duration::from_millis(1500))
  .connect_with(lab::opts("pool-timeout"))
  .await?;

let held = pool.acquire().await?;
println!("holding: size={} num_idle={}", pool.size(), pool.num_idle());

for _ in 0..3 {
  let started = Instant::now();
  let res = sqlx::query_scalar::<_, i32>("select 1").fetch_one(&pool).await;
  let err = res.unwrap_err();
  println!("after {:?}: {err} / {err:?}", started.elapsed());
}

drop(held);
let started = Instant::now();
let one: i32 = sqlx::query_scalar("select 1").fetch_one(&pool).await?;
println!("released, query -> {one} after {:?}", started.elapsed());

holding: size=1 num_idle=0
after 1.500789042s: pool timed out while waiting for an open connection / PoolTimedOut
after 1.501650375s: pool timed out while waiting for an open connection / PoolTimedOut
after 1.501728792s: pool timed out while waiting for an open connection / PoolTimedOut
released, query -> 1 after 3.001542ms

        let acquired_after = acquire_started_at.elapsed();

        let acquire_slow_level = self
            .acquire_slow_level
            .filter(|_| acquired_after > self.options.acquire_slow_threshold);

        if let Some(level) = acquire_slow_level {
            private_tracing_dynamic_event!(
                target: "sqlx::pool::acquire",
                level,
                acquired_after_secs = acquired_after.as_secs_f64(),
                slow_acquire_threshold_secs = self.options.acquire_slow_threshold.as_secs_f64(),
                "acquired connection, but time to acquire exceeded slow threshold"
            );
        } else if let Some(level) = self.acquire_time_level {
            private_tracing_dynamic_event!(
                target: "sqlx::pool::acquire",
                level,
                acquired_after_secs = acquired_after.as_secs_f64(),
                "acquired connection"
            );
        }

let pool = PgPoolOptions::new()
  .max_connections(1)
  .acquire_slow_threshold(Duration::from_millis(200))
  .connect_with(lab::opts("pool-slow"))
  .await?;

println!("--- fast acquire ---");
drop(pool.acquire().await?);
tokio::time::sleep(Duration::from_millis(100)).await;

println!("--- someone holds the connection for 500ms ---");
let held = pool.acquire().await?;
tokio::spawn(async move {
  tokio::time::sleep(Duration::from_millis(500)).await;
  drop(held);
});
drop(pool.acquire().await?);

--- fast acquire ---
--- someone holds the connection for 500ms ---
 WARN sqlx::pool::acquire: acquired connection, but time to acquire exceeded slow threshold acquired_after_secs=0.503702334 slow_acquire_threshold_secs=0.2

closed port: after 1.005184208s: pool timed out while waiting for an open connection
57P03: after 1.003278709s: pool timed out while waiting for an open connection
  attempts at ms: [0, 14, 38, 82, 165, 329, 532, 736, 939]
28P01: after 425.542µs: error returned from database: password authentication failed for user "lab"
  attempts at ms: [0]