diff --git a/MODULE.bazel b/MODULE.bazel index fe2b1a6d83..5bab9d4221 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -66,10 +66,16 @@ crate.annotation( # other workers idle. One patch file holds both because crate_universe applies # an annotation's patches in sorted label order. Delete this annotation and the # patch once the pinned or published asupersync contains both fixes. +# The blocking-pool stack patch touches other files, so its order does not +# matter: it gives pool threads the runtime's thread_stack_size (16 MiB in the +# CLI) instead of the standard library's 2 MiB. crate.annotation( crate = "asupersync", patch_args = ["-p1"], - patches = ["//scripts:bazel/patches/asupersync-scheduler.patch"], + patches = [ + "//scripts:bazel/patches/asupersync-blocking-pool-stack.patch", + "//scripts:bazel/patches/asupersync-scheduler.patch", + ], version = "0.6.0", ) diff --git a/crates/tracedecay-cli/src/main.rs b/crates/tracedecay-cli/src/main.rs index 25e09fc13d..a23092ea61 100644 --- a/crates/tracedecay-cli/src/main.rs +++ b/crates/tracedecay-cli/src/main.rs @@ -233,6 +233,7 @@ impl Drop for Spinner { /// an explicit stack size gives every platform the same headroom. const ASYNC_STACK_BYTES: usize = 16 * 1024 * 1024; const MAX_ASYNC_WORKER_THREADS: usize = 16; +const MIN_LONG_LIVED_WORKER_THREADS: usize = 2; /// The runtimes a command's process owns. The native runtime drives every /// command's future. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -256,9 +257,14 @@ fn async_worker_threads(command: Option<&Commands>) -> usize { if is_one_shot_daemon_client(command) { return 1; } + // At least two, even on a one-CPU host: the reactor handoff patch + // (asupersync-scheduler.patch) promotes a parked follower only when + // the runtime has two or more parkers (`count < 2` returns early), so on + // one worker a long poll in `daemon run` or `serve` holds every socket + // wake and timer until it returns. std::thread::available_parallelism() .map_or(1, usize::from) - .clamp(1, MAX_ASYNC_WORKER_THREADS) + .clamp(MIN_LONG_LIVED_WORKER_THREADS, MAX_ASYNC_WORKER_THREADS) } /// Every command but `daemon run` is a daemon client or a local store tool @@ -572,10 +578,7 @@ fn async_main() -> tracedecay_domain::errors::Result { .thread_name_prefix("tracedecay-native") .worker_threads(worker_threads) .thread_stack_size(ASYNC_STACK_BYTES) - .blocking_threads( - 1, - blocking_threads.max(native_runtime::MAX_BLOCKING_THREADS), - ), + .blocking_threads(1, native_runtime::MAX_BLOCKING_THREADS), ) .map_err(|error| tracedecay_domain::errors::TraceDecayError::Config { message: error.to_string(), diff --git a/crates/tracedecay-cli/src/startup_tests.rs b/crates/tracedecay-cli/src/startup_tests.rs index 4f49b78a07..d0c6a53bb0 100644 --- a/crates/tracedecay-cli/src/startup_tests.rs +++ b/crates/tracedecay-cli/src/startup_tests.rs @@ -1,10 +1,11 @@ use super::{ AnalyticsAction, Cli, CommandFamily, CommandRuntime, Commands, DAEMON_CPU_THREADS_ENV, DEFAULT_MAX_DAEMON_CPU_THREADS, HostBundleCliOptions, PackageHookAction, RAYON_NUM_THREADS_ENV, - ScoopPackageHookAction, StderrTracingDefault, command_profile_label, command_runtime, - daemon_cpu_threads_from, is_one_shot_daemon_client, normalize_tool_reserved_global_flags, - runs_worldwide_counter_flush, should_skip_agent_install_check, should_skip_startup_maintenance, - stderr_tracing_default, try_print_cli_version, validate_host_bundle_options, + ScoopPackageHookAction, StderrTracingDefault, async_worker_threads, command_profile_label, + command_runtime, daemon_cpu_threads_from, is_one_shot_daemon_client, + normalize_tool_reserved_global_flags, runs_worldwide_counter_flush, + should_skip_agent_install_check, should_skip_startup_maintenance, stderr_tracing_default, + try_print_cli_version, validate_host_bundle_options, }; use clap::{CommandFactory, Parser}; use std::iter; @@ -390,6 +391,23 @@ fn status_is_a_one_shot_daemon_client() { ])))); } +#[test] +fn long_lived_commands_run_at_least_two_native_workers() { + for args in [&["daemon", "run"][..], &["serve"]] { + assert!( + async_worker_threads(Some(&parse_command(args))) >= 2, + "{args:?}" + ); + } + for args in [&["status"][..], &["tool", "tracedecay_status"]] { + assert_eq!( + async_worker_threads(Some(&parse_command(args))), + 1, + "{args:?}" + ); + } +} + #[test] fn only_the_foreground_daemon_owns_a_tokio_runtime() { for args in [ diff --git a/crates/tracedecay-code-index-runtime/src/code_index_task_support.rs b/crates/tracedecay-code-index-runtime/src/code_index_task_support.rs index 3e2b508d8a..836e9dd92a 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_task_support.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_task_support.rs @@ -102,6 +102,7 @@ pub async fn settle_owned_blocking_task( /// Starts `work` on the native runtime and detaches it. For work that never /// reaches Tokio. A refused start is logged, and `work` is dropped. +#[track_caller] pub(crate) fn spawn_detached(event: &'static str, work: impl Future + Send + 'static) { if let Err(error) = native_runtime::NativeTask::spawn(work) { tracing::warn!(event, %error, "native task did not start"); diff --git a/crates/tracedecay-lsp/src/gateway/runtime.rs b/crates/tracedecay-lsp/src/gateway/runtime.rs index 201a8a62ff..b421075861 100644 --- a/crates/tracedecay-lsp/src/gateway/runtime.rs +++ b/crates/tracedecay-lsp/src/gateway/runtime.rs @@ -42,6 +42,7 @@ pub trait LspRuntimeTask: Send + Sync { /// Runtime injection used by broker policy without taking a Tokio dependency. pub trait LspRuntimeSpawner: Send + Sync { + #[track_caller] fn spawn(&self, future: LspRuntimeFuture<()>) -> Box; } diff --git a/crates/tracedecay-rusqlite-runtime/Cargo.toml b/crates/tracedecay-rusqlite-runtime/Cargo.toml index e674286a74..248e15b6ee 100644 --- a/crates/tracedecay-rusqlite-runtime/Cargo.toml +++ b/crates/tracedecay-rusqlite-runtime/Cargo.toml @@ -32,7 +32,7 @@ tracedecay-contracts = { path = "../tracedecay-contracts" } tracedecay-store = { path = "../tracedecay-store" } tracing = "0.1" -# Thread CPU time for the debug-build long-poll guard on native tasks. +# Thread CPU time for the long-poll count on native tasks. [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs index b0ae1d4c9b..6062c3b582 100644 --- a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs +++ b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs @@ -24,8 +24,10 @@ use asupersync::types::PanicPayload; use asupersync::{Budget, Cx}; use futures_util::future::{AbortHandle, Aborted, BoxFuture, FutureExt, Shared, abortable}; -/// Tokio's default blocking-pool ceiling. Shard closes and SQL reads park a -/// blocking thread for their whole duration. +/// The native blocking pool's thread ceiling. Shard closes, SQL reads, and +/// [`off_native_thread`] handlers each hold a pool thread for their whole +/// duration, mostly parked, so this bounds concurrent waits rather than CPU +/// use. The value matches Tokio's `max_blocking_threads` default. pub const MAX_BLOCKING_THREADS: usize = 512; static PROCESS_RUNTIME: OnceLock> = OnceLock::new(); @@ -73,6 +75,11 @@ pub fn block_on(future: F) -> Result { /// be cancelled with that region before it starts, which would strand anything /// the caller had already queued for it. /// +/// Called on a blocking-pool thread that already runs work queued here (an +/// [`off_native_thread`] handler, say), it runs `work` inline at the call: +/// queuing would hold this pool thread while waiting for another, and a full +/// pool would deadlock on that wait. +/// /// Boxed so callers prove the context and channel wait `Send` once, here, /// instead of inside every future that awaits blocking work /// (rust-lang/rust#159228). @@ -81,6 +88,11 @@ where T: Send + 'static, F: FnOnce() -> T + Send + 'static, { + if ON_BLOCKING_POOL.get() { + let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(work)) + .map_err(|_| "native blocking work panicked".to_owned()); + return Box::pin(std::future::ready(ran)); + } let queued = queue_blocking(work); Box::pin(async move { let (cx, mut receiver) = queued?; @@ -139,12 +151,20 @@ where let (sender, receiver) = asupersync::channel::oneshot::channel(); runtime .spawn_blocking(move || { + ON_BLOCKING_POOL.set(true); let _ = sender.send_blocking(work()); }) .ok_or("native blocking pool unavailable")?; Ok((cx, receiver)) } +thread_local! { + /// Set for good once a blocking-pool thread runs work [`queue_blocking`] + /// or [`NativeTask::spawn_off_native_thread`] queued; such a thread serves + /// the pool for its whole life. + static ON_BLOCKING_POOL: std::cell::Cell = const { std::cell::Cell::new(false) }; +} + /// Drives `settlement` on the caller's executor. A caller that stops waiting /// hands the unfinished settlement to the runtime, so accepted work still /// outlives a cancelled request. Spawning a native task for every call instead @@ -281,46 +301,77 @@ impl Future for Timeout { } } -/// The most CPU time one poll of a [`NativeTask`] may take in a debug build. +/// The wall time past which one poll of a [`NativeTask`] counts as long. +/// +/// While a worker is inside a poll, no other task queued on that worker runs. +/// A one-worker runtime also has no follower to take the reactor, and a +/// `block_on` root never turns it, so there every socket wake and timer waits +/// for the poll to return (#3994, upstream Dicklesworthstone/asupersync#83, +/// whose multi-worker half Bazel patches in). The waiters see wall time, +/// whether the poll computed, parked on a lock or a writer reply, or was +/// descheduled, so wall time decides what counts; thread CPU time is logged +/// beside it on unix to tell computing from parking. /// -/// A one-worker runtime has no follower to take the reactor while its worker -/// is inside a poll, and a `block_on` root never turns the reactor. Every -/// socket wake and timer on that runtime waits for the poll to return (#3994, -/// upstream Dicklesworthstone/asupersync#83, whose multi-worker half Bazel -/// patches in). -pub const LONG_POLL_CPU: Duration = Duration::from_millis(250); - -/// Fails a debug build's native task when one poll exceeds [`LONG_POLL_CPU`] -/// of thread CPU time. CPU time, not wall time, so a worker descheduled on a -/// loaded host does not trip it. -#[cfg(all(debug_assertions, unix))] +/// 100 ms is where a reply delayed behind the poll becomes noticeable to the +/// person waiting on a hook or MCP call. A poll's cost moves with host load +/// and opt level (a 7 ms fastbuild dispatch measured 490-590 ms of CPU at load +/// average ~1000), so this bound sets the floor of a latency report, not a +/// correctness line: a long poll is logged and counted, never failed. +pub const LONG_POLL: Duration = Duration::from_millis(100); + +static LONG_POLLS: Mutex, u64>> = + Mutex::new(BTreeMap::new()); + +/// Long native-task polls this process ran, by the site that spawned the +/// task. Each entry names work that held a native worker past [`LONG_POLL`] +/// and is a candidate for [`spawn_blocking`] when its latency matters. +#[must_use] +pub fn long_native_polls() -> BTreeMap<&'static std::panic::Location<'static>, u64> { + LONG_POLLS + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone() +} + +/// Times every poll of a [`NativeTask`] and counts the ones past +/// [`LONG_POLL`] against the task's spawn site. It never alters the poll's +/// result. Per poll it reads the monotonic clock twice and the thread CPU +/// clock once, plus once more after a long poll. struct LongPollGuard { work: F, site: &'static std::panic::Location<'static>, } -#[cfg(all(debug_assertions, unix))] impl Future for LongPollGuard { type Output = F::Output; fn poll(mut self: Pin<&mut Self>, task: &mut Context<'_>) -> Poll { - let started = thread_cpu_time(); + let started = Instant::now(); + let cpu_started = thread_cpu_time(); let poll = Pin::new(&mut self.work).poll(task); - if let (Some(started), Some(ended)) = (started, thread_cpu_time()) { - let spent = ended.saturating_sub(started); - assert!( - spent <= LONG_POLL_CPU, - "native task spawned at {} spent {spent:?} of CPU in one poll, past \ - {LONG_POLL_CPU:?}; move its synchronous work to native_runtime::spawn_blocking \ - (#3994)", - self.site, + let wall = started.elapsed(); + if wall >= LONG_POLL { + let cpu = cpu_started + .zip(thread_cpu_time()) + .map(|(started, ended)| ended.saturating_sub(started)); + *LONG_POLLS + .lock() + .unwrap_or_else(PoisonError::into_inner) + .entry(self.site) + .or_default() += 1; + tracing::warn!( + spawn_site = %self.site, + ?wall, + ?cpu, + "native task held its worker for one poll past {LONG_POLL:?}; \ + native_runtime::spawn_blocking keeps that work off the worker" ); } poll } } -#[cfg(all(debug_assertions, unix))] +#[cfg(unix)] fn thread_cpu_time() -> Option { let mut now = libc::timespec { tv_sec: 0, @@ -336,6 +387,68 @@ fn thread_cpu_time() -> Option { }) } +#[cfg(not(unix))] +fn thread_cpu_time() -> Option { + None +} + +/// How long the scheduling-lateness canary sleeps between wakes. +const CANARY_TICK: Duration = Duration::from_millis(100); + +/// What the scheduling-lateness canary measured in this process. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct SchedulingLateness { + pub ticks: u64, + /// Wakes that ran [`LONG_POLL`] or more past their deadline. + pub late_ticks: u64, + pub worst: Duration, +} + +static SCHEDULING_LATENESS: Mutex = Mutex::new(SchedulingLateness { + ticks: 0, + late_ticks: 0, + worst: Duration::ZERO, +}); + +#[must_use] +pub fn scheduling_lateness() -> SchedulingLateness { + *SCHEDULING_LATENESS + .lock() + .unwrap_or_else(PoisonError::into_inner) +} + +/// Starts a native task that sleeps on a native timer again and again and +/// records how late each wake runs. A wake runs late when no worker turns the +/// timer or picks up the woken task: a long poll on a one-worker runtime, or +/// every worker busy. That is the delay a socket reply waits behind, measured +/// directly rather than inferred from [`long_native_polls`]. A wake +/// [`LONG_POLL`] late or worse is logged. Abort the task to stop it. +#[track_caller] +pub fn spawn_lateness_canary() -> Result, String> { + NativeTask::spawn(async { + loop { + let deadline = Instant::now() + CANARY_TICK; + if let Err(error) = sleep_until(deadline).await { + tracing::warn!(%error, "scheduling-lateness canary stopped"); + return; + } + let late = deadline.elapsed(); + let mut lateness = SCHEDULING_LATENESS + .lock() + .unwrap_or_else(PoisonError::into_inner); + lateness.ticks += 1; + lateness.worst = lateness.worst.max(late); + if late >= LONG_POLL { + lateness.late_ticks += 1; + tracing::warn!( + ?late, + "native runtime woke a timer late; a long poll or busy workers held it" + ); + } + } + }) +} + /// How a [`NativeTask`] ended. #[derive(Clone, Debug, PartialEq, Eq)] pub enum NativeTaskOutcome { @@ -362,8 +475,9 @@ impl NativeTask { /// runtime. The runtime owns it, not the caller's region, so it outlives /// the request that started it. /// - /// `work` must not run CPU-heavy synchronous code inside one poll; see - /// [`LONG_POLL_CPU`]. Queue that part with [`spawn_blocking`]. + /// A poll of `work` past [`LONG_POLL`] is logged and counted against the + /// caller's location in [`long_native_polls`]. Queue blocking or + /// CPU-heavy parts with [`spawn_blocking`] when their latency matters. #[track_caller] pub fn spawn(work: F) -> Result where @@ -373,7 +487,6 @@ impl NativeTask { // Boxed so the runtime's own spawn wrappers do not nest `work` in // their `Send` proof (rust-lang/rust#159228). let work: BoxFuture<'static, T> = Box::pin(work); - #[cfg(all(debug_assertions, unix))] let work = LongPollGuard { work, site: std::panic::Location::caller(), @@ -412,6 +525,7 @@ impl NativeTask { handle() .ok_or("native execution runtime unavailable")? .spawn_blocking(move || { + ON_BLOCKING_POOL.set(true); let polled = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { poll_on_current_thread(work) })); @@ -699,10 +813,6 @@ fn start(builder: RuntimeBuilder) -> Result { } #[cfg(test)] -#[allow( - clippy::disallowed_methods, - reason = "tests run on an ambient Tokio runtime" -)] mod tests { use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -1292,7 +1402,6 @@ mod tests { #[cfg(unix)] const WAKE_BOUND: Duration = Duration::from_millis(40); - #[cfg(unix)] fn spin(duration: Duration) { let spun = Instant::now(); while spun.elapsed() < duration { @@ -1370,34 +1479,133 @@ mod tests { ); } - #[cfg(all(debug_assertions, unix))] + #[track_caller] + fn spawned_here( + work: impl Future + Send + 'static, + ) -> (NativeTask, &'static std::panic::Location<'static>) { + ( + NativeTask::spawn(work).expect("native task"), + std::panic::Location::caller(), + ) + } + + /// The guard's window encloses the spin, so a descheduled worker only + /// lengthens the measured poll. + #[test] + fn a_long_native_poll_completes_and_is_counted_at_its_spawn_site() { + let runtime = RuntimeBuilder::multi_thread() + .worker_threads(1) + .build() + .expect("test runtime"); + let (long, short) = runtime.block_on(async { + let (task, site) = spawned_here(async { + spin(LONG_POLL + Duration::from_millis(50)); + 7 + }); + let long = (task.wait().await, site); + let (task, site) = spawned_here(async { 8 }); + (long, (task.wait().await, site)) + }); + assert_eq!(long.0, NativeTaskOutcome::Completed(7)); + assert_eq!(long_native_polls().get(long.1), Some(&1), "{}", long.1); + assert_eq!(short.0, NativeTaskOutcome::Completed(8)); + assert_eq!(long_native_polls().get(short.1), None, "{}", short.1); + } + + /// Queued instead, the inner work would wait for a second pool thread + /// while the outer one holds the first; a full pool never frees one. #[test] - fn a_native_task_poll_past_the_cpu_bound_fails_in_debug_builds() { + fn blocking_work_spawned_from_the_blocking_pool_runs_on_the_same_thread() { + let (outer, inner) = block_on(spawn_blocking(|| { + let inner = spawn_blocking(|| std::thread::current().id()).now_or_never(); + (std::thread::current().id(), inner) + })) + .expect("process runtime") + .expect("outer blocking work"); + assert_eq!(inner, Some(Ok(outer))); + } + + /// A fresh one-thread pool, so the thread has run no other queued work. + #[test] + fn blocking_work_from_a_task_spawned_off_native_threads_runs_on_its_thread() { let runtime = RuntimeBuilder::multi_thread() .worker_threads(1) + .blocking_threads(1, 1) .build() .expect("test runtime"); let outcome = runtime.block_on(async { - // Spins on the thread's CPU clock, the one the guard reads, so a - // descheduled worker on a loaded host still crosses the bound. - NativeTask::spawn(async { - let started = thread_cpu_time().expect("thread CPU clock"); - while thread_cpu_time().expect("thread CPU clock") - started - <= LONG_POLL_CPU + Duration::from_millis(50) - { - std::hint::spin_loop(); - } + NativeTask::spawn_off_native_thread(async { + spawn_blocking(|| std::thread::current().id()).now_or_never() + == Some(Ok(std::thread::current().id())) }) - .expect("native task") + .expect("blocking-pool task") .wait() .await }); - let NativeTaskOutcome::Panicked(payload) = outcome else { - panic!("a {LONG_POLL_CPU:?} poll must fail the task, got {outcome:?}"); - }; + assert_eq!(outcome, NativeTaskOutcome::Completed(true)); + } + + #[cfg(target_os = "linux")] + fn current_thread_stack_size() -> usize { + let mut attributes = std::mem::MaybeUninit::::uninit(); + let mut size = 0; + // SAFETY: `pthread_getattr_np` initializes `attributes` for the calling + // thread before `pthread_attr_getstacksize` reads it, and they are + // destroyed once, after that read. + unsafe { + assert_eq!( + libc::pthread_getattr_np(libc::pthread_self(), attributes.as_mut_ptr()), + 0 + ); + assert_eq!( + libc::pthread_attr_getstacksize(attributes.as_ptr(), &raw mut size), + 0 + ); + libc::pthread_attr_destroy(attributes.as_mut_ptr()); + } + size + } + + #[cfg(target_os = "linux")] + #[test] + fn blocking_pool_threads_get_the_runtime_thread_stack_size() { + const STACK: usize = 16 * 1024 * 1024; + let runtime = RuntimeBuilder::multi_thread() + .worker_threads(1) + .thread_stack_size(STACK) + .blocking_threads(1, 2) + .build() + .expect("test runtime"); + let stack = runtime + .block_on(async { spawn_blocking(current_thread_stack_size).await }) + .expect("blocking work"); + assert!(stack >= STACK, "pool thread stack is {stack} bytes"); + } + + #[test] + fn the_lateness_canary_records_a_wake_held_behind_a_long_poll() { + let runtime = RuntimeBuilder::multi_thread() + .worker_threads(1) + .build() + .expect("test runtime"); + runtime.block_on(async { + let canary = spawn_lateness_canary().expect("canary"); + sleep_until(Instant::now() + Duration::from_millis(20)) + .await + .expect("let the canary arm its first timer"); + NativeTask::spawn(async { spin(CANARY_TICK + LONG_POLL * 3) }) + .expect("spinning task") + .wait() + .await; + sleep_until(Instant::now() + CANARY_TICK) + .await + .expect("let the canary record its overdue wake"); + canary.abort(); + }); + let lateness = scheduling_lateness(); assert!( - payload.to_string().contains("spawn_blocking"), - "the failure names the remedy: {payload}" + lateness.late_ticks >= 1 && lateness.worst >= LONG_POLL, + "{lateness:?}" ); } diff --git a/crates/tracedecay/src/daemon/bootstrap.rs b/crates/tracedecay/src/daemon/bootstrap.rs index e7385609b0..ac5f6790e2 100644 --- a/crates/tracedecay/src/daemon/bootstrap.rs +++ b/crates/tracedecay/src/daemon/bootstrap.rs @@ -193,6 +193,7 @@ async fn run_foreground_loopback( }, ) .await; + let lateness_canary = spawn_lateness_canary(); let spooled_hook_opener = hook_v2_replay_consumer::spawn_spooled_hook_opener( hook_v2_replay_consumer::PortableSpoolOpenerOwners { lifecycle: lifecycle.clone(), @@ -279,6 +280,9 @@ async fn run_foreground_loopback( if let Some(opener) = spooled_hook_opener { opener.abort(); } + if let Some(canary) = lateness_canary { + canary.abort(); + } cancel_retained_session_history(&store_administration).await; let shutdown_deadline = Instant::now() + DAEMON_SHUTDOWN_COORDINATION_DEADLINE; let endpoint_cleanup = authority.cleanup_owned_endpoint(); @@ -493,8 +497,15 @@ fn log_background_shutdown_receipt(receipt: &tracedecay_daemon_service::shutdown /// The daemon's lifetime count of Tokio-only paths skipped for want of a Tokio /// runtime, and of writer and reader waits that parked a native thread. Zero -/// is the only healthy value for each. +/// is the only healthy value for each. Long native polls are listed by spawn +/// site, and the lateness canary reports timer wakes that ran late. fn log_runtime_hazards() { + let long_native_polls = native_runtime::long_native_polls() + .into_iter() + .map(|(site, polls)| format!("{site}={polls}")) + .collect::>() + .join(","); + let lateness = native_runtime::scheduling_lateness(); log_daemon_event( "daemon_runtime_hazards", &[ @@ -510,10 +521,28 @@ fn log_runtime_hazards() { "native_thread_reader_waits", tracedecay_rusqlite_runtime::exact_sql::native_thread_reader_waits().to_string(), ), + ("long_native_polls", long_native_polls), + ("late_timer_wakes", lateness.late_ticks.to_string()), + ( + "worst_timer_lateness_ms", + lateness.worst.as_millis().to_string(), + ), ], ); } +/// Measures, for the daemon's lifetime, how late native timer wakes run. +fn spawn_lateness_canary() -> Option> { + native_runtime::spawn_lateness_canary() + .inspect_err(|error| { + log_daemon_event( + "daemon_lateness_canary_unavailable", + &[("error", error.clone())], + ); + }) + .ok() +} + fn log_project_server_shutdown_receipt(receipt: &tracedecay_store_runtime::ShutdownTaskReceipt) { if receipt.is_clean() { return; @@ -774,6 +803,7 @@ async fn run_foreground_unix( .with_pr_autotrack_task(pr_autotrack_task) .await; let mut shutdown_signals = ShutdownSignals::listen()?; + let lateness_canary = spawn_lateness_canary(); let spooled_hook_opener = hook_v2_replay_consumer::spawn_spooled_hook_opener(engine.clone(), profile.clone()); let admission = DaemonClientAdmission::new(MAX_CONCURRENT_DAEMON_CLIENTS); @@ -854,6 +884,9 @@ async fn run_foreground_unix( if let Some(rotation) = stderr_log_rotation { rotation.abort(); } + if let Some(canary) = lateness_canary { + canary.abort(); + } // Stop accepting and unlink the socket before draining so clients that // connect during shutdown get NotFound/ConnectionRefused (which they retry // via `connect_with_restart_grace`) instead of a queued connection that diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index db2c58b059..300da67891 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -70,6 +70,7 @@ async fn registered_project_stores( /// Starts one maintenance owner on the native runtime. A refusal is logged /// with the owner it leaves unstarted. +#[track_caller] fn spawn_maintenance_task( owner: &'static str, work: impl std::future::Future + Send + 'static, diff --git a/crates/tracedecay/src/doctor.rs b/crates/tracedecay/src/doctor.rs index 052ec06b91..6b8fb5e20a 100644 --- a/crates/tracedecay/src/doctor.rs +++ b/crates/tracedecay/src/doctor.rs @@ -286,6 +286,7 @@ where /// A daemon round trip running on the native runtime beside the blocking /// checks. The result travels over a channel because a `NativeTask` output /// must be `Clone`, and the doctor's error type is not. +#[track_caller] fn spawn_doctor_task( work: impl Future + Send + 'static, ) -> DoctorCheck { diff --git a/scripts/bazel/patches/asupersync-blocking-pool-stack.patch b/scripts/bazel/patches/asupersync-blocking-pool-stack.patch new file mode 100644 index 0000000000..c601b520c2 --- /dev/null +++ b/scripts/bazel/patches/asupersync-blocking-pool-stack.patch @@ -0,0 +1,66 @@ +Blocking-pool threads take the runtime's thread_stack_size instead of the +standard library default (2 MiB), so work moved off a worker keeps the stack +the worker had. + +diff --git a/src/runtime/blocking_pool.rs b/src/runtime/blocking_pool.rs +index b6bac66..a13ad99 100644 +--- a/src/runtime/blocking_pool.rs ++++ b/src/runtime/blocking_pool.rs +@@ -346,6 +346,8 @@ struct BlockingPoolInner { + sleep_fn: SleepFn, + /// Thread name prefix. + thread_name_prefix: String, ++ /// Stack size for each pool thread; zero keeps the standard library's. ++ thread_stack_size: usize, + /// Callback when a thread starts. + on_thread_start: Option>, + /// Callback when a thread stops. +@@ -567,6 +569,7 @@ impl BlockingPool { + time_getter: options.time_getter, + sleep_fn: options.sleep_fn, + thread_name_prefix: options.thread_name_prefix, ++ thread_stack_size: options.thread_stack_size, + on_thread_start: options.on_thread_start, + on_thread_stop: options.on_thread_stop, + thread_handles: Mutex::new(Vec::with_capacity(max_threads)), +@@ -1189,6 +1192,8 @@ pub struct BlockingPoolOptions { + pub sleep_fn: SleepFn, + /// Thread name prefix. + pub thread_name_prefix: String, ++ /// Stack size for each pool thread; zero keeps the standard library's. ++ pub thread_stack_size: usize, + /// Callback when a thread starts. + pub on_thread_start: Option>, + /// Callback when a thread stops. +@@ -1206,6 +1211,7 @@ impl Default for BlockingPoolOptions { + time_getter: wall_clock_now, + sleep_fn: blocking_thread_sleep, + thread_name_prefix: "asupersync".to_string(), ++ thread_stack_size: 0, + on_thread_start: None, + on_thread_stop: None, + affinity_profile: BlockingPoolAffinityProfile::Disabled, +@@ -1326,7 +1332,10 @@ fn spawn_thread_on_inner(inner: &Arc) { + // panic and strand the pool in an inconsistent state. + let thread_id = inner.next_thread_id.fetch_add(1, Ordering::Relaxed); + let name = format!("{}-blocking-{}", inner.thread_name_prefix, thread_id); +- let builder = thread::Builder::new().name(name); ++ let mut builder = thread::Builder::new().name(name); ++ if inner.thread_stack_size > 0 { ++ builder = builder.stack_size(inner.thread_stack_size); ++ } + let assigned_cohort = inner + .affinity + .as_ref() +diff --git a/src/runtime/builder.rs b/src/runtime/builder.rs +index 6bddecd..bba4600 100644 +--- a/src/runtime/builder.rs ++++ b/src/runtime/builder.rs +@@ -6348,6 +6348,7 @@ impl RuntimeInner { + let options = crate::runtime::blocking_pool::BlockingPoolOptions { + idle_timeout: Duration::from_secs(10), + thread_name_prefix: format!("{}-blocking", config.thread_name_prefix), ++ thread_stack_size: config.thread_stack_size, + on_thread_start: config.on_thread_start.clone(), + on_thread_stop: config.on_thread_stop.clone(), + affinity_profile: config.blocking.affinity_profile,