Dispatcher

pub(crate) struct Dispatcher<D, Bs: Body, I, T> {
    conn: Conn<I, Bs::Data, T>,
    dispatch: D,
    body_tx: SenderDropGuard,
    body_rx: Pin<Box<Option<Bs>>>,
    is_closing: bool,
}

pub(crate) trait Dispatch {
    type PollItem;
    type PollBody;
    type PollError;
    type RecvItem;
    fn poll_msg(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Self::PollError>>>;
    fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, IncomingBody)>)
        -> crate::Result<()>;
    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), ()>>;
    fn should_poll(&self) -> bool;
}

        fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, IncomingBody)>) -> crate::Result<()> {
            let (msg, body) = msg?;
            let mut req = Request::new(body);
            *req.method_mut() = msg.subject.0;
            *req.uri_mut() = msg.subject.1;
            *req.headers_mut() = msg.headers;
            *req.version_mut() = msg.version;
            *req.extensions_mut() = msg.extensions;
            let fut = self.service.call(req);
            self.in_flight.set(Some(fut));
            Ok(())
        }

        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), ()>> {
            if self.in_flight.is_some() {
                Poll::Pending
            } else {
                Poll::Ready(Ok(()))
            }
        }

    fn poll_loop(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<()>> {
        // Limit the looping on this connection, in case it is ready far too
        // often, so that other futures don't starve.
        //
        // 16 was chosen arbitrarily, as that is number of pipelined requests
        // benchmarks often use. Perhaps it should be a config option instead.
        for _ in 0..16 {
            let _ = self.poll_read(cx)?;
            let write_ready = self.poll_write(cx)?.is_ready();
            let flush_ready = self.poll_flush(cx)?.is_ready();

            // If we can write more body and the connection is ready, we should
            // write again. If we return `Ready(Ok(())` here, we will yield
            // without a guaranteed wake-up from the write side of the connection.
            // This would lead to a deadlock if we also don't expect reads.
            let wants_write_again = self.can_write_again() && (write_ready || flush_ready);

            // ...
            let wants_read_again = self.conn.wants_read_again();

            // If we cannot write or read again, we yield and rely on the
            // wake-up from the connection futures.
            if !(wants_write_again || wants_read_again) {
                return Poll::Ready(Ok(()));
            }

            // ...
            if !wants_read_again && wants_write_again {
                // ...
                if self.poll_write(cx)?.is_pending() {
                    // ...
                    if self.conn.has_buffered_write() {
                        let _ = self.poll_flush(cx)?;
                    }
                    return Poll::Ready(Ok(()));
                }
            }
        }
        trace!("poll_loop yielding (self = {:p})", self);
        task::yield_now(cx).map(|never| match never {})
    }

pub(crate) fn yield_now(cx: &mut Context<'_>) -> Poll<std::convert::Infallible> {
    cx.waker().wake_by_ref();
    Poll::Pending
}

        loop {
            if self.is_closing {
                return Poll::Ready(Ok(()));
            } else if self.conn.can_read_head() {
                ready!(self.poll_read_head(cx))?;
            } else if let Some(mut body) = self.body_tx.take() {
                if self.conn.can_read_body() {
                    match body.poll_ready(cx) {
                        Poll::Ready(Ok(())) => (),
                        Poll::Pending => {
                            self.body_tx.set(body);
                            return Poll::Pending;
                        }
                        Poll::Ready(Err(_canceled)) => {
                            // user doesn't care about the body
                            // so we should stop reading
                            trace!("body receiver dropped before eof, draining or closing");
                            self.conn.poll_drain_or_close_read(cx);
                            continue;
                        }
                    }
                    match self.conn.poll_read_body(cx) {
                        // ... 数据块 try_send_data，trailers try_send_trailers
                    }
                } else {
                    // just drop, the body will close automatically
                }
            } else {
                return self.conn.poll_read_keep_alive(cx);
            }
        }

    fn poll_read_head(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<()>> {
        // can dispatch receive, or does it still care about other incoming message?
        if let Ok(()) = ready!(self.dispatch.poll_ready(cx)) {
        } else {
            trace!("dispatch no longer receiving messages");
            self.close();
            return Poll::Ready(Ok(()));
        }

        // dispatch is ready for a message, try to read one
        match ready!(self.conn.poll_read_head(cx)) {
            Some(Ok((mut head, body_len, wants))) => {
                let body = match body_len {
                    DecodedLength::ZERO => IncomingBody::empty(),
                    other => {
                        let (tx, rx) =
                            IncomingBody::new_channel(other, wants.contains(Wants::EXPECT));
                        self.body_tx.set(tx);
                        rx
                    }
                };
                if wants.contains(Wants::UPGRADE) {
                    let upgrade = self.conn.on_upgrade();
                    // ...
                    head.extensions.insert(upgrade);
                }
                self.dispatch.recv_msg(Ok((head, body)))?;
                Poll::Ready(Ok(()))
            }
            // ...
        }
    }

            } else if self.body_rx.is_none()
                && self.conn.can_write_head()
                && self.dispatch.should_poll()
            {
                if let Some(msg) = ready!(Pin::new(&mut self.dispatch).poll_msg(cx)) {
                    let (head, body) = msg.map_err(crate::Error::new_user_service)?;

                    let body_type = if body.is_end_stream() {
                        self.body_rx.set(None);
                        None
                    } else {
                        let btype = body
                            .size_hint()
                            .exact()
                            .map(BodyLength::Known)
                            .or(Some(BodyLength::Unknown));
                        self.body_rx.set(Some(body));
                        btype
                    };
                    self.conn.write_head(head, body_type);
                } else {
                    self.close();
                    return Poll::Ready(Ok(()));
                }
            } else if !self.conn.can_buffer_body() {
                ready!(self.poll_flush(cx))?;
            } else {
                // ... 从 body_rx poll_frame，数据块 write_body / write_body_and_end，
                // trailers write_trailers，结束 end_body
            }

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

use std::{convert::Infallible, 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::{io::{AsyncReadExt, AsyncWriteExt}, net::{TcpListener, TcpStream}};

#[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();

  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();
      println!("{:>4}ms server: handler {path} start", ms());
      tokio::time::sleep(Duration::from_millis(200)).await;
      println!("{:>4}ms server: handler {path} done", ms());
      Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(path))))
    });
    http1::Builder::new().serve_connection(TokioIo::new(stream), svc).await.unwrap();
  });

  let mut client = TcpStream::connect(addr).await.unwrap();
  // three requests in a single write: pipelining
  client
    .write_all(b"GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\n\r\nGET /c HTTP/1.1\r\nHost: x\r\n\r\n")
    .await
    .unwrap();
  println!("{:>4}ms client: wrote 3 requests", ms());
  let mut buf = [0u8; 4096];
  let mut seen = 0;
  while seen < 3 {
    let n = client.read(&mut buf).await.unwrap();
    let text = String::from_utf8_lossy(&buf[..n]);
    let bodies: Vec<&str> = text.split("\r\n\r\n").skip(1).map(|s| s.split("HTTP/").next().unwrap()).collect();
    seen += bodies.len();
    println!("{:>4}ms client: read {n} bytes, bodies {bodies:?}", ms());
  }
}

   0ms client: wrote 3 requests
   0ms server: handler /a start
 203ms server: handler /a done
 203ms server: handler /b start
 203ms client: read 77 bytes, bodies ["/a"]
 407ms server: handler /b done
 407ms server: handler /c start
 407ms client: read 77 bytes, bodies ["/b"]
 610ms server: handler /c done
 611ms client: read 77 bytes, bodies ["/c"]

hyper = { version = "=1.12.0", features = ["server", "http1"] }
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::stream;
use http_body_util::StreamBody;
use hyper::{body::{Bytes, Frame, Incoming}, server::conn::http1, service::service_fn, Request, Response};
use hyper_util::rt::TokioIo;
use tokio::{io::{AsyncReadExt, AsyncWriteExt}, net::{TcpListener, TcpStream}};

#[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();

  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();
      println!("{:>4}ms server: handler {path} called", ms());
      // /a: two chunks, the second one 300ms later; /b: one chunk
      let n = if path == "/a" { 2 } else { 1 };
      let body = stream::unfold(0, move |i| async move {
        if i == n {
          return None;
        }
        if i > 0 {
          tokio::time::sleep(Duration::from_millis(300)).await;
        }
        println!("{:>4}ms server: body chunk {i}", ms());
        Some((Ok::<_, Infallible>(Frame::data(Bytes::from(format!("{i};")))), i + 1))
      });
      Ok::<_, Infallible>(Response::new(StreamBody::new(body)))
    });
    http1::Builder::new().serve_connection(TokioIo::new(stream), svc).await.unwrap();
  });

  let mut client = TcpStream::connect(addr).await.unwrap();
  client
    .write_all(b"GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\n\r\n")
    .await
    .unwrap();
  let mut buf = [0u8; 4096];
  let mut all = String::new();
  while !all.ends_with("0;\r\n0\r\n\r\n") || all.matches("HTTP/1.1").count() < 2 {
    let n = client.read(&mut buf).await.unwrap();
    all.push_str(&String::from_utf8_lossy(&buf[..n]));
  }
  println!("{:>4}ms client: both responses complete", ms());
}

   0ms server: handler /a called
   0ms server: body chunk 0
 303ms server: body chunk 1
 304ms server: handler /b called
 304ms server: body chunk 0
 304ms client: both responses complete

    fn is_done(&self) -> bool {
        if self.is_closing {
            return true;
        }

        let read_done = self.conn.is_read_closed();

        if !T::should_read_first() && read_done {
            // a client that cannot read may was well be done.
            true
        } else {
            let write_done = self.conn.is_write_closed()
                || (!self.dispatch.should_poll() && self.body_rx.is_none());
            read_done && write_done
        }
    }

        if self.is_done() {
            if let Some(pending) = self.conn.pending_upgrade() {
                self.conn.take_error()?;
                return Poll::Ready(Ok(Dispatched::Upgrade(pending)));
            } else if should_shutdown {
                ready!(self.conn.poll_shutdown(cx)).map_err(crate::Error::new_shutdown)?;
            }
            self.conn.take_error()?;
            Poll::Ready(Ok(Dispatched::Shutdown))
        } else {
            Poll::Pending
        }

poll #1: 16 requests, woke itself: true, Pending
poll #2: 16 requests, woke itself: true, Pending
poll #3: 8 requests, woke itself: false, Ready
connection: Ok(()), client got 40 responses