From 7d4bc14c1d40b62fdde1ed732fe520b363dc2c5f Mon Sep 17 00:00:00 2001 From: dotCooCoo Date: Sat, 24 Oct 2026 08:23:14 -0700 Subject: [PATCH] Control server: refused connections drain on their own threads The accepting thread drained each connection it refused as busy for up to 250 ms before accepting the next, so refused peers that stayed connected without writing delayed every connection after them. Each drain now runs on a thread of its own, up to 16 at once; past that, a refused connection gets the busy reply and is closed without a drain. connection_limit_refuses_then_frees_slot now uses a 30 s idle timeout: its first client sends nothing, and under load the 300 ms timeout closed it before the third client connected. --- apps/deckcraft/src/control_server.rs | 58 ++++++++++++++++++++++++---- 1 file changed, 51 insertions(+), 7 deletions(-) diff --git a/apps/deckcraft/src/control_server.rs b/apps/deckcraft/src/control_server.rs index 1773729d..6088f343 100644 --- a/apps/deckcraft/src/control_server.rs +++ b/apps/deckcraft/src/control_server.rs @@ -66,10 +66,13 @@ pub fn start(port: u16, ctx: egui::Context) -> Receiver { /// core busy until it recovers. const ACCEPT_RETRY: Duration = Duration::from_millis(100); -/// The longest the accepting thread drains a connection it refused as busy before closing it. -/// It accepts no other connection meanwhile. +/// The longest a connection refused as busy is drained before it is closed. const REFUSED_GRACE: Duration = Duration::from_millis(250); +/// Refused connections drained at once, each on a thread of its own. Past this many, a refused +/// connection gets the busy reply and is closed without a drain. +const MAX_REFUSING: usize = 16; + /// The connections `incoming` accepts, calling `wait` after each failed accept. fn accepted(incoming: impl Iterator>, mut wait: impl FnMut()) -> impl Iterator { incoming.filter_map(move |stream| stream.map_err(|_| wait()).ok()) @@ -79,16 +82,14 @@ fn accepted(incoming: impl Iterator>, mut wait: imp fn serve_listener(listener: TcpListener, limits: Limits, ctx: egui::Context) -> Receiver { let (tx, rx) = sync_channel::(limits.queue.max(1)); let active = Arc::new(AtomicUsize::new(0)); + let refusing = Arc::new(AtomicUsize::new(0)); std::thread::spawn(move || { for mut stream in accepted(listener.incoming(), || std::thread::sleep(ACCEPT_RETRY)) { let _ = stream.set_write_timeout(Some(limits.write_timeout)); if active.fetch_add(1, Ordering::SeqCst) >= limits.max_clients { active.fetch_sub(1, Ordering::SeqCst); let _ = writeln!(stream, "{}", json!({"ok": false, "error": "busy: too many control connections"})); - // A peer that wrote a line before it read this reply would see only a reset. - if let Ok(read) = stream.try_clone() { - close_politely(&stream, &mut BufReader::new(read), limits.write_timeout.min(REFUSED_GRACE)); - } + refuse(stream, &refusing, limits.write_timeout.min(REFUSED_GRACE)); continue; } let slot = Slot(Arc::clone(&active)); @@ -106,6 +107,27 @@ fn serve_listener(listener: TcpListener, limits: Limits, ctx: egui::Context) -> rx } +/// Close a connection that was told it is refused. A peer that wrote a line before it read the +/// reply would see only a reset, so the connection is drained for up to `grace` first, on a +/// thread of its own: the accepting thread goes straight back to accepting. Past +/// [`MAX_REFUSING`] drains at once, the connection is closed without one. +fn refuse(stream: TcpStream, refusing: &Arc, grace: Duration) { + if refusing.fetch_add(1, Ordering::SeqCst) >= MAX_REFUSING { + refusing.fetch_sub(1, Ordering::SeqCst); + return; + } + let slot = Slot(Arc::clone(refusing)); + let spawned = std::thread::Builder::new().name("control-refused".into()).spawn(move || { + let _slot = slot; + if let Ok(read) = stream.try_clone() { + close_politely(&stream, &mut BufReader::new(read), grace); + } + }); + if spawned.is_err() { + log::warn!("control server could not start a thread to close a refused connection"); + } +} + /// Holds one connection slot; gives it back when the client thread ends, however it ends. struct Slot(Arc); @@ -436,6 +458,26 @@ mod tests { } } + /// Peers refused as busy that stay connected without sending anything do not hold up the + /// connections after them: each one's drain runs to its deadline on a thread of its own, and + /// past `MAX_REFUSING` drains at once a refused connection is closed without one. + #[test] + fn silent_refused_peers_do_not_delay_the_next_refusal() { + let (addr, rx) = server(Limits { max_clients: 1, idle_timeout: Duration::from_secs(30), ..small() }); + ui(rx); + let (mut hr, mut hw) = client(addr); + writeln!(hw, r#"{{"id":1,"method":"m"}}"#).unwrap(); + assert!(reply(&mut hr).is_some()); + // Twenty drains of 250 ms each, one after another, would take five seconds. + let silent: Vec<_> = (0..MAX_REFUSING + 4).map(|_| client(addr)).collect(); + let start = Instant::now(); + let (mut r, _w) = client(addr); + let refused = reply(&mut r).unwrap(); + assert!(refused["error"].as_str().unwrap_or_default().starts_with("busy"), "{refused}"); + assert!(start.elapsed() < Duration::from_secs(1), "refused after {:?}", start.elapsed()); + drop(silent); + } + /// A peer that keeps its connection open after its line was rejected gives its slot back at /// the drain deadline (the write timeout), not at the idle timeout. #[test] @@ -483,7 +525,9 @@ mod tests { #[test] fn connection_limit_refuses_then_frees_slot() { - let (addr, rx) = server(small()); + // `a` sends nothing. Under a 300 ms idle timeout, a slow run could close it before the + // third peer connected, and that peer then took its slot instead of reading `busy`. + let (addr, rx) = server(Limits { idle_timeout: Duration::from_secs(30), ..small() }); ui(rx); let a = client(addr); let (mut br, mut bw) = client(addr);