HTTP/2 客户端

/// An unbounded version.
///
/// Cannot poll the Giver, but can still use it to determine if the Receiver
/// has been dropped. However, this version can be cloned.
#[cfg(feature = "http2")]
pub(crate) struct UnboundedSender<T, U> {
    /// Only used for `is_closed`, since `mpsc::UnboundedSender` cannot be checked.
    giver: want::SharedGiver,
    inner: mpsc::UnboundedSender<Envelope<T, U>>,
}

    pub fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<crate::Result<()>> {
        if self.is_closed() {
            Poll::Ready(Err(crate::Error::new_closed()))
        } else {
            Poll::Ready(Ok(()))
        }
    }

    /// Dropping the returned future is the supported way to cancel an
    /// in-flight HTTP/2 request. The stream is reset with `RST_STREAM`
    /// (`CANCEL` error code); the shared connection remains usable for
    /// other in-flight and future requests. The peer is notified
    /// immediately rather than continuing to send a response body that
    /// would be discarded.

    let (h2_tx, mut conn) = new_builder(config)
        .handshake::<_, SendBuf<B::Data>>(Compat::new(io))
        .await
        .map_err(crate::Error::new_h2)?;

    // An mpsc channel is used entirely to detect when the
    // 'Client' has been dropped. This is to get around a bug
    // in h2 where dropping all SendRequests won't notify a
    // parked Connection.
    let (conn_drop_ref, conn_drop_rx) = mpsc::channel(1);
    let (cancel_tx, conn_eof) = oneshot::channel();

    let ping_config = new_ping_config(config);

    let (conn, ping) = if ping_config.is_enabled() {
        let pp = conn.ping_pong().expect("conn.ping_pong");
        let (recorder, ponger) = ping::channel(pp, ping_config, timer);

        let conn: Conn<_, B> = Conn::new(ponger, conn);
        (Either::left(conn), recorder)
    } else {
        (Either::right(conn), ping::disabled())
    };
    // ...
    exec.execute_h2_future(H2ClientFuture::Task {
        task: ConnTask::new(conn, conn_drop_rx, cancel_tx),
    });

    Ok(ClientTask {
        ping,
        conn_drop_ref,
        conn_eof,
        executor: exec,
        h2_tx,
        req_rx,
        fut_ctx: None,
        marker: PhantomData,
    })

        loop {
            match ready!(self.h2_tx.poll_ready(cx)) {
                Ok(()) => (),
                Err(err) => {
                    self.ping.ensure_not_timed_out()?;
                    return if err.reason() == Some(::h2::Reason::NO_ERROR) {
                        trace!("connection gracefully shutdown");
                        Poll::Ready(Ok(Dispatched::Shutdown))
                    } else {
                        Poll::Ready(Err(crate::Error::new_h2(err)))
                    };
                }
            }

            // If we were waiting on pending open
            // continue where we left off.
            if let Some(f) = self.fut_ctx.take() {
                self.poll_pipe(f, cx);
                continue;
            }

            match self.req_rx.poll_recv(cx) {
                Poll::Ready(Some((req, cb))) => {
                    // check that future hasn't been canceled already
                    if cb.is_canceled() {
                        trace!("request callback is canceled");
                        continue;
                    }
                    let (head, body) = req.into_parts();
                    let mut req = ::http::Request::from_parts(head, ());
                    super::strip_connection_headers(req.headers_mut(), super::MessageKind::Request);
                    if let Some(len) = body.size_hint().exact() {
                        if len != 0 || headers::method_has_defined_payload_semantics(req.method()) {
                            headers::set_content_length_if_missing(req.headers_mut(), len);
                        }
                    }

                    let is_connect = req.method() == Method::CONNECT;
                    let eos = body.is_end_stream();

                    // ... CONNECT 带非零 Content-Length：回调直接给错误

                    let (fut, body_tx) = match self.h2_tx.send_request(req, !is_connect && eos) {
                        Ok(ok) => ok,
                        Err(err) => {
                            // ... 回调给错误，continue
                        }
                    };

                    let f = FutCtx {
                        is_connect,
                        eos,
                        fut,
                        body_tx,
                        body,
                        cb,
                    };

                    // Check poll_ready() again.
                    // If the call to send_request() resulted in the new stream being pending open
                    // we have to wait for the open to complete before accepting new requests.
                    match self.h2_tx.poll_ready(cx) {
                        Poll::Pending => {
                            // Save Context
                            self.fut_ctx = Some(f);
                            return Poll::Pending;
                        }
                        // ...
                    }
                    self.poll_pipe(f, cx);
                }

                Poll::Ready(None) => {
                    trace!("client::dispatch::Sender dropped");
                    return Poll::Ready(Ok(Dispatched::Shutdown));
                }

                Poll::Pending => match ready!(Pin::new(&mut self.conn_eof).poll(cx)) {
                    // ...
                    Err(_conn_is_eof) => {
                        trace!("connection task is closed, closing dispatch task");
                        return Poll::Ready(Ok(Dispatched::Shutdown));
                    }
                },
            }
        }

        let send_stream = if !f.is_connect {
            if !f.eos {
                let mut pipe = PipeToSendStream::new(f.body, f.body_tx);

                // eagerly see if the body pipe is ready and
                // can thus skip allocating in the executor
                match Pin::new(&mut pipe).poll(cx) {
                    Poll::Ready(_) => (),
                    Poll::Pending => {
                        let conn_drop_ref = self.conn_drop_ref.clone();
                        // keep the ping recorder's knowledge of an
                        // "open stream" alive while this body is
                        // still sending...
                        let ping = ping.clone();

                        let pipe = PipeMap {
                            pipe,
                            conn_drop_ref: Some(conn_drop_ref),
                            ping: Some(ping),
                            cancel_rx: Some(cancel_rx),
                        };
                        // Clear send task
                        self.executor
                            .execute_h2_future(H2ClientFuture::Pipe { pipe });
                    }
                }
            }

            None
        } else {
            Some(f.body_tx)
        };

        self.executor.execute_h2_future(H2ClientFuture::Send {
            send_when: SendWhen {
                when: ResponseFutMap {
                    fut: f.fut,
                    ping: Some(ping),
                    send_stream: Some(send_stream),
                    exec: self.executor.clone(),
                    cancel_tx: Some(cancel_tx),
                },
                call_back: Some(f.cb),
            },
        });

       client execute H2ClientFuture -> task 2
       client: Connection future -> task 3
       client: request /a -> task 4
       client: request /b -> task 5
       client: request /c -> task 6
       client execute H2ClientFuture -> task 7
       client execute H2ClientFuture -> task 8
       client execute H2ClientFuture -> task 9

        match Pin::new(&mut this.when).poll(cx) {
            Poll::Ready(Ok(res)) => {
                call_back.send(Ok(res));
                Poll::Ready(())
            }
            Poll::Pending => {
                // check if the callback is canceled
                match call_back.poll_canceled(cx) {
                    Poll::Ready(v) => v,
                    Poll::Pending => {
                        // Move call_back back to struct before return
                        this.call_back.set(Some(call_back));
                        return Poll::Pending;
                    }
                }
                trace!("send_when canceled");
                // Tell pipe_task to reset the h2 stream so that
                // RST_STREAM is sent and flow-control capacity freed.
                this.when.as_mut().cancel();
                Poll::Ready(())
            }
            // ...
        }

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

use std::{convert::Infallible, time::{Duration, Instant}};
use futures::channel::mpsc;
use http_body_util::{combinators::BoxBody, BodyExt, Empty, StreamBody};
use hyper::{body::{Bytes, Frame, Incoming}, client, server, service::service_fn, Request, Response};
use hyper_util::rt::{TokioExecutor, TokioIo};
use tokio::net::{TcpListener, TcpStream};

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

// Prints when the handler's future is dropped.
struct Guard(String, Instant);
impl Drop for Guard {
  fn drop(&mut self) {
    println!("{:>5}ms server: handler {} future dropped", self.1.elapsed().as_millis(), self.0);
  }
}

type Body = BoxBody<Bytes, Infallible>;

#[tokio::main(flavor = "current_thread")]
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 show = &["HEADERS", "DATA", "RST_STREAM", "GOAWAY"];
  let proxy = sniff::proxy(listener.local_addr().unwrap(), start, show).await;

  tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    let svc = service_fn(move |req: Request<Incoming>| async move {
      let path = req.uri().path().to_string();
      let _guard = Guard(path.clone(), start);
      if path == "/upload" {
        let mut body = req.into_body();
        while let Some(frame) = body.frame().await {
          match frame {
            Ok(f) => println!("{:>5}ms server: /upload got {:?}", ms(), f.into_data().unwrap()),
            Err(e) => {
              let source = std::error::Error::source(&e).unwrap();
              println!("{:>5}ms server: /upload body error: {e}: {source}", ms());
              break;
            }
          }
        }
      } else if path == "/slow" {
        tokio::time::sleep(Duration::from_secs(1)).await;
      }
      Ok::<_, Infallible>(Response::new(Empty::<Bytes>::new()))
    });
    let r = server::conn::http2::Builder::new(TokioExecutor::new())
      .serve_connection(TokioIo::new(stream), svc)
      .await;
    println!("{:>5}ms server conn: {r:?}", ms());
  });

  let stream = TcpStream::connect(proxy).await.unwrap();
  let (mut sender, conn) = client::conn::http2::handshake::<_, _, Body>(TokioExecutor::new(), TokioIo::new(stream))
    .await
    .unwrap();
  let client_conn = tokio::spawn(conn);
  let timeout = Duration::from_millis(100);

  // 1: GET whose response never comes in time
  let req = Request::get("http://x/slow").body(Empty::new().boxed()).unwrap();
  let r = tokio::time::timeout(timeout, sender.send_request(req)).await;
  println!("{:>5}ms client: /slow timed out = {}", ms(), r.is_err());
  tokio::time::sleep(Duration::from_millis(50)).await;

  // 2: POST whose body sends one chunk and then never ends
  let (mut tx, rx) = mpsc::channel::<Result<Frame<Bytes>, Infallible>>(1);
  tx.try_send(Ok(Frame::data(Bytes::from_static(b"part1")))).unwrap();
  let req = Request::post("http://x/upload").body(StreamBody::new(rx).boxed()).unwrap();
  let r = tokio::time::timeout(timeout, sender.send_request(req)).await;
  println!("{:>5}ms client: /upload timed out = {}", ms(), r.is_err());
  tokio::time::sleep(Duration::from_millis(50)).await;

  // 3: the same connection is still usable
  let req = Request::get("http://x/ok").body(Empty::new().boxed()).unwrap();
  let res = sender.send_request(req).await;
  println!("{:>5}ms client: /ok -> {:?}", ms(), res.map(|r| r.status()));
  println!("{:>5}ms client: conn finished = {}", ms(), client_conn.is_finished());
  drop(tx);
}

    1ms c->s HEADERS stream=1 END_STREAM
  104ms client: /slow timed out = true
  104ms c->s RST_STREAM stream=1 CANCEL
  104ms server: handler /slow future dropped
  157ms c->s HEADERS stream=3 
  157ms c->s DATA stream=3 len=5
  157ms server: /upload got b"part1"
  258ms client: /upload timed out = true
  258ms c->s RST_STREAM stream=3 CANCEL
  258ms server: /upload body error: error reading a body from connection: stream error received: stream no longer needed
  258ms server: handler /upload future dropped
  311ms c->s HEADERS stream=5 END_STREAM
  311ms server: handler /ok future dropped
  311ms s->c HEADERS stream=5 END_STREAM
  311ms client: /ok -> Ok(200)
  311ms client: conn finished = false

  202ms client: is_closed=false
  202ms client conn: Ok(())
  202ms client: /late -> Err(hyper::Error(Canceled, "connection closed"))

handshake: 1
POST empty : +1 (server read b"0" bytes)
POST full  : +2 (server read b"5" bytes)
POST stream: +2 (server read b"6" bytes)
POST nodata: +1 (server read b"0" bytes)