From a1f5b1ffd7211c29f4fa1a75266f9c198e8a82ff Mon Sep 17 00:00:00 2001 From: Levi Neuwirth Date: Tue, 7 Jul 2026 12:54:33 -0400 Subject: [PATCH] fix(worker): deterministic pool teardown; EditorState::drop signals workers Every EditorState leaked its entire worker pool (cores-1 threads, each waking every 100ms): the Rc is cloned into dozens of Lua closures, and registries those closures capture store mlua::Function values -- reference cycles through the Lua VM that keep the Rc from ever reaching zero, so WorkerPool::Drop never ran. Harmless for one editor per process; in the m4 acceptance suite (54 editor-building tests) it accumulated 1000+ live threads (observed: 60 complete pmacs-worker-0..14 pools at once) plus ~9k spurious wakeups/second. Fix: WorkerPool::signal_shutdown() -- set the shutdown flag + wake the parkers from a shared reference; parked workers exit within their 100ms park timeout. Exposed as AsyncRuntime::shutdown_workers(), called from a new impl Drop for EditorState. Deliberately signal-only, NO join on the drop path: a worker can be blocked publishing its reply onto the bus only the main thread drains, and a first cut that joined in Drop deadlocked the m4 suite at teardown (fake-LSP tests wedged 2h+). WorkerPool::shutdown() (signal + join) remains for owners with no bus consumer to deadlock against; Drop delegates to it as before. Measured on the m4 suite: peak live threads 1026 -> 122. Regression: tests/worker_shutdown_acceptance.rs (thread-count probe is /proc-based, Linux-gated; the idempotence test runs everywhere). Co-Authored-By: Claude Fable 5 --- src/async_runtime.rs | 14 +++++ src/editor.rs | 26 ++++++++++ src/worker.rs | 77 ++++++++++++++++++++++------ tests/completion_popup_acceptance.rs | 2 +- tests/worker_shutdown_acceptance.rs | 47 +++++++++++++++++ 5 files changed, 149 insertions(+), 17 deletions(-) create mode 100644 tests/worker_shutdown_acceptance.rs diff --git a/src/async_runtime.rs b/src/async_runtime.rs index cad4235..94ae001 100644 --- a/src/async_runtime.rs +++ b/src/async_runtime.rs @@ -602,6 +602,20 @@ impl AsyncRuntime { Self::with_pool(WorkerPool::new(size)) } + /// Signal the worker-pool threads to exit (see + /// [`crate::worker::WorkerPool::signal_shutdown`]). Idempotent, + /// non-blocking. Called from [`crate::editor::EditorState`]'s + /// `Drop`: the runtime's `Rc` is captured into Lua-VM reference + /// cycles and never reaches a zero refcount, so without this + /// explicit teardown every editor instance leaks its whole pool + /// --- one leaked pool per test in a suite that builds real + /// editors. Signal-only (no join): a worker blocked handing its + /// reply to the main thread must not deadlock the main thread's + /// drop. + pub fn shutdown_workers(&self) { + self.pool.signal_shutdown(); + } + /// Returns the number of in-flight or settled-but-not-yet-taken /// pending entries. #[must_use] diff --git a/src/editor.rs b/src/editor.rs index c913191..88bdac7 100644 --- a/src/editor.rs +++ b/src/editor.rs @@ -102,6 +102,32 @@ pub struct EditorState { mouse_click: Option, } +impl Drop for EditorState { + /// Tear down the worker-pool threads. + /// + /// The `Rc` is cloned into dozens of Lua closures + /// (`pmacs.workers`, LSP request wrappers, ...), and several of + /// the registries those closures capture themselves store + /// `mlua::Function` values --- reference cycles through the Lua + /// VM that keep the `Rc` from ever reaching zero. Harmless for a + /// single editor per process (the OS reclaims at exit), but a + /// test binary that builds one `EditorState` per test would leak + /// one full worker pool (`cores - 1` threads, each waking every + /// 100ms) per test --- observed as 1000+ live threads in the m4 + /// acceptance suite. Dropping the editor reaches the pool through + /// its own `Rc` clone and signals the threads down regardless of + /// the cycle; parked workers exit within their 100ms wakeup. + /// + /// Signal-only, NO join: a worker can be blocked publishing its + /// reply onto the bus that this (main) thread drains --- joining + /// here deadlocked the m4 suite at teardown for hours. A worker + /// stuck mid-handoff stays alive (bounded by its job), which is + /// still a ~15x improvement over leaking every pool whole. + fn drop(&mut self) { + self.async_runtime.shutdown_workers(); + } +} + #[derive(Copy, Clone)] struct MouseClickState { frontend_id: FrontendId, diff --git a/src/worker.rs b/src/worker.rs index 03a41a4..ade0a2d 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -175,7 +175,17 @@ impl PoolShared { /// thread). pub struct WorkerPool { shared: Arc, - workers: Vec>, + /// Join handles, drained exactly once by [`Self::shutdown`] + /// (directly or via `Drop`). Behind a `Mutex` so shutdown works + /// from a shared reference: the pool's owner is typically an + /// `Rc` cloned into Lua closures, and those clones + /// form VM reference cycles that keep the `Rc` from ever + /// reaching zero --- an embedder that merely *drops* its handle + /// would leak every worker thread. `EditorState::drop` calls + /// `shutdown()` explicitly instead. + workers: Mutex>>, + /// Thread count at construction (stable across shutdown). + size: usize, } impl WorkerPool { @@ -194,7 +204,7 @@ impl WorkerPool { next_id: AtomicU64::new(0), parker: (Mutex::new(()), Condvar::new()), }); - let workers = local_queues + let workers: Vec> = local_queues .into_iter() .enumerate() .map(|(idx, local)| { @@ -205,7 +215,11 @@ impl WorkerPool { .expect("spawn worker thread") }) .collect(); - Self { shared, workers } + Self { + shared, + size, + workers: Mutex::new(workers), + } } /// Build a pool sized at `available_parallelism - 1`, with a @@ -218,10 +232,48 @@ impl WorkerPool { Self::new(cores.saturating_sub(1)) } - /// Number of worker threads owned by this pool. + /// Number of worker threads this pool was built with. #[must_use] pub fn size(&self) -> usize { - self.workers.len() + self.size + } + + /// Signal every worker to exit, without joining. Idle (parked) + /// workers observe the flag within their 100ms park timeout and + /// return; a worker mid-job exits when its job finishes. Queued + /// jobs that haven't been picked up are dropped without running; + /// jobs dispatched *after* the signal are never picked up. + /// + /// Exists as an explicit method (not just `Drop`) because the + /// pool's owning `Rc` is captured into Lua-VM + /// reference cycles and may never be reclaimed --- callers that + /// know the editor is going away (`EditorState::drop`) signal the + /// threads down regardless. + /// + /// Deliberately does NOT join: a worker can be blocked publishing + /// its reply onto the message bus that only the *main thread* + /// drains, so a main-thread join here is a deadlock (observed as + /// the m4 acceptance suite wedging for hours at teardown). Callers + /// that own the whole world and want the join use + /// [`Self::shutdown`] (or just drop the pool). + pub fn signal_shutdown(&self) { + self.shared.shutdown.store(true, Ordering::Release); + self.shared.notify_all(); + } + + /// [`Self::signal_shutdown`] plus a join of every worker thread. + /// Idempotent: the second call finds no handles and returns + /// immediately. Only safe where no worker can be blocked on the + /// caller's own thread (see `signal_shutdown`); `Drop` uses it + /// because a pool being dropped has no live bus consumer to + /// deadlock against in the bare-pool case. + pub fn shutdown(&self) { + self.signal_shutdown(); + let handles: Vec> = + std::mem::take(&mut *self.workers.lock().expect("worker pool mutex poisoned")); + for handle in handles { + let _ = handle.join(); + } } /// Submit `work` to be run on a worker. Returns a [`JobHandle`] @@ -263,18 +315,11 @@ impl WorkerPool { } impl Drop for WorkerPool { - /// Shutdown semantics: dropping the pool signals every worker - /// to exit at its next idle wakeup and joins them. Running - /// jobs run to completion (or to their own cancellation - /// check); queued jobs that haven't been picked up are dropped - /// without running. Tests and callers that want explicit - /// shutdown just `drop(pool)`. + /// Dropping the pool is an implicit [`Self::shutdown`]: every + /// worker is signalled to exit at its next idle wakeup and + /// joined. No-op when `shutdown` already ran. fn drop(&mut self) { - self.shared.shutdown.store(true, Ordering::Release); - self.shared.notify_all(); - for handle in self.workers.drain(..) { - let _ = handle.join(); - } + self.shutdown(); } } diff --git a/tests/completion_popup_acceptance.rs b/tests/completion_popup_acceptance.rs index 6243c25..774f204 100644 --- a/tests/completion_popup_acceptance.rs +++ b/tests/completion_popup_acceptance.rs @@ -3,7 +3,7 @@ //! the Q#C3 partial dispatcher shadow, and Q#C7 validated accept, all //! driven through `dispatch_key` exactly as a terminal user would. //! -//! The LSP provider's request path is covered by the m4_5 fake-LSP +//! The LSP provider's request path is covered by the `m4_5` fake-LSP //! suite; these tests run hermetic (dabbrev + custom Lua providers) //! so no server binary is needed. //! diff --git a/tests/worker_shutdown_acceptance.rs b/tests/worker_shutdown_acceptance.rs new file mode 100644 index 0000000..116aae4 --- /dev/null +++ b/tests/worker_shutdown_acceptance.rs @@ -0,0 +1,47 @@ +//! Worker-pool teardown regression: dropping an `EditorState` must +//! release its worker threads even though the `Rc` is +//! trapped in Lua-VM reference cycles and never reaches refcount +//! zero (`EditorState::drop` → `AsyncRuntime::shutdown_workers`). +//! +//! Before the fix, every `EditorState` ever constructed leaked a +//! full `cores - 1` worker pool: the m4 acceptance suite (54 editor- +//! building tests) accumulated 1000+ live threads, each waking every +//! 100ms. + +use pmacs::editor::EditorState; + +/// Thread-count probe via /proc; Linux-only (macOS CI skips --- the +/// leak and the fix are platform-independent, the *probe* isn't). +#[cfg(target_os = "linux")] +fn live_threads() -> usize { + std::fs::read_dir("/proc/self/task").map_or(0, std::iter::Iterator::count) +} + +#[cfg(target_os = "linux")] +#[test] +fn editor_state_drop_releases_worker_threads() { + let baseline = live_threads(); + for _ in 0..3 { + let s = EditorState::new(); + drop(s); + } + // Joins are synchronous in drop; the small sleep only covers + // detached per-process reaper threads finishing up. + std::thread::sleep(std::time::Duration::from_millis(300)); + let after = live_threads(); + assert!( + after <= baseline + 2, + "worker threads leak across EditorState drop: \ + baseline {baseline}, after 3 create/drop cycles {after}" + ); +} + +/// Platform-independent variant: the pool reports itself dead after +/// an explicit shutdown, and shutdown is idempotent. +#[test] +fn explicit_shutdown_is_idempotent() { + let s = EditorState::new(); + s.async_runtime.shutdown_workers(); + s.async_runtime.shutdown_workers(); // second call must not hang or panic + drop(s); // drop runs shutdown a third time +}