client::pool：四块积木

    pub enum State<F, S> {
        Empty,
        Making(Batch<F, S>),
        Made(S),
    }

            State::Made(ref svc) => SingletonFuture::Made {
                svc: Some(svc.clone()),
                state: Arc::downgrade(&self.state),
            },

            if self.driver.as_ref().is_some_and(|driver| driver.id == id) {
                if let Some(waiter) = self.waiters.pop() {
                    let waker = waiter.waker;
                    self.driver = Some(Driver {
                        id: waiter.id,
                        waker,
                    });
                    self.wake_driver();
                    return false;
                }

                self.driver = None;
                self.future = None;
                return true;
            }

== Singleton: the driver is dropped halfway
  b ok=true c ok=true makes=1

    impl<F, S> Drop for SingletonFuture<F, S> {
        fn drop(&mut self) {
            if let SingletonFuture::Participating { id, state, .. } = self {
                if let Ok(mut locked) = state.lock() {
                    if let State::Making(batch) = &mut *locked {
                        if batch.remove(*id) {
                            *locked = State::Empty;
                        }
                    }
                }
            }
        }
    }

                        State::Empty => {
                            unreachable!("singleton participant polled after making was canceled")
                        }

  let mut s = Singleton::new(Maker(Arc::new(AtomicUsize::new(0))));
  let mut a = s.call(());
  let ra = (&mut a).await; // a finishes with an error, but is not dropped yet
  println!("a: {:?}", ra.map(|_| ()).map_err(|e| e.to_string()));
  let b = s.call(()); // starts a new batch
  drop(a); // drop the finished participant of the old batch
  let rb = tokio::spawn(async move { b.await.map(|_| ()).map_err(|e| e.to_string()) }).await;
  println!("b: {rb:?}");
  let rc = tokio::spawn(async move { s.call(()).await.map(|_| ()).map_err(|e| e.to_string()) }).await;
  println!("c: {rc:?}");

a: Err("singleton connection error")

thread 'tokio-rt-worker' (2127285) panicked at .../hyper-util-0.1.21/src/client/pool/singleton.rs:288:29:
internal error: entered unreachable code: singleton participant polled after making was canceled
b: Err(JoinError::Panic(Id(13), "internal error: entered unreachable code: singleton participant polled after making was canceled", ...))

thread 'tokio-rt-worker' (2127285) panicked at .../hyper-util-0.1.21/src/client/pool/singleton.rs:111:44:
called `Result::unwrap()` on an `Err` value: PoisonError { .. }
c: Err(JoinError::Panic(Id(14), "called `Result::unwrap()` on an `Err` value: PoisonError { .. }", ...))

    pub struct Shared<S> {
        services: Vec<S>,
        waiters: VecDeque<Waiter>,
        reservations: Vec<(WaiterId, S)>,
        next_waiter: usize,
    }

                        if let Some(pool_got) = locked.take_reserved(*waiter) {
                            events.on_race_lost(BackgroundConnect {
                                future: future.take().expect("racing future polled after done"),
                                shared: Arc::downgrade(&shared),
                            });
                            return Poll::Ready(Ok(Cached::new(pool_got, Arc::downgrade(&shared))));
                        }

== Cache, race lost, builder().executor: false
  first: conn #1
  second got conn #1
  connections finished by the connector: 1
== Cache, race lost, builder().executor: true
  first: conn #1
  second got conn #1
  connections finished by the connector: 2

            // States:
            // - `Eager`: try the "right" path first; on `UseOther` sentinel, fall back to left.
            // - `Fallback`: try the left path; on `UseOther` sentinel, upgrade back to right.
            // - `Upgrade`: retry the right path after a fallback.
            // If all fail, give up.

    pub fn service(&mut self, req: &Req) -> &mut T::Service {
        let key = self.targeter.key(req);
        self.map
            .entry(key)
            .or_insert_with(|| self.targeter.service(req))
    }

hyper = { version = "1", features = ["full"] }   # 1.12.0
hyper-util = { version = "=0.1.21", features = ["full"] }
tokio = { version = "=1.54.1", features = ["full"] }
tower = { version = "=0.5.3", features = ["full"] }
http-body-util = "0.1"
bytes = "1"
http = "1"

#[derive(Clone)]
struct H2Svc(http2::SendRequest<Empty<Bytes>>);
impl Service<Req> for H2Svc {
  type Response = Response<Incoming>;
  type Error = hyper::Error;
  type Future = Pin<Box<dyn Future<Output = Result<Response<Incoming>, hyper::Error>> + Send>>;
  fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), hyper::Error>> { self.0.poll_ready(cx) }
  fn call(&mut self, req: Req) -> Self::Future { Box::pin(self.0.send_request(req)) }
}

#[derive(Clone)]
struct H2Make<S>(S);
impl<S, D> Service<D> for H2Make<S>
where
  S: Service<D, Response = TokioIo<TcpStream>>,
  S::Error: Into<BoxError>,
  S::Future: Send + 'static,
{
  type Response = H2Svc;
  type Error = BoxError;
  type Future = BoxFut<H2Svc>;
  fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), BoxError>> { Poll::Ready(Ok(())) }
  fn call(&mut self, dst: D) -> Self::Future {
    let fut = self.0.call(dst);
    Box::pin(async move {
      let io = fut.await.map_err(Into::into)?;
      let (tx, conn) = http2::handshake(TokioExecutor::new(), io).await?;
      tokio::spawn(conn);
      Ok(H2Svc(tx))
    })
  }
}

  // stand-in for ALPN: the connection "negotiated h2" if it goes to the h2 server
  let make_pool = move |_dst: &Uri| {
    negotiate::builder()
      .connect(HttpConnector::new())
      .inspect(move |io: &TokioIo<TcpStream>| io.inner().peer_addr().map(|a| a.port() == h2_port).unwrap_or(false))
      .fallback(layer_fn(|s| cache::builder().build(H1Make(s))))
      .upgrade(layer_fn(|s| Singleton::new(H2Make(s))))
      .build::<Uri>()
  };
  let map = Arc::new(Mutex::new(
    Map::builder::<Uri>()
      .keys(|uri: &Uri| (uri.scheme().cloned(), uri.authority().cloned()))
      .values(make_pool)
      .build(),
  ));

  let send = |uri: Uri| {
    let map = map.clone();
    async move {
      let mut pool = map.lock().unwrap().service(&uri).clone();
      let mut conn = pool.ready().await?.call(uri.clone()).await?;
      let kind = if conn.upgraded_ref().is_some() { "upgraded(h2)" } else { "fallback(h1)" };
      let req = Request::get(uri).body(Empty::new()).unwrap();
      let resp = conn.ready().await?.call(req).await?;
      let v = resp.version();
      resp.into_body().collect().await?;
      Ok::<_, BoxError>(format!("{kind} {v:?}"))
    }
  };

-- run 1
h1 sequential #1 -> Ok("fallback(h1) HTTP/1.1"), server accepts: 1
h1 sequential #2 -> Ok("fallback(h1) HTTP/1.1"), server accepts: 1
h1 sequential #3 -> Ok("fallback(h1) HTTP/1.1"), server accepts: 1
h2 sequential #1 -> Ok("upgraded(h2) HTTP/2.0"), server accepts: 1
h2 sequential #2 -> Ok("upgraded(h2) HTTP/2.0"), server accepts: 1
h2 sequential #3 -> Ok("upgraded(h2) HTTP/2.0"), server accepts: 1
h2 8 concurrent -> {"panic": 8}, new accepts: 6
h1 8 concurrent -> {"fallback(h1) HTTP/1.1": 4, "panic": 4}, new accepts: 4
-- run 2
h2 8 concurrent -> {"panic": 8}, new accepts: 3
h1 8 concurrent -> {"fallback(h1) HTTP/1.1": 8}, new accepts: 8
-- run 3
h2 8 concurrent -> {"panic": 8}, new accepts: 3
h1 8 concurrent -> {"fallback(h1) HTTP/1.1": 6, "panic": 2}, new accepts: 6