Executor 与 Timer

pub trait Executor<Fut> {
    /// Place the future into the executor to be run.
    fn execute(&self, fut: Fut);
}

impl<Fut> Executor<Fut> for TokioExecutor
where
    Fut: Future + Send + 'static,
    Fut::Output: Send + 'static,
{
    fn execute(&self, fut: Fut) {
        #[cfg(feature = "rt-tracing-exec-force")]
        tokio::spawn(fut.in_current_span());

        #[cfg(not(feature = "rt-tracing-exec-force"))]
        tokio::spawn(fut);
    }
}

    pub trait Http2ServerConnExec<F, B: Body>:
        super::Http2UpgradedExec<B::Data> + sealed::Sealed<(F, B)> + Clone
    {
        #[doc(hidden)]
        fn execute_h2stream(&mut self, fut: H2Stream<F, B, Self>);
    }

    #[doc(hidden)]
    impl<E, F, B> Http2ServerConnExec<F, B> for E
    where
        E: Clone,
        E: Executor<H2Stream<F, B, E>>,
        E: super::Http2UpgradedExec<B::Data>,
        H2Stream<F, B, E>: Future<Output = ()>,
        B: Body,
    {
        fn execute_h2stream(&mut self, fut: H2Stream<F, B, E>) {
            self.execute(fut);
        }
    }

pub trait Timer {
    /// Return a future that resolves in `duration` time.
    fn sleep(&self, duration: Duration) -> Pin<Box<dyn Sleep>>;

    /// Return a future that resolves at `deadline`.
    fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn Sleep>>;

    /// Return an `Instant` representing the current time.
    ///
    /// The default implementation returns [`Instant::now()`].
    fn now(&self) -> Instant {
        Instant::now()
    }

    /// Reset a future to resolve at `new_deadline` instead.
    fn reset(&self, sleep: &mut Pin<Box<dyn Sleep>>, new_deadline: Instant) {
        *sleep = self.sleep_until(new_deadline);
    }
}

pub trait Sleep: Send + Sync + Future<Output = ()> {
    #[doc(hidden)]
    /// This method is private and can not be implemented by downstream crates.
    fn __type_id(&self, _: private::Sealed) -> TypeId
    where
        Self: 'static,
    {
        TypeId::of::<Self>()
    }
}

    fn reset(&self, sleep: &mut Pin<Box<dyn Sleep>>, new_deadline: Instant) {
        if let Some(sleep) = sleep.as_mut().downcast_mut_pin::<TokioSleep>() {
            sleep.reset(new_deadline)
        }
    }

pub(crate) enum Time {
    Timer(Arc<dyn Timer + Send + Sync>),
    Empty,
}

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

use std::convert::Infallible;
use std::sync::Mutex;
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{Empty, Full};
use hyper::{client, server, service::service_fn, Request, Response};
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use tokio::net::{TcpListener, TcpStream};

static PANIC_MSG: Mutex<Option<String>> = Mutex::new(None);

async fn hello(_req: Request<hyper::body::Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
  Ok(Response::new(Full::new(Bytes::from_static(b"ok"))))
}

fn report<T: std::fmt::Debug, E: std::fmt::Display>(r: Result<Result<T, String>, E>, is_panic: bool) -> String {
  match r {
    Ok(Ok(v)) => format!("Ok({v:?})"),
    Ok(Err(e)) => format!("Err({e})"),
    Err(e) if is_panic => format!("panic: {}", PANIC_MSG.lock().unwrap().take().unwrap_or(e.to_string())),
    Err(e) => format!("join error: {e}"),
  }
}

async fn run_case(name: &str, srv: server::conn::http2::Builder<TokioExecutor>, cli: client::conn::http2::Builder<TokioExecutor>) {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let server = tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    srv.serve_connection(TokioIo::new(stream), service_fn(hello)).await.map_err(|e| e.to_string())
  });
  let client = tokio::spawn(async move {
    let stream = TcpStream::connect(addr).await.unwrap();
    let (mut tx, conn) = cli.handshake(TokioIo::new(stream)).await.map_err(|e| e.to_string())?;
    let conn_task = tokio::spawn(conn);
    let req = Request::get("http://x/").body(Empty::<Bytes>::new()).unwrap();
    let res = tx.send_request(req).await.map_err(|e| e.to_string())?;
    drop(tx);
    let _ = conn_task.await;
    Ok::<_, String>(res.status())
  });
  let c = client.await;
  let c_panic = c.as_ref().err().is_some_and(|e| e.is_panic());
  let c = report(c, c_panic);
  let s = tokio::time::timeout(Duration::from_secs(3), server).await;
  let s = match s {
    Ok(r) => {
      let p = r.as_ref().err().is_some_and(|e| e.is_panic());
      report(r, p)
    }
    Err(_) => "still running after 3s".to_string(),
  };
  println!("{name}\n  client: {c}\n  server: {s}");
}

#[tokio::main]
async fn main() {
  std::panic::set_hook(Box::new(|info| {
    let loc = info.location().map(|l| {
      let file = l.file();
      let short = file.find("hyper-1.12.0/").map_or(file, |i| &file[i..]);
      format!("{short}:{}", l.line())
    });
    let msg = info
      .payload()
      .downcast_ref::<String>()
      .cloned()
      .or_else(|| info.payload().downcast_ref::<&str>().map(|s| s.to_string()))
      .unwrap_or_default();
    *PANIC_MSG.lock().unwrap() = Some(format!("{msg} at {}", loc.unwrap_or_default()));
  }));
  let exec = TokioExecutor::new();
  let plain_cli = || client::conn::http2::Builder::new(TokioExecutor::new());
  let plain_srv = || server::conn::http2::Builder::new(TokioExecutor::new());

  let mut s = plain_srv();
  s.keep_alive_interval(Duration::from_secs(1));
  run_case("1 server keep_alive_interval, no timer", s, plain_cli()).await;

  let mut s = plain_srv();
  s.adaptive_window(true);
  run_case("2 server adaptive_window, no timer", s, plain_cli()).await;

  let mut c = client::conn::http2::Builder::new(exec.clone());
  c.keep_alive_interval(Duration::from_secs(1));
  run_case("3 client keep_alive_interval, no timer", plain_srv(), c).await;

  let mut s = plain_srv();
  s.keep_alive_interval(Duration::from_secs(1)).timer(TokioTimer::new());
  run_case("4 server keep_alive_interval + TokioTimer", s, plain_cli()).await;
}

1 server keep_alive_interval, no timer
  client: Err(connection error)
  server: panic: You must supply a timer. at hyper-1.12.0/src/common/time.rs:38
2 server adaptive_window, no timer
  client: Ok(200)
  server: Ok(())
3 client keep_alive_interval, no timer
  client: panic: You must supply a timer. at hyper-1.12.0/src/common/time.rs:38
  server: Ok(())
4 server keep_alive_interval + TokioTimer
  client: Ok(200)
  server: Ok(())

pub(crate) enum Dur {
    Default(Option<Duration>),
    Configured(Option<Duration>),
}

    pub(crate) fn check(&self, dur: Dur, name: &'static str) -> Option<Duration> {
        match dur {
            Dur::Default(Some(dur)) => match self {
                Time::Empty => {
                    warn!("timeout `{}` has default, but no timer set", name,);
                    None
                }
                Time::Timer(..) => Some(dur),
            },
            Dur::Configured(Some(dur)) => match self {
                #[allow(clippy::panic)]
                Time::Empty => panic!("timeout `{name}` set, but no timer set"),
                Time::Timer(..) => Some(dur),
            },
            Dur::Default(None) | Dur::Configured(None) => None,
        }
    }

A default, no timer        client: still open after 35.0s, got 0 bytes
                           server: Err(connection closed before message completed (is_timeout=false))
B default, timer           client: closed after 30.0s, got 0 bytes
                           server: Err(read header from client timeout (is_timeout=true))
C 1s, no timer             client: read error after 0.0s: Connection reset by peer (os error 54)
                           server: panic: timeout `header_read_timeout` set, but no timer set at hyper-1.12.0/src/common/time.rs:84
D 1s, timer, partial       client: closed after 1.0s, got 0 bytes
                           server: Err(read header from client timeout (is_timeout=true))
E 1s, timer, idle          client: closed after 1.0s, got 77 bytes
                           server: Err(read header from client timeout (is_timeout=true))
F None, no timer           client: still open after 35.0s, got 0 bytes
                           server: Err(connection closed before message completed (is_timeout=false))