126 lines
4.5 KiB
Rust
126 lines
4.5 KiB
Rust
#[allow(clippy::too_many_arguments)]
|
|
async fn http2_connection<I>(
|
|
stream: I,
|
|
router: Router,
|
|
peer: SocketAddr,
|
|
local: SocketAddr,
|
|
transport: Transport,
|
|
tls_sni: Option<String>,
|
|
extended_connect: bool,
|
|
frontend_state: Option<Arc<App>>,
|
|
lease: Arc<ConnectionLease>,
|
|
stop: CancellationToken,
|
|
grace: Duration,
|
|
settings: Http2Connections,
|
|
) where
|
|
I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
|
|
{
|
|
let request_stop = stop.clone();
|
|
let connection_metrics = lease.metrics.clone();
|
|
let activity = H2ConnectionState::new();
|
|
let service_activity = activity.clone();
|
|
let service = service_fn(move |mut request: Request<Incoming>| {
|
|
let stream_lease = H2StreamLease::begin(service_activity.clone());
|
|
request.extensions_mut().insert(PeerAddr(peer));
|
|
request.extensions_mut().insert(LocalAddr(local));
|
|
request.extensions_mut().insert(transport);
|
|
request.extensions_mut().insert(lease.clone());
|
|
request
|
|
.extensions_mut()
|
|
.insert(ServerShutdown(request_stop.clone()));
|
|
let router = router.clone();
|
|
async move {
|
|
router
|
|
.oneshot(request.map(Body::new))
|
|
.await
|
|
.map(|response| {
|
|
response.map(|body| Body::new(H2TrackedBody::new(body, stream_lease)))
|
|
})
|
|
}
|
|
});
|
|
let _active_http2 = ActiveHttp2Connection::begin(connection_metrics);
|
|
let stream = H2ActivityIo::new(stream, activity.clone());
|
|
let mut builder = http2::Builder::new(TokioExecutor::new());
|
|
builder
|
|
.timer(TokioTimer::new())
|
|
.max_concurrent_streams(64)
|
|
.max_header_list_size(32_768)
|
|
.max_send_buf_size(65_536)
|
|
.initial_stream_window_size(65_535)
|
|
.initial_connection_window_size(1_048_576);
|
|
if extended_connect {
|
|
builder.enable_connect_protocol();
|
|
}
|
|
let conn = builder.serve_connection(TokioIo::new(stream), service);
|
|
tokio::pin!(conn);
|
|
|
|
let max_lifetime = settings.max_connection_lifetime();
|
|
let lifetime = async move {
|
|
if let Some(duration) = max_lifetime {
|
|
tokio::time::sleep(duration).await;
|
|
} else {
|
|
std::future::pending::<()>().await;
|
|
}
|
|
};
|
|
tokio::pin!(lifetime);
|
|
|
|
let cause = loop {
|
|
tokio::select! {
|
|
result = &mut conn => {
|
|
if let Err(error) = result {
|
|
log_protocol_error(
|
|
frontend_state.as_ref(), "http2_connection_error", &error, peer, local,
|
|
if transport.tls { "tls" } else { "tcp" }, "2", 0, tls_sni.clone(),
|
|
);
|
|
tracing::debug!(%error, "HTTP/2 connection closed");
|
|
}
|
|
return;
|
|
},
|
|
_ = stop.cancelled() => break H2ShutdownCause::Process,
|
|
_ = &mut lifetime => break H2ShutdownCause::Lifetime,
|
|
_ = wait_for_h2_idle(activity.clone(), settings.idle_timeout()) => {
|
|
// Re-check immediately before GOAWAY so activity racing the timer
|
|
// cannot turn a busy connection into an idle shutdown.
|
|
if activity.is_idle(settings.idle_timeout()) {
|
|
break H2ShutdownCause::Idle;
|
|
}
|
|
},
|
|
}
|
|
};
|
|
|
|
// Hyper's graceful shutdown sends GOAWAY and refuses newer streams while
|
|
// continuing to poll streams that were already accepted.
|
|
conn.as_mut().graceful_shutdown();
|
|
let configured_drain = settings.drain_timeout();
|
|
let drain = if matches!(cause, H2ShutdownCause::Process) {
|
|
std::cmp::min(configured_drain, grace)
|
|
} else {
|
|
configured_drain
|
|
};
|
|
match tokio::time::timeout(drain, &mut conn).await {
|
|
Ok(Ok(())) => tracing::debug!(?cause, "HTTP/2 graceful shutdown drained"),
|
|
Ok(Err(error)) => {
|
|
log_protocol_error(
|
|
frontend_state.as_ref(),
|
|
"http2_drain_protocol_error",
|
|
&error,
|
|
peer,
|
|
local,
|
|
if transport.tls { "tls" } else { "tcp" },
|
|
"2",
|
|
0,
|
|
tls_sni.clone(),
|
|
);
|
|
tracing::debug!(?cause, %error, "HTTP/2 graceful shutdown ended with protocol error")
|
|
}
|
|
Err(_) => tracing::debug!(
|
|
?cause,
|
|
drain_seconds = drain.as_secs_f64(),
|
|
"HTTP/2 drain timeout reached; closing transport"
|
|
),
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
include!("serve/tests.rs");
|