From 13dcfe7eee3c262137e2f64354af35081ca417eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 09:03:27 +0200 Subject: [PATCH] A long-occupancy job class for the blocking pool --- DESIGN.md | 8 +- crates/turnloop-contract/src/extended.rs | 306 +++++++++++ crates/turnloop-contract/src/lib.rs | 8 + crates/turnloop-contract/tests/windows.rs | 2 + crates/turnloop/src/blocking.rs | 623 ++++++++++++++++++++-- crates/turnloop/src/driver.rs | 60 +++ crates/turnloop/src/lib.rs | 8 +- 7 files changed, 971 insertions(+), 44 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index 9105cb7..0162cd0 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -205,6 +205,12 @@ There are two ways a host drives a loop: - A bounded pool: default 4 threads, configurable, lazily started, shared per process or per loop (config). - It runs file operations, DNS resolution and host-submitted `FnOnce() -> BlockingResult + Send` jobs, and completes through the notifier. - Cancellation is best-effort: a job that has started runs to the end and completes as `Cancelled` if its cancel won the race. +- **Two occupancy classes, declared by the job rather than by the pool** (`Loop::blocking_with`, issue #42). The bounded set above is right for work whose cost is a computation — hashing, compression, key derivation, a file request — and wrong for work whose cost is a *connection*: an accept loop, a per-connection protocol runtime, a synchronous client driven from its own thread. N of those saturate an N-thread pool and every later short job is refused. + - `Occupancy::Bounded` is the default and is exactly the behaviour above: the fixed worker set, the shared queue, `ResourceLimit` when that queue is full. + - `Occupancy::Long` is served by a separate worker set that starts threads on demand up to `PoolConfig::long_threads_max` (default 512) and retires them again after `PoolConfig::long_idle_timeout` (default 10 s). **The class has no queue**: a submission is given a thread immediately or refused with `ResourceLimit`, so an accepted long job never waits behind another long job — a long job that waits on a queued long job's progress would otherwise deadlock. + - The classes share the process, the completion port and the loop's operation table, and **no thread, queue slot or reservation**. Saturating either therefore refuses and delays nothing in the other, in both directions. + - Every job submitted through `blocking_with` receives a `Cancellation`, true once the job was cancelled or its loop was dropped. Nothing interrupts a running worker on any platform, so a long job that never observes it never stops; that is the price of hosting one, and it is the same price tokio's blocking pool charges. +- **`turnloop::pool_stats()`** reports a deliberately stale, lock-free `PoolStats` — bounded `threads`/`busy`/`queued`/`reserved` and `long_threads`/`long_busy` (issue #44). It answers "should I shed load or run this inline?" before a submission is built, rather than after `ResourceLimit` has already been returned. It is not a consistent snapshot and no invariant may be built on it; the result of a submission remains the exact answer. - File I/O goes through the pool on every native platform in 0.x. io_uring and IOCP file I/O are later optimisations behind the same API. - **Typed requests** (`Loop::fs`: open, read, write, metadata, directories, namespace changes) use this one shared pool on Linux, macOS/BSD **and Windows**. Requests on one handle run FIFO; only the head is ever queued on the pool, and each accepted request reserves its queue slot, so a full queue rejects before acceptance and a queued successor always starts. - **Adopted descriptors** (stdio, `Detached::from_fd`/`from_handle`) keep their stream paths: reusable pool jobs on Unix, and one synchronous worker per handle on Windows. Such a handle may be a pipe or console whose read blocks indefinitely and cannot be cancelled from a shared thread; it must not occupy the pool. This is the only per-handle file worker. @@ -538,7 +544,7 @@ No other Perry change is needed for P0. - Today's `spawn_async` / `spawn_blocking` / `run_pending` assume an ambient tokio `Handle` (`perry-ffi/src/async_runtime.rs:383, 431, 469, 82`), and ext crates call `Handle::current().block_on`. - **v2** is token-based on turnloop: - `spawn_async` runs on the calling thread's loop executor - - `spawn_blocking` goes to the shared pool + - `spawn_blocking` goes to the shared pool, at `Occupancy::Bounded` for a computation and `Occupancy::Long` for a caller that holds its thread for a connection's lifetime (the `node:http2` accept loop, the HTTP/2 client and request runtimes, and every database binding running `Handle::current().block_on`, until §5b replaces them) - `run_pending` becomes a bounded `turn` - **v1 signatures** are kept as shims over v2 where the semantics carry over. Ext crates that call `Handle::current().block_on` are rewritten onto the protocol crates (§5b); there is no tokio fallback. diff --git a/crates/turnloop-contract/src/extended.rs b/crates/turnloop-contract/src/extended.rs index 02fca73..1b5d336 100644 --- a/crates/turnloop-contract/src/extended.rs +++ b/crates/turnloop-contract/src/extended.rs @@ -1,4 +1,8 @@ use super::*; +use std::sync::{ + Arc, Condvar, Mutex, + atomic::{AtomicBool, AtomicUsize, Ordering}, +}; pub fn detach_inflight() { let mut source = Driver::::new(Config::default()).expect("loop"); let (_, a, b) = pair(&mut source); @@ -299,3 +303,305 @@ pub fn handoff_distribution() { } assert_eq!(accepted, CONNECTIONS); } + +/// A shared release point for fixture jobs, so a held job parks instead of +/// polling: the tests below hold more threads than a machine has cores. +struct Gate { + open: Mutex, + changed: Condvar, +} +impl Gate { + fn new() -> Arc { + Arc::new(Self { + open: Mutex::new(false), + changed: Condvar::new(), + }) + } + fn wait(&self) { + let mut open = self.open.lock().expect("gate"); + while !*open { + open = self.changed.wait(open).expect("gate"); + } + } + fn release(&self) { + *self.open.lock().expect("gate") = true; + self.changed.notify_all(); + } + fn released(&self) -> bool { + *self.open.lock().expect("gate") + } +} +#[track_caller] +fn until(what: &str, mut ready: impl FnMut() -> bool) { + let deadline = Instant::now() + Duration::from_secs(30); + while !ready() { + assert!(Instant::now() < deadline, "{what}"); + thread::yield_now(); + } +} +/// Collect exactly one completion for `op`, then prove nothing follows it. +#[track_caller] +fn settle(l: &mut Driver, op: OpId, out: &mut Completions) -> OpResult { + let until = l.now() + Duration::from_secs(30); + let mut result = None; + while result.is_none() { + assert!(l.now() < until, "the job never completed"); + l.turn(Timeout::Until(until), out).expect("turn"); + for c in out.drain() { + assert_eq!(c.op, Some(op), "an unexpected completion arrived"); + assert!(c.terminal); + assert!(result.replace(c.result).is_none(), "delivered twice"); + } + } + for _ in 0..5 { + l.turn(Timeout::Now, out).expect("drain"); + assert!(out.is_empty(), "a second completion followed the first"); + } + result.expect("settled") +} +/// Neither occupancy class can exhaust the other (issue #42). +/// +/// Both halves turn on an **ordering** assertion rather than a deadline: the +/// starved class's job must complete while the saturating class is still held. +/// A slow machine cannot satisfy that by accident, and a pool with one class +/// cannot satisfy it at all — 32 connection-lifetime jobs on a four-thread pool +/// is precisely the deadlock this class exists to remove. +pub fn occupancy_classes_do_not_starve_each_other() { + let pool = Config::default().blocking_pool; + let mut l = Driver::::new(Config::default()).expect("loop"); + let mut out = Completions::default(); + // A quiet baseline: another test's worker may still have been retiring. + until("the shared pool is quiet", || { + let s = pool_stats(); + s.busy == 0 && s.queued == 0 && s.reserved == 0 + }); + + // Direction 1: a saturated long class refuses and delays no bounded work. + let gate = Gate::new(); + let held = pool.threads * 8; + let entered = Arc::new(AtomicUsize::new(0)); + let long: Vec = (0..held) + .map(|i| { + let (gate, entered) = (gate.clone(), entered.clone()); + l.blocking_with( + move |_| { + entered.fetch_add(1, Ordering::Release); + gate.wait(); + Ok(Payload::U64(i as u64)) + }, + Occupancy::Long, + Token(1000 + i as u64), + ) + .expect("a long job is accepted past the bounded set's size") + }) + .collect(); + // Every one is *running*: the class hands out threads rather than queueing, + // so this cannot be reached unless `held` threads exist at once. + until("every long job started", || { + entered.load(Ordering::Acquire) == held + }); + let stats = pool_stats(); + // Simultaneously busy is the load-bearing one: it can only be reached with + // `held` distinct threads, so it is what says the class really grew. + assert_eq!(stats.long_busy, held, "each holds a long worker of its own"); + assert!( + stats.long_threads >= held, + "at least that many long workers" + ); + assert_eq!(stats.busy, 0, "no long job occupies a bounded worker"); + assert_eq!(stats.queued, 0, "no long job entered the bounded queue"); + assert_eq!(stats.reserved, 0, "no long job took a queue reservation"); + assert_eq!(stats.threads, pool.threads, "the bounded set is intact"); + + let short = l + .blocking(|| Ok(Payload::U64(7)), Token(1)) + .expect("bounded work is still accepted"); + assert!( + matches!( + settle(&mut l, short, &mut out), + OpResult::Blocking(Payload::U64(7)) + ), + "a bounded job ran while eight times the bounded set was held long" + ); + assert!( + !gate.released(), + "it finished before anything released the long jobs" + ); + + gate.release(); + let mut settled = 0; + let deadline = l.now() + Duration::from_secs(30); + while settled < held { + assert!(l.now() < deadline, "released long jobs never completed"); + l.turn(Timeout::Until(deadline), &mut out).expect("turn"); + for c in out.drain() { + assert!(long.contains(&c.op.expect("job completion"))); + assert!(matches!(c.result, OpResult::Blocking(Payload::U64(_)))); + settled += 1; + } + } + assert_eq!( + settled, held, + "every accepted long job settled exactly once" + ); + + // Direction 2: a saturated bounded class refuses and delays no long work. + let gate = Gate::new(); + let running = Arc::new(AtomicUsize::new(0)); + let holders: Vec = (0..pool.threads) + .map(|_| { + let (gate, running) = (gate.clone(), running.clone()); + l.blocking( + move || { + running.fetch_add(1, Ordering::Release); + gate.wait(); + Ok(Payload::U64(0)) + }, + Token(2), + ) + .expect("bounded holder") + }) + .collect(); + until("every bounded worker is occupied", || { + running.load(Ordering::Acquire) == pool.threads + }); + let mut queued = Vec::new(); + let refused = loop { + match l.blocking(|| Ok(Payload::U64(0)), Token(3)) { + Ok(op) => queued.push(op), + Err(e) => break e, + } + assert!(queued.len() <= pool.queue_capacity, "queue bound enforced"); + }; + assert_eq!(refused.kind, ErrorKind::ResourceLimit); + assert_eq!(queued.len(), pool.queue_capacity, "the queue is full"); + assert_eq!(pool_stats().queued, pool.queue_capacity); + + let long = l + .blocking_with(|_| Ok(Payload::U64(9)), Occupancy::Long, Token(4)) + .expect("a long job is accepted while the bounded class refuses"); + assert!(matches!( + settle(&mut l, long, &mut out), + OpResult::Blocking(Payload::U64(9)) + )); + assert!( + l.blocking(|| Ok(Payload::U64(0)), Token(3)).is_err(), + "the bounded class was still saturated when the long job finished" + ); + assert!(!gate.released(), "nothing released the bounded holders"); + + gate.release(); + let mut settled = 0; + let expected = holders.len() + queued.len(); + let deadline = l.now() + Duration::from_secs(60); + while settled < expected { + assert!(l.now() < deadline, "released bounded jobs never completed"); + l.turn(Timeout::Until(deadline), &mut out).expect("turn"); + for c in out.drain() { + assert!(matches!(c.result, OpResult::Blocking(Payload::U64(0)))); + settled += 1; + } + } + assert_eq!(settled, expected, "every accepted bounded job settled once"); + assert!(!l.alive()); +} +/// A long job delivers exactly one completion on every path that is not its own +/// return: cancellation, a panic, and the loop going away underneath it. +pub fn long_jobs_settle_once_on_cancel_panic_and_shutdown() { + let mut l = Driver::::new(Config::default()).expect("loop"); + let mut out = Completions::default(); + + // A running long job stops because it was asked to, not because it was + // interrupted: nothing can take a thread back from a job that never looks. + let started = Arc::new(AtomicBool::new(false)); + let observed = Arc::new(AtomicBool::new(false)); + let (running, stopped) = (started.clone(), observed.clone()); + let op = l + .blocking_with( + move |stop| { + running.store(true, Ordering::Release); + while !stop.requested() { + thread::park_timeout(Duration::from_millis(1)); + } + stopped.store(true, Ordering::Release); + Ok(Payload::U64(1)) + }, + Occupancy::Long, + Token(1), + ) + .expect("long job"); + until("the long job is running", || { + started.load(Ordering::Acquire) + }); + assert!(l.cancel(op)); + assert!(!l.cancel(op), "a second cancel finds it already cancelled"); + assert!(matches!(settle(&mut l, op, &mut out), OpResult::Cancelled)); + assert!( + observed.load(Ordering::Acquire), + "the job returned of its own accord after seeing the request" + ); + + // A panicking long job is reported to its host and leaves the class usable. + // The hook is silenced so an expected panic does not read as a CI failure. + let hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + let op = l + .blocking_with( + |_| panic!("a long job may panic"), + Occupancy::Long, + Token(2), + ) + .expect("long job"); + let result = settle(&mut l, op, &mut out); + std::panic::set_hook(hook); + assert!( + matches!( + result, + OpResult::Err(Error { + kind: ErrorKind::Other, + .. + }) + ), + "unexpected {result:?}" + ); + let op = l + .blocking_with(|_| Ok(Payload::U64(3)), Occupancy::Long, Token(3)) + .expect("the long class still serves work"); + assert!(matches!( + settle(&mut l, op, &mut out), + OpResult::Blocking(Payload::U64(3)) + )); + + // A dropped loop asks its outstanding long jobs to stop. Without that, a + // connection-lifetime job would hold its thread for the process's life and + // wait forever for a host that no longer exists. + let started = Arc::new(AtomicBool::new(false)); + let finished = Arc::new(AtomicBool::new(false)); + let (running, exited) = (started.clone(), finished.clone()); + let mut doomed = Driver::::new(Config::default()).expect("second loop"); + doomed + .blocking_with( + move |stop| { + running.store(true, Ordering::Release); + while !stop.requested() { + thread::park_timeout(Duration::from_millis(1)); + } + exited.store(true, Ordering::Release); + Ok(Payload::U64(4)) + }, + Occupancy::Long, + Token(4), + ) + .expect("long job"); + until("the long job is running", || { + started.load(Ordering::Acquire) + }); + assert!(doomed.alive(), "an accepted job keeps its loop alive"); + drop(doomed); + until("shutdown released the long worker", || { + finished.load(Ordering::Acquire) + }); + until("the worker rejoined the idle set", || { + pool_stats().long_busy == 0 + }); +} diff --git a/crates/turnloop-contract/src/lib.rs b/crates/turnloop-contract/src/lib.rs index cb03fa2..0c29ce6 100644 --- a/crates/turnloop-contract/src/lib.rs +++ b/crates/turnloop-contract/src/lib.rs @@ -259,6 +259,14 @@ mod native { pool_and_dns::(); } #[test] + fn occupancy_classes_are_independent() { + occupancy_classes_do_not_starve_each_other::(); + } + #[test] + fn long_job_delivery() { + long_jobs_settle_once_on_cancel_panic_and_shutdown::(); + } + #[test] fn reuse_port_distribution() { reuse_port::(); } diff --git a/crates/turnloop-contract/tests/windows.rs b/crates/turnloop-contract/tests/windows.rs index 384a0de..51693a8 100644 --- a/crates/turnloop-contract/tests/windows.rs +++ b/crates/turnloop-contract/tests/windows.rs @@ -22,6 +22,8 @@ contract!( udp_round_trip, detach_inflight, pool_and_dns, + occupancy_classes_do_not_starve_each_other, + long_jobs_settle_once_on_cancel_panic_and_shutdown, handoff_distribution, writev_and_shutdown, capacity_and_stale_ids, diff --git a/crates/turnloop/src/blocking.rs b/crates/turnloop/src/blocking.rs index a995186..40f9bc1 100644 --- a/crates/turnloop/src/blocking.rs +++ b/crates/turnloop/src/blocking.rs @@ -1,4 +1,17 @@ -//! Lazily started process-wide bounded pool, with a completion port per loop. +//! Lazily started process-wide pool, with a completion port per loop. +//! +//! The pool serves two job classes that share a process and nothing else +//! (DESIGN D8). [`Occupancy::Bounded`] work is short and CPU-bound: it is +//! counted against the fixed worker set and its shared queue, and a full queue +//! is backpressure. [`Occupancy::Long`] work holds its thread for as long as a +//! connection, stream or nested runtime lives: it is served by a separately +//! accounted worker set that grows on demand to a ceiling. +//! +//! Neither class can exhaust the other, because they share no thread, no queue +//! and no reservation. A long job never occupies a bounded worker, so a host +//! may hold more threads than the bounded set has without refusing a single +//! short job; a bounded backlog never delays a long job, so a connection is +//! accepted while a hash queue is full. use crate::{BlockingResult, Error, ErrorKind, Notifier, OpId, Payload, Result, queue::Queue}; #[cfg(not(target_arch = "wasm32"))] use std::sync::atomic::AtomicUsize; @@ -8,24 +21,153 @@ use std::{ Arc, atomic::{AtomicBool, Ordering}, }, + time::Duration, }; #[derive(Clone, Copy, Debug, Eq, PartialEq)] /// Configuration fixed by the first process-wide blocking-pool submission. pub struct PoolConfig { - /// Number of lazily started process-wide blocking workers. + /// Number of lazily started process-wide blocking workers serving + /// [`Occupancy::Bounded`] jobs. Started as one set on first use, whichever + /// class asked for it. pub threads: usize, - /// Maximum queued blocking jobs before ResourceLimit is returned. + /// Maximum queued bounded jobs before ResourceLimit is returned. Long jobs + /// never enter this queue and never consume its reservations. pub queue_capacity: usize, + /// Ceiling on [`Occupancy::Long`] workers, which are started on demand and + /// are not preallocated. A long submission that would need worker number + /// `long_threads_max + 1` is refused with `ResourceLimit`; it is never + /// queued behind another long job, so an accepted long job always has a + /// thread of its own from the moment it is accepted. + /// + /// Matching tokio's `max_blocking_threads`, the default is 512. A host that + /// hosts one connection per thread should raise it to its own connection + /// ceiling and size [`Config::max_operations`](crate::Config::max_operations) + /// to match, or lower it to place a hard bound on the process. + pub long_threads_max: usize, + /// How long an idle long worker waits for more long work before exiting. + /// + /// The long set shrinks back to zero threads after a burst, so a host pays + /// for peak long occupancy only while it lasts. Reuse within the window + /// costs no thread creation. + pub long_idle_timeout: Duration, } impl Default for PoolConfig { fn default() -> Self { Self { threads: 4, queue_capacity: 1024, + long_threads_max: 512, + long_idle_timeout: Duration::from_secs(10), } } } +#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)] +/// How long a submitted job will hold the worker thread that runs it. +/// +/// The class is a property of the job, not of the pool: one process-wide +/// configuration serves both, and a host does not have to prove a global thread +/// budget before it can host work of either shape. +pub enum Occupancy { + /// Short CPU-bound work: hashing, compression, key derivation, a file + /// request, a name lookup. + /// + /// Runs on the fixed worker set ([`PoolConfig::threads`]), queued behind + /// other bounded work when every worker is busy. A full queue + /// ([`PoolConfig::queue_capacity`]) refuses the submission with + /// [`ErrorKind::ResourceLimit`] rather than growing. + /// + /// This is the default and the only class before this API existed; + /// [`Driver::blocking`](crate::Driver::blocking) submits it. + #[default] + Bounded, + /// Work that holds its thread for as long as a connection, stream or nested + /// runtime lives, rather than for as long as a computation takes: an accept + /// loop, a per-connection protocol runtime, a synchronous client library + /// driven from its own thread. + /// + /// Served by a separate worker set that starts threads on demand up to + /// [`PoolConfig::long_threads_max`] and retires them again after + /// [`PoolConfig::long_idle_timeout`]. **The class has no queue:** a long + /// submission is either given a thread immediately or refused with + /// [`ErrorKind::ResourceLimit`], so an accepted long job never waits behind + /// another long job and can never deadlock on one. + /// + /// A long job should observe [`Cancellation::requested`] and return when it + /// is set; nothing can take a thread back from a job that does not. + Long, +} +#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)] +/// A snapshot of the process-wide pool, as it looked at some point during the +/// call (issue #44). +/// +/// Deliberately **not** a consistent view: each field is read separately and +/// without taking the pool's lock, so the fields may belong to slightly +/// different instants and any of them may be stale the moment it is returned. +/// That is the useful trade for the question this answers — "is the pool near +/// saturation, should I shed load or run this inline?" — and a consistent +/// snapshot would cost a lock round-trip on the submitting thread. Do not build +/// an invariant on it; use the result of [`Driver::blocking`](crate::Driver::blocking) +/// for that, which is exact. +/// +/// Every field is zero before the first submission starts the pool. +pub struct PoolStats { + /// Bounded workers, once started: [`PoolConfig::threads`]. + pub threads: usize, + /// Bounded workers currently running a job. + pub busy: usize, + /// Bounded jobs waiting for a worker. + pub queued: usize, + /// Queue slots promised to accepted operations that have not queued their + /// work yet (typed filesystem requests). `queued + reserved` is what is + /// measured against [`PoolConfig::queue_capacity`]. + pub reserved: usize, + /// Live long workers, including idle ones that have not retired yet. + pub long_threads: usize, + /// Long workers currently running a job. Equal to the number of accepted + /// long jobs that have not finished, because the class never queues. + pub long_busy: usize, +} +/// Read the process-wide pool's [`PoolStats`]. Never blocks and never fails; +/// reports zeros before the first submission has started the pool. +pub fn pool_stats() -> PoolStats { + #[cfg(not(target_arch = "wasm32"))] + { + native::stats() + } + #[cfg(target_arch = "wasm32")] + { + PoolStats::default() + } +} +/// The cooperative stop signal of a running pool job. +/// +/// Handed to every closure submitted through +/// [`Driver::blocking_with`](crate::Driver::blocking_with). Cancellation of a +/// job that has already started is cooperative on every platform: nothing +/// interrupts a running worker, so a job that never looks at this never stops. +pub struct Cancellation { + cancel: Arc, + port: Arc, +} +impl Cancellation { + pub(crate) fn new(cancel: Arc, port: Arc) -> Self { + Self { cancel, port } + } + /// Whether the host has asked this job to stop. + /// + /// True once [`Driver::cancel`](crate::Driver::cancel) was called for this + /// job, or the loop that owns it was dropped. Returning promptly afterwards + /// is what makes a long job's thread reusable and a host's shutdown finite. + /// + /// Delivery is unchanged by observing it: a job that returns after a cancel + /// still delivers exactly one `Cancelled` completion, and one whose loop is + /// gone delivers nothing to anybody, because there is no longer a host to + /// deliver to. + pub fn requested(&self) -> bool { + self.cancel.load(Ordering::Acquire) || self.port.is_closed() + } +} #[derive(Debug)] /// Owned hostname and port for a pool-backed native address lookup. pub struct DnsRequest { @@ -67,6 +209,11 @@ impl WorkPort { pub fn close(&self) { self.closed.store(true, Ordering::Release); } + /// Whether the owning loop is gone, so a result pushed here is discarded. + /// A running job reads this through [`Cancellation::requested`]. + pub(crate) fn is_closed(&self) -> bool { + self.closed.load(Ordering::Acquire) + } // The owning agent is already collecting this result. Do not leave a stale // notification that would turn its next future deadline into a Now poll. #[cfg(target_arch = "wasm32")] @@ -104,7 +251,7 @@ mod native { sync::{Condvar, Mutex, OnceLock}, thread, }; - struct Job { + pub(super) struct Job { op: OpId, cancel: Arc, port: Arc, @@ -114,11 +261,45 @@ mod native { Boxed(Job), Reusable(Arc), } + /// Counts a worker as busy for as long as it holds a task, restoring the + /// count on an unwind that escapes the task itself. + struct Busy<'a>(&'a AtomicUsize); + impl Busy<'_> { + fn start(counter: &AtomicUsize) -> Busy<'_> { + counter.fetch_add(1, Ordering::Relaxed); + Busy(counter) + } + } + impl Drop for Busy<'_> { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::Relaxed); + } + } + /// Run one job to its single completion, whichever class it belongs to. + /// + /// A cancel that arrived before the job started replaces the work; a panic + /// inside it is caught and reported. Either way the port receives exactly + /// one result for this operation. + fn run(job: Job, busy: &AtomicUsize) { + let _busy = Busy::start(busy); + let result = if job.cancel.load(Ordering::Acquire) { + Err(Error::new(ErrorKind::Cancelled)) + } else { + catch_unwind(AssertUnwindSafe(job.f)) + .unwrap_or_else(|_| Err(Error::new(ErrorKind::Other))) + }; + job.port.complete(WorkResult { op: job.op, result }); + } struct State { jobs: Mutex>, // Queue capacity promised to accepted operations that will push later. // Modified only while holding `jobs`: queued + reserved <= queue_capacity. reserved: AtomicUsize, + // Lock-free mirrors of the queue length and of the workers running a + // task, for PoolStats. Written under `jobs` (queued) or by the worker + // that owns the task (busy); read without the lock and allowed to be stale. + queued: AtomicUsize, + busy: AtomicUsize, ready: Condvar, stopping: AtomicBool, /// Workers that finished thread startup; `start` returns only once all have. @@ -127,6 +308,7 @@ mod native { } struct Pool { state: Arc, + long: Arc, config: PoolConfig, } static POOL: OnceLock> = OnceLock::new(); @@ -134,6 +316,8 @@ mod native { let state = Arc::new(State { jobs: Mutex::new(VecDeque::with_capacity(config.queue_capacity)), reserved: AtomicUsize::new(0), + queued: AtomicUsize::new(0), + busy: AtomicUsize::new(0), ready: Condvar::new(), stopping: AtomicBool::new(false), started: Mutex::new(0), @@ -166,22 +350,17 @@ mod native { if s.stopping.load(Ordering::Acquire) { return; } - jobs.pop_front().expect("nonempty job queue") + let task = jobs.pop_front().expect("nonempty job queue"); + s.queued.store(jobs.len(), Ordering::Relaxed); + task }; - let job = match job { - Task::Boxed(job) => job, + match job { + Task::Boxed(job) => run(job, &s.busy), Task::Reusable(work) => { + let _busy = Busy::start(&s.busy); work.run(); - continue; } - }; - let result = if job.cancel.load(Ordering::Acquire) { - Err(Error::new(ErrorKind::Cancelled)) - } else { - catch_unwind(AssertUnwindSafe(job.f)) - .unwrap_or_else(|_| Err(Error::new(ErrorKind::Other))) - }; - job.port.complete(WorkResult { op: job.op, result }); + } } }); match worker { @@ -211,42 +390,60 @@ mod native { } drop(started); // Workers live for the process lifetime, shared by every submitting loop. - Ok(Pool { state, config }) + Ok(Pool { + state, + long: LongSet::new(config.long_threads_max, config.long_idle_timeout), + config, + }) } pub(super) fn submit( config: PoolConfig, + occupancy: Occupancy, op: OpId, cancel: Arc, port: Arc, f: Box Result + Send>, ) -> Result<()> { - if config.threads == 0 || config.queue_capacity == 0 { - return Err(Error::new(ErrorKind::InvalidInput)); - } - let pool = POOL - .get_or_init(|| start(config)) - .as_ref() - .map_err(|&e| e)?; - if pool.config != config { - return Err(Error::new(ErrorKind::InvalidInput)); - } - let mut jobs = pool - .state - .jobs - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if jobs.len() + pool.state.reserved.load(Ordering::Relaxed) >= pool.config.queue_capacity { - return Err(Error::new(ErrorKind::ResourceLimit)); - } - jobs.push_back(Task::Boxed(Job { + let pool = pool(config)?; + let job = Job { op, cancel, port, f, - })); - drop(jobs); - pool.state.ready.notify_one(); - Ok(()) + }; + match occupancy { + // A long job takes nothing from the bounded class: not a worker, not + // a queue slot, not a reservation. Saturating one therefore cannot + // refuse or delay the other, in either direction. + Occupancy::Long => pool.long.submit(job), + Occupancy::Bounded => { + let mut jobs = lock(pool); + if jobs.len() + pool.state.reserved.load(Ordering::Relaxed) + >= pool.config.queue_capacity + { + return Err(Error::new(ErrorKind::ResourceLimit)); + } + jobs.push_back(Task::Boxed(job)); + pool.state.queued.store(jobs.len(), Ordering::Relaxed); + drop(jobs); + pool.state.ready.notify_one(); + Ok(()) + } + } + } + pub(super) fn stats() -> PoolStats { + let Some(Ok(pool)) = POOL.get().map(|p| p.as_ref()) else { + return PoolStats::default(); + }; + let (long_threads, long_busy) = pool.long.stats(); + PoolStats { + threads: pool.config.threads, + busy: pool.state.busy.load(Ordering::Relaxed), + queued: pool.state.queued.load(Ordering::Relaxed), + reserved: pool.state.reserved.load(Ordering::Relaxed), + long_threads, + long_busy, + } } fn pool(config: PoolConfig) -> Result<&'static Pool> { if config.threads == 0 || config.queue_capacity == 0 { @@ -275,6 +472,7 @@ mod native { return Err(Error::new(ErrorKind::ResourceLimit)); } jobs.push_back(Task::Reusable(work)); + pool.state.queued.store(jobs.len(), Ordering::Relaxed); pool.state.ready.notify_one(); Ok(()) } @@ -311,12 +509,343 @@ mod native { pool.state.reserved.store(reserved - 1, Ordering::Relaxed); debug_assert!(jobs.len() < pool.config.queue_capacity); jobs.push_back(Task::Reusable(work)); + pool.state.queued.store(jobs.len(), Ordering::Relaxed); drop(jobs); pool.state.ready.notify_one(); } + /// The [`Occupancy::Long`] worker set: no queue, threads on demand. + /// + /// A long job holds its thread for as long as a connection lives, so the + /// two rules here are that a long job must never wait for another long job + /// (that is a deadlock whenever the first waits on the second's progress), + /// and that the set must still be bounded. Both follow from the same + /// decision: **the class has no backlog.** A submission is handed to a + /// parked worker, or starts one of its own, or is refused; there is no + /// third outcome and nothing to queue behind. + pub(super) struct LongSet { + handoff: Mutex, + ready: Condvar, + /// Lock-free mirror of `Long::threads`, written under `handoff`, and the + /// count of workers running a job. Both are read without the lock for + /// PoolStats and are allowed to be stale. + live: AtomicUsize, + busy: AtomicUsize, + max: usize, + idle_timeout: Duration, + } + struct Long { + /// Jobs given to a specific parked worker, never a backlog: a job is + /// pushed only while `jobs.len() < idle` holds, so every pushed job has + /// a parked worker that no other pushed job has been promised to. + jobs: VecDeque, + /// Live workers, counted from the moment one is decided on rather than + /// from the moment its thread starts, so two submissions racing at the + /// ceiling cannot both pass it. + threads: usize, + /// Parked workers not yet promised a job. + idle: usize, + /// Monotone worker ordinal, for thread names. + started: usize, + } + impl LongSet { + pub(super) fn new(max: usize, idle_timeout: Duration) -> Arc { + Arc::new(Self { + handoff: Mutex::new(Long { + jobs: VecDeque::new(), + threads: 0, + idle: 0, + started: 0, + }), + ready: Condvar::new(), + live: AtomicUsize::new(0), + busy: AtomicUsize::new(0), + max, + idle_timeout, + }) + } + fn lock(&self) -> std::sync::MutexGuard<'_, Long> { + self.handoff + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + /// Start `job` now on a long worker, or refuse it. + /// + /// Refusal is `ResourceLimit` at the ceiling, and the spawn's own error + /// if the OS cannot start a thread. Both refuse the submission before + /// it becomes a job, so the caller owes no completion either way. + pub(super) fn submit(self: &Arc, job: Job) -> Result<()> { + let mut long = self.lock(); + if long.jobs.len() < long.idle { + long.jobs.push_back(job); + drop(long); + self.ready.notify_one(); + return Ok(()); + } + if long.threads >= self.max { + return Err(Error::new(ErrorKind::ResourceLimit)); + } + long.threads += 1; + self.live.store(long.threads, Ordering::Relaxed); + let ordinal = long.started; + long.started += 1; + drop(long); + let set = self.clone(); + // The job is the new worker's first task rather than a queue entry: + // a failed spawn then drops it here, with no window in which another + // worker could have taken work this submission is about to disown. + match thread::Builder::new() + .name(format!("turnloop-long-{ordinal}")) + .spawn(move || set.worker(job)) + { + Ok(_detached) => Ok(()), + Err(e) => { + let mut long = self.lock(); + long.threads -= 1; + self.live.store(long.threads, Ordering::Relaxed); + Err(e.into()) + } + } + } + /// Run `first`, then serve handed-off jobs until the idle timeout. + fn worker(self: Arc, first: Job) { + // Retiring under the same lock that admits work is what keeps the + // ceiling honest through a panic that escapes a job's own guard. + struct Retire<'a>(&'a LongSet); + impl Drop for Retire<'_> { + fn drop(&mut self) { + let mut long = self.0.lock(); + long.threads -= 1; + self.0.live.store(long.threads, Ordering::Relaxed); + } + } + let _retire = Retire(&self); + let mut next = Some(first); + while let Some(job) = next.take() { + run(job, &self.busy); + let mut long = self.lock(); + long.idle += 1; + loop { + if let Some(job) = long.jobs.pop_front() { + long.idle -= 1; + next = Some(job); + break; + } + let (guard, wait) = self + .ready + .wait_timeout(long, self.idle_timeout) + .unwrap_or_else(std::sync::PoisonError::into_inner); + long = guard; + // A job that arrived during the wait is taken even if the + // wait also timed out: it was promised to this worker. + if wait.timed_out() && long.jobs.is_empty() { + long.idle -= 1; + // Explicitly, because `Retire` takes this same lock: + // scope-exit order already releases it first, and this + // keeps that true through any later edit. + drop(long); + return; + } + } + } + } + /// Live and busy long workers. Read without the lock, so the two may be + /// a moment apart; `live` never lags a submission that returned `Ok`. + pub(super) fn stats(&self) -> (usize, usize) { + ( + self.live.load(Ordering::Relaxed), + self.busy.load(Ordering::Relaxed), + ) + } + /// Parked workers, read under the lock. Tests use it to reach a state + /// the stale counters cannot describe: a worker that is not merely done + /// with its job but actually available to be handed the next one. + #[cfg(all(test, not(loom)))] + fn idle(&self) -> usize { + self.lock().idle + } + } + #[cfg(all(test, not(loom)))] + mod long_tests { + use super::*; + use std::{ + sync::mpsc, + time::{Duration, Instant}, + }; + struct NoWake; + impl crate::backend::Wake for NoWake { + fn wake(&self) -> Result<()> { + Ok(()) + } + fn syscall_count(&self) -> u64 { + 0 + } + } + fn port() -> Arc { + WorkPort::new(64, Notifier::new(std::sync::Arc::new(NoWake))) + } + fn job( + port: &Arc, + key: u64, + cancel: Arc, + f: impl FnOnce() -> Result + Send + 'static, + ) -> Job { + Job { + op: OpId { owner: 1, key }, + cancel, + port: port.clone(), + f: Box::new(f), + } + } + fn live() -> Arc { + Arc::new(AtomicBool::new(false)) + } + #[track_caller] + fn until(what: &str, mut ready: impl FnMut() -> bool) { + let deadline = Instant::now() + Duration::from_secs(10); + while !ready() { + assert!(Instant::now() < deadline, "{what}"); + thread::yield_now(); + } + } + #[track_caller] + fn failure(result: Result) -> ErrorKind { + match result { + Ok(_) => panic!("expected a failed job"), + Err(e) => e.kind, + } + } + /// Poll for the next published result. `is_empty` is not a substitute: + /// a producer that has reserved its slot is already non-empty, and the + /// value is only readable once publication finishes. + #[track_caller] + fn take(port: &Arc) -> WorkResult { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if let Some(result) = port.pop() { + return result; + } + assert!(Instant::now() < deadline, "a result arrived"); + thread::yield_now(); + } + } + /// One worker serves job after job, and the set shrinks back to nothing + /// when the work stops: the ceiling bounds a peak, not a floor. + #[test] + fn long_workers_are_reused_and_retire_when_idle() { + let set = LongSet::new(4, Duration::from_millis(50)); + let port = port(); + set.submit(job(&port, 1, live(), || { + Ok(WorkOutput::Blocking(Payload::U64(1))) + })) + .expect("first long job"); + assert!(matches!( + take(&port).result, + Ok(WorkOutput::Blocking(Payload::U64(1))) + )); + assert_eq!(set.stats().0, 1, "one worker started for one long job"); + until("the worker parked", || set.idle() == 1); + set.submit(job(&port, 2, live(), || { + Ok(WorkOutput::Blocking(Payload::U64(2))) + })) + .expect("second long job"); + assert!(matches!( + take(&port).result, + Ok(WorkOutput::Blocking(Payload::U64(2))) + )); + assert_eq!( + set.stats().0, + 1, + "the parked worker served it instead of a new thread" + ); + until("the idle worker retired", || set.stats() == (0, 0)); + } + /// At the ceiling a long submission is refused, not queued: a long job + /// that waits for another long job's progress would never finish. + #[test] + fn long_submissions_at_the_ceiling_are_refused_rather_than_queued() { + let set = LongSet::new(2, Duration::from_secs(30)); + let port = port(); + let entered = Arc::new(AtomicUsize::new(0)); + let (release, held) = mpsc::channel::<()>(); + let held = Arc::new(Mutex::new(held)); + for key in 0..2 { + let (entered, held) = (entered.clone(), held.clone()); + set.submit(job(&port, key, live(), move || { + entered.fetch_add(1, Ordering::Release); + let _ = held.lock().expect("held").recv(); + Ok(WorkOutput::Blocking(Payload::U64(key))) + })) + .expect("within the ceiling"); + } + until("both long jobs are running", || { + entered.load(Ordering::Acquire) == 2 + }); + assert_eq!(set.stats(), (2, 2), "both workers are occupied"); + let refused = set + .submit(job(&port, 9, live(), || { + unreachable!("a refused submission never runs") + })) + .expect_err("above the ceiling"); + assert_eq!(refused.kind, ErrorKind::ResourceLimit); + assert!( + port.is_empty(), + "a refused submission owes no completion at all" + ); + drop(release); + assert!(matches!(take(&port).result, Ok(WorkOutput::Blocking(_)))); + assert!(matches!(take(&port).result, Ok(WorkOutput::Blocking(_)))); + // The ceiling bounds what runs at once, not what may ever run. + until("a worker is free again", || set.idle() > 0); + set.submit(job(&port, 10, live(), || { + Ok(WorkOutput::Blocking(Payload::U64(10))) + })) + .expect("the released capacity is usable"); + assert!(matches!( + take(&port).result, + Ok(WorkOutput::Blocking(Payload::U64(10))) + )); + } + /// Exactly one result per accepted long job, on both paths that do not + /// return one from the closure. + #[test] + fn a_cancelled_or_panicking_long_job_delivers_exactly_one_result() { + let set = LongSet::new(2, Duration::from_millis(50)); + let port = port(); + let ran = Arc::new(AtomicUsize::new(0)); + let counter = ran.clone(); + let cancelled = Arc::new(AtomicBool::new(true)); + set.submit(job(&port, 1, cancelled, move || { + counter.fetch_add(1, Ordering::Release); + Ok(WorkOutput::Blocking(Payload::U64(1))) + })) + .expect("accepted"); + let result = take(&port); + assert_eq!(result.op.key, 1); + assert_eq!(failure(result.result), ErrorKind::Cancelled); + assert_eq!(ran.load(Ordering::Acquire), 0, "the work never started"); + set.submit(job(&port, 2, live(), || panic!("a job may panic"))) + .expect("accepted"); + let result = take(&port); + assert_eq!(result.op.key, 2); + assert_eq!(failure(result.result), ErrorKind::Other); + // The worker survived the panic and still serves the class. + set.submit(job(&port, 3, live(), || { + Ok(WorkOutput::Blocking(Payload::U64(3))) + })) + .expect("accepted"); + let result = take(&port); + assert_eq!(result.op.key, 3); + assert!(matches!( + result.result, + Ok(WorkOutput::Blocking(Payload::U64(3))) + )); + assert!(port.is_empty(), "three jobs delivered three results"); + } + } } pub(crate) fn submit( config: PoolConfig, + occupancy: Occupancy, op: OpId, cancel: Arc, port: Arc, @@ -324,11 +853,11 @@ pub(crate) fn submit( ) -> Result<()> { #[cfg(not(target_arch = "wasm32"))] { - native::submit(config, op, cancel, port, f) + native::submit(config, occupancy, op, cancel, port, f) } #[cfg(target_arch = "wasm32")] { - let _ = (config, op, cancel, port, f); + let _ = (config, occupancy, op, cancel, port, f); Err(Error::new(ErrorKind::Unsupported)) } } @@ -337,6 +866,16 @@ pub(crate) fn blocking( ) -> Box Result + Send> { Box::new(move || f().map(WorkOutput::Blocking)) } +/// A job that can observe its own cancellation. The signal is captured once, +/// when the job is built, so running it costs exactly what `blocking` costs. +pub(crate) fn blocking_cancellable( + f: impl FnOnce(&Cancellation) -> BlockingResult + Send + 'static, + cancel: Arc, + port: Arc, +) -> Box Result + Send> { + let stop = Cancellation::new(cancel, port); + Box::new(move || f(&stop).map(WorkOutput::Blocking)) +} pub(crate) fn resolve(request: DnsRequest) -> Box Result + Send> { Box::new(move || { // std's native ToSocketAddrs implementation uses getaddrinfo (Unix) or diff --git a/crates/turnloop/src/driver.rs b/crates/turnloop/src/driver.rs index 885536a..2f4f290 100644 --- a/crates/turnloop/src/driver.rs +++ b/crates/turnloop/src/driver.rs @@ -1167,6 +1167,10 @@ impl Driver { self.signal_stop(h, token) } /// Submit an owned Send closure to the bounded shared blocking pool. + /// + /// Equivalent to [`blocking_with`](Self::blocking_with) at + /// [`Occupancy::Bounded`], for work that does not need to observe its own + /// cancellation. pub fn blocking BlockingResult + Send + 'static>( &mut self, f: F, @@ -1174,6 +1178,49 @@ impl Driver { ) -> Result { self.submit_work(crate::blocking::blocking(f), token) } + /// Submit an owned Send closure with an explicit [`Occupancy`] class, and + /// hand it the [`Cancellation`] signal of its own job. + /// + /// [`Occupancy::Bounded`] behaves exactly like [`blocking`](Self::blocking): + /// the fixed worker set, the shared queue, `ResourceLimit` when that queue + /// is full. [`Occupancy::Long`] is for work that holds its thread for as + /// long as a connection lives; it is served by a separate, on-demand worker + /// set, so neither class can exhaust the other. The two share this loop's + /// operation table, as every operation does, and nothing else. + /// + /// Delivery is the same single completion for both classes: `Blocking` with + /// the payload, `Cancelled` if [`cancel`](Self::cancel) won, or `Err` if the + /// job panicked. A refused submission returns here and never completes. + /// + /// ```no_run + /// # use turnloop::*; + /// # fn f(l: &mut Loop, accept_one: impl Fn() -> BlockingResult + Send + 'static) -> Result<()> { + /// // An accept loop holds its thread until the host asks it to stop. + /// l.blocking_with( + /// move |stop| { + /// while !stop.requested() { + /// accept_one()?; + /// } + /// Ok(Payload::U64(0)) + /// }, + /// Occupancy::Long, + /// Token(1), + /// )?; + /// # Ok(()) } + /// ``` + pub fn blocking_with BlockingResult + Send + 'static>( + &mut self, + f: F, + occupancy: Occupancy, + token: Token, + ) -> Result { + let port = self.work_port.clone(); + self.submit_job( + move |cancel| crate::blocking::blocking_cancellable(f, cancel.clone(), port), + occupancy, + token, + ) + } /// Resolve through a host resolver, or the shared native blocking pool. pub fn resolve(&mut self, request: crate::DnsRequest, token: Token) -> Result { let op = self.new_op(None, token)?; @@ -1200,11 +1247,24 @@ impl Driver { f: Box Result + Send>, token: Token, ) -> Result { + self.submit_job(move |_| f, Occupancy::Bounded, token) + } + /// Take an operation slot, then hand the pool a job built around that + /// operation's cancellation flag. A refused submission retires the slot, so + /// a caller that sees an error is owed no completion. + fn submit_job(&mut self, make: M, occupancy: Occupancy, token: Token) -> Result + where + M: FnOnce( + &std::sync::Arc, + ) -> Box Result + Send>, + { let op = self.new_op(None, token)?; let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); self.ops.get_mut(op.key).expect("new op").job_cancel = Some(cancel.clone()); + let f = make(&cancel); if let Err(e) = crate::blocking::submit( self.config.blocking_pool, + occupancy, op, cancel, self.work_port.clone(), diff --git a/crates/turnloop/src/lib.rs b/crates/turnloop/src/lib.rs index 1e571e9..5af95bc 100644 --- a/crates/turnloop/src/lib.rs +++ b/crates/turnloop/src/lib.rs @@ -28,6 +28,12 @@ //! # fn main() {} //! ``` //! +//! Blocking work declares how long it will hold its worker. `Occupancy::Bounded` +//! is the fixed, queued, backpressured set that short CPU-bound jobs belong on; +//! `Occupancy::Long` is a separately accounted, on-demand set for work that holds +//! a thread for a connection's lifetime. Neither class can exhaust the other, and +//! `pool_stats` reports both without taking a lock. +//! //! Enable the `executor` feature for `executor::LocalExecutor` and futures-io //! adapters. Each loop/executor belongs to its creating thread; its notifier and //! poster may be cloned onto other threads. @@ -71,7 +77,7 @@ pub use backend::iocp::Detached; pub use backend::unix::Detached; mod blocking; -pub use blocking::{DnsRequest, PoolConfig}; +pub use blocking::{Cancellation, DnsRequest, Occupancy, PoolConfig, PoolStats, pool_stats}; mod fs; pub use fs::{