1 use std::error::Error as StdError;
2 #[cfg(feature = "runtime")]
3 use std::time::Duration;
4
5 use bytes::Bytes;
6 use futures_channel::{mpsc, oneshot};
7 use futures_util::future::{self, Either, FutureExt as _, TryFutureExt as _};
8 use futures_util::stream::StreamExt as _;
9 use h2::client::{Builder, SendRequest};
10 use http::{Method, StatusCode};
11 use tokio::io::{AsyncRead, AsyncWrite};
12 use tracing::{debug, trace, warn};
13
14 use super::{ping, H2Upgraded, PipeToSendStream, SendBuf};
15 use crate::body::HttpBody;
16 use crate::common::{exec::Exec, task, Future, Never, Pin, Poll};
17 use crate::headers;
18 use crate::proto::h2::UpgradedSendStream;
19 use crate::proto::Dispatched;
20 use crate::upgrade::Upgraded;
21 use crate::{Body, Request, Response};
22
23 type ClientRx<B> = crate::client::dispatch::Receiver<Request<B>, Response<Body>>;
24
25 ///// An mpsc channel is used to help notify the `Connection` task when *all*
26 ///// other handles to it have been dropped, so that it can shutdown.
27 type ConnDropRef = mpsc::Sender<Never>;
28
29 ///// A oneshot channel watches the `Connection` task, and when it completes,
30 ///// the "dispatch" task will be notified and can shutdown sooner.
31 type ConnEof = oneshot::Receiver<Never>;
32
33 // Our defaults are chosen for the "majority" case, which usually are not
34 // resource constrained, and so the spec default of 64kb can be too limiting
35 // for performance.
36 const DEFAULT_CONN_WINDOW: u32 = 1024 * 1024 * 5; // 5mb
37 const DEFAULT_STREAM_WINDOW: u32 = 1024 * 1024 * 2; // 2mb
38 const DEFAULT_MAX_FRAME_SIZE: u32 = 1024 * 16; // 16kb
39
40 #[derive(Clone, Debug)]
41 pub(crate) struct Config {
42 pub(crate) adaptive_window: bool,
43 pub(crate) initial_conn_window_size: u32,
44 pub(crate) initial_stream_window_size: u32,
45 pub(crate) max_frame_size: u32,
46 #[cfg(feature = "runtime")]
47 pub(crate) keep_alive_interval: Option<Duration>,
48 #[cfg(feature = "runtime")]
49 pub(crate) keep_alive_timeout: Duration,
50 #[cfg(feature = "runtime")]
51 pub(crate) keep_alive_while_idle: bool,
52 pub(crate) max_concurrent_reset_streams: Option<usize>,
53 }
54
55 impl Default for Config {
default() -> Config56 fn default() -> Config {
57 Config {
58 adaptive_window: false,
59 initial_conn_window_size: DEFAULT_CONN_WINDOW,
60 initial_stream_window_size: DEFAULT_STREAM_WINDOW,
61 max_frame_size: DEFAULT_MAX_FRAME_SIZE,
62 #[cfg(feature = "runtime")]
63 keep_alive_interval: None,
64 #[cfg(feature = "runtime")]
65 keep_alive_timeout: Duration::from_secs(20),
66 #[cfg(feature = "runtime")]
67 keep_alive_while_idle: false,
68 max_concurrent_reset_streams: None,
69 }
70 }
71 }
72
new_builder(config: &Config) -> Builder73 fn new_builder(config: &Config) -> Builder {
74 let mut builder = Builder::default();
75 builder
76 .initial_window_size(config.initial_stream_window_size)
77 .initial_connection_window_size(config.initial_conn_window_size)
78 .max_frame_size(config.max_frame_size)
79 .enable_push(false);
80 if let Some(max) = config.max_concurrent_reset_streams {
81 builder.max_concurrent_reset_streams(max);
82 }
83 builder
84 }
85
new_ping_config(config: &Config) -> ping::Config86 fn new_ping_config(config: &Config) -> ping::Config {
87 ping::Config {
88 bdp_initial_window: if config.adaptive_window {
89 Some(config.initial_stream_window_size)
90 } else {
91 None
92 },
93 #[cfg(feature = "runtime")]
94 keep_alive_interval: config.keep_alive_interval,
95 #[cfg(feature = "runtime")]
96 keep_alive_timeout: config.keep_alive_timeout,
97 #[cfg(feature = "runtime")]
98 keep_alive_while_idle: config.keep_alive_while_idle,
99 }
100 }
101
handshake<T, B>( io: T, req_rx: ClientRx<B>, config: &Config, exec: Exec, ) -> crate::Result<ClientTask<B>> where T: AsyncRead + AsyncWrite + Send + Unpin + 'static, B: HttpBody, B::Data: Send + 'static,102 pub(crate) async fn handshake<T, B>(
103 io: T,
104 req_rx: ClientRx<B>,
105 config: &Config,
106 exec: Exec,
107 ) -> crate::Result<ClientTask<B>>
108 where
109 T: AsyncRead + AsyncWrite + Send + Unpin + 'static,
110 B: HttpBody,
111 B::Data: Send + 'static,
112 {
113 let (h2_tx, mut conn) = new_builder(config)
114 .handshake::<_, SendBuf<B::Data>>(io)
115 .await
116 .map_err(crate::Error::new_h2)?;
117
118 // An mpsc channel is used entirely to detect when the
119 // 'Client' has been dropped. This is to get around a bug
120 // in h2 where dropping all SendRequests won't notify a
121 // parked Connection.
122 let (conn_drop_ref, rx) = mpsc::channel(1);
123 let (cancel_tx, conn_eof) = oneshot::channel();
124
125 let conn_drop_rx = rx.into_future().map(|(item, _rx)| {
126 if let Some(never) = item {
127 match never {}
128 }
129 });
130
131 let ping_config = new_ping_config(&config);
132
133 let (conn, ping) = if ping_config.is_enabled() {
134 let pp = conn.ping_pong().expect("conn.ping_pong");
135 let (recorder, mut ponger) = ping::channel(pp, ping_config);
136
137 let conn = future::poll_fn(move |cx| {
138 match ponger.poll(cx) {
139 Poll::Ready(ping::Ponged::SizeUpdate(wnd)) => {
140 conn.set_target_window_size(wnd);
141 conn.set_initial_window_size(wnd)?;
142 }
143 #[cfg(feature = "runtime")]
144 Poll::Ready(ping::Ponged::KeepAliveTimedOut) => {
145 debug!("connection keep-alive timed out");
146 return Poll::Ready(Ok(()));
147 }
148 Poll::Pending => {}
149 }
150
151 Pin::new(&mut conn).poll(cx)
152 });
153 (Either::Left(conn), recorder)
154 } else {
155 (Either::Right(conn), ping::disabled())
156 };
157 let conn = conn.map_err(|e| debug!("connection error: {}", e));
158
159 exec.execute(conn_task(conn, conn_drop_rx, cancel_tx));
160
161 Ok(ClientTask {
162 ping,
163 conn_drop_ref,
164 conn_eof,
165 executor: exec,
166 h2_tx,
167 req_rx,
168 })
169 }
170
conn_task<C, D>(conn: C, drop_rx: D, cancel_tx: oneshot::Sender<Never>) where C: Future + Unpin, D: Future<Output = ()> + Unpin,171 async fn conn_task<C, D>(conn: C, drop_rx: D, cancel_tx: oneshot::Sender<Never>)
172 where
173 C: Future + Unpin,
174 D: Future<Output = ()> + Unpin,
175 {
176 match future::select(conn, drop_rx).await {
177 Either::Left(_) => {
178 // ok or err, the `conn` has finished
179 }
180 Either::Right(((), conn)) => {
181 // mpsc has been dropped, hopefully polling
182 // the connection some more should start shutdown
183 // and then close
184 trace!("send_request dropped, starting conn shutdown");
185 drop(cancel_tx);
186 let _ = conn.await;
187 }
188 }
189 }
190
191 pub(crate) struct ClientTask<B>
192 where
193 B: HttpBody,
194 {
195 ping: ping::Recorder,
196 conn_drop_ref: ConnDropRef,
197 conn_eof: ConnEof,
198 executor: Exec,
199 h2_tx: SendRequest<SendBuf<B::Data>>,
200 req_rx: ClientRx<B>,
201 }
202
203 impl<B> Future for ClientTask<B>
204 where
205 B: HttpBody + Send + 'static,
206 B::Data: Send,
207 B::Error: Into<Box<dyn StdError + Send + Sync>>,
208 {
209 type Output = crate::Result<Dispatched>;
210
poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output>211 fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
212 loop {
213 match ready!(self.h2_tx.poll_ready(cx)) {
214 Ok(()) => (),
215 Err(err) => {
216 self.ping.ensure_not_timed_out()?;
217 return if err.reason() == Some(::h2::Reason::NO_ERROR) {
218 trace!("connection gracefully shutdown");
219 Poll::Ready(Ok(Dispatched::Shutdown))
220 } else {
221 Poll::Ready(Err(crate::Error::new_h2(err)))
222 };
223 }
224 };
225
226 match self.req_rx.poll_recv(cx) {
227 Poll::Ready(Some((req, cb))) => {
228 // check that future hasn't been canceled already
229 if cb.is_canceled() {
230 trace!("request callback is canceled");
231 continue;
232 }
233 let (head, body) = req.into_parts();
234 let mut req = ::http::Request::from_parts(head, ());
235 super::strip_connection_headers(req.headers_mut(), true);
236 if let Some(len) = body.size_hint().exact() {
237 if len != 0 || headers::method_has_defined_payload_semantics(req.method()) {
238 headers::set_content_length_if_missing(req.headers_mut(), len);
239 }
240 }
241
242 let is_connect = req.method() == Method::CONNECT;
243 let eos = body.is_end_stream();
244 let ping = self.ping.clone();
245
246 if is_connect {
247 if headers::content_length_parse_all(req.headers())
248 .map_or(false, |len| len != 0)
249 {
250 warn!("h2 connect request with non-zero body not supported");
251 cb.send(Err((
252 crate::Error::new_h2(h2::Reason::INTERNAL_ERROR.into()),
253 None,
254 )));
255 continue;
256 }
257 }
258
259 let (fut, body_tx) = match self.h2_tx.send_request(req, !is_connect && eos) {
260 Ok(ok) => ok,
261 Err(err) => {
262 debug!("client send request error: {}", err);
263 cb.send(Err((crate::Error::new_h2(err), None)));
264 continue;
265 }
266 };
267
268 let send_stream = if !is_connect {
269 if !eos {
270 let mut pipe =
271 Box::pin(PipeToSendStream::new(body, body_tx)).map(|res| {
272 if let Err(e) = res {
273 debug!("client request body error: {}", e);
274 }
275 });
276
277 // eagerly see if the body pipe is ready and
278 // can thus skip allocating in the executor
279 match Pin::new(&mut pipe).poll(cx) {
280 Poll::Ready(_) => (),
281 Poll::Pending => {
282 let conn_drop_ref = self.conn_drop_ref.clone();
283 // keep the ping recorder's knowledge of an
284 // "open stream" alive while this body is
285 // still sending...
286 let ping = ping.clone();
287 let pipe = pipe.map(move |x| {
288 drop(conn_drop_ref);
289 drop(ping);
290 x
291 });
292 self.executor.execute(pipe);
293 }
294 }
295 }
296
297 None
298 } else {
299 Some(body_tx)
300 };
301
302 let fut = fut.map(move |result| match result {
303 Ok(res) => {
304 // record that we got the response headers
305 ping.record_non_data();
306
307 let content_length = headers::content_length_parse_all(res.headers());
308 if let (Some(mut send_stream), StatusCode::OK) =
309 (send_stream, res.status())
310 {
311 if content_length.map_or(false, |len| len != 0) {
312 warn!("h2 connect response with non-zero body not supported");
313
314 send_stream.send_reset(h2::Reason::INTERNAL_ERROR);
315 return Err((
316 crate::Error::new_h2(h2::Reason::INTERNAL_ERROR.into()),
317 None,
318 ));
319 }
320 let (parts, recv_stream) = res.into_parts();
321 let mut res = Response::from_parts(parts, Body::empty());
322
323 let (pending, on_upgrade) = crate::upgrade::pending();
324 let io = H2Upgraded {
325 ping,
326 send_stream: unsafe { UpgradedSendStream::new(send_stream) },
327 recv_stream,
328 buf: Bytes::new(),
329 };
330 let upgraded = Upgraded::new(io, Bytes::new());
331
332 pending.fulfill(upgraded);
333 res.extensions_mut().insert(on_upgrade);
334
335 Ok(res)
336 } else {
337 let res = res.map(|stream| {
338 let ping = ping.for_stream(&stream);
339 crate::Body::h2(stream, content_length.into(), ping)
340 });
341 Ok(res)
342 }
343 }
344 Err(err) => {
345 ping.ensure_not_timed_out().map_err(|e| (e, None))?;
346
347 debug!("client response error: {}", err);
348 Err((crate::Error::new_h2(err), None))
349 }
350 });
351 self.executor.execute(cb.send_when(fut));
352 continue;
353 }
354
355 Poll::Ready(None) => {
356 trace!("client::dispatch::Sender dropped");
357 return Poll::Ready(Ok(Dispatched::Shutdown));
358 }
359
360 Poll::Pending => match ready!(Pin::new(&mut self.conn_eof).poll(cx)) {
361 Ok(never) => match never {},
362 Err(_conn_is_eof) => {
363 trace!("connection task is closed, closing dispatch task");
364 return Poll::Ready(Ok(Dispatched::Shutdown));
365 }
366 },
367 }
368 }
369 }
370 }
371