连接池

pub trait Poolable: Unpin + Send + Sized + 'static {
    fn is_open(&self) -> bool;
    /// Reserve this connection.
    ///
    /// Allows for HTTP/2 to return a shared reservation.
    fn reserve(self) -> Reservation<Self>;
    fn can_share(&self) -> bool;
}

impl<T: Poolable, K: Key> Drop for Pooled<T, K> {
    fn drop(&mut self) {
        if let Some(value) = self.value.take() {
            if !value.is_open() {
                // If we *already* know the connection is done here,
                // it shouldn't be re-inserted back into the pool.
                return;
            }

            if let Some(pool) = self.pool.upgrade() {
                if let Ok(mut inner) = pool.lock() {
                    inner.put(self.key.clone(), value, &pool);
                }
            } else if !value.can_share() {
                trace!("pool dropped, dropping pooled ({:?})", self.key);
            }
            // Ver::Http2 is already in the Pool (or dead), so we wouldn't
            // have an actual reference to the Pool.
        }
    }
}

        if pooled.is_http2() || !pooled.is_pool_enabled() || pooled.is_ready() {
            drop(pooled);
        } else {
            let on_idle = poll_fn(move |cx| pooled.poll_ready(cx)).map(|_| ());
            self.exec.execute(on_idle);
        }

hyper = { version = "=1.12.0", features = ["server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
http-body-util = "0.1"
tokio = { version = "=1.54.1", features = ["full"] }

// common.rs: numbers every accepted connection, logs accept/close,
// and puts `x-conn: N` on every response.
use std::{
  convert::Infallible,
  net::SocketAddr,
  sync::{Arc, atomic::{AtomicUsize, Ordering}},
  time::{Duration, Instant},
};
use http_body_util::Full;
use hyper::{body::{Bytes, Incoming}, server::conn::http1, service::service_fn, Request, Response};
use hyper_util::rt::TokioIo;
use tokio::net::TcpListener;

pub fn clock() -> impl Fn() -> u128 + Clone + Send + 'static {
  let start = Instant::now();
  move || start.elapsed().as_millis()
}

pub async fn server(ms: impl Fn() -> u128 + Clone + Send + Sync + 'static) -> (SocketAddr, Arc<AtomicUsize>) {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let accepts = Arc::new(AtomicUsize::new(0));
  let counter = accepts.clone();
  tokio::spawn(async move {
    loop {
      let (stream, _) = listener.accept().await.unwrap();
      let n = counter.fetch_add(1, Ordering::SeqCst) + 1;
      let ms = ms.clone();
      println!("{:>4}ms server: accept #{n}", ms());
      tokio::spawn(async move {
        let svc = service_fn(move |req: Request<Incoming>| async move {
          let body = match req.uri().path() {
            "/slow" => {
              tokio::time::sleep(Duration::from_millis(200)).await;
              Bytes::from_static(b"slow")
            }
            "/big" => Bytes::from(vec![b'x'; 1 << 20]),
            _ => Bytes::from_static(b"hello"),
          };
          let mut res = Response::new(Full::new(body));
          res.headers_mut().insert("x-conn", n.into());
          Ok::<_, Infallible>(res)
        });
        let _ = http1::Builder::new().serve_connection(TokioIo::new(stream), svc).await;
        println!("{:>4}ms server: close #{n}", ms());
      });
    }
  });
  (addr, accepts)
}

#[path = "../common.rs"]
mod common;

use std::{sync::atomic::Ordering, time::Duration};
use http_body_util::{BodyExt, Empty};
use hyper::{body::Bytes, Uri};
use hyper_util::{
  client::legacy::{connect::HttpConnector, Client},
  rt::TokioExecutor,
};

#[tokio::main]
async fn main() {
  for case in ["read to end", "drop small", "drop 1 MiB", "hold 1 MiB"] {
    println!("--- {case}");
    let ms = common::clock();
    let (addr, accepts) = common::server(ms.clone()).await;
    let client: Client<HttpConnector, Empty<Bytes>> =
      Client::builder(TokioExecutor::new()).build_http();
    let path = if case.contains("MiB") { "/big" } else { "/" };
    let first = client.get(format!("http://{addr}{path}").parse::<Uri>().unwrap()).await.unwrap();
    let mut held = None;
    match case {
      "read to end" => { first.into_body().collect().await.unwrap(); }
      "hold 1 MiB" => held = Some(first),
      _ => drop(first),
    }
    tokio::time::sleep(Duration::from_millis(50)).await;
    let second = client.get(format!("http://{addr}/").parse::<Uri>().unwrap()).await.unwrap();
    println!(
      "{:>4}ms client: second request on conn #{}, accepts: {}",
      ms(),
      second.headers()["x-conn"].to_str().unwrap(),
      accepts.load(Ordering::SeqCst)
    );
    drop((held, second, client));
    tokio::time::sleep(Duration::from_millis(20)).await;
  }
}

--- read to end
   0ms server: accept #1
  54ms client: second request on conn #1, accepts: 1
  54ms server: close #1
--- drop small
   0ms server: accept #1
  53ms client: second request on conn #1, accepts: 1
  53ms server: close #1
--- drop 1 MiB
   0ms server: accept #1
   1ms server: close #1
  53ms server: accept #2
  53ms client: second request on conn #2, accepts: 2
  53ms server: close #2
--- hold 1 MiB
   0ms server: accept #1
  53ms server: accept #2
  53ms client: second request on conn #2, accepts: 2
  54ms server: close #2
  54ms server: close #1

    fn pop(self, expiration: &Expiration, now: Instant) -> Option<Idle<T>> {
        while let Some(entry) = self.list.pop() {
            // If the connection has been closed, or is older than our idle
            // timeout, simply drop it and keep looking...
            if !entry.value.is_open() {
                trace!("removing closed connection for {:?}", self.key);
                continue;
            }
            // ...
            if expiration.expires(entry.idle_at, now) {
                trace!("removing expired connection for {:?}", self.key);
                continue;
            }

            let value = match entry.value.reserve() {
                // ...
                Reservation::Unique(unique) => unique,
            };

            return Some(Idle {
                idle_at: entry.idle_at,
                value,
            });
        }

        None
    }

use std::time::{Duration, Instant};
use http_body_util::{BodyExt, Empty};
use hyper::{body::Bytes, Uri};
use hyper_util::{
  client::legacy::{connect::HttpConnector, Client},
  rt::TokioExecutor,
};
use tokio::{io::{AsyncReadExt, AsyncWriteExt}, net::TcpListener};

#[tokio::main]
async fn main() {
  let start = Instant::now();
  let ms = move || start.elapsed().as_millis();
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  // Raw server: answer one request with a keep-alive response,
  // then close the socket 100ms later.
  tokio::spawn(async move {
    for n in 1.. {
      let (mut s, _) = listener.accept().await.unwrap();
      println!("{:>4}ms server: accept #{n}", ms());
      tokio::spawn(async move {
        let mut buf = [0u8; 1024];
        let _ = s.read(&mut buf).await.unwrap();
        let res = format!("HTTP/1.1 200 OK\r\ncontent-length: 2\r\nx-conn: {n}\r\n\r\nok");
        s.write_all(res.as_bytes()).await.unwrap();
        tokio::time::sleep(Duration::from_millis(100)).await;
        drop(s);
        println!("{:>4}ms server: closed #{n}", ms());
      });
    }
  });

  // No retries: if the dead connection were handed out, we'd see an error.
  let client: Client<HttpConnector, Empty<Bytes>> = Client::builder(TokioExecutor::new())
    .retry_canceled_requests(false)
    .build_http();
  let uri: Uri = format!("http://{addr}/").parse().unwrap();
  for _ in 0..2 {
    let res = client.get(uri.clone()).await.unwrap();
    let conn = res.headers()["x-conn"].to_str().unwrap().to_string();
    res.into_body().collect().await.unwrap();
    println!("{:>4}ms client: ok on conn #{conn}", ms());
    tokio::time::sleep(Duration::from_millis(300)).await;
  }
}

   0ms server: accept #1
   1ms client: ok on conn #1
 104ms server: closed #1
 304ms server: accept #2
 304ms client: ok on conn #2
 407ms server: closed #2

        let checkout = self.pool.checkout(pool_key.clone());
        let connect = self.connect_to(pool_key);
        let is_ver_h2 = self.config.ver == Ver::Http2;

        // The order of the `select` is depended on below...

        match future::select(checkout, connect).await {
            // Checkout won, connect future may have been started or not.
            //
            // If it has, let it finish and insert back into the pool,
            // so as to not waste the socket...
            Either::Left((Ok(checked_out), connecting)) => {
                // ...
                if connecting.started() {
                    let bg = connecting
                        .map_err(|err| {
                            trace!("background connect error: {}", err);
                        })
                        .map(|_pooled| {
                            // dropping here should just place it in
                            // the Pool for us...
                        });
                    // An execute error here isn't important, we're just trying
                    // to prevent a waste of a socket...
                    self.exec.execute(bg);
                }
                Ok(checked_out)
            }
            // Connect won, checkout can just be dropped.
            Either::Right((Ok(connected), _checkout)) => Ok(connected),
            // ...
        }

        if let Some(waiters) = self.waiters.get_mut(&key) {
            while let Some(tx) = waiters.pop_front() {
                if !tx.is_canceled() {
                    let reserved = value.take().expect("value already sent");
                    let reserved = match reserved.reserve() {
                        // ...
                        Reservation::Unique(uniq) => uniq,
                    };
                    match tx.send(reserved) {
                        Ok(()) => {
                            if value.is_none() {
                                break;
                            } else {
                                continue;
                            }
                        }
                        Err(e) => {
                            value = Some(e);
                        }
                    }
                }

                trace!("put; removing canceled waiter for {:?}", key);
            }
            // ...
        }

hyper = { version = "=1.12.0", features = ["server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
http-body-util = "0.1"
tokio = { version = "=1.54.1", features = ["full"] }
tower = { version = "=0.5.3", features = ["util"] }

#[path = "../common.rs"]
mod common;

use std::{future::Future, pin::Pin, sync::atomic::Ordering, time::Duration};
use http_body_util::{BodyExt, Empty};
use hyper::{body::Bytes, Uri};
use hyper_util::{
  client::legacy::{connect::HttpConnector, Client},
  rt::TokioExecutor,
};
use tower::{service_fn, ServiceExt};

type Connecting = Pin<Box<dyn Future<Output = Result<hyper_util::rt::TokioIo<tokio::net::TcpStream>, Box<dyn std::error::Error + Send + Sync>>> + Send>>;

#[tokio::main]
async fn main() {
  let ms = common::clock();
  let (addr, accepts) = common::server(ms.clone()).await;

  // Every connect takes 300ms longer than it should.
  let ms2 = ms.clone();
  let connector = service_fn(move |dst: Uri| -> Connecting {
    let ms = ms2.clone();
    Box::pin(async move {
      println!("{:>4}ms client: connect start", ms());
      tokio::time::sleep(Duration::from_millis(300)).await;
      let io = HttpConnector::new().oneshot(dst).await?;
      println!("{:>4}ms client: connect done", ms());
      Ok(io)
    })
  });
  let client: Client<_, Empty<Bytes>> = Client::builder(TokioExecutor::new()).build(connector);

  let get = |path: &'static str| {
    let client = client.clone();
    let ms = ms.clone();
    async move {
      let uri: Uri = format!("http://{addr}{path}").parse().unwrap();
      let res = client.get(uri).await.unwrap();
      let conn = res.headers()["x-conn"].to_str().unwrap().to_string();
      res.into_body().collect().await.unwrap();
      println!("{:>4}ms client: {path} done on conn #{conn}", ms());
    }
  };

  let a = tokio::spawn(get("/slow"));
  tokio::time::sleep(Duration::from_millis(350)).await;
  let b = tokio::spawn(get("/"));
  a.await.unwrap();
  b.await.unwrap();
  tokio::time::sleep(Duration::from_millis(300)).await;
  get("/").await;
  println!("accepts: {}", accepts.load(Ordering::SeqCst));
}

   0ms client: connect start
 303ms client: connect done
 303ms server: accept #1
 353ms client: connect start
 507ms client: /slow done on conn #1
 507ms client: / done on conn #1
 655ms client: connect done
 655ms server: accept #2
 812ms client: / done on conn #2
accepts: 2

    fn spawn_idle_interval(&mut self, pool_ref: &Arc<Mutex<PoolInner<T, K>>>) {
        if self.idle_interval_ref.is_some() {
            return;
        }
        let dur = if let Some(dur) = self.timeout {
            dur
        } else {
            return;
        };
        if dur == Duration::ZERO {
            return;
        }
        let timer = if let Some(timer) = self.timer.clone() {
            timer
        } else {
            return;
        };

        // While someone might want a shorter duration, and it will be respected
        // at checkout time, there's no need to wake up and proactively evict
        // faster than this.
        const MIN_CHECK: Duration = Duration::from_millis(90);

        let dur = dur.max(MIN_CHECK);
        // ...
        self.exec.execute(interval.run());
    }

    async fn run(self) {
        use futures_util::future;

        let mut sleep = self.timer.sleep_until(self.timer.now() + self.duration);
        let mut on_pool_drop = self.pool_drop_notifier;
        loop {
            match future::select(&mut on_pool_drop, &mut sleep).await {
                future::Either::Left(_) => {
                    // pool dropped, bah-bye
                    break;
                }
                future::Either::Right(((), _)) => {
                    if let Some(inner) = self.pool.upgrade() {
                        if let Ok(mut inner) = inner.lock() {
                            trace!("idle interval checking for expired");
                            inner.clear_expired();
                            if inner.idle.is_empty() {
                                inner.idle_interval_ref = None;
                                trace!("pool empty, canceling idle interval");
                                return;
                            }
                        }
                    }

                    let deadline = self.timer.now() + self.duration;
                    self.timer.reset(&mut sleep, deadline);
                }
            }
        }
        // ...
    }

                    let idle_list = self.idle.entry(key.clone()).or_default();
                    if self.max_idle_per_host <= idle_list.len() {
                        trace!("max idle per host for {:?}, dropping connection", key);
                        return;
                    }

#[path = "../common.rs"]
mod common;

use std::{sync::atomic::Ordering, time::Duration};
use futures::future::join_all;
use http_body_util::{BodyExt, Empty};
use hyper::{body::Bytes, Uri};
use hyper_util::{
  client::legacy::{connect::HttpConnector, Client},
  rt::TokioExecutor,
};

#[tokio::main]
async fn main() {
  for max_idle in [usize::MAX, 1, 0] {
    let label = if max_idle == usize::MAX { "usize::MAX".to_string() } else { max_idle.to_string() };
    println!("--- pool_max_idle_per_host({label})");
    let ms = common::clock();
    let (addr, accepts) = common::server(ms.clone()).await;
    let client: Client<HttpConnector, Empty<Bytes>> = Client::builder(TokioExecutor::new())
      .pool_max_idle_per_host(max_idle)
      .build_http();
    let uri: Uri = format!("http://{addr}/slow").parse().unwrap();
    for round in 1..=2 {
      // three requests at the same time
      join_all((0..3).map(|_| async {
        let res = client.get(uri.clone()).await.unwrap();
        res.into_body().collect().await.unwrap();
      }))
      .await;
      tokio::time::sleep(Duration::from_millis(50)).await;
      println!("{:>4}ms client: round {round} done, accepts so far: {}", ms(), accepts.load(Ordering::SeqCst));
    }
    drop(client);
    tokio::time::sleep(Duration::from_millis(50)).await;
    println!("{:>4}ms client: dropped", ms());
  }
}

--- pool_max_idle_per_host(usize::MAX)
   0ms server: accept #1
   0ms server: accept #2
   0ms server: accept #3
 257ms client: round 1 done, accepts so far: 3
 514ms client: round 2 done, accepts so far: 3
 514ms server: close #1
 514ms server: close #2
 514ms server: close #3
 565ms client: dropped
--- pool_max_idle_per_host(1)
   0ms server: accept #1
   0ms server: accept #2
   0ms server: accept #3
 203ms server: close #2
 203ms server: close #1
 256ms client: round 1 done, accepts so far: 3
 257ms server: accept #4
 257ms server: accept #5
 461ms server: close #5
 461ms server: close #4
 514ms client: round 2 done, accepts so far: 5
 514ms server: close #3
 566ms client: dropped
--- pool_max_idle_per_host(0)
   0ms server: accept #1
   0ms server: accept #2
   0ms server: accept #3
 203ms server: close #2
 203ms server: close #3
 204ms server: close #1
 257ms client: round 1 done, accepts so far: 3
 257ms server: accept #4
 257ms server: accept #5
 257ms server: accept #6
 459ms server: close #6
 459ms server: close #4
 459ms server: close #5
 510ms client: round 2 done, accepts so far: 6
 564ms client: dropped

--- pool_idle_timeout(200ms), pool_timer: true
   0ms server: accept #1
   0ms client: done on conn #1
 204ms server: close #1
 605ms server: accept #2
 605ms client: done on conn #2
 808ms server: close #2
accepts: 2
--- pool_idle_timeout(200ms), pool_timer: false
   0ms server: accept #1
   0ms client: done on conn #1
 603ms server: close #1
 603ms server: accept #2
 604ms client: done on conn #2
accepts: 2