Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,14 @@ The Rust HTTP server also reads these environment variables, which are not expos
See [Runtime Configuration](runtime-configuration.mdx) for `DYN_TCP_MAX_MESSAGE_SIZE`.
</ParamField>

<ParamField path="DYN_HTTP_LISTEN_BACKLOG" type="integer" default="4096">
Listen backlog of the frontend socket, for both plain HTTP and TLS, capped by
the kernel at `net.core.somaxconn`. Raise it when many clients open connections
within a few seconds (for example a load client with one connection per
session): connections that arrive while the accept queue is full are delayed
or, depending on the host's TCP settings, fail.
</ParamField>
Comment thread
coderabbitai[bot] marked this conversation as resolved.

<ParamField path="DYN_HTTP_GRACEFUL_SHUTDOWN_TIMEOUT_SECS" type="integer" default="5">
Maximum time the Frontend waits for admitted HTTP response bodies and `/v1/realtime` WebSocket
tasks to finish after shutdown begins. New inference requests are rejected while draining.
Expand Down
86 changes: 79 additions & 7 deletions lib/llm/src/http/service/service_v2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -872,9 +872,7 @@ impl HttpService {

/// Like [`spawn`], but uses a caller-provided pre-bound listener. Closes the TOCTOU
/// port-allocation gap for tests that need to know the bound port up front. Not
/// supported in TLS mode: TLS uses `axum_server::bind_rustls`, which owns its own
/// bind, so a pre-bound listener cannot be threaded through and dropping it before
/// `bind_rustls` would just re-open the same race. Returns an error if invoked on a
/// supported in TLS mode, which binds internally. Returns an error if invoked on a
/// service built with `enable_tls(true)`.
///
/// [`spawn`]: HttpService::spawn
Expand Down Expand Up @@ -924,8 +922,7 @@ impl HttpService {
if self.enable_tls {
if listener.is_some() {
return Err(anyhow::anyhow!(
"Pre-bound listener is not supported in TLS mode; \
axum_server::bind_rustls owns its own bind. \
"Pre-bound listener is not supported in TLS mode. \
Use run()/spawn() (which bind internally) when enable_tls is set."
));
}
Expand All @@ -950,7 +947,18 @@ impl HttpService {
let config = RustlsConfig::from_config(Arc::new(server_config));

let handle = tls_handle.unwrap_or_default();
let server = axum_server::bind_rustls(addr, config)
let std_listener = bind_listener(addr)
.and_then(|l| l.into_std())
.map_err(|e| {
tracing::error!(
protocol = %protocol,
address = %address,
error = %e,
"Failed to bind server to address"
);
anyhow::anyhow!("Failed to start {} server on {}: {}", protocol, address, e)
})?;
let server = axum_server::from_tcp_rustls(std_listener, config)
.handle(handle.clone())
.serve(router.into_make_service());

Expand Down Expand Up @@ -1002,7 +1010,7 @@ impl HttpService {
let addr: SocketAddr = address
.parse()
.map_err(|e| anyhow::anyhow!("Invalid address '{}': {}", address, e))?;
tokio::net::TcpListener::bind(addr).await.map_err(|e| {
bind_listener(addr).map_err(|e| {
tracing::error!(
protocol = %protocol,
address = %address,
Expand Down Expand Up @@ -1132,6 +1140,35 @@ fn get_graceful_shutdown_timeout() -> usize {
.unwrap_or(5)
}

const DEFAULT_LISTEN_BACKLOG: u32 = 4096;

// `listen(2)` takes an `int`; anything above `i32::MAX` would go negative.
fn parse_listen_backlog(value: Result<String, std::env::VarError>) -> u32 {
Comment thread
rmccorm4 marked this conversation as resolved.
Comment thread
rmccorm4 marked this conversation as resolved.
value
.ok()
.and_then(|s| s.trim().parse::<u32>().ok())
.filter(|n| (1..=i32::MAX as u32).contains(n))
.unwrap_or(DEFAULT_LISTEN_BACKLOG)
}

fn listen_backlog() -> u32 {
parse_listen_backlog(std::env::var(env_llm::DYN_HTTP_LISTEN_BACKLOG))
}

/// `tokio::net::TcpListener::bind` listens with a backlog of 128. A few thousand
/// clients connecting within seconds overflow that, and overflowed connections
/// are delayed or, depending on host TCP settings, fail.
fn bind_listener(addr: SocketAddr) -> std::io::Result<tokio::net::TcpListener> {
let socket = if addr.is_ipv4() {
tokio::net::TcpSocket::new_v4()?
} else {
tokio::net::TcpSocket::new_v6()?
};
socket.set_reuseaddr(true)?;
socket.bind(addr)?;
socket.listen(listen_backlog())
}

/// Environment variable to set the metrics endpoint path (default: `/metrics`)
static HTTP_SVC_METRICS_PATH_ENV: &str = "DYN_HTTP_SVC_METRICS_PATH";
/// Environment variable to set the models endpoint path (default: `/v1/models`)
Expand Down Expand Up @@ -2702,6 +2739,41 @@ mod tests {
assert_eq!(parse_sse_keep_alive(Ok(u64::MAX.to_string())), expected);
}

#[test]
fn test_listen_backlog_env_var() {
assert_eq!(
parse_listen_backlog(Err(std::env::VarError::NotPresent)),
DEFAULT_LISTEN_BACKLOG
);
assert_eq!(
parse_listen_backlog(Ok("0".to_string())),
DEFAULT_LISTEN_BACKLOG
);
assert_eq!(
parse_listen_backlog(Ok("invalid".to_string())),
DEFAULT_LISTEN_BACKLOG
);
assert_eq!(
parse_listen_backlog(Ok((i32::MAX as u32 + 1).to_string())),
DEFAULT_LISTEN_BACKLOG
);
assert_eq!(
parse_listen_backlog(Ok(i32::MAX.to_string())),
i32::MAX as u32
);
assert_eq!(parse_listen_backlog(Ok(" 8192 ".to_string())), 8192);
}

#[tokio::test]
async fn test_bind_listener_accepts_connections() {
let listener = bind_listener("127.0.0.1:0".parse().unwrap()).unwrap();
let addr = listener.local_addr().unwrap();
assert_ne!(addr.port(), 0);
let client = tokio::net::TcpStream::connect(addr).await.unwrap();
let (_server_side, peer) = listener.accept().await.unwrap();
assert_eq!(peer, client.local_addr().unwrap());
}

#[test]
fn test_sse_keep_alive_for_deferred_response() {
let configured = Duration::from_millis(5000);
Expand Down
4 changes: 4 additions & 0 deletions lib/runtime/src/config/environment_names.rs
Original file line number Diff line number Diff line change
Expand Up @@ -356,6 +356,9 @@ pub mod llm {
/// HTTP body size limit in MB
pub const DYN_HTTP_BODY_LIMIT_MB: &str = "DYN_HTTP_BODY_LIMIT_MB";

/// Listen backlog of the frontend HTTP/HTTPS socket (default 4096).
pub const DYN_HTTP_LISTEN_BACKLOG: &str = "DYN_HTTP_LISTEN_BACKLOG";

pub const DYN_HTTP_GRACEFUL_SHUTDOWN_TIMEOUT_SECS: &str =
"DYN_HTTP_GRACEFUL_SHUTDOWN_TIMEOUT_SECS";

Expand Down Expand Up @@ -1070,6 +1073,7 @@ mod tests {
kvbm::leader::DYN_KVBM_LEADER_ZMQ_ACK_PORT,
// LLM
llm::DYN_HTTP_BODY_LIMIT_MB,
llm::DYN_HTTP_LISTEN_BACKLOG,
llm::DYN_HTTP_GRACEFUL_SHUTDOWN_TIMEOUT_SECS,
llm::DYN_HTTP_OVERLOAD_STATUS_CODE,
llm::DYN_HTTP_BACKEND_STREAM_TIMEOUT_SECS,
Expand Down
Loading