use std::error::Error as StdError; use bytes::{Buf, Bytes}; use http::Request; use tokio::io::{AsyncRead, AsyncWrite}; use tracing::{debug, trace}; use super::{Http1Transaction, Wants}; use crate::body::{Body, DecodedLength, HttpBody}; use crate::common::{task, Future, Pin, Poll, Unpin}; use crate::proto::{ BodyLength, Conn, Dispatched, MessageHead, RequestHead, }; use crate::upgrade::OnUpgrade; pub(crate) struct Dispatcher { conn: Conn, dispatch: D, body_tx: Option, body_rx: Pin>>, is_closing: bool, } pub(crate) trait Dispatch { type PollItem; type PollBody; type PollError; type RecvItem; fn poll_msg( self: Pin<&mut Self>, cx: &mut task::Context<'_>, ) -> Poll>>; fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()>; fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll>; fn should_poll(&self) -> bool; } cfg_server! { use crate::service::HttpService; pub(crate) struct Server, B> { in_flight: Pin>>, pub(crate) service: S, } } cfg_client! { pin_project_lite::pin_project! { pub(crate) struct Client { callback: Option, http::Response>>, #[pin] rx: ClientRx, rx_closed: bool, } } type ClientRx = crate::client::dispatch::Receiver, http::Response>; } impl Dispatcher where D: Dispatch< PollItem = MessageHead, PollBody = Bs, RecvItem = MessageHead, > + Unpin, D::PollError: Into>, I: AsyncRead + AsyncWrite + Unpin, T: Http1Transaction + Unpin, Bs: HttpBody + 'static, Bs::Error: Into>, { pub(crate) fn new(dispatch: D, conn: Conn) -> Self { Dispatcher { conn, dispatch, body_tx: None, body_rx: Box::pin(None), is_closing: false, } } #[cfg(feature = "server")] pub(crate) fn disable_keep_alive(&mut self) { self.conn.disable_keep_alive(); if self.conn.is_write_closed() { self.close(); } } pub(crate) fn into_inner(self) -> (I, Bytes, D) { let (io, buf) = self.conn.into_inner(); (io, buf, self.dispatch) } /// Run this dispatcher until HTTP says this connection is done, /// but don't call `AsyncWrite::shutdown` on the underlying IO. /// /// This is useful for old-style HTTP upgrades, but ignores /// newer-style upgrade API. pub(crate) fn poll_without_shutdown( &mut self, cx: &mut task::Context<'_>, ) -> Poll> where Self: Unpin, { Pin::new(self).poll_catch(cx, false).map_ok(|ds| { if let Dispatched::Upgrade(pending) = ds { pending.manual(); } }) } fn poll_catch( &mut self, cx: &mut task::Context<'_>, should_shutdown: bool, ) -> Poll> { Poll::Ready(ready!(self.poll_inner(cx, should_shutdown)).or_else(|e| { // An error means we're shutting down either way. // We just try to give the error to the user, // and close the connection with an Ok. If we // cannot give it to the user, then return the Err. self.dispatch.recv_msg(Err(e))?; Ok(Dispatched::Shutdown) })) } fn poll_inner( &mut self, cx: &mut task::Context<'_>, should_shutdown: bool, ) -> Poll> { T::update_date(); ready!(self.poll_loop(cx))?; 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 } } fn poll_loop(&mut self, cx: &mut task::Context<'_>) -> Poll> { // 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 _ = self.poll_write(cx)?; let _ = self.poll_flush(cx)?; // This could happen if reading paused before blocking on IO, // such as getting to the end of a framed message, but then // writing/flushing set the state back to Init. In that case, // if the read buffer still had bytes, we'd want to try poll_read // again, or else we wouldn't ever be woken up again. // // Using this instead of task::current() and notify() inside // the Conn is noticeably faster in pipelined benchmarks. if !self.conn.wants_read_again() { //break; return Poll::Ready(Ok(())); } } trace!("poll_loop yielding (self = {:p})", self); task::yield_now(cx).map(|never| match never {}) } fn poll_read(&mut self, cx: &mut task::Context<'_>) -> Poll> { 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 = Some(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) { Poll::Ready(Some(Ok(chunk))) => match body.try_send_data(chunk) { Ok(()) => { self.body_tx = Some(body); } Err(_canceled) => { if self.conn.can_read_body() { trace!("body receiver dropped before eof, closing"); self.conn.close_read(); } } }, Poll::Ready(None) => { // just drop, the body will close automatically } Poll::Pending => { self.body_tx = Some(body); return Poll::Pending; } Poll::Ready(Some(Err(e))) => { body.send_error(crate::Error::new_body(e)); } } } 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 task::Context<'_>) -> Poll> { // can dispatch receive, or does it still care about, an incoming message? match ready!(self.dispatch.poll_ready(cx)) { Ok(()) => (), Err(()) => { 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 => Body::empty(), other => { let (tx, rx) = Body::new_channel(other, wants.contains(Wants::EXPECT)); self.body_tx = Some(tx); rx } }; if wants.contains(Wants::UPGRADE) { let upgrade = self.conn.on_upgrade(); debug_assert!(!upgrade.is_none(), "empty upgrade"); debug_assert!(head.extensions.get::().is_none(), "OnUpgrade already set"); head.extensions.insert(upgrade); } self.dispatch.recv_msg(Ok((head, body)))?; Poll::Ready(Ok(())) } Some(Err(err)) => { debug!("read_head error: {}", err); self.dispatch.recv_msg(Err(err))?; // if here, the dispatcher gave the user the error // somewhere else. we still need to shutdown, but // not as a second error. self.close(); Poll::Ready(Ok(())) } None => { // read eof, the write side will have been closed too unless // allow_read_close was set to true, in which case just do // nothing... debug_assert!(self.conn.is_read_closed()); if self.conn.is_write_closed() { self.close(); } Poll::Ready(Ok(())) } } } fn poll_write(&mut self, cx: &mut task::Context<'_>) -> Poll> { loop { if self.is_closing { return 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, mut body) = msg.map_err(crate::Error::new_user_service)?; // Check if the body knows its full data immediately. // // If so, we can skip a bit of bookkeeping that streaming // bodies need to do. if let Some(full) = crate::body::take_full_data(&mut body) { self.conn.write_full_msg(head, full); return Poll::Ready(Ok(())); } 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_else(|| 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 { // A new scope is needed :( if let (Some(mut body), clear_body) = OptGuard::new(self.body_rx.as_mut()).guard_mut() { debug_assert!(!*clear_body, "opt guard defaults to keeping body"); if !self.conn.can_write_body() { trace!( "no more write body allowed, user body is_end_stream = {}", body.is_end_stream(), ); *clear_body = true; continue; } let item = ready!(body.as_mut().poll_data(cx)); if let Some(item) = item { let chunk = item.map_err(|e| { *clear_body = true; crate::Error::new_user_body(e) })?; let eos = body.is_end_stream(); if eos { *clear_body = true; if chunk.remaining() == 0 { trace!("discarding empty chunk"); self.conn.end_body()?; } else { self.conn.write_body_and_end(chunk); } } else { if chunk.remaining() == 0 { trace!("discarding empty chunk"); continue; } self.conn.write_body(chunk); } } else { *clear_body = true; self.conn.end_body()?; } } else { return Poll::Pending; } } } } fn poll_flush(&mut self, cx: &mut task::Context<'_>) -> Poll> { self.conn.poll_flush(cx).map_err(|err| { debug!("error writing: {}", err); crate::Error::new_body_write(err) }) } fn close(&mut self) { self.is_closing = true; self.conn.close_read(); self.conn.close_write(); } 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 } } } impl Future for Dispatcher where D: Dispatch< PollItem = MessageHead, PollBody = Bs, RecvItem = MessageHead, > + Unpin, D::PollError: Into>, I: AsyncRead + AsyncWrite + Unpin, T: Http1Transaction + Unpin, Bs: HttpBody + 'static, Bs::Error: Into>, { type Output = crate::Result; #[inline] fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll { self.poll_catch(cx, true) } } // ===== impl OptGuard ===== /// A drop guard to allow a mutable borrow of an Option while being able to /// set whether the `Option` should be cleared on drop. struct OptGuard<'a, T>(Pin<&'a mut Option>, bool); impl<'a, T> OptGuard<'a, T> { fn new(pin: Pin<&'a mut Option>) -> Self { OptGuard(pin, false) } fn guard_mut(&mut self) -> (Option>, &mut bool) { (self.0.as_mut().as_pin_mut(), &mut self.1) } } impl<'a, T> Drop for OptGuard<'a, T> { fn drop(&mut self) { if self.1 { self.0.set(None); } } } // ===== impl Server ===== cfg_server! { impl Server where S: HttpService, { pub(crate) fn new(service: S) -> Server { Server { in_flight: Box::pin(None), service, } } pub(crate) fn into_service(self) -> S { self.service } } // Service is never pinned impl, B> Unpin for Server {} impl Dispatch for Server where S: HttpService, S::Error: Into>, Bs: HttpBody, { type PollItem = MessageHead; type PollBody = Bs; type PollError = S::Error; type RecvItem = RequestHead; fn poll_msg( mut self: Pin<&mut Self>, cx: &mut task::Context<'_>, ) -> Poll>> { let mut this = self.as_mut(); let ret = if let Some(ref mut fut) = this.in_flight.as_mut().as_pin_mut() { let resp = ready!(fut.as_mut().poll(cx)?); let (parts, body) = resp.into_parts(); let head = MessageHead { version: parts.version, subject: parts.status, headers: parts.headers, extensions: parts.extensions, }; Poll::Ready(Some(Ok((head, body)))) } else { unreachable!("poll_msg shouldn't be called if no inflight"); }; // Since in_flight finished, remove it this.in_flight.set(None); ret } fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> 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 task::Context<'_>) -> Poll> { if self.in_flight.is_some() { Poll::Pending } else { self.service.poll_ready(cx).map_err(|_e| { // FIXME: return error value. trace!("service closed"); }) } } fn should_poll(&self) -> bool { self.in_flight.is_some() } } } // ===== impl Client ===== cfg_client! { impl Client { pub(crate) fn new(rx: ClientRx) -> Client { Client { callback: None, rx, rx_closed: false, } } } impl Dispatch for Client where B: HttpBody, { type PollItem = RequestHead; type PollBody = B; type PollError = crate::common::Never; type RecvItem = crate::proto::ResponseHead; fn poll_msg( mut self: Pin<&mut Self>, cx: &mut task::Context<'_>, ) -> Poll>> { let mut this = self.as_mut(); debug_assert!(!this.rx_closed); match this.rx.poll_recv(cx) { Poll::Ready(Some((req, mut cb))) => { // check that future hasn't been canceled already match cb.poll_canceled(cx) { Poll::Ready(()) => { trace!("request canceled"); Poll::Ready(None) } Poll::Pending => { let (parts, body) = req.into_parts(); let head = RequestHead { version: parts.version, subject: crate::proto::RequestLine(parts.method, parts.uri), headers: parts.headers, extensions: parts.extensions, }; this.callback = Some(cb); Poll::Ready(Some(Ok((head, body)))) } } } Poll::Ready(None) => { // user has dropped sender handle trace!("client tx closed"); this.rx_closed = true; Poll::Ready(None) } Poll::Pending => Poll::Pending, } } fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()> { match msg { Ok((msg, body)) => { if let Some(cb) = self.callback.take() { let res = msg.into_response(body); cb.send(Ok(res)); Ok(()) } else { // Getting here is likely a bug! An error should have happened // in Conn::require_empty_read() before ever parsing a // full message! Err(crate::Error::new_unexpected_message()) } } Err(err) => { if let Some(cb) = self.callback.take() { cb.send(Err((err, None))); Ok(()) } else if !self.rx_closed { self.rx.close(); if let Some((req, cb)) = self.rx.try_recv() { trace!("canceling queued request with connection error: {}", err); // in this case, the message was never even started, so it's safe to tell // the user that the request was completely canceled cb.send(Err((crate::Error::new_canceled().with(err), Some(req)))); Ok(()) } else { Err(err) } } else { Err(err) } } } } fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll> { match self.callback { Some(ref mut cb) => match cb.poll_canceled(cx) { Poll::Ready(()) => { trace!("callback receiver has dropped"); Poll::Ready(Err(())) } Poll::Pending => Poll::Ready(Ok(())), }, None => Poll::Ready(Err(())), } } fn should_poll(&self) -> bool { self.callback.is_none() } } } #[cfg(test)] mod tests { use super::*; use crate::proto::h1::ClientTransaction; use std::time::Duration; #[test] fn client_read_bytes_before_writing_request() { let _ = pretty_env_logger::try_init(); tokio_test::task::spawn(()).enter(|cx, _| { let (io, mut handle) = tokio_test::io::Builder::new().build_with_handle(); // Block at 0 for now, but we will release this response before // the request is ready to write later... let (mut tx, rx) = crate::client::dispatch::channel(); let conn = Conn::<_, bytes::Bytes, ClientTransaction>::new(io); let mut dispatcher = Dispatcher::new(Client::new(rx), conn); // First poll is needed to allow tx to send... assert!(Pin::new(&mut dispatcher).poll(cx).is_pending()); // Unblock our IO, which has a response before we've sent request! // handle.read(b"HTTP/1.1 200 OK\r\n\r\n"); let mut res_rx = tx .try_send(crate::Request::new(crate::Body::empty())) .unwrap(); tokio_test::assert_ready_ok!(Pin::new(&mut dispatcher).poll(cx)); let err = tokio_test::assert_ready_ok!(Pin::new(&mut res_rx).poll(cx)) .expect_err("callback should send error"); match (err.0.kind(), err.1) { (&crate::error::Kind::Canceled, Some(_)) => (), other => panic!("expected Canceled, got {:?}", other), } }); } #[tokio::test] async fn client_flushing_is_not_ready_for_next_request() { let _ = pretty_env_logger::try_init(); let (io, _handle) = tokio_test::io::Builder::new() .write(b"POST / HTTP/1.1\r\ncontent-length: 4\r\n\r\n") .read(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n") .wait(std::time::Duration::from_secs(2)) .build_with_handle(); let (mut tx, rx) = crate::client::dispatch::channel(); let mut conn = Conn::<_, bytes::Bytes, ClientTransaction>::new(io); conn.set_write_strategy_queue(); let dispatcher = Dispatcher::new(Client::new(rx), conn); let _dispatcher = tokio::spawn(async move { dispatcher.await }); let req = crate::Request::builder() .method("POST") .body(crate::Body::from("reee")) .unwrap(); let res = tx.try_send(req).unwrap().await.expect("response"); drop(res); assert!(!tx.is_ready()); } #[tokio::test] async fn body_empty_chunks_ignored() { let _ = pretty_env_logger::try_init(); let io = tokio_test::io::Builder::new() // no reading or writing, just be blocked for the test... .wait(Duration::from_secs(5)) .build(); let (mut tx, rx) = crate::client::dispatch::channel(); let conn = Conn::<_, bytes::Bytes, ClientTransaction>::new(io); let mut dispatcher = tokio_test::task::spawn(Dispatcher::new(Client::new(rx), conn)); // First poll is needed to allow tx to send... assert!(dispatcher.poll().is_pending()); let body = { let (mut tx, body) = crate::Body::channel(); tx.try_send_data("".into()).unwrap(); body }; let _res_rx = tx.try_send(crate::Request::new(body)).unwrap(); // Ensure conn.write_body wasn't called with the empty chunk. // If it is, it will trigger an assertion. assert!(dispatcher.poll().is_pending()); } }