[rust-server] fix TCP-layer TTFT stalls (#33026)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: Cheng Wan <54331508+ch-wan@users.noreply.github.com>
This commit is contained in:
co-authored by
Claude Fable 5
Cheng Wan
parent
00a219f6c9
commit
574ead753a
@@ -31,7 +31,7 @@ struct AppState {
|
||||
}
|
||||
|
||||
pub async fn serve(
|
||||
listener: std::net::TcpListener,
|
||||
socket: tokio::net::TcpSocket,
|
||||
senders: Senders,
|
||||
egress_buf: usize,
|
||||
server_args: Arc<ServerArgs>,
|
||||
@@ -61,12 +61,12 @@ pub async fn serve(
|
||||
.with_state(state);
|
||||
let app = log::apply(app, &server_args);
|
||||
|
||||
// The listener was already bound synchronously in `runtime::start` (so a port
|
||||
// conflict fails startup); adopt it into the tokio reactor here.
|
||||
let listener = match tokio::net::TcpListener::from_std(listener) {
|
||||
// The socket was already bound synchronously in `runtime::start` (so a port
|
||||
// conflict fails startup); start listening here, on the api runtime.
|
||||
let listener = match socket.listen(1024) {
|
||||
Ok(l) => l,
|
||||
Err(e) => {
|
||||
tracing::error!(error = %e, "failed to adopt pre-bound listener");
|
||||
tracing::error!(error = %e, "listen failed");
|
||||
return;
|
||||
}
|
||||
};
|
||||
@@ -79,6 +79,14 @@ pub async fn serve(
|
||||
// runtime drops → detached handlers cancel → their `AbortGuard`s fire, release
|
||||
// `Senders` clones → tok/detok channels close → workers exit. Full drain is
|
||||
// deferred (see `request_shutdown`).
|
||||
// Match Python (asyncio sets TCP_NODELAY); avoids a ~13 ms
|
||||
// Nagle/delayed-ACK penalty on keep-alive connections.
|
||||
use axum::serve::ListenerExt;
|
||||
let listener = listener.tap_io(|io| {
|
||||
if let Err(e) = io.set_nodelay(true) {
|
||||
tracing::debug!(error = %e, "set_nodelay failed");
|
||||
}
|
||||
});
|
||||
// `with_connect_info` exposes the peer address to the access-log middleware.
|
||||
let serve = axum::serve(
|
||||
listener,
|
||||
|
||||
@@ -94,11 +94,21 @@ impl Drop for Runtime {
|
||||
pub fn start(cfg: RuntimeConfig) -> Result<Runtime, String> {
|
||||
// Bind the API server port before spawning any thread, so an unavailable
|
||||
// port (EADDRINUSE) is a hard startup error.
|
||||
let listener = std::net::TcpListener::bind(cfg.rust_server_args.http_addr)
|
||||
.map_err(|e| format!("bind {} failed: {e}", cfg.rust_server_args.http_addr))?;
|
||||
listener
|
||||
.set_nonblocking(true)
|
||||
.map_err(|e| format!("listener set_nonblocking failed: {e}"))?;
|
||||
let addr = cfg.rust_server_args.http_addr;
|
||||
let socket = match addr {
|
||||
std::net::SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4(),
|
||||
std::net::SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6(),
|
||||
}
|
||||
.map_err(|e| format!("socket for {addr} failed: {e}"))?;
|
||||
socket
|
||||
.set_reuseaddr(true)
|
||||
.map_err(|e| format!("set_reuseaddr failed: {e}"))?;
|
||||
if let Err(e) = socket.set_recv_buffer_size(16 * 1024 * 1024) {
|
||||
eprintln!("warning: set_recv_buffer_size on listener failed: {e}");
|
||||
}
|
||||
socket
|
||||
.bind(addr)
|
||||
.map_err(|e| format!("bind {addr} failed: {e}"))?;
|
||||
|
||||
let (shutdown_tx, shutdown_rx) = flume::unbounded::<()>();
|
||||
let mut threads = Vec::new();
|
||||
@@ -267,7 +277,7 @@ pub fn start(cfg: RuntimeConfig) -> Result<Runtime, String> {
|
||||
}
|
||||
let rt = builder.build().expect("build api runtime");
|
||||
rt.block_on(api_server::serve(
|
||||
listener,
|
||||
socket,
|
||||
senders,
|
||||
cfg.rust_server_args.channel_cap,
|
||||
cfg.server_args.clone(),
|
||||
|
||||
Reference in New Issue
Block a user