From 68031b430beba00e5b48c75deb205fe7a3b8be6a Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Wed, 29 Jul 2026 20:41:29 +0700 Subject: [PATCH] =?UTF-8?q?feat(server):=20task-exit=20parking=20for=20idl?= =?UTF-8?q?e=20connections=20=E2=80=94=2019.8=20=E2=86=92=203.25=20KB/conn?= =?UTF-8?q?=20(c1M=20P1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A monoio plain-TCP connection idle past --conn-park-secs (default 60 s, 0 = off) now has its handler task EXIT instead of sitting parked inside the read future. The task returns MonoioHandlerResult::ParkIdle carrying {Box, RegistryGuard}; conn_accept spawns a tiny boxed watcher future owning {TcpStream, state, guard, ConnectionContext} that awaits stream.readable(false) (race-free on both io_uring and kqueue/epoll — no vendor patch needed) or shard shutdown. On wake it drops the registry guard synchronously and respawns the full handler through the migration-restore path (can_migrate=false, can_park=true), so a resumed connection can park again indefinitely. Mechanism details: - Stage-2 arm in handler_monoio/mod.rs splits: parkable requires can_park && S::SUPPORTS_TASK_PARK && park_after_ms() > 0, empty read/write buffers, !in_multi, empty command_queue, and no active cross-shard txn. Subscribers, tracking conns, and `timeout N` conns are structurally excluded (their select arms precede the park arm). - idle_park sweep gains a stage2 flag per slot: stage-1 cancels at IDLE_DOWNSHIFT_MS (1 s, probe-buffer downshift), stage-2 cancels at conn_park_after_ms (task exit). TLS keeps SUPPORTS_TASK_PARK=false (follow-up: readable() passthrough in the vendored wrapper). - RegistryGuard hoisted to module scope and moved through ParkIdle so deregister-before-fd-close holds on every path; CLIENT KILL's shutdown(2) makes the parked fd read-ready, the watcher wakes, and the resumed handler reads EOF. record_connection_closed fires exactly once per connection (original task skips it on park; the watcher or resumed task owns it). - can_park is opt-in per call site: only sites that route ParkIdle pass true (plain-TCP accept + resumed helper); TLS, fail-open, and migrated-spawn sites pass false so a ParkIdle can never be silently dropped (= closed). Follow-up noted to wire the migrated-spawn site. - Known limitation (independent review, Low): on data-wake the deregister→re-register window spans a task-scheduling boundary, so a racing CLIENT LIST briefly misses the waking conn and CLIENT KILL ID returns 0 (same observable as a reconnect race). Closing it needs registration handoff into the handler; documented at the drop site. - New flag plumbed via moon::runtime atomic (set_conn_park_secs); tokio runtime warns and ignores it. Measured (E10, moon-dev VM, 10 k idle conns, shards=2, same-binary flag A/B, tmp/c10k/e10_park_rss.sh): - PARK_OFF (W11 baseline): 19.8 KB/conn idle - PARK_ON: 3.25 KB/conn idle; re-parks at 3.75 after a full wake sweep - Wake sweep wire-correct: bad=0; 0.25 s vs 0.10 s for 10 k PINGs (~15 µs/conn extra wake cost, off the hot path) - Campaign total: 56.5 → 3.3 KB/idle-conn (−94 %); 1 M idle ≈ 3.3 GB Gates: - tests/parked_idle_parity.rs (new, 3 tests): multi-cycle park/wake parity (probe, 100-deep pipeline, 4 KiB value), CLIENT LIST/KILL on a parked conn, active-sibling isolation — green on kqueue (macOS) AND io_uring (VM), plus tokio (degenerate parity) - idle_park unit tests 10/10 incl. new stage2_uses_park_after_threshold - VM monoio lib 4466 pass; tokio lib 3609 pass; W11 + TLS idle parity suites still green - fmt, clippy --all-targets (both matrices, 0 lints), unsafe/unwrap audits pass refs: .planning/rfcs/c1m-connection-plane.md (sequencing item 6, P1) author: Tin Dang --- CHANGELOG.md | 17 ++ src/config.rs | 7 + src/main.rs | 15 ++ src/runtime/mod.rs | 19 ++ src/server/conn/handler_monoio/idle_park.rs | 58 ++++- src/server/conn/handler_monoio/mod.rs | 118 ++++++++-- src/shard/conn_accept.rs | 159 ++++++++++++++ tests/mq_integration.rs | 1 + tests/parked_idle_parity.rs | 230 ++++++++++++++++++++ tests/txn_kv_wiring.rs | 1 + tests/workspace_integration.rs | 2 + 11 files changed, 608 insertions(+), 19 deletions(-) create mode 100644 tests/parked_idle_parity.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 867756bf..5b83aca8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **Task-exit parking for idle connections (c1M P1, `--conn-park-secs`, + default 60 s).** A plain-TCP monoio connection that stays idle past the + W11 downshift now has its handler task exit entirely: only a tiny + readiness watcher (boxed future holding the stream, ~100 B of session + state, and the client-registry guard) remains, reclaiming the ~6 KB task + state machine plus the remaining per-task buffers. The watcher wakes on + read-readiness (`readable(false)`, race-free on io_uring and + epoll/kqueue) or server shutdown and rehydrates a fresh handler through + the migration-restore path — wire-invisible across repeated park/wake + cycles. Parked connections stay in CLIENT LIST, keep their maxclients + slot, and CLIENT KILL still works (its `shutdown(2)` wakes the watcher; + the resumed handler sees EOF). Exclusions: subscriber/tracking/timeout + connections (structurally never reach the park arm), MULTI/EXEC or + cross-store-txn sessions, partial frames, TLS (keeps W11+P4b buffer + downshift), and the tokio runtime. `--conn-park-secs 0` disables. + ### Fixed - **c10k connection-plane wave (PR #TBD)** — from the 2026-07-29 empirical review (`tmp/C10K-REVIEW.md`: 10k/25k live-connection ramps, idle-CPU diff --git a/src/config.rs b/src/config.rs index b3577cad..65132891 100644 --- a/src/config.rs +++ b/src/config.rs @@ -304,6 +304,13 @@ pub struct ServerConfig { #[arg(long = "uring-entries")] pub uring_entries: Option, + /// c1M P1: seconds a downshifted idle connection waits before its handler + /// task exits entirely, leaving only a tiny readiness watcher (task-exit + /// parking; plain-TCP monoio connections only — TLS and the tokio runtime + /// keep the buffer-downshift behavior). 0 disables task parking. + #[arg(long = "conn-park-secs", default_value_t = 60)] + pub conn_park_secs: u64, + /// I/O driver for the monoio runtime. "auto" lets FusionDriver pick /// (io_uring on Linux when available, else epoll/kqueue); "epoll" forces /// the legacy poller. Measured on GCE ARM (c4a Axion, 2026-07): epoll is diff --git a/src/main.rs b/src/main.rs index 856dd3eb..9c0f60dd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1097,6 +1097,21 @@ fn main() -> anyhow::Result<()> { tracing::warn!("--uring-entries has no effect under the tokio runtime"); } + // c1M P1: stage-2 task-parking threshold — set BEFORE shard threads spawn + // (same contract as set_uring_entries above). Default 60s; 0 disables. + moon::runtime::set_conn_park_secs(config.conn_park_secs); + #[cfg(feature = "runtime-monoio")] + if config.conn_park_secs > 0 { + tracing::info!( + "Idle task-parking: downshifted connections park after {}s (--conn-park-secs)", + config.conn_park_secs + ); + } + #[cfg(not(feature = "runtime-monoio"))] + if config.conn_park_secs != 60 { + tracing::warn!("--conn-park-secs has no effect under the tokio runtime"); + } + // Graph traversal timeout default — set BEFORE shard threads spawn so every // TraversalGuard::with_default_timeout observes it (per-query TIMEOUT overrides). #[cfg(feature = "graph")] diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 36bdb535..326042dd 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -72,6 +72,25 @@ pub fn uring_entries() -> Option { } } +/// c1M P1: stage-2 idle task-parking threshold (`--conn-park-secs`), in ms. +/// 0 = task parking disabled. Set once from main BEFORE shard threads spawn +/// (same contract as [`set_uring_entries`]); read by the monoio idle-park +/// sweep and stage-2 read arm. +static CONN_PARK_AFTER_MS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(60_000); + +/// Configure the task-parking threshold from `--conn-park-secs`. +pub fn set_conn_park_secs(secs: u64) { + CONN_PARK_AFTER_MS.store( + secs.saturating_mul(1000), + std::sync::atomic::Ordering::Release, + ); +} + +/// Current task-parking threshold in ms; 0 = disabled. +pub fn conn_park_after_ms() -> u64 { + CONN_PARK_AFTER_MS.load(std::sync::atomic::Ordering::Acquire) +} + /// True when the epoll busy-poll park is configured — via the /// `--io-busy-poll-us` flag (the caller passes the config value) or the /// `MOON_EPOLL_SPIN_US` env fallback the vendored driver also honors. Gates diff --git a/src/server/conn/handler_monoio/idle_park.rs b/src/server/conn/handler_monoio/idle_park.rs index ad06a852..435091eb 100644 --- a/src/server/conn/handler_monoio/idle_park.rs +++ b/src/server/conn/handler_monoio/idle_park.rs @@ -39,6 +39,13 @@ use monoio::io::{AsyncReadRent, CancelHandle, Canceller}; /// Park duration after which the sweep cancels a stage-1 read (ms). pub(super) const IDLE_DOWNSHIFT_MS: u64 = 1000; + +/// c1M P1: stage-2 threshold — a downshifted connection parked this long has +/// its read cancelled so the handler TASK can exit (task-exit parking, see +/// `conn_accept::spawn_parked_idle_watcher`). 0 = task parking disabled. +/// The value lives in `crate::runtime` (set once from main before shards +/// spawn, same contract as `set_uring_entries`); this is a local alias. +pub(crate) use crate::runtime::conn_park_after_ms as park_after_ms; /// Rent-buffer size while downshifted. A typical idle→active wake (one /// command) fits; larger bursts spill into the next full-size read. pub(super) const IDLE_PROBE_BUF: usize = 512; @@ -55,6 +62,10 @@ pub(super) struct IdleSlot { /// Shard-cached-clock ms when the connection parked its stage-1 read; /// 0 = not parked in stage 1 (processing, downshifted, or gone). parked_since_ms: Cell, + /// c1M P1: true while the park is a STAGE-2 (downshifted, task-parkable) + /// read — the sweep then applies [`park_after_ms`] instead of + /// [`IDLE_DOWNSHIFT_MS`], and the woken handler exits its task. + stage2: Cell, } impl IdleSlot { @@ -66,6 +77,14 @@ impl IdleSlot { /// Mark the connection parked (stage 1). `now_ms` comes from the shard /// cached clock; clamped to ≥1 so 0 stays the "not parked" sentinel. pub(super) fn mark_parked(&self, now_ms: u64) { + self.stage2.set(false); + self.parked_since_ms.set(now_ms.max(1)); + } + + /// Mark the connection parked in stage 2 (c1M P1): downshifted, session + /// verified parkable, cancel = task exit. + pub(super) fn mark_parked_stage2(&self, now_ms: u64) { + self.stage2.set(true); self.parked_since_ms.set(now_ms.max(1)); } @@ -98,6 +117,7 @@ pub(super) fn register(client_id: u64) -> IdleParkRegistration { let slot = Rc::new(IdleSlot { canceller: RefCell::new(Canceller::new()), parked_since_ms: Cell::new(0), + stage2: Cell::new(false), }); REGISTRY.with(|r| r.borrow_mut().insert(client_id, slot.clone())); IdleParkRegistration { slot, client_id } @@ -111,11 +131,23 @@ pub(super) fn register(client_id: u64) -> IdleParkRegistration { /// downshifts, and re-parks in stage 2 with `parked_since_ms == 0`, so each /// idle connection is cancelled exactly once per idle period. pub(crate) fn sweep(now_ms: u64) -> usize { + let park_after = park_after_ms(); REGISTRY.with(|r| { let mut cancelled = 0usize; for slot in r.borrow().values() { let parked = slot.parked_since_ms.get(); - if parked != 0 && now_ms.saturating_sub(parked) >= IDLE_DOWNSHIFT_MS { + // Stage-2 (task-park) entries use the operator threshold; a 0 + // threshold means task parking is off and stage-2 entries are + // never registered, but guard anyway. + let threshold = if slot.stage2.get() { + if park_after == 0 { + continue; + } + park_after + } else { + IDLE_DOWNSHIFT_MS + }; + if parked != 0 && now_ms.saturating_sub(parked) >= threshold { // Consume-and-replace: cancel() fires every associated op and // returns a fresh canceller for the connection's next park. let old = slot.canceller.replace(Canceller::new()); @@ -159,6 +191,12 @@ pub(super) fn downshift_idle_buffers( pub(crate) trait IdleParkRead: AsyncReadRent { const SUPPORTS_IDLE_PARK: bool = false; + /// c1M P1: streams whose task can exit while the connection stays open + /// (requires a race-free standalone readiness await — plain TCP only; + /// TLS keeps W11+P4b behavior because rustls session state lives in the + /// stream and the wrapper exposes no readiness API). + const SUPPORTS_TASK_PARK: bool = false; + fn idle_park_read( &mut self, buf: Vec, @@ -175,6 +213,7 @@ pub(crate) trait IdleParkRead: AsyncReadRent { impl IdleParkRead for monoio::net::TcpStream { const SUPPORTS_IDLE_PARK: bool = true; + const SUPPORTS_TASK_PARK: bool = true; fn idle_park_read( &mut self, @@ -238,6 +277,23 @@ mod idle_park_tests { assert_eq!(sweep(60_000 + IDLE_DOWNSHIFT_MS), 1); } + #[test] + fn stage2_uses_park_after_threshold() { + let reg = register(90_003); + // Stage-2 park: the 1s downshift threshold must NOT fire it... + reg.slot.mark_parked_stage2(100_000); + assert_eq!( + sweep(100_000 + IDLE_DOWNSHIFT_MS), + 0, + "stage-2 park must outlive the stage-1 threshold" + ); + // ...only the operator threshold does (default 60s). + assert_eq!(sweep(100_000 + park_after_ms()), 1); + // A later stage-1 park on the same slot reverts to the 1s threshold. + reg.slot.mark_parked(300_000); + assert_eq!(sweep(300_000 + IDLE_DOWNSHIFT_MS), 1); + } + #[test] fn registration_drop_deregisters() { { diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 5fdc74ee..4efc5f23 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -47,6 +47,20 @@ use crate::shard::dispatch::ShardMessage; // AtomicWaker rides that mechanism. Transaction (txn.rs) and blocking-write // (write.rs) paths remain on oneshots — they are off the hot path. +/// RAII client-registry entry: deregisters on drop. Lives as a handler +/// local, EXCEPT for a task-parked connection (c1M P1), where it travels +/// inside [`MonoioHandlerResult::ParkIdle`] to the readiness watcher so the +/// CLIENT LIST/KILL entry and maxclients slot survive the parked lifetime. +#[cfg(feature = "runtime-monoio")] +pub struct RegistryGuard(pub(crate) u64); + +#[cfg(feature = "runtime-monoio")] +impl Drop for RegistryGuard { + fn drop(&mut self) { + crate::client_registry::deregister(self.0); + } +} + /// Result of `handle_connection_sharded_monoio` execution. /// /// Same purpose as the Tokio handler's `HandlerResult`: the generic handler cannot @@ -60,6 +74,19 @@ pub enum MonoioHandlerResult { state: MigratedConnectionState, target_shard: usize, }, + /// c1M P1: the connection idled past `--conn-park-secs` in the + /// downshifted state; the handler task exits and the caller parks the + /// stream behind a tiny readiness watcher + /// (`conn_accept::spawn_parked_idle_watcher`). Only returned when the + /// caller opted in via `can_park` — every other call site would drop the + /// stream and silently close a healthy connection. `registry_guard` + /// keeps the CLIENT LIST/KILL entry (and its maxclients slot) alive for + /// the parked lifetime; the watcher drops it just before re-registering + /// on wake. + ParkIdle { + state: Box, + registry_guard: RegistryGuard, + }, /// PSYNC arrived on this connection. Caller must hand the underlying /// `monoio::net::TcpStream` to /// `crate::replication::master::handle_psync_inline_single_shard` / @@ -183,6 +210,10 @@ pub(crate) async fn handle_connection_sharded_monoio< // (non-unix). Threaded from the concrete spawn site; the generic `S` here // has no `AsRawFd` bound. kill_fd: i32, + // c1M P1: only call sites that ROUTE `ParkIdle` (spawning the readiness + // watcher) may pass true — anywhere else a park return would drop the + // stream and silently close a healthy idle connection. + can_park: bool, ) -> (MonoioHandlerResult, Option) { use monoio::io::AsyncWriteRentExt; @@ -225,13 +256,7 @@ pub(crate) async fn handle_connection_sharded_monoio< ctx.shard_id, kill_fd, ); - struct RegistryGuard(u64); - impl Drop for RegistryGuard { - fn drop(&mut self) { - crate::client_registry::deregister(self.0); - } - } - let _registry_guard = RegistryGuard(client_id); + let registry_guard = RegistryGuard(client_id); // Functions API registry — LAZY per connection (P-1 footprint): built on // first FUNCTION/FCALL/FCALL_RO via `ensure_function_registry`, so the @@ -665,18 +690,75 @@ pub(crate) async fn handle_connection_sharded_monoio< continue; } } else if downshifted { - // c10k W11 stage 2: park with the probe buffer, no chore state. - // Real data restores the full working set (lazily, via the - // pre-park sizing above) and re-arms stage 1. - let (result, returned_buf) = stream.read(tmp_buf).await; - tmp_buf = returned_buf; - match result { - Ok(0) => break, - Ok(n) => { - read_buf.extend_from_slice(&tmp_buf[..n]); - downshifted = false; + // c10k W11 stage 2: park with the probe buffer. Real data + // restores the full working set (lazily, via the pre-park sizing + // above) and re-arms stage 1. + // + // c1M P1: when task parking is enabled and the session is + // parkable, the stage-2 read is ALSO cancelable and registered + // with the sweep under the longer `--conn-park-secs` threshold — + // a cancel here means "exit the task", leaving only a tiny + // readiness watcher (conn_accept::spawn_parked_idle_watcher). + // The predicate is stable while parked in read (no commands can + // execute), and reaching this arm already excludes subscriber / + // tracking / idle-timeout connections (each takes its own arm). + let parkable = can_park + && S::SUPPORTS_TASK_PARK + && idle_park::park_after_ms() > 0 + && read_buf.is_empty() + && write_buf.is_empty() + && !conn.in_multi + && conn.command_queue.is_empty() + && conn.active_cross_txn.is_none(); + if let (true, Some(reg)) = (parkable, idle_reg.as_ref()) { + let handle = reg.slot.handle(); + reg.slot.mark_parked_stage2(ctx.cached_clock.ms()); + let (result, returned_buf) = stream.idle_park_read(tmp_buf, handle).await; + reg.slot.mark_unparked(); + tmp_buf = returned_buf; + match result { + Ok(0) => break, + Ok(n) => { + read_buf.extend_from_slice(&tmp_buf[..n]); + downshifted = false; + } + Err(_) => { + // Cancelled by the stage-2 sweep (or a real socket + // error, which the resumed handler's first read will + // surface as EOF/err): exit the task. read_buf is + // empty (predicate), so no partial frame is at risk. + let state = Box::new(MigratedConnectionState { + selected_db: conn.selected_db, + authenticated: conn.authenticated, + client_name: conn.client_name.clone(), + protocol_version: conn.protocol_version, + current_user: conn.current_user.clone(), + flags: 0, + read_buf_remainder: read_buf.split(), + client_id, + peer_addr: peer_addr.clone(), + workspace_id: conn.workspace_id, + }); + return ( + MonoioHandlerResult::ParkIdle { + state, + registry_guard, + }, + Some(stream), + ); + } + } + } else { + let (result, returned_buf) = stream.read(tmp_buf).await; + tmp_buf = returned_buf; + match result { + Ok(0) => break, + Ok(n) => { + read_buf.extend_from_slice(&tmp_buf[..n]); + downshifted = false; + } + Err(_) => break, } - Err(_) => break, } } else if let Some(reg) = idle_reg.as_ref() { // c10k W11 stage 1: full-size read, registered for the shard diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index e12cae51..864be8d9 100644 --- a/src/shard/conn_accept.rs +++ b/src/shard/conn_accept.rs @@ -727,6 +727,7 @@ pub(crate) fn spawn_monoio_connection( BytesMut::new(), None, // fresh connection kill_fd, + false, // can_park: TLS keeps W11+P4b (no task park) ) .await; } @@ -760,6 +761,9 @@ pub(crate) fn spawn_monoio_connection( // locally if a migration hand-off cannot be delivered. #[cfg(target_os = "linux")] let (sd2, peer2) = (sd.clone(), peer_addr.clone()); + // c1M P1: kept for the ParkIdle routing below (`sd` moves + // into the handler call). + let sd_park = sd.clone(); let _result = handle_connection_sharded_monoio( tcp_stream, peer_addr, @@ -770,6 +774,7 @@ pub(crate) fn spawn_monoio_connection( BytesMut::new(), None, // fresh connection kill_fd, + true, // can_park: this site routes ParkIdle below ) .await; @@ -835,6 +840,32 @@ pub(crate) fn spawn_monoio_connection( } let _ = _hijacked_psync; let _result = (_result_outcome, _result_stream); + // c1M P1: task-exit parking. Hand the stream + ~100 B of + // session state to a tiny readiness watcher and end this + // (fat) task WITHOUT recording a close — the connection + // is still live; the watcher chain owns the close metric + // from here. + let _result = match _result { + ( + crate::server::conn::handler_monoio::MonoioHandlerResult::ParkIdle { + state, + registry_guard, + }, + Some(stream), + ) => { + spawn_parked_idle_watcher( + stream, + state, + registry_guard, + Box::new(conn_ctx), + sd_park, + cid, + kill_fd, + ); + return; + } + other => other, + }; // Handle migration result: extract FD via dup() and send via SPSC. // libc::dup is only available on Linux (target-specific dependency). #[cfg(target_os = "linux")] @@ -862,6 +893,7 @@ pub(crate) fn spawn_monoio_connection( false, // can_migrate BytesMut::new(), Some(&state), kill_fd, + false, // can_park: result is discarded here ) .await; } else { @@ -918,6 +950,7 @@ pub(crate) fn spawn_monoio_connection( false, // can_migrate: pin locally, no retry loop BytesMut::new(), Some(&payload.state), kill_fd, + false, // can_park: result is discarded here ) .await; } @@ -957,6 +990,131 @@ pub(crate) fn spawn_monoio_connection( // `unix`-gated alongside `runtime-monoio` for the same reason as the tokio twin // above: the `RawFd` signature + `from_raw_fd` are Unix-only. Always true on // supported targets (Linux + macOS). +/// c1M P1: park an idle connection out of the task model. The fat handler +/// task has exited; this tiny watcher owns the stream, the ~100 B session +/// state, and the client-registry guard (CLIENT LIST/KILL entry + maxclients +/// slot stay live). It wakes on read-readiness (`readable(false)` is +/// race-free on both drivers: uring PollAdd / legacy readiness+poll) or +/// server shutdown. CLIENT KILL's `shutdown(2)` makes the fd read-ready, so +/// a parked kill flows through the wake path and the resumed handler's first +/// read sees EOF. +/// +/// The future is boxed as `dyn` for two reasons: the parked footprint IS +/// this future, so it must stay small and measurable; and it breaks the +/// park→resume→park type cycle (the resumed handler task constructs this +/// watcher again on re-park). +#[cfg(all(feature = "runtime-monoio", unix))] +fn spawn_parked_idle_watcher( + stream: monoio::net::TcpStream, + state: Box, + registry_guard: crate::server::conn::handler_monoio::RegistryGuard, + conn_ctx: Box, + shutdown: CancellationToken, + client_id: u64, + kill_fd: i32, +) { + let fut: std::pin::Pin>> = Box::pin(async move { + let woke = monoio::select! { + res = stream.readable(false) => { + // Err (fd error) also resumes: the handler's first read + // surfaces the real error and tears down cleanly. + let _ = res; + true + } + _ = shutdown.cancelled() => false, + }; + if !woke { + // Server shutdown: this watcher owns the close accounting. + drop(registry_guard); + drop(stream); + crate::admin::metrics_setup::record_connection_closed(); + return; + } + // Wake: the handler re-registers this client_id at entry, so drop + // the parked entry first. The deregistered window lasts until the + // executor first polls the resumed task (a task-scheduling boundary, + // not just a few instructions): a racing CLIENT LIST misses the + // conn and CLIENT KILL ID returns 0 — same observable as a + // reconnect race, and a kill_flag set in that window is superseded + // by the shutdown(2) the killer already issued, which is what woke + // us. Closing the gap needs registration handoff into the handler + // (pass the guard through instead of drop/re-register); not worth + // it for a transient-invisibility race on an actively-waking conn. + drop(registry_guard); + spawn_resumed_parked_conn(stream, state, conn_ctx, shutdown, client_id, kill_fd); + }); + monoio::spawn(fut); +} + +/// c1M P1: rehydrate a parked connection into a full handler task, reusing +/// the migration-rehydration shape (`migrated_state` + initial bytes). +/// Resumed connections run with `can_migrate: false` (affinity re-sampling +/// would need the primary spawn site's full migration routing) and route +/// their own re-parks back to [`spawn_parked_idle_watcher`]. +#[cfg(all(feature = "runtime-monoio", unix))] +fn spawn_resumed_parked_conn( + stream: monoio::net::TcpStream, + mut state: Box, + conn_ctx: Box, + shutdown: CancellationToken, + client_id: u64, + kill_fd: i32, +) { + use crate::server::connection::handle_connection_sharded_monoio; + monoio::spawn(async move { + let initial = std::mem::take(&mut state.read_buf_remainder); + let peer = state.peer_addr.clone(); + let (outcome, stream_back) = handle_connection_sharded_monoio( + stream, + peer, + &conn_ctx, + shutdown.clone(), + client_id, + false, // can_migrate + initial, + Some(&state), + kill_fd, + true, // can_park: this site routes ParkIdle below + ) + .await; + match (outcome, stream_back) { + ( + crate::server::conn::handler_monoio::MonoioHandlerResult::ParkIdle { + state, + registry_guard, + }, + Some(stream), + ) => { + spawn_parked_idle_watcher( + stream, + state, + registry_guard, + conn_ctx, + shutdown, + client_id, + kill_fd, + ); + } + ( + crate::server::conn::handler_monoio::MonoioHandlerResult::HijackForPsync { .. }, + _, + ) => { + // A replica PSYNCs immediately after connecting — it never + // idles past --conn-park-secs first — so the full inline-PSYNC + // routing is not wired here. Loud so a real occurrence shows. + tracing::warn!( + "connection {}: PSYNC on a resumed parked connection is unsupported; closing", + client_id + ); + crate::admin::metrics_setup::record_connection_closed(); + } + _ => { + crate::admin::metrics_setup::record_connection_closed(); + } + } + }); +} + #[cfg(all(feature = "runtime-monoio", unix))] #[allow(clippy::too_many_arguments)] pub(crate) fn spawn_migrated_monoio_connection( @@ -1117,6 +1275,7 @@ pub(crate) fn spawn_migrated_monoio_connection( migration_buf, Some(&state), kill_fd, + false, // can_park: this site discards the result (follow-up: route ParkIdle here too) ) .await; // Migrated connection: the source shard's wrapper skipped the diff --git a/tests/mq_integration.rs b/tests/mq_integration.rs index ea6ac01c..83462557 100644 --- a/tests/mq_integration.rs +++ b/tests/mq_integration.rs @@ -91,6 +91,7 @@ async fn start_mq_server(num_shards: usize) -> (u16, CancellationToken) { io_driver: "auto".to_string(), io_busy_poll_us: 0, uring_entries: None, + conn_park_secs: 0, ft_search_workers: None, admin_port: 0, slowlog_log_slower_than: 10000, diff --git a/tests/parked_idle_parity.rs b/tests/parked_idle_parity.rs new file mode 100644 index 00000000..267a192d --- /dev/null +++ b/tests/parked_idle_parity.rs @@ -0,0 +1,230 @@ +//! c1M P1: task-exit parking must be invisible on the wire. With +//! `--conn-park-secs 2`, an idle plain-TCP connection downshifts (~1s), then +//! its handler task EXITS (~2s more), leaving only a readiness watcher. Every +//! byte sent afterwards must be served exactly as before on the same +//! connection, across multiple park/wake cycles; the parked connection must +//! stay visible to CLIENT LIST and killable by CLIENT KILL. +//! +//! (Runtime note: parking is monoio-plain-TCP only; under the tokio runtime +//! the suite degenerates to a plain parity check — green either way.) + +mod common; + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Command, Stdio}; +use std::time::Duration; + +/// Past downshift (1s) + park threshold (2s) + sweep cadence (1s) + margin. +const PARK_WAIT: Duration = Duration::from_millis(4600); + +fn spawn_moon(dir: &std::path::Path, port: u16) -> std::process::Child { + Command::new(common::find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--shards", + "1", + "--dir", + dir.to_str().unwrap(), + "--disk-free-min-pct", + "0", + "--conn-park-secs", + "2", + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") +} + +fn read_exact_deadline(stream: &mut TcpStream, want: usize) -> Vec { + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("set timeout"); + let mut out = Vec::with_capacity(want); + let mut chunk = [0u8; 4096]; + while out.len() < want { + match stream.read(&mut chunk) { + Ok(0) => break, + Ok(n) => out.extend_from_slice(&chunk[..n]), + Err(e) => panic!("read failed after {} of {} bytes: {}", out.len(), want, e), + } + } + out +} + +fn ping(stream: &mut TcpStream) { + stream.write_all(b"PING\r\n").expect("write PING"); + let r = read_exact_deadline(stream, 7); + assert_eq!(&r, b"+PONG\r\n"); +} + +/// Read one full RESP reply (bulk string or simple line) — enough for the +/// CLIENT LIST / CLIENT KILL helper commands below. +fn command_reply(stream: &mut TcpStream, cmd: &str) -> String { + stream.write_all(cmd.as_bytes()).expect("write cmd"); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("set timeout"); + let mut buf = Vec::new(); + let mut chunk = [0u8; 65536]; + loop { + let n = stream.read(&mut chunk).expect("read reply"); + assert!(n > 0, "connection closed mid-reply"); + buf.extend_from_slice(&chunk[..n]); + // Bulk replies announce their length; simple replies end at CRLF. + if buf.starts_with(b"$") { + if let Some(pos) = buf.iter().position(|&b| b == b'\n') { + let len: usize = std::str::from_utf8(&buf[1..pos - 1]) + .unwrap() + .trim() + .parse() + .unwrap(); + if buf.len() >= pos + 1 + len + 2 { + break; + } + } + } else if buf.ends_with(b"\r\n") { + break; + } + } + String::from_utf8_lossy(&buf).into_owned() +} + +/// Multiple park/wake cycles on one connection: probe wake, >512 B pipeline, +/// 4 KiB value round trip. +#[test] +fn parked_connection_serves_all_traffic_after_wake() { + let dir = tempfile::tempdir().expect("tempdir"); + let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p)); + + let mut conn = TcpStream::connect(("127.0.0.1", port)).expect("connect"); + conn.set_nodelay(true).ok(); + ping(&mut conn); + + // Cycle 1: park, then a probe-sized wake. + std::thread::sleep(PARK_WAIT); + ping(&mut conn); + + // Cycle 2: park again (the resumed handler must re-arm both stages), + // then a 100-deep pipeline. + std::thread::sleep(PARK_WAIT); + let pipeline = b"PING\r\n".repeat(100); + conn.write_all(&pipeline).expect("write pipeline"); + let replies = read_exact_deadline(&mut conn, 7 * 100); + assert_eq!(replies.len(), 700, "all 100 pipelined replies must arrive"); + assert!( + replies.chunks(7).all(|c| c == b"+PONG\r\n"), + "every pipelined reply must be +PONG" + ); + + // Cycle 3: park again, then a 4 KiB SET + GET round trip. + std::thread::sleep(PARK_WAIT); + let payload = vec![b'x'; 4096]; + let mut set_cmd = Vec::new(); + set_cmd.extend_from_slice(b"*3\r\n$3\r\nSET\r\n$8\r\nbigvalue\r\n$4096\r\n"); + set_cmd.extend_from_slice(&payload); + set_cmd.extend_from_slice(b"\r\n"); + conn.write_all(&set_cmd).expect("write SET"); + let r = read_exact_deadline(&mut conn, 5); + assert_eq!(&r, b"+OK\r\n"); + conn.write_all(b"*2\r\n$3\r\nGET\r\n$8\r\nbigvalue\r\n") + .expect("write GET"); + let want = format!("$4096\r\n{}\r\n", String::from_utf8(payload).unwrap()); + let r = read_exact_deadline(&mut conn, want.len()); + assert_eq!( + r, + want.as_bytes(), + "4 KiB value must survive the park cycle" + ); + + let _ = child.kill(); + let _ = child.wait(); +} + +/// A parked connection must stay in CLIENT LIST, and CLIENT KILL must close +/// it (the kill's shutdown(2) wakes the readiness watcher; the resumed +/// handler sees EOF). +#[test] +fn parked_connection_visible_and_killable() { + let dir = tempfile::tempdir().expect("tempdir"); + let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p)); + + // Victim: name itself so the control conn can find its id, then park. + let mut victim = TcpStream::connect(("127.0.0.1", port)).expect("connect victim"); + let r = command_reply(&mut victim, "CLIENT SETNAME parkvictim\r\n"); + assert!(r.starts_with("+OK"), "SETNAME failed: {r}"); + std::thread::sleep(PARK_WAIT); + + // Control connection stays active. + let mut control = TcpStream::connect(("127.0.0.1", port)).expect("connect control"); + let list = command_reply(&mut control, "CLIENT LIST\r\n"); + let victim_line = list + .lines() + .find(|l| l.contains("name=parkvictim")) + .unwrap_or_else(|| panic!("parked connection missing from CLIENT LIST:\n{list}")); + let victim_id: u64 = victim_line + .split_whitespace() + .find_map(|kv| kv.strip_prefix("id=")) + .expect("id= field") + .parse() + .expect("numeric id"); + + // Kill the parked victim by id. + let r = command_reply(&mut control, &format!("CLIENT KILL ID {victim_id}\r\n")); + assert!( + r.starts_with(":1"), + "CLIENT KILL must report 1 killed conn, got: {r}" + ); + + // The victim's socket must observe the close (EOF or reset) promptly. + victim + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("set timeout"); + let mut buf = [0u8; 16]; + match victim.read(&mut buf) { + Ok(0) => {} // EOF — clean close + Err(_) => {} // RST — also a close + Ok(n) => panic!("expected close, got {n} bytes"), + } + + // And it must be gone from CLIENT LIST. + let list = command_reply(&mut control, "CLIENT LIST\r\n"); + assert!( + !list.contains("name=parkvictim"), + "killed parked connection still listed:\n{list}" + ); + + let _ = child.kill(); + let _ = child.wait(); +} + +/// An active sibling must be undisturbed while its neighbor parks and wakes. +#[test] +fn active_sibling_undisturbed_by_parking() { + let dir = tempfile::tempdir().expect("tempdir"); + let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p)); + + let mut idle = TcpStream::connect(("127.0.0.1", port)).expect("connect idle"); + ping(&mut idle); + + let mut active = TcpStream::connect(("127.0.0.1", port)).expect("connect active"); + active.set_nodelay(true).ok(); + + // ~5s of continuous activity spanning downshift + park of the sibling. + let deadline = std::time::Instant::now() + PARK_WAIT + Duration::from_millis(500); + let mut rounds = 0u32; + while std::time::Instant::now() < deadline { + ping(&mut active); + rounds += 1; + std::thread::sleep(Duration::from_millis(5)); + } + assert!(rounds > 100, "active connection must keep full throughput"); + + // The parked sibling still answers. + ping(&mut idle); + + let _ = child.kill(); + let _ = child.wait(); +} diff --git a/tests/txn_kv_wiring.rs b/tests/txn_kv_wiring.rs index a46eeb40..c4782c54 100644 --- a/tests/txn_kv_wiring.rs +++ b/tests/txn_kv_wiring.rs @@ -103,6 +103,7 @@ async fn start_txn_server(num_shards: usize, persistence_dir: &str) -> (u16, Can io_driver: "auto".to_string(), io_busy_poll_us: 0, uring_entries: None, + conn_park_secs: 0, ft_search_workers: None, admin_port: 0, slowlog_log_slower_than: 10000, diff --git a/tests/workspace_integration.rs b/tests/workspace_integration.rs index 8e9eba56..5e7524d8 100644 --- a/tests/workspace_integration.rs +++ b/tests/workspace_integration.rs @@ -84,6 +84,7 @@ async fn start_workspace_server(num_shards: usize) -> (u16, CancellationToken) { io_driver: "auto".to_string(), io_busy_poll_us: 0, uring_entries: None, + conn_park_secs: 0, ft_search_workers: None, admin_port: 0, slowlog_log_slower_than: 10000, @@ -327,6 +328,7 @@ async fn start_workspace_server_with_auth( io_driver: "auto".to_string(), io_busy_poll_us: 0, uring_entries: None, + conn_park_secs: 0, ft_search_workers: None, admin_port: 0, slowlog_log_slower_than: 10000,