diff --git a/DESIGN.md b/DESIGN.md index 0162cd0..1385547 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -552,7 +552,7 @@ No other Perry change is needed for P0. **Hard rules**, enforced by tests and CI benchmarks: -1. **Allocations:** zero heap allocations per read, write, timer or accept after warm-up (checked with a counting allocator in tests). +1. **Allocations:** zero heap allocations per read, write, timer or accept after warm-up (checked with a counting allocator in tests). *Warm-up* includes reaching a high-water mark: `Config::max_handles` and `Config::max_operations` are ceilings whose slot storage is built in pages as the mark rises (§10a), so passing a new mark allocates a page and a loop at its mark allocates nothing. **Contract tests:** a loop built with a 1024x larger ceiling makes the same allocations, and requests the same bytes, as the smaller one; a loop pinned at a high-water mark 500 handles above page zero turns 1000 times without allocating. 2. **Wake:** zero syscalls on `notify()` while the loop is running (checked with a syscall-counting harness: `strace -c` / `ktrace` / ETW in CI smoke tests). 3. **OS waits and discovery** (amended by tl-i01b, see the rationale below): 1. A `turn` makes at most one OS wait. @@ -566,6 +566,44 @@ No other Perry change is needed for P0. 4a. **No spin.** A turn with nothing ready and a future deadline blocks until that deadline, at the precision in §7.6. It never returns immediately and never degrades into a zero-timeout poll loop. Hosts must pass exact deadlines (`Instant`, not truncated milliseconds). The Linux A/B measured what happens otherwise: 37,607 turns for 50 timers, ~70× user instructions. A backend whose timeout is implemented by a private wakeup source — epoll's timerfd, the IOCP deadline packet, the WASI 0.2 deadline pollable, the WASI 0.3 deadline subtask — reports that wake as a **zero-event** wait, exactly as a timed OS wait that returned nothing; real I/O or notifier events arriving in the same call still make it non-empty. **Contract tests:** with an idle registered socket and a 0.5 ms / 2 ms / 10 ms timer, turns per expiry ≤ 2 and zero-event OS waits ≤ 1 per expiry, on every backend; and (tl-i02) each of the sixty expiries at those delays costs exactly one blocking wait that observed no native event, identically with the loop idle and with a registered-but-idle native operation, with no allocation, while a call that carries real bytes is not counted empty. 5. **Instruction budgets per operation** (Linux, cgu=1, `perf stat -e instructions:u,instructions:k`): TCP read / write / accept, timer start + cancel, notify + turn round trip, blocking job round trip, idle turn. **Values to be set from the attribution run of today's tokio bridge**, with a target below the tokio-bridge cost and within X % of a hand-written epoll loop. The CI gate compares against a committed baseline, with a control probe that must not move. +### 10a. Capacities are ceilings, not reservations + +`max_handles` and `max_operations` name the largest number of slots a loop may +ever hold. They are not preallocations: slots live in fixed-size pages built on +demand, page zero with the loop and another when the high-water mark crosses into +it. An idle loop therefore costs one page of each structure whatever its ceiling, +which is what makes a large default ceiling and a loop per agent (§5a) affordable +at once — the alternative forced hosts to choose between refusing connections and +paying for a loop they mostly do not use. + +Substituting pages for a flat vector is safe because of how slots are addressed. +A slot is named by its index and reused under a generation, so **adding a page +moves no existing slot and invalidates no handle or operation id**; pages are +separately allocated, so a slot's *address* is stable too, which the backends that +hand a slot's address to the kernel rely on. Indices are handed out +lowest-free-first, so the materialised prefix tracks the loop's high-water mark, +not its ceiling. + +Two things stay contiguous and sized by their ceiling. The Windows `OVERLAPPED` +slab (`kernel`, and the `bridges` parallel to it) maps a completion packet's +pointer back to an operation index by pointer arithmetic over one allocation; +paging it needs a different reverse map, which is its own change. The cross-thread +result rings are lock-free and index by a power-of-two mask, where the capacity is +also the backpressure bound. `pooled_buffers` is a separate question, tracked in +#43: it is the remaining per-loop cost that scales with configuration. + +**Reaching the ceiling is backpressure.** An operation whose completion creates a +handle — an accept, a handle receive — reserves its handle slot when it is +submitted. An accept the kernel has been asked to perform therefore always has +somewhere to put its connection, and at the ceiling the *submission* is refused +with `ResourceLimit` instead, before the kernel is asked. The pending connection +stays in the listener's backlog, which is the queue meant to absorb it. This +replaces accepting a connection and then destroying it for want of a slot, which +is what a host at its ceiling did before, and is why it surfaced as a connection +refusal rather than a delay. A multishot accept holds one such reservation, so it +is protected for one connection at a time; bounding a whole batch needs a per-turn +native event budget on `Backend::poll`, which is a separate change. + **Rule 3 rationale (tl-i01b, spec-owner decision, 2026-09-15).** The original rule 3 ("none when completions are already queued") forbade even a zero-timeout poll. Backend revision 2 has no separate no-wait discovery primitive: `poll(Duration::ZERO)` is the only way to learn about fresh readiness or completions, and on Unix, after cached readiness reaches `EAGAIN`, only the poller marks a resource ready again. libuv and Node make the same trade: `uv_run` computes `uv_backend_timeout()`, which is zero while pending, idle or closing work exists, and still calls `uv__io_poll` with that zero timeout (see the libuv [loop API](https://docs.libuv.org/en/v1.x/loop.html) and [`src/unix/core.c`](https://github.com/libuv/libuv/blob/v1.x/src/unix/core.c)). The [tl-i01 probes](docs/lanes/tl-i01.md#evidence--specification-decision) showed that skipping discovery breaks the unchanged fairness contract: skipping the native step whenever work was queued failed the timer/I/O/post fairness test after two seconds, and a guard that only drained cached work delivered 64 posts with zero reads and failed the same fairness test. A separate no-wait collection API on all six backends is not justified before measurement. Rule 4a and its raw zero-event accounting are unchanged: an empty discovery poll still counts toward the same no-spin bound. **Methodology** (lessons already paid for): diff --git a/crates/turnloop-contract/src/lib.rs b/crates/turnloop-contract/src/lib.rs index 0c29ce6..dc424f2 100644 --- a/crates/turnloop-contract/src/lib.rs +++ b/crates/turnloop-contract/src/lib.rs @@ -219,6 +219,18 @@ mod native { capacity_and_stale_ids::(); } #[test] + fn growth_preserves_handles() { + paged_growth_preserves_handles::(); + } + #[test] + fn accept_reserves_a_handle_slot() { + accept_reserves_its_handle_slot::(); + } + #[test] + fn an_armed_accept_keeps_its_reserved_slot() { + an_armed_accept_keeps_its_slot::(); + } + #[test] fn cancellation_close() { cancel_close_ordering::(); } @@ -894,6 +906,273 @@ pub fn capacity_and_stale_ids() { assert_eq!(count, 3); } +/// A key taken before the handle table grew still names its own resource after. +/// +/// `max_handles` is a ceiling served by pages built on demand, so a loop crosses +/// page boundaries as its high-water mark rises. Growth must leave every live +/// handle addressing what it addressed: an index means the same slot before and +/// after, and no live slot is ever handed out twice. +pub fn paged_growth_preserves_handles() { + // A ceiling many pages wide, so reaching the high-water mark below grows the + // table repeatedly while every handle taken from an earlier page is live. + let config = Config { + max_handles: 4096, + max_operations: 4096, + events_per_turn: 64, + ..Config::default() + }; + let mut l = Driver::::new(config).expect("loop"); + let at = l.now() + Duration::from_secs(30); + const LIVE: u64 = 600; + let mut handles = Vec::new(); + for i in 0..LIVE { + handles.push(l.timer(at, None, Token(i)).expect("within the ceiling")); + } + let mut keys: Vec = handles.iter().map(|h| h.key()).collect(); + keys.sort_unstable(); + keys.dedup(); + assert_eq!( + keys.len(), + handles.len(), + "growth handed out a slot that was still live" + ); + let mut indices: Vec = handles.iter().map(|h| h.index()).collect(); + indices.sort_unstable(); + indices.dedup(); + assert_eq!( + indices.len(), + handles.len(), + "two live handles share a slot" + ); + // Slots are handed out lowest-free-first, so holding LIVE of them at once + // means the table really did grow through every page boundary below it — + // without this the rest of the assertions would pass on an ungrown table. + assert_eq!( + (indices[0], *indices.last().expect("handles")), + (0, LIVE as usize - 1), + "the high-water mark did not reach the slot count under test" + ); + // Every handle still resolves, the first page's included, and each still + // names the timer it was created with rather than a neighbour's. + for (i, h) in handles.iter().enumerate() { + let op = l + .timer_op(*h) + .expect("a handle taken before growth still names its timer"); + assert_eq!(op.owner(), h.owner()); + l.set_ref(*h, true) + .expect("a live handle is still addressable"); + assert_eq!(h.index(), indices[i], "a handle changed slot across growth"); + } + // Closing every handle delivers exactly one Closed per handle, carrying the + // token that handle was closed with: growth aliased no slot onto another. + let mut out = Completions::with_capacity(64); + let mut closed = vec![false; LIVE as usize]; + for h in &handles { + l.close(*h, Token(h.index() as u64)).expect("close"); + } + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while closed.iter().any(|done| !done) { + assert!( + std::time::Instant::now() < deadline, + "Closed never delivered" + ); + l.turn(Timeout::After(Duration::from_millis(50)), &mut out) + .expect("drain"); + for c in out.drain() { + if matches!(c.result, OpResult::Closed) { + let slot = c.token.0 as usize; + assert!(!closed[slot], "slot {slot} delivered Closed twice"); + closed[slot] = true; + } + } + } + // A stale key never reaches the slot it used to name, even after reuse. + let reused = l.timer(at, None, Token(9999)).expect("freed slots return"); + assert!( + handles.iter().all(|h| h.key() != reused.key()), + "a retired key was handed out again" + ); + assert!( + l.set_ref(handles[0], true).is_err(), + "stale handle rejected" + ); +} + +/// A loop at its handle ceiling refuses to *ask* for a connection, rather than +/// accepting one and destroying it. +/// +/// An accept reserves its handle slot when it is submitted, so an accept the +/// kernel has been asked to perform always has somewhere to put its connection. +/// At the ceiling the submission is refused instead, and the connection stays in +/// the listener's backlog: the accept queue is the queue that absorbs it. +pub fn accept_reserves_its_handle_slot() { + // Two handles: the listener, and room for exactly one accepted connection. + let mut l = Driver::::new(Config { + max_handles: 2, + max_operations: 8, + ..Config::default() + }) + .expect("loop"); + let listener = l + .tcp_listen("127.0.0.1:0".parse().expect("addr"), &ListenOpts::default()) + .expect("listen"); + let addr = l.local_addr(listener).expect("addr"); + // The reserve is taken at submission, so a second concurrent accept is + // refused before it reaches the kernel even though nothing has connected. + let first = l.accept(listener, Token(1)).expect("one slot is free"); + assert!( + matches!( + l.accept(listener, Token(2)), + Err(Error { + kind: ErrorKind::ResourceLimit, + .. + }) + ), + "a second accept has no slot to reserve and must not reach the kernel" + ); + // The reserved slot is honoured: the accept that was armed delivers. + let _client = std::net::TcpStream::connect(addr).expect("client"); + let mut out = Completions::with_capacity(8); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + let mut conn = None; + while conn.is_none() { + assert!( + std::time::Instant::now() < deadline, + "accept never delivered" + ); + l.turn(Timeout::After(Duration::from_millis(50)), &mut out) + .expect("turn"); + for c in out.drain() { + match c.result { + OpResult::Accepted { conn: h, .. } => conn = Some(h), + other => panic!("unexpected {other:?}"), + } + } + } + let conn = conn.expect("accepted"); + assert_eq!(l.timer_op(conn), None, "an accepted socket is not a timer"); + let _ = first; + // Now at the ceiling: no slot to reserve, so no accept is armed at all. + assert!( + matches!( + l.accept(listener, Token(3)), + Err(Error { + kind: ErrorKind::ResourceLimit, + .. + }) + ), + "at the ceiling an accept is refused before the kernel is asked" + ); + // Traffic on handles that already exist is not gated by the reservation: a + // loop at its ceiling still serves the connections it has. + static HELLO: [u8; 5] = *b"hello"; + // SAFETY: HELLO is static immutable memory, valid through the completion. + let buf = unsafe { IoBuf::from_raw_parts(HELLO.as_ptr(), HELLO.len()) }; + l.write(conn, WriteBuf::Provided(buf), Token(6)) + .expect("a loop at its ceiling still serves its open connections"); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + let mut wrote = 0; + while wrote == 0 { + assert!( + std::time::Instant::now() < deadline, + "write never completed" + ); + l.turn(Timeout::After(Duration::from_millis(50)), &mut out) + .expect("turn"); + for c in out.drain() { + if let OpResult::Wrote(n) = c.result { + wrote = n; + } + } + } + assert_eq!(wrote, HELLO.len()); + // Releasing the connection returns its slot, and accepts arm again. + l.close(conn, Token(4)).expect("close"); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + let mut released = false; + while !released { + assert!( + std::time::Instant::now() < deadline, + "close never delivered" + ); + l.turn(Timeout::After(Duration::from_millis(50)), &mut out) + .expect("turn"); + for c in out.drain() { + released |= matches!(c.result, OpResult::Closed); + } + } + l.accept(listener, Token(5)) + .expect("a freed slot makes the loop acceptive again"); +} + +/// A slot promised to an armed accept cannot be spent by anything else. +/// +/// The reservation is only worth something if the loop refuses to hand that slot +/// to a handle the host creates directly. Without that, a host can fill its +/// ceiling while an accept is in flight, and the accept arrives with nowhere to +/// put its connection — which is the failure this whole mechanism exists to +/// remove, because the connection is destroyed rather than deferred. +pub fn an_armed_accept_keeps_its_slot() { + // Three slots: the listener, one held for the armed accept, one to spend. + let mut l = Driver::::new(Config { + max_handles: 3, + max_operations: 16, + ..Config::default() + }) + .expect("loop"); + let listener = l + .tcp_listen("127.0.0.1:0".parse().expect("addr"), &ListenOpts::default()) + .expect("listen"); + let addr = l.local_addr(listener).expect("addr"); + l.accept(listener, Token(1)).expect("one slot to reserve"); + let at = l.now() + Duration::from_secs(30); + let spare = l.timer(at, None, Token(2)).expect("the unreserved slot"); + assert!( + matches!( + l.timer(at, None, Token(3)), + Err(Error { + kind: ErrorKind::ResourceLimit, + .. + }) + ), + "the armed accept's slot must not be available to a new handle" + ); + // And the accept therefore still has somewhere to put its connection. + let _client = std::net::TcpStream::connect(addr).expect("client"); + let mut out = Completions::with_capacity(8); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + let mut conn = None; + while conn.is_none() { + assert!( + std::time::Instant::now() < deadline, + "accept never delivered" + ); + l.turn(Timeout::After(Duration::from_millis(50)), &mut out) + .expect("turn"); + for c in out.drain() { + match c.result { + OpResult::Accepted { conn: h, .. } => conn = Some(h), + OpResult::Err(e) => panic!("the reserved slot was spent elsewhere: {e:?}"), + other => panic!("unexpected {other:?}"), + } + } + } + let conn = conn.expect("accepted"); + // Every slot is now live, and the reservation is gone with the accept. + assert!( + matches!( + l.timer(at, None, Token(4)), + Err(Error { + kind: ErrorKind::ResourceLimit, + .. + }) + ), + "the loop is at its ceiling" + ); + l.close(conn, Token(5)).expect("close"); + l.close(spare, Token(6)).expect("close"); +} + pub fn pooled_lease_backpressure() { let mut l = Driver::::new(Config { pooled_buffers: 1, diff --git a/crates/turnloop-contract/tests/allocations.rs b/crates/turnloop-contract/tests/allocations.rs index 0220edf..2e9c2de 100644 --- a/crates/turnloop-contract/tests/allocations.rs +++ b/crates/turnloop-contract/tests/allocations.rs @@ -27,6 +27,9 @@ use turnloop::*; struct Counting; use std::cell::Cell; thread_local! { static ACTIVE: Cell = const { Cell::new(false) }; static ALLOCS: Cell = const { Cell::new(0) }; } +// Bytes requested while counting. A single oversized reservation is one +// allocation, so gates about a capacity *ceiling* have to weigh it, not count it. +thread_local! { static BYTES: Cell = const { Cell::new(0) }; } #[cfg(windows)] static ALL_THREADS_ACTIVE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); @@ -50,10 +53,9 @@ fn record(layout: Layout) { slot[2].store(layout.align(), Relaxed); } } - #[cfg(not(windows))] - let _ = layout; if ACTIVE.try_with(Cell::get).unwrap_or(false) { let _ = ALLOCS.try_with(|n| n.set(n.get() + 1)); + let _ = BYTES.try_with(|n| n.set(n.get() + layout.size())); } } // SAFETY: all allocation calls are forwarded unchanged to System. Counters only @@ -82,6 +84,112 @@ unsafe impl GlobalAlloc for Counting { #[global_allocator] static ALLOCATOR: Counting = Counting; +/// `max_handles` is a ceiling, not a preallocation: raising it must not make an +/// idle loop cost more. +/// +/// The handle table, the timer index and every backend array addressed by a slot +/// index build page zero with the loop and another page only as the high-water +/// mark rises. Weighed in bytes as well as counted, because one oversized +/// reservation is a single allocation — which is exactly how the cost used to +/// hide. +#[test] +fn an_idle_loop_costs_the_same_at_any_ceiling() { + // Take the process-wide lazy initialisation out of the measurement. + drop(Loop::new(Config::default()).expect("warm")); + let mut measured = Vec::new(); + for max_handles in [1024usize, 1024 * 1024] { + let config = Config { + max_handles, + ..Config::default() + }; + ALLOCS.with(|n| n.set(0)); + BYTES.with(|n| n.set(0)); + ACTIVE.with(|v| v.set(true)); + let l = Loop::new(config).expect("loop"); + ACTIVE.with(|v| v.set(false)); + measured.push((ALLOCS.with(Cell::get), BYTES.with(Cell::get))); + drop(l); + } + let (small, large) = (measured[0], measured[1]); + assert!( + small.0 > 0 && small.1 > 0, + "the allocator must see the loop being built" + ); + assert_eq!( + small.0, large.0, + "a 1024x larger ceiling made {} allocations instead of {}", + large.0, small.0 + ); + assert_eq!( + small.1, large.1, + "a 1024x larger ceiling reserved {} bytes instead of {}", + large.1, small.1 + ); +} + +/// A loop pinned at its high-water mark allocates nothing per turn. +/// +/// Pages are built as the mark rises, so growth allocates and a steady state +/// does not. This is the gate that keeps paging honest: without it, a structure +/// that quietly rebuilt itself per turn would still satisfy the ceiling gate above. +#[test] +fn a_loop_at_its_high_water_mark_allocates_nothing_per_turn() { + // A ceiling far above the working set, so the mark below is nowhere near it. + let mut l = Loop::new(Config { + max_handles: 1024 * 1024, + max_operations: 1024 * 1024, + ..Config::default() + }) + .expect("loop"); + let mut out = Completions::with_capacity(64); + // Raise the high-water mark well past page zero, and hold it there. + let held: Vec = (0..500) + .map(|i| { + l.timer(l.now() + Duration::from_secs(600), None, Token(i)) + .expect("within the ceiling") + }) + .collect(); + assert!( + held.iter().map(|h| h.index()).max().expect("held") >= 499, + "the mark did not rise past page zero, so this would gate nothing" + ); + // One churn cycle at the mark, to settle any first-use storage the loop + // reaches only when a slot is reused rather than built. + let churn = |l: &mut Loop, out: &mut Completions, token: u64| { + let h = l + .timer(l.now() + Duration::from_secs(600), None, Token(token)) + .expect("a slot at the mark"); + let op = l.timer_op(h).expect("timer op"); + assert!(l.cancel(op)); + l.close(h, Token(token)).expect("close"); + let mut terminals = 0; + while terminals < 2 { + let info = l.turn(Timeout::Now, out).expect("turn"); + assert_eq!(info.os_waits, 0, "queued work must not wait"); + for c in out.drain() { + assert!(c.terminal); + terminals += 1; + } + } + }; + churn(&mut l, &mut out, 1_000); + ALLOCS.with(|n| n.set(0)); + BYTES.with(|n| n.set(0)); + ACTIVE.with(|v| v.set(true)); + for token in 0..1000u64 { + churn(&mut l, &mut out, 2_000 + token); + } + ACTIVE.with(|v| v.set(false)); + assert_eq!( + (ALLOCS.with(Cell::get), BYTES.with(Cell::get)), + (0, 0), + "a loop at its high-water mark allocated while turning" + ); + for h in held { + l.close(h, Token(0)).expect("close"); + } +} + #[test] fn queued_posts_and_terminals_with_idle_udp_allocate_nothing() { ALLOCS.set(0); diff --git a/crates/turnloop-contract/tests/wasi.rs b/crates/turnloop-contract/tests/wasi.rs index a43f38c..577eaae 100644 --- a/crates/turnloop-contract/tests/wasi.rs +++ b/crates/turnloop-contract/tests/wasi.rs @@ -126,6 +126,10 @@ fn capacity_stale_ids() { contract::capacity_and_stale_ids::(); } #[test] +fn growth_preserves_handles() { + contract::paged_growth_preserves_handles::(); +} +#[test] fn pooled_backpressure() { contract::pooled_lease_backpressure::(); } diff --git a/crates/turnloop-contract/tests/windows.rs b/crates/turnloop-contract/tests/windows.rs index 51693a8..67952df 100644 --- a/crates/turnloop-contract/tests/windows.rs +++ b/crates/turnloop-contract/tests/windows.rs @@ -27,6 +27,9 @@ contract!( handoff_distribution, writev_and_shutdown, capacity_and_stale_ids, + paged_growth_preserves_handles, + accept_reserves_its_handle_slot, + an_armed_accept_keeps_its_slot, ready_timer_liveness, pooled_lease_backpressure, io_and_posts_progress_with_repeating_timers diff --git a/crates/turnloop/Cargo.toml b/crates/turnloop/Cargo.toml index 9ccfb5e..c4d5fb2 100644 --- a/crates/turnloop/Cargo.toml +++ b/crates/turnloop/Cargo.toml @@ -44,7 +44,7 @@ loom.workspace = true role = "core" wasi-lib-tests = true loom-filters = ["models"] -miri-filters = ["timer::tests", "table::tests"] +miri-filters = ["timer::tests", "table::tests", "slots::tests"] [target.'cfg(all(target_os = "wasi", target_env = "p2"))'.dependencies] wasip2 = "1.0.3" diff --git a/crates/turnloop/src/backend/files.rs b/crates/turnloop/src/backend/files.rs index 5bf11bf..06c5c05 100644 --- a/crates/turnloop/src/backend/files.rs +++ b/crates/turnloop/src/backend/files.rs @@ -1,5 +1,6 @@ //! Reusable regular-file jobs on the shared blocking pool; never file I/O in poll. use super::{poller::last_error, unix::Detached}; +use crate::slots::{Slots, page_reserve}; use crate::{ backend::{Event, Operation, Outcome, Request}, blocking::ReusableWork, @@ -93,17 +94,33 @@ struct Active { awaiting_pool: bool, } pub(super) struct Files { - slots: Vec>, - active: Vec>, - heads: Vec>, - tails: Vec>, + /// Per-operation job slots, built when an operation index is first used. + slots: Slots>, + active: Slots, + heads: Slots, + tails: Slots, port: Arc, config: PoolConfig, pool: BufferPool, - leases: Vec>, + leases: Slots, ready: VecDeque, awaiting_pool: VecDeque, } +/// Darwin's std mutex allocates its pthread storage on first lock, so the slot +/// takes its own lock here: publishing a job later uses only reserved storage. +fn build_slot(port: &Arc) -> Arc { + let slot = Arc::new(Slot { + job: Mutex::new(None), + cancel: AtomicBool::new(false), + port: port.clone(), + }); + drop( + slot.job + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ); + slot +} impl Files { pub fn new(config: &Config, pool: BufferPool) -> Self { let port = Arc::new(Port { @@ -113,33 +130,16 @@ impl Files { quiescent: Condvar::new(), }); Self { - slots: (0..config.max_operations) - .map(|_| { - let slot = Arc::new(Slot { - job: Mutex::new(None), - cancel: AtomicBool::new(false), - port: port.clone(), - }); - // Darwin's std mutex allocates its pthread storage on first - // lock. Reserve it now for every op slot, including slots - // first reached later through backpressure/cancellation. - drop( - slot.job - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner), - ); - slot - }) - .collect(), - active: (0..config.max_operations).map(|_| None).collect(), - heads: vec![None; config.max_handles], - tails: vec![None; config.max_handles], + slots: Slots::filled(config.max_operations, || build_slot(&port)), + active: Slots::new(config.max_operations), + heads: Slots::new(config.max_handles), + tails: Slots::new(config.max_handles), port, config: config.blocking_pool, pool, - leases: (0..config.max_operations).map(|_| None).collect(), - ready: VecDeque::with_capacity(config.max_handles), - awaiting_pool: VecDeque::with_capacity(config.max_handles), + leases: Slots::new(config.max_operations), + ready: VecDeque::with_capacity(page_reserve(config.max_handles)), + awaiting_pool: VecDeque::with_capacity(page_reserve(config.max_handles)), } } pub fn set_notifier(&mut self, notifier: Notifier) { @@ -181,6 +181,16 @@ impl Files { }); Ok(()) } + /// This operation's job slot, built on first use. + /// + /// Darwin's std mutex allocates its pthread storage on first lock, so the + /// slot takes its own lock here: a job publication from the loop thread only + /// ever uses storage this reserved, including slots first reached later + /// through backpressure or cancellation. + fn slot(&mut self, i: usize) -> &Arc { + let port = self.port.clone(); + self.slots.get_or_insert_with(i, || build_slot(&port)) + } pub fn cancel(&mut self, op: OpId) -> bool { let Some(active) = self .active @@ -197,7 +207,11 @@ impl Files { self.ready.push_back(op.index()); } if active.submitted { - self.slots[op.index()].cancel.store(true, Ordering::Release); + self.slots[op.index()] + .as_ref() + .expect("a submitted request built its slot") + .cancel + .store(true, Ordering::Release); } true } @@ -216,6 +230,7 @@ impl Files { } } while let Some(i) = self.ready.pop_front() { + let slot = self.slot(i).clone(); let active = self.active[i].as_mut().expect("file head"); debug_assert!(!active.submitted); if active.cancelled { @@ -255,7 +270,6 @@ impl Files { self.leases[i] = Some(lease); } let result = active.fd.try_clone().map_err(Error::from).and_then(|fd| { - let slot = &self.slots[i]; slot.cancel.store(false, Ordering::Release); *slot .job @@ -352,9 +366,10 @@ impl Files { impl Drop for Files { fn drop(&mut self) { for a in self.active.iter().flatten() { - self.slots[a.op.index()] - .cancel - .store(true, Ordering::Release); + // A request that never started has no slot, and nothing to cancel. + if let Some(slot) = self.slots[a.op.index()].as_ref() { + slot.cancel.store(true, Ordering::Release); + } } let mut running = self .port diff --git a/crates/turnloop/src/backend/fsevents.rs b/crates/turnloop/src/backend/fsevents.rs index 6e53f8a..0036ff6 100644 --- a/crates/turnloop/src/backend/fsevents.rs +++ b/crates/turnloop/src/backend/fsevents.rs @@ -255,10 +255,10 @@ pub(super) struct Streams { queue: Ref, } impl Streams { - pub fn new(capacity: usize) -> Self { + pub fn new() -> Self { let inbox = Arc::new(Inbox { pending: AtomicBool::new(false), - ready: Mutex::new(VecDeque::with_capacity(capacity * 2)), + ready: Mutex::new(VecDeque::with_capacity(crate::slots::PAGE)), notifier: Mutex::new(None), }); drop(lock(&inbox.ready)); diff --git a/crates/turnloop/src/backend/iocp/mod.rs b/crates/turnloop/src/backend/iocp/mod.rs index 4da7082..288e9fd 100644 --- a/crates/turnloop/src/backend/iocp/mod.rs +++ b/crates/turnloop/src/backend/iocp/mod.rs @@ -1,5 +1,6 @@ //! Native IOCP completion backend. Kernel storage is pinned separately from Rust //! operation metadata and survives cancellation until the OS acknowledgement. +use crate::slots::{Slots, page_reserve}; mod bridge; mod integration; mod ipc; @@ -271,9 +272,12 @@ impl crate::backend::Wake for IocpWake { } /// One IOCP per host-driven loop, with fixed operation storage and an opt-in GUI helper. pub struct Iocp { - resources: Vec>, - ops: Vec>, + resources: Slots, + ops: Slots, kernel: Box<[UnsafeCell]>, + /// One per `kernel` slot, so it is sized with that slab rather than paged: + /// a bridge holds its slot's pinned address, and the reverse map from an + /// `OVERLAPPED` pointer back to an operation index needs `kernel` contiguous. bridges: Vec>, ready: VecDeque, pool_waiting: VecDeque, @@ -285,7 +289,7 @@ pub struct Iocp { event: Option, notifier: Option, services: services::Services, - workers: Vec>, + workers: Slots<[sync_io::Worker; 2]>, deadline: Option, failure: Option, next_listener_key: usize, @@ -1113,12 +1117,12 @@ unsafe impl Backend for Iocp { .collect::>>()?; let watches = watch::Watches::new(config, pool.clone(), Arc::clone(&port)); Ok(Self { - resources: (0..config.max_handles).map(|_| None).collect(), - ops: (0..config.max_operations).map(|_| None).collect(), + resources: Slots::new(config.max_handles), + ops: Slots::new(config.max_operations), kernel, bridges, - ready: VecDeque::with_capacity(config.max_operations), - pool_waiting: VecDeque::with_capacity(config.max_operations), + ready: VecDeque::with_capacity(page_reserve(config.max_operations)), + pool_waiting: VecDeque::with_capacity(page_reserve(config.max_operations)), pool, wake: Arc::new(IocpWake { port: Arc::clone(&port), @@ -1130,7 +1134,7 @@ unsafe impl Backend for Iocp { event: None, notifier: None, services: services::Services::new(config.max_handles), - workers: (0..config.max_handles).map(|_| None).collect(), + workers: Slots::new(config.max_handles), deadline: None, failure: None, next_listener_key: pipes::FIRST_KEY, @@ -1683,7 +1687,7 @@ impl Drop for Iocp { } } self.failure = None; // teardown no longer arms host deadlines - for i in 0..self.ops.len() { + for i in 0..self.ops.materialised() { if let Some(p) = &self.ops[i] { let op = p.request.op; if self.cancel(op).is_err() { diff --git a/crates/turnloop/src/backend/iocp/services.rs b/crates/turnloop/src/backend/iocp/services.rs index 3a10606..f151353 100644 --- a/crates/turnloop/src/backend/iocp/services.rs +++ b/crates/turnloop/src/backend/iocp/services.rs @@ -1,4 +1,5 @@ use super::{Detached, invalid, process::Child, signals::Subscription, unsupported}; +use crate::slots::Slots; use crate::{ backend::{Event, Operation, Outcome, Request}, *, @@ -15,13 +16,13 @@ struct Entry { closing: bool, } pub(super) struct Services { - entries: Vec>, + entries: Slots, pending: usize, } impl Services { pub(super) fn new(capacity: usize) -> Self { Self { - entries: (0..capacity).map(|_| None).collect(), + entries: Slots::new(capacity), pending: 0, } } diff --git a/crates/turnloop/src/backend/iocp/watch.rs b/crates/turnloop/src/backend/iocp/watch.rs index b8e4140..84295e0 100644 --- a/crates/turnloop/src/backend/iocp/watch.rs +++ b/crates/turnloop/src/backend/iocp/watch.rs @@ -9,6 +9,7 @@ use super::{ Detached, os_error, port::{Entry, Port}, }; +use crate::slots::{Slots, page_reserve}; use crate::{ backend::{Event, Operation, Outcome, Request}, fs::watch::Ring, @@ -149,11 +150,11 @@ fn same_name(a: &[u16], b: &[u16]) -> bool { } pub(super) struct Watches { - entries: Vec>, + entries: Slots, active: Vec, /// Released watches whose last request awaits its acknowledgement. retired: Vec, - ops: Vec>, + ops: Slots, ready: VecDeque, finished: VecDeque<(OpId, Result<()>)>, pool: BufferPool, @@ -162,12 +163,12 @@ pub(super) struct Watches { impl Watches { pub fn new(config: &Config, pool: BufferPool, port: Arc) -> Self { Self { - entries: (0..config.max_handles).map(|_| None).collect(), - active: Vec::with_capacity(config.max_handles), - retired: Vec::with_capacity(config.max_handles), - ops: vec![None; config.max_operations], - ready: VecDeque::with_capacity(config.max_handles), - finished: VecDeque::with_capacity(config.max_operations), + entries: Slots::new(config.max_handles), + active: Vec::with_capacity(page_reserve(config.max_handles)), + retired: Vec::with_capacity(page_reserve(config.max_handles)), + ops: Slots::new(config.max_operations), + ready: VecDeque::with_capacity(page_reserve(config.max_handles)), + finished: VecDeque::with_capacity(page_reserve(config.max_operations)), pool, port, } diff --git a/crates/turnloop/src/backend/services.rs b/crates/turnloop/src/backend/services.rs index a23939d..453d4b8 100644 --- a/crates/turnloop/src/backend/services.rs +++ b/crates/turnloop/src/backend/services.rs @@ -4,6 +4,7 @@ use super::{ signals::{self, Subscription}, unix::Detached, }; +use crate::slots::{Slots, page_reserve}; use crate::{ backend::{Event, Operation, Outcome, Request}, sync::{AtomicBool, Ordering}, @@ -76,15 +77,17 @@ mod readiness_models { } struct ReadyState { queue: VecDeque, - queued: Vec, + /// Whether each handle is already in `queue`, so a repeat publication does + /// not enqueue it twice. Vacant reads as "not queued". + queued: Slots<()>, } impl ReadyQueue { fn new(capacity: usize) -> Self { let ready = Self { pending: AtomicBool::new(false), state: Mutex::new(ReadyState { - queue: VecDeque::with_capacity(capacity), - queued: vec![false; capacity], + queue: VecDeque::with_capacity(page_reserve(capacity)), + queued: Slots::new(capacity), }), }; // Initialize Darwin's lazily allocated pthread mutex during loop setup, @@ -102,8 +105,8 @@ impl ReadyQueue { .state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if !s.queued[h.index()] { - s.queued[h.index()] = true; + if s.queued[h.index()].is_none() { + s.queued[h.index()] = Some(()); s.queue.push_back(h); self.pending.store(true, Ordering::Release); } @@ -117,7 +120,7 @@ impl ReadyQueue { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let h = s.queue.pop_front()?; - s.queued[h.index()] = false; + s.queued[h.index()] = None; self.pending.store(!s.queue.is_empty(), Ordering::Release); Some(h) } @@ -127,7 +130,7 @@ impl ReadyQueue { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); s.queue.retain(|&queued| queued != h); - s.queued[h.index()] = false; + s.queued[h.index()] = None; self.pending.store(!s.queue.is_empty(), Ordering::Release); } fn has_work(&self) -> bool { @@ -194,16 +197,16 @@ struct Entry { cancelled: bool, } pub(super) struct Services { - entries: Vec>, - operations: Vec>, + entries: Slots, + operations: Slots, ready: Arc, notifier: Option, } impl Services { pub fn new(config: &Config) -> Self { Self { - entries: (0..config.max_handles).map(|_| None).collect(), - operations: vec![None; config.max_operations], + entries: Slots::new(config.max_handles), + operations: Slots::new(config.max_operations), ready: Arc::new(ReadyQueue::new(config.max_handles)), notifier: None, } diff --git a/crates/turnloop/src/backend/unix.rs b/crates/turnloop/src/backend/unix.rs index fa91004..1d63448 100644 --- a/crates/turnloop/src/backend/unix.rs +++ b/crates/turnloop/src/backend/unix.rs @@ -8,6 +8,7 @@ use super::{ poller::{Poller, Ready, last_error}, socket::{self, Addr}, }; +use crate::slots::{Slots, page_reserve}; use crate::{ backend::{Backend, Event, Operation, Outcome, PollInfo, Request}, *, @@ -135,8 +136,8 @@ struct Pending { /// Completion engine shared by kqueue and epoll, with native process and signal services. pub struct Unix { poller: SystemPoller, - resources: Vec>, - ops: Vec>, + resources: Slots, + ops: Slots, ready: VecDeque, cancelled: VecDeque, polled: Vec, @@ -298,10 +299,10 @@ unsafe impl Backend for Unix { fn new(config: &Config, pool: BufferPool) -> Result { Ok(Self { poller: SystemPoller::new(config.events_per_turn)?, - resources: (0..config.max_handles).map(|_| None).collect(), - ops: (0..config.max_operations).map(|_| None).collect(), - ready: VecDeque::with_capacity(config.max_handles), - cancelled: VecDeque::with_capacity(config.max_operations), + resources: Slots::new(config.max_handles), + ops: Slots::new(config.max_operations), + ready: VecDeque::with_capacity(page_reserve(config.max_handles)), + cancelled: VecDeque::with_capacity(page_reserve(config.max_operations)), polled: Vec::with_capacity(config.events_per_turn), files: super::files::Files::new(config, pool.clone()), watches: super::watch::Watches::new(config, pool.clone()), diff --git a/crates/turnloop/src/backend/wasi_fs.rs b/crates/turnloop/src/backend/wasi_fs.rs index d95a84e..8213be9 100644 --- a/crates/turnloop/src/backend/wasi_fs.rs +++ b/crates/turnloop/src/backend/wasi_fs.rs @@ -6,6 +6,7 @@ //! //! Paths resolve only against preopened directories: an absolute path uses the //! preopen with the longest matching name, a relative path the preopen named `.`. +use crate::slots::{Slots, page_reserve}; use crate::{ backend::{Event, Outcome}, fs::{ @@ -133,10 +134,10 @@ struct Pending { } pub(super) struct Files { preopens: Option>, - objects: Vec>>, - ops: Vec>, - heads: Vec>, - tails: Vec>, + objects: Slots>, + ops: Slots, + heads: Slots, + tails: Slots, ready: VecDeque, waiting: VecDeque, cancelled: VecDeque, @@ -146,13 +147,13 @@ impl Files { pub fn new(config: &Config, pool: BufferPool) -> Self { Self { preopens: None, - objects: (0..config.max_handles).map(|_| None).collect(), - ops: (0..config.max_operations).map(|_| None).collect(), - heads: vec![None; config.max_handles], - tails: vec![None; config.max_handles], - ready: VecDeque::with_capacity(config.max_operations), - waiting: VecDeque::with_capacity(config.max_operations), - cancelled: VecDeque::with_capacity(config.max_operations), + objects: Slots::new(config.max_handles), + ops: Slots::new(config.max_operations), + heads: Slots::new(config.max_handles), + tails: Slots::new(config.max_handles), + ready: VecDeque::with_capacity(page_reserve(config.max_operations)), + waiting: VecDeque::with_capacity(page_reserve(config.max_operations)), + cancelled: VecDeque::with_capacity(page_reserve(config.max_operations)), pool, } } @@ -362,7 +363,7 @@ fn resolve<'a, D>(preopens: &'a [(D, String)], path: &'a FsPath) -> Result<(&'a } fn execute( preopens: &[(A::Descriptor, String)], - objects: &mut [Option>], + objects: &mut Slots>, handle: Option, request: FsRequest, ) -> Result { diff --git a/crates/turnloop/src/backend/wasi_p2.rs b/crates/turnloop/src/backend/wasi_p2.rs index 0d8917e..2770c03 100644 --- a/crates/turnloop/src/backend/wasi_p2.rs +++ b/crates/turnloop/src/backend/wasi_p2.rs @@ -1,5 +1,6 @@ //! WASI 0.2 completion backend. One poll import per turn, reusable canonical //! lists, and synchronous nonblocking I/O with generational cancellation. +use crate::slots::{Slots, page_reserve}; mod abi; mod fs; mod sockopt; @@ -116,13 +117,13 @@ enum PollOwner { } /// WASI 0.2 pollable driver with retained canonical buffers. pub struct WasiP2 { - resources: Vec>, - ops: Vec>, + resources: Slots, + ops: Slots, ready: VecDeque, cancelled: VecDeque, handles: Vec, owners: Vec, - lookups: Vec>, + lookups: Slots, indices: Vec, poll_storage: Vec, scratch: Vec, @@ -388,22 +389,27 @@ unsafe impl Backend for WasiP2 { self.files.submit(op, handle, request) } fn new(config: &Config, pool: BufferPool) -> Result { + // The most pollables this loop could ever subscribe at once. Still + // computed, because an overflow here is a configuration this backend + // cannot serve; the poll batch below reserves a page of it and grows + // with the pollables actually subscribed, not with the ceiling. let polls = config .max_handles .checked_mul(2) .and_then(|n| n.checked_add(config.max_operations)) .and_then(|n| n.checked_add(1)) .ok_or(Error::new(ErrorKind::ResourceLimit))?; + let batch = page_reserve(polls); Ok(Self { - resources: (0..config.max_handles).map(|_| None).collect(), - ops: (0..config.max_operations).map(|_| None).collect(), - ready: VecDeque::with_capacity(config.max_handles), - cancelled: VecDeque::with_capacity(config.max_operations), - handles: Vec::with_capacity(polls), - owners: Vec::with_capacity(polls), - lookups: (0..config.max_operations).map(|_| None).collect(), - indices: Vec::with_capacity(polls), - poll_storage: vec![0; polls], + resources: Slots::new(config.max_handles), + ops: Slots::new(config.max_operations), + ready: VecDeque::with_capacity(page_reserve(config.max_handles)), + cancelled: VecDeque::with_capacity(page_reserve(config.max_operations)), + handles: Vec::with_capacity(batch), + owners: Vec::with_capacity(batch), + lookups: Slots::new(config.max_operations), + indices: Vec::with_capacity(batch), + poll_storage: vec![0; batch], scratch: vec![0; 16400], files: super::wasi_fs::Files::new(config, pool.clone()), pool, @@ -704,6 +710,13 @@ unsafe impl Backend for WasiP2 { // No native source can ever wake a Forever wait on this single agent. return Err(Error::new(ErrorKind::Unsupported)); } + // `poll` lowers its result into `poll_storage` as the canonical return + // arena, and returns at most one index per subscribed pollable, so the + // arena has to cover exactly that many `u32`s. It grows with the poll + // batch; at a steady set of pollables it is already large enough. + if self.poll_storage.len() < self.handles.len() { + self.poll_storage.resize(self.handles.len(), 0); + } abi::poll(&self.handles, &mut self.poll_storage, &mut self.indices); // The private deadline pollable implements this wait's timeout, so its // readiness has the meaning of the poll returning nothing (see `PollInfo` diff --git a/crates/turnloop/src/backend/wasi_p3.rs b/crates/turnloop/src/backend/wasi_p3.rs index 297c822..b329533 100644 --- a/crates/turnloop/src/backend/wasi_p3.rs +++ b/crates/turnloop/src/backend/wasi_p3.rs @@ -2,6 +2,7 @@ //! and fixed request return areas avoid a fresh block_on or executor per turn. //! Experimental: host-yield boundedness is unproven. Allocation gates require //! release on the pinned p3 compiler; see docs/upstream/wasi-p3-wait.md. +use crate::slots::{Slots, page_reserve}; mod abi; mod fs; mod return_storage; @@ -132,8 +133,8 @@ impl Wake for WasiWake { } /// Experimental WASI 0.3 driver with a persistent waitable set. pub struct WasiP3 { - resources: Vec>, - ops: Vec>, + resources: Slots, + ops: Slots, ready: VecDeque, cancelled: VecDeque, pool: BufferPool, @@ -353,10 +354,10 @@ unsafe impl Backend for WasiP3 { } fn new(config: &Config, pool: BufferPool) -> Result { Ok(Self { - resources: (0..config.max_handles).map(|_| None).collect(), - ops: (0..config.max_operations).map(|_| None).collect(), - ready: VecDeque::with_capacity(config.max_handles), - cancelled: VecDeque::with_capacity(config.max_operations), + resources: Slots::new(config.max_handles), + ops: Slots::new(config.max_operations), + ready: VecDeque::with_capacity(page_reserve(config.max_handles)), + cancelled: VecDeque::with_capacity(page_reserve(config.max_operations)), files: super::wasi_fs::Files::new(config, pool.clone()), pool, wake: Arc::new(WasiWake), @@ -648,7 +649,9 @@ unsafe impl Backend for WasiP3 { } } if kind != 0 && !deadline_event { - for i in 0..self.ops.len() { + // The materialised prefix, not the ceiling: an occupied slot is + // always below the high-water mark, and this runs per event. + for i in 0..self.ops.materialised() { let Some(p) = &mut self.ops[i] else { continue; }; diff --git a/crates/turnloop/src/backend/watch.rs b/crates/turnloop/src/backend/watch.rs index e6ef98c..6b1fb8b 100644 --- a/crates/turnloop/src/backend/watch.rs +++ b/crates/turnloop/src/backend/watch.rs @@ -12,6 +12,7 @@ use super::{ poller::{Poller, Ready}, unix::Detached, }; +use crate::slots::{Slots, page_reserve}; use crate::{ backend::{Event, Operation, Outcome, Request}, fs::watch::Ring, @@ -55,10 +56,10 @@ impl Entry { } } pub(super) struct Watches { - entries: Vec>, + entries: Slots, /// Indices of occupied entries, so event dispatch never scans every handle slot. active: Vec, - ops: Vec>, + ops: Slots, ready: VecDeque, cancelled: VecDeque, pool: BufferPool, @@ -90,16 +91,16 @@ fn basename(path: &FsPath) -> Box<[u8]> { impl Watches { pub fn new(config: &Config, pool: BufferPool) -> Self { Self { - entries: (0..config.max_handles).map(|_| None).collect(), - active: Vec::with_capacity(config.max_handles), - ops: vec![None; config.max_operations], - ready: VecDeque::with_capacity(config.max_handles), - cancelled: VecDeque::with_capacity(config.max_operations), + entries: Slots::new(config.max_handles), + active: Vec::with_capacity(page_reserve(config.max_handles)), + ops: Slots::new(config.max_operations), + ready: VecDeque::with_capacity(page_reserve(config.max_handles)), + cancelled: VecDeque::with_capacity(page_reserve(config.max_operations)), pool, #[cfg(any(target_os = "linux", target_os = "android"))] inotify: None, #[cfg(target_os = "macos")] - fsevents: fsevents::Streams::new(config.max_handles), + fsevents: fsevents::Streams::new(), } } #[cfg_attr(not(target_os = "macos"), allow(unused_variables))] diff --git a/crates/turnloop/src/backend/web.rs b/crates/turnloop/src/backend/web.rs index c8ac1b3..c59c803 100644 --- a/crates/turnloop/src/backend/web.rs +++ b/crates/turnloop/src/backend/web.rs @@ -6,6 +6,7 @@ //! `TCP_NODELAY`, keep-alive schedule, linger, buffer size or group membership to //! set or read. `ListenOpts` never reaches this backend either, because listening //! sockets are themselves unsupported (DESIGN §7.5). +use crate::slots::Slots; use crate::{ backend::{Backend, Event, Operation, Outcome, PollInfo, Request, Wake}, *, @@ -94,8 +95,8 @@ struct Pending { pub struct Web { id: u32, wake: Arc, - resources: Vec>, - ops: Vec>, + resources: Slots, + ops: Slots, pool: BufferPool, failure: Option, #[cfg(feature = "web-worker")] @@ -194,8 +195,8 @@ unsafe impl Backend for Web { Ok(Self { id, wake: Arc::new(WebWake { id }), - resources: (0..config.max_handles).map(|_| None).collect(), - ops: (0..config.max_operations).map(|_| None).collect(), + resources: Slots::new(config.max_handles), + ops: Slots::new(config.max_operations), pool, failure: None, #[cfg(feature = "web-worker")] diff --git a/crates/turnloop/src/driver.rs b/crates/turnloop/src/driver.rs index 2f4f290..45303f4 100644 --- a/crates/turnloop/src/driver.rs +++ b/crates/turnloop/src/driver.rs @@ -49,6 +49,9 @@ struct Op { external_wait: bool, /// A typed filesystem request (pool service or backend, per `B::FILESYSTEM`). fs: bool, + /// Handle slots this operation holds against the ceiling, released when it + /// retires. Non-zero only for operations whose completion creates a handle. + reserved_handles: usize, /// Counted in `native_pending`: a socket-handle operation or a request the /// backend accepted natively (such as WASI DNS). DESIGN §10 rule 3 keys /// queued-turn discovery on these operations. @@ -88,10 +91,16 @@ pub struct Driver { timers: TimerQueue, connect_deadlines: TimerQueue, queued: VecDeque, + /// Largest number of completions that may be queued at once. Was implicit in + /// `queued`'s reserved capacity, which is now a page that grows on demand. + completion_capacity: usize, buffered: [usize; 3], events: Vec>, refs: usize, outstanding: usize, + /// Handle slots promised to accepts and handle receives that have been asked + /// for but not yet delivered. See [`Driver::submit`]. + reserved_handles: usize, native_pending: usize, config: Config, _local: PhantomData>, @@ -152,16 +161,35 @@ impl Driver { // multishot events, repeating timers and posts each have their own // bounded reserve, so no source can consume cancellation capacity or // prevent another source from making progress with small host output. - queued: VecDeque::with_capacity(completion_capacity), + // The bound is `completion_capacity`, checked on every push; the + // queue reserves a page of it and grows as a turn's backlog does. + queued: VecDeque::with_capacity(crate::slots::page_reserve(completion_capacity)), + completion_capacity, buffered: [0; 3], events: Vec::with_capacity(config.events_per_turn), refs: 0, outstanding: 0, + reserved_handles: 0, native_pending: 0, config, _local: PhantomData, }) } + /// Reserved slots are a subset of free ones. + /// + /// Every path that moves either side keeps this: a reserving submission + /// checks it before promising, `new_handle` refuses to spend a promised + /// slot, and a delivery releases its promise before taking the slot. Drift + /// here means an armed accept can arrive with nowhere to put its connection, + /// which is silent until a connection is destroyed for it. + fn assert_reservations(&self) { + debug_assert!( + self.reserved_handles <= self.handles.remaining(), + "{} handle slots promised but only {} free", + self.reserved_handles, + self.handles.remaining() + ); + } fn assert_owner(&self) { debug_assert_eq!( self.thread, @@ -178,7 +206,17 @@ impl Driver { .filter(|r| !r.hidden) .ok_or(Error::new(ErrorKind::NotFound)) } + /// Take a handle slot for a resource the host is creating. + /// + /// Slots promised to armed accepts are not available here: an accept that the + /// kernel has been asked to perform must still have somewhere to put its + /// connection when it arrives, so a host creating handles cannot spend the + /// last one out from under it. The completion path consumes its own + /// reservation first, in [`Driver::attach_reserved`]. fn new_handle(&mut self, kind: Kind) -> Result { + if self.handles.remaining() <= self.reserved_handles { + return Err(Error::new(ErrorKind::ResourceLimit)); + } let key = self .handles .insert(Resource { @@ -195,6 +233,7 @@ impl Driver { if matches!(kind, Kind::Socket | Kind::File) { self.refs += 1; } + self.assert_reservations(); Ok(Handle { owner: self.owner, key, @@ -219,6 +258,7 @@ impl Driver { job_cancel: None, external_wait: false, fs: false, + reserved_handles: 0, native, previous, next: None, @@ -254,6 +294,7 @@ impl Driver { fn retire(&mut self, id: OpId) -> Option { let op = self.ops.remove(id.key)?; self.connect_deadlines.cancel(id.key); + self.reserved_handles -= op.reserved_handles; if op.native { self.native_pending -= 1; } @@ -314,7 +355,7 @@ impl Driver { debug_assert!(self.buffered[class] < self.config.events_per_turn); self.buffered[class] += 1; } - debug_assert!(self.queued.len() < self.queued.capacity()); + debug_assert!(self.queued.len() < self.completion_capacity); self.refs += usize::from(referenced); self.queued.push_back(Queued { completion, @@ -784,12 +825,33 @@ impl Driver { } self.backend.raw_transport(h) } + /// Submit one operation on a socket handle. + /// + /// An operation whose completion creates a handle (an accept, a handle + /// receive) reserves its handle slot here, before the backend is asked. + /// Refusing at the ceiling is therefore a refusal to *ask*: the connection + /// stays in the listener's backlog, which is the queue that is meant to + /// absorb it. The alternative, which this replaces, was to let the kernel + /// hand over a connection and then destroy it for want of a slot, so a host + /// at its ceiling refused connections instead of deferring them. fn submit(&mut self, h: Handle, operation: Operation, token: Token) -> Result { let r = self.resource(h)?; if r.closing.is_some() || !matches!(r.kind, Kind::Socket) { return Err(Error::new(ErrorKind::InvalidInput)); } + let reserve = usize::from(matches!( + operation, + Operation::Accept { .. } | Operation::RecvHandle + )); + // Only a reserving operation is gated: reads, writes and everything else + // on an existing handle must still work at the ceiling. + if reserve != 0 && self.handles.remaining() < self.reserved_handles + reserve { + return Err(Error::new(ErrorKind::ResourceLimit)); + } let op = self.new_op(Some(h), token)?; + self.reserved_handles += reserve; + self.ops.get_mut(op.key).expect("new op").reserved_handles = reserve; + self.assert_reservations(); if let Err(e) = self.backend.submit(Request { op, handle: h, @@ -992,7 +1054,43 @@ impl Driver { } Ok(d) } + /// House a transport an accept has already taken from the kernel, spending + /// the slot that operation reserved when it was submitted. + /// + /// Releasing the reservation before creating the handle is what makes it + /// real: `new_handle` refuses slots that are still promised, so the only + /// thing that can spend this one is the accept that reserved it. A multishot + /// accept stays armed and reserves again for its next connection; if the + /// loop is at its ceiling it holds nothing, and its next connection is + /// refused the way a fresh submission would be. + fn attach_reserved( + &mut self, + transport: B::Detached, + id: OpId, + token: Token, + terminal: bool, + ) -> Result { + let held = self + .ops + .get_mut(id.key) + .map_or(0, |op| std::mem::take(&mut op.reserved_handles)); + self.reserved_handles -= held; + let attached = self.attach(transport, token); + if !terminal && held != 0 && self.handles.remaining() > self.reserved_handles { + self.reserved_handles += held; + if let Some(op) = self.ops.get_mut(id.key) { + op.reserved_handles = held; + } else { + self.reserved_handles -= held; + } + } + self.assert_reservations(); + attached + } /// Register an owning transport on this loop; failure drops the rejected transport. + /// + /// Reports `ResourceLimit` at the handle ceiling, and also when the only + /// remaining slots are promised to accepts this loop has already armed. pub fn attach(&mut self, d: B::Detached, _token: Token) -> Result { let h = self.new_handle(Kind::Socket)?; if let Err(e) = self.backend.attach(h, d) { @@ -1371,22 +1469,27 @@ impl Driver { Ok(Outcome::Resolved(addresses)) => OpResult::Resolved(addresses), Ok(Outcome::Exited(status)) => OpResult::Exited(status), Ok(Outcome::Signal(signal)) => OpResult::Signal(signal), - Ok(Outcome::PipeAccepted(d)) => match self.attach(d, op.token) { - Ok(conn) => OpResult::PipeAccepted { conn }, - Err(e) => OpResult::Err(e), - }, - Ok(Outcome::HandleReceived(d)) => match self.attach(d, op.token) { - Ok(handle) => OpResult::HandleReceived { handle }, - Err(e) => OpResult::Err(e), - }, + Ok(Outcome::PipeAccepted(d)) => { + match self.attach_reserved(d, e.op, op.token, e.terminal) { + Ok(conn) => OpResult::PipeAccepted { conn }, + Err(e) => OpResult::Err(e), + } + } + Ok(Outcome::HandleReceived(d)) => { + match self.attach_reserved(d, e.op, op.token, e.terminal) { + Ok(handle) => OpResult::HandleReceived { handle }, + Err(e) => OpResult::Err(e), + } + } Ok(Outcome::HandleSent) => OpResult::HandleSent, Err(e) => OpResult::Err(e), Ok(Outcome::Connected) => OpResult::Connected, - Ok(Outcome::Accepted { transport, peer }) => match self.attach(transport, op.token) - { - Ok(conn) => OpResult::Accepted { conn, peer }, - Err(e) => OpResult::Err(e), - }, + Ok(Outcome::Accepted { transport, peer }) => { + match self.attach_reserved(transport, e.op, op.token, e.terminal) { + Ok(conn) => OpResult::Accepted { conn, peer }, + Err(e) => OpResult::Err(e), + } + } Ok(Outcome::Read { n, lease }) => OpResult::Read { n, lease }, Ok(Outcome::Eof) => OpResult::Eof, Ok(Outcome::Wrote(n)) => OpResult::Wrote(n), diff --git a/crates/turnloop/src/fs/service.rs b/crates/turnloop/src/fs/service.rs index e8a8b45..a98e9db 100644 --- a/crates/turnloop/src/fs/service.rs +++ b/crates/turnloop/src/fs/service.rs @@ -6,6 +6,7 @@ //! Only the head of a handle's FIFO is ever on the pool; a request needing a //! pooled lease waits (without spinning) until one is available. use super::{FileMetadata, FsOutput, FsRequest}; +use crate::slots::{Slots, page_reserve}; use crate::{ BufLease, BufferPool, Config, Error, ErrorKind, Handle, IoBufMut, OpId, PoolConfig, ReadBuf, Result, @@ -145,16 +146,37 @@ struct Active { } pub(crate) struct Service { - slots: Vec>, - objects: Vec>>, - active: Vec>, - heads: Vec>, - tails: Vec>, + /// Per-operation job slots, built when an operation index is first used. + slots: Slots>, + /// Per-handle object cells, built when a handle index is first used. + objects: Slots>>, + active: Slots, + heads: Slots, + tails: Slots, waiting: VecDeque, shared: Arc, pool: BufferPool, config: PoolConfig, } +/// Darwin's std mutex allocates its storage on first lock, so both locks are +/// taken here: publishing a job later uses only storage reserved with the slot. +fn build_slot(shared: &Arc) -> Arc { + let slot = Arc::new(Slot { + job: Mutex::new(None), + metadata: Mutex::new(None), + cancel: AtomicBool::new(false), + shared: shared.clone(), + }); + drop(lock(&slot.job)); + drop(lock(&slot.metadata)); + slot +} +/// See [`build_slot`]: the cell's lock storage is reserved with the cell. +fn build_object() -> Arc> { + let object = Arc::new(Mutex::new(Object::Empty)); + drop(lock(&object)); + object +} impl Service { pub fn new(config: &Config, work: Arc, pool: BufferPool) -> Self { let shared = Arc::new(Shared { @@ -164,38 +186,31 @@ impl Service { }); // Darwin's std mutex allocates its storage on first lock: do it at setup. drop(lock(&shared.running)); - let slots = (0..config.max_operations) - .map(|_| { - let slot = Arc::new(Slot { - job: Mutex::new(None), - metadata: Mutex::new(None), - cancel: AtomicBool::new(false), - shared: shared.clone(), - }); - drop(lock(&slot.job)); - drop(lock(&slot.metadata)); - slot - }) - .collect(); - let objects = (0..config.max_handles) - .map(|_| { - let object = Arc::new(Mutex::new(Object::Empty)); - drop(lock(&object)); - object - }) - .collect(); Self { - slots, - objects, - active: (0..config.max_operations).map(|_| None).collect(), - heads: vec![None; config.max_handles], - tails: vec![None; config.max_handles], - waiting: VecDeque::with_capacity(config.max_operations), + slots: Slots::filled(config.max_operations, || build_slot(&shared)), + objects: Slots::filled(config.max_handles, build_object), + active: Slots::new(config.max_operations), + heads: Slots::new(config.max_handles), + tails: Slots::new(config.max_handles), + waiting: VecDeque::with_capacity(page_reserve(config.max_operations)), shared, pool, config: config.blocking_pool, } } + /// This operation's job slot, built on first use. + /// + /// Darwin's std mutex allocates its storage on first lock, so both locks are + /// taken here: the slot's storage is reserved when the slot is built, never + /// on a later publication from the loop thread. + fn slot(&mut self, i: usize) -> &Arc { + let shared = self.shared.clone(); + self.slots.get_or_insert_with(i, || build_slot(&shared)) + } + /// This handle's object cell, built on first use. See [`Service::slot`]. + fn object(&mut self, i: usize) -> &Arc> { + self.objects.get_or_insert_with(i, build_object) + } /// Accept a request, or reject it without retaining or touching its buffers. pub fn submit(&mut self, op: OpId, handle: Option, request: FsRequest) -> Result<()> { crate::blocking::reserve(self.config)?; @@ -246,12 +261,18 @@ impl Service { active.lease = Some(lease); } active.stage = Stage::Running; - let slot = &self.slots[i]; + let (op, request, handle) = ( + active.op, + active.request.take().expect("unstarted request"), + active.handle, + ); + let object = handle.map(|h| self.object(h.index()).clone()); + let slot = self.slot(i).clone(); slot.cancel.store(false, Ordering::Release); *lock(&slot.job) = Some(Job { - op: active.op, - request: active.request.take().expect("unstarted request"), - object: active.handle.map(|h| self.objects[h.index()].clone()), + op, + request, + object, }); *lock(&self.shared.running) += 1; crate::blocking::push_reserved(slot.clone()); @@ -292,7 +313,7 @@ impl Service { return; }; match active.stage { - Stage::Running => self.slots[i].cancel.store(true, Ordering::Release), + Stage::Running => self.slot(i).cancel.store(true, Ordering::Release), Stage::Withdrawn => {} Stage::Queued | Stage::Waiting => { let was_head = active.previous.is_none(); @@ -328,13 +349,13 @@ impl Service { self.start(next); } } - let metadata = lock(&self.slots[i].metadata).take(); + let metadata = lock(&self.slot(i).metadata).take(); (self.active[i].take().and_then(|a| a.lease), metadata) } /// Reset a released handle's state; a still-open descriptor closes here. pub fn release(&mut self, h: Handle) { debug_assert!(self.heads[h.index()].is_none()); - let object = std::mem::take(&mut *lock(&self.objects[h.index()])); + let object = std::mem::take(&mut *lock(self.object(h.index()))); drop(object); } } @@ -345,8 +366,11 @@ impl Drop for Service { let Some(active) = active else { continue }; match active.stage { Stage::Running => { - self.slots[i].cancel.store(true, Ordering::Release); - if lock(&self.slots[i].job).take().is_some() { + let slot = self.slots[i] + .as_ref() + .expect("a running request built its slot"); + slot.cancel.store(true, Ordering::Release); + if lock(&slot.job).take().is_some() { withdrawn += 1; } } diff --git a/crates/turnloop/src/lib.rs b/crates/turnloop/src/lib.rs index 5af95bc..b70b503 100644 --- a/crates/turnloop/src/lib.rs +++ b/crates/turnloop/src/lib.rs @@ -50,6 +50,7 @@ mod native; pub use native::*; mod completion; mod driver; +mod slots; mod table; #[doc(hidden)] pub mod timer; diff --git a/crates/turnloop/src/slots.rs b/crates/turnloop/src/slots.rs new file mode 100644 index 0000000..e573c4f --- /dev/null +++ b/crates/turnloop/src/slots.rs @@ -0,0 +1,367 @@ +//! Index-addressed storage that allocates in pages, so a capacity ceiling costs +//! nothing until it is used. +//! +//! A loop's capacities (`Config::max_handles`, `Config::max_operations`) name the +//! largest number of slots it may ever hold. Building every slot at construction +//! makes that number a preallocation instead of a ceiling, which forces hosts to +//! choose between refusing work and paying for a loop they mostly do not use. +//! +//! [`Slots`] keeps the same addressing — a slot is named by its index, exactly as +//! in the `Vec>` it replaces — and materialises pages on demand. Two +//! properties make that safe to substitute: +//! +//! * **A page never moves.** Each page is a separately allocated boxed slice, so +//! growth appends to the page directory and leaves every existing slot at its +//! address. Backends that hand a slot's address to the kernel (the IOCP +//! `OVERLAPPED` slab) depend on this; a `Vec` that reallocates on growth does +//! not have it. +//! * **A page boundary is invisible to a key.** An index means the same slot +//! before and after growth, so generations, handles and operation ids stay +//! valid across it. +//! +//! Pages are a dense prefix: page `n` exists only if pages `0..n` do. Slot +//! indices are handed out lowest-free-first, so the materialised prefix tracks +//! the loop's high-water mark rather than its ceiling. +//! +//! Page zero is built at construction. A loop that exists will use its first +//! slots, and building them with it keeps the reserve the allocation gates +//! check: a loop at its high-water mark allocates nothing per turn, and only +//! passing a new high-water mark builds a page. +use std::ops::{Index, IndexMut}; + +/// Slots per page. A power of two so the index split is a shift and a mask. +/// +/// The page is the unit of growth, so this trades the per-loop floor (one page +/// of every paged structure) against how often a growing loop allocates. At 64, +/// the largest paged element in the crate keeps a page under 16 KiB. +pub(crate) const PAGE: usize = 64; +const SHIFT: u32 = PAGE.trailing_zeros(); +const MASK: usize = PAGE - 1; + +/// One page's worth of a ceiling. +/// +/// Queues bounded by a capacity reserved that whole capacity, which made the +/// ceiling a preallocation for the same reason the slot arrays did. They reserve +/// this instead and grow from it, so the first page of work still costs no +/// allocation and the reserve no longer scales with the ceiling. +pub(crate) fn page_reserve(ceiling: usize) -> usize { + PAGE.min(ceiling) +} + +/// Paged, index-addressed storage for at most `ceiling` values. +/// +/// Substitutable for `Vec>` at the call site: [`Index`], [`IndexMut`], +/// [`get`](Self::get), [`get_mut`](Self::get_mut) and [`len`](Self::len) all +/// report what the flat vector reported. Indexing for write materialises the +/// page; indexing for read never allocates. +pub(crate) struct Slots { + /// Materialised pages, in index order. Page `p` covers `p * PAGE..(p + 1) * PAGE`. + pages: Vec]>>, + /// The configured ceiling. Indices at or above it are out of bounds. + ceiling: usize, +} + +impl Slots { + /// A vacant slot to hand out for reads of an unmaterialised index. Borrowing + /// a `None` const promotes to `'static`, so this costs no storage and no + /// allocation. + const VACANT: Option = None; + + /// Storage for at most `ceiling` slots, with page zero built. + pub fn new(ceiling: usize) -> Self { + let mut slots = Self { + pages: Vec::new(), + ceiling, + }; + slots.grow(); + slots + } + + /// Storage for at most `ceiling` slots, with page zero built *and filled*. + /// + /// For slabs whose elements were all constructed up front so that later use + /// could not allocate — a per-operation job slot holding a mutex whose + /// storage Darwin allocates on first lock. Page zero's elements are built + /// here; later pages build theirs in + /// [`get_or_insert_with`](Self::get_or_insert_with) as the ceiling is used. + // Serves the native pool slabs (`fs::Service`, `backend::files`), which wasm + // targets do not build. + #[cfg_attr(target_arch = "wasm32", allow(dead_code))] + pub fn filled(ceiling: usize, mut build: impl FnMut() -> T) -> Self { + let mut slots = Self::new(ceiling); + for slot in slots.iter_mut() { + *slot = Some(build()); + } + slots + } + + /// Build the next page, clamped to the ceiling. Refused at the ceiling. + fn grow(&mut self) -> bool { + let base = self.pages.len() * PAGE; + if base >= self.ceiling { + return false; + } + let len = PAGE.min(self.ceiling - base); + self.pages.push( + (0..len) + .map(|_| None) + .collect::>() + .into_boxed_slice(), + ); + true + } + + /// The ceiling, matching the length of the `Vec>` this replaces. + pub fn len(&self) -> usize { + self.ceiling + } + + /// The materialised prefix: every index at or above it is vacant. + /// + /// Scans that walked `0..len()` looking for occupied slots want this instead, + /// so the cost tracks what the loop has used rather than what it may use. + pub fn materialised(&self) -> usize { + (self.pages.len() * PAGE).min(self.ceiling) + } + + /// The vacant slot for an unmaterialised read, or `None` past the ceiling. + fn read(&self, i: usize) -> Option<&Option> { + if i >= self.len() { + return None; + } + Some(match self.pages.get(i >> SHIFT) { + Some(page) => &page[i & MASK], + None => &Self::VACANT, + }) + } + + /// Materialise pages through `i`'s, so the slot can be written. + /// + /// Pages are a dense prefix, so reaching a high index materialises the pages + /// below it. Slot indices are handed out lowest-free-first, so that only + /// happens when a host addresses a slot the loop never allocated. + fn materialise(&mut self, i: usize) -> &mut Option { + assert!( + i < self.len(), + "slot {i} is beyond the configured ceiling {}", + self.ceiling + ); + let page = i >> SHIFT; + while self.pages.len() <= page { + assert!(self.grow(), "the ceiling admits index {i}"); + } + debug_assert!(i < self.materialised(), "pages are a dense prefix"); + &mut self.pages[page][i & MASK] + } + + /// The slot at `i`, or `None` past the ceiling. Never allocates. + pub fn get(&self, i: usize) -> Option<&Option> { + self.read(i) + } + + /// The slot at `i` for writing, or `None` past the ceiling. Materialises the page. + pub fn get_mut(&mut self, i: usize) -> Option<&mut Option> { + (i < self.len()).then(|| self.materialise(i)) + } + + /// The value at `i`, building it on first use. + /// + /// For slabs whose slots are always present once reached — a per-operation + /// job slot, a per-handle object cell — where the flat vector built every + /// element at construction. The element is built when its index is first + /// reached rather than when the loop is created. + // Serves the native pool slabs (`fs::Service`, `backend::files`), which wasm + // targets do not build. + #[cfg_attr(target_arch = "wasm32", allow(dead_code))] + pub fn get_or_insert_with(&mut self, i: usize, build: impl FnOnce() -> T) -> &mut T { + self.materialise(i).get_or_insert_with(build) + } + + /// The materialised prefix, in index order. Every slot beyond it is vacant, + /// so `flatten`, `any(Option::is_some)` and `position` see the same sequence + /// the flat vector produced, and `enumerate` yields the same indices. + /// + /// Returns a named type rather than `impl Iterator` on purpose: an opaque + /// return type is assumed to have drop glue, which holds the borrow of the + /// whole owner to the end of the loop and rejects bodies that a slice + /// iterator allows. + pub fn iter(&self) -> Iter<'_, T> { + self.into_iter() + } + + /// The materialised prefix for mutation. See [`iter`](Self::iter). + // Serves the native pool slabs (`fs::Service`, `backend::files`), which wasm + // targets do not build. + #[cfg_attr(target_arch = "wasm32", allow(dead_code))] + pub fn iter_mut(&mut self) -> IterMut<'_, T> { + self.into_iter() + } +} + +/// Borrowed iteration over the materialised prefix. See [`Slots::iter`]. +pub(crate) type Iter<'a, T> = std::iter::Flatten]>>>; +/// Mutable iteration over the materialised prefix. See [`Slots::iter`]. +pub(crate) type IterMut<'a, T> = std::iter::Flatten]>>>; + +impl<'a, T> IntoIterator for &'a Slots { + type Item = &'a Option; + type IntoIter = Iter<'a, T>; + /// The materialised prefix. See [`Slots::iter`]. + fn into_iter(self) -> Self::IntoIter { + self.pages.iter().flatten() + } +} + +impl<'a, T> IntoIterator for &'a mut Slots { + type Item = &'a mut Option; + type IntoIter = IterMut<'a, T>; + /// The materialised prefix. See [`Slots::iter`]. + fn into_iter(self) -> Self::IntoIter { + self.pages.iter_mut().flatten() + } +} + +impl IntoIterator for Slots { + type Item = Option; + type IntoIter = std::iter::FlatMap< + std::vec::IntoIter]>>, + std::vec::Vec>, + fn(Box<[Option]>) -> std::vec::Vec>, + >; + /// The materialised prefix, by value. See [`Slots::iter`]. + fn into_iter(self) -> Self::IntoIter { + self.pages.into_iter().flat_map(Vec::from) + } +} + +impl Index for Slots { + type Output = Option; + fn index(&self, i: usize) -> &Option { + self.read(i) + .unwrap_or_else(|| panic!("slot {i} is beyond the configured ceiling {}", self.ceiling)) + } +} + +impl IndexMut for Slots { + fn index_mut(&mut self, i: usize) -> &mut Option { + self.materialise(i) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_unmaterialised_slot_reads_vacant_without_allocating() { + let slots: Slots = Slots::new(1_000_000); + assert_eq!( + slots.materialised(), + PAGE, + "exactly page zero is built at construction, whatever the ceiling" + ); + assert_eq!(slots.len(), 1_000_000, "the ceiling is still the length"); + assert!(slots[999_999].is_none()); + assert!(slots.get(999_999).expect("within the ceiling").is_none()); + assert_eq!(slots.materialised(), PAGE, "reading materialises nothing"); + assert!(slots.get(1_000_000).is_none(), "past the ceiling"); + } + + #[test] + fn growth_leaves_every_earlier_slot_at_its_address() { + let mut slots: Slots = Slots::new(4096); + let mut addresses = Vec::new(); + for i in 0..PAGE { + slots[i] = Some(i as u32); + addresses.push(std::ptr::from_ref(&slots[i])); + } + assert_eq!(slots.materialised(), PAGE); + for i in PAGE..PAGE * 8 { + slots[i] = Some(i as u32); + } + assert_eq!(slots.materialised(), PAGE * 8, "seven more pages"); + for (i, address) in addresses.iter().enumerate() { + assert_eq!(slots[i], Some(i as u32), "value survived growth"); + assert_eq!( + std::ptr::from_ref(&slots[i]), + *address, + "slot {i} moved across a page boundary" + ); + } + for i in 0..PAGE * 8 { + assert_eq!(slots[i], Some(i as u32), "index still names its slot"); + } + } + + #[test] + fn iteration_matches_the_flat_vector_over_occupied_slots() { + let mut slots: Slots = Slots::new(4096); + for i in [0, 5, 63, 64, 130] { + slots[i] = Some(i as u32); + } + let seen: Vec = slots.iter().flatten().copied().collect(); + assert_eq!(seen, vec![0, 5, 63, 64, 130]); + let indices: Vec = slots + .iter() + .enumerate() + .filter_map(|(i, slot)| slot.map(|_| i)) + .collect(); + assert_eq!(indices, vec![0, 5, 63, 64, 130], "enumerate keeps indices"); + assert!(slots.iter().any(Option::is_some)); + for slot in slots.iter_mut() { + *slot = None; + } + assert!(slots.iter().all(Option::is_none)); + } + + #[test] + fn a_page_holds_exactly_its_own_index_range() { + let mut slots: Slots = Slots::new(PAGE * 3); + assert_eq!(slots.materialised(), PAGE); + slots[PAGE * 3 - 1] = Some(7); + assert_eq!(slots.materialised(), PAGE * 3); + assert_eq!(slots[PAGE * 3 - 1], Some(7)); + assert!( + slots.iter().filter(|s| s.is_some()).count() == 1, + "materialising a page leaves its other slots vacant" + ); + } + + #[test] + #[should_panic(expected = "beyond the configured ceiling")] + fn writing_past_the_ceiling_panics_like_a_flat_vector() { + let mut slots: Slots = Slots::new(8); + slots[8] = Some(1); + } + + #[test] + fn a_ceiling_below_a_page_builds_only_the_slots_it_allows() { + let mut slots: Slots = Slots::new(3); + assert_eq!(slots.materialised(), 3, "page zero stops at the ceiling"); + assert_eq!(slots.iter().count(), 3); + slots[2] = Some(9); + assert_eq!(slots[2], Some(9)); + } + + #[test] + fn filled_builds_page_zeros_elements_and_later_pages_on_demand() { + let mut built = 0; + let mut slots: Slots = Slots::filled(4096, || { + built += 1; + Some(built).map(|n| n as u32).expect("counter") + }); + assert_eq!(built, PAGE, "exactly page zero is filled at construction"); + assert!(slots[0].is_some()); + assert!(slots[PAGE - 1].is_some()); + assert!(slots[PAGE].is_none(), "a later page starts vacant"); + let value = *slots.get_or_insert_with(PAGE, || 4242); + assert_eq!(value, 4242); + } + + #[test] + #[should_panic(expected = "beyond the configured ceiling")] + fn reading_past_the_ceiling_panics_like_a_flat_vector() { + let slots: Slots = Slots::new(8); + let _ = &slots[8]; + } +} diff --git a/crates/turnloop/src/table.rs b/crates/turnloop/src/table.rs index d3aa2fa..6da1def 100644 --- a/crates/turnloop/src/table.rs +++ b/crates/turnloop/src/table.rs @@ -1,44 +1,99 @@ -//! Fixed-capacity generational storage. Exhausted generations retire their slot. +//! Paged generational storage. Exhausted generations retire their slot. +//! +//! The capacity passed to [`Table::new`] is a ceiling, not a preallocation: the +//! table builds page zero with itself and another page whenever the free list +//! empties and the ceiling still allows it, so an idle table costs one page +//! whatever its ceiling. +//! Slots are named by index and pages never move, so growth leaves every live +//! entry where it was and invalidates no key. See [`crate::slots`] for the +//! addressing properties this relies on. +use crate::slots::PAGE; struct Slot { generation: u32, value: Option, } pub(crate) struct Table { - slots: Vec>, + /// Materialised pages, in index order. Page `p` covers `p * PAGE..(p + 1) * PAGE`. + pages: Vec]>>, + /// Reusable indices, most recently retired last. free: Vec, + /// Largest number of slots this table may ever hold. + ceiling: usize, + /// Occupied slots, so remaining capacity does not need a scan. + live: usize, } impl Table { pub fn new(capacity: usize) -> Self { - Self { - slots: (0..capacity) + let mut table = Self { + pages: Vec::new(), + free: Vec::new(), + ceiling: capacity, + live: 0, + }; + // Page zero is built with the table: a loop that exists will use its + // first slots, and the allocation gates require that use to be free. + table.grow(); + table + } + /// Slots neither occupied nor retired: what `insert` can still hand out. + /// + /// Counts capacity the table has not built yet, because a page is built on + /// demand. Retired slots (a saturated generation) are excluded, since they + /// are counted live and never return to the free list. + pub fn remaining(&self) -> usize { + self.ceiling - self.live + } + /// Build the next page and offer its indices, lowest first. Refused at the ceiling. + fn grow(&mut self) -> bool { + let base = self.pages.len() * PAGE; + if base >= self.ceiling { + return false; + } + let len = PAGE.min(self.ceiling - base); + self.pages.push( + (0..len) .map(|_| Slot { generation: 1, value: None, }) - .collect(), - free: (0..capacity).rev().collect(), - } + .collect::>() + .into_boxed_slice(), + ); + // `pop` takes the last element, so push descending to hand out ascending. + self.free.extend((base..base + len).rev()); + true + } + fn slot(&self, i: usize) -> Option<&Slot> { + self.pages.get(i / PAGE)?.get(i % PAGE) + } + fn slot_mut(&mut self, i: usize) -> Option<&mut Slot> { + self.pages.get_mut(i / PAGE)?.get_mut(i % PAGE) } pub fn insert(&mut self, value: T) -> Option { + if self.free.is_empty() && !self.grow() { + return None; + } let i = self.free.pop()?; - self.slots[i].value = Some(value); - Some((u64::from(self.slots[i].generation) << 32) | i as u64) + self.live += 1; + let slot = self.slot_mut(i).expect("free index names a built slot"); + slot.value = Some(value); + Some((u64::from(slot.generation) << 32) | i as u64) } pub fn get(&self, key: u64) -> Option<&T> { - let s = self.slots.get(key as u32 as usize)?; + let s = self.slot(key as u32 as usize)?; (s.generation == (key >> 32) as u32) .then_some(s.value.as_ref()) .flatten() } pub fn get_mut(&mut self, key: u64) -> Option<&mut T> { - let s = self.slots.get_mut(key as u32 as usize)?; + let s = self.slot_mut(key as u32 as usize)?; (s.generation == (key >> 32) as u32) .then_some(s.value.as_mut()) .flatten() } pub fn remove(&mut self, key: u64) -> Option { let i = key as u32 as usize; - let s = self.slots.get_mut(i)?; + let s = self.slot_mut(i)?; if s.generation != (key >> 32) as u32 { return None; } @@ -46,6 +101,7 @@ impl Table { if let Some(generation) = s.generation.checked_add(1) { s.generation = generation; self.free.push(i); + self.live -= 1; } Some(value) } @@ -64,4 +120,83 @@ mod tests { assert_eq!(t.remove(a), None); assert_eq!(t.get(b), Some(&34)); } + #[test] + fn a_key_taken_before_growth_still_names_its_value_after_it() { + let mut t = Table::new(4096); + // Keys from the first page, taken before any later page exists. + let early: Vec = (0..PAGE).map(|i| t.insert(i).expect("capacity")).collect(); + assert_eq!(t.remaining(), 4096 - PAGE); + // Force seven more pages, which is where a flat vector would reallocate. + let late: Vec = (PAGE..PAGE * 8) + .map(|i| t.insert(i).expect("capacity")) + .collect(); + for (i, key) in early.iter().enumerate() { + assert_eq!( + t.get(*key), + Some(&i), + "key {key} lost its value across growth" + ); + assert_eq!(*key as u32 as usize, i, "key {key} changed slot"); + } + for (n, key) in late.iter().enumerate() { + assert_eq!(t.get(*key), Some(&(PAGE + n))); + } + // And the earlier keys still remove exactly their own value. + for (i, key) in early.iter().enumerate() { + assert_eq!(t.remove(*key), Some(i)); + } + } + #[test] + fn the_ceiling_is_a_ceiling_and_a_partial_page_does_not_exceed_it() { + // Not a multiple of PAGE: the last page must stop at the ceiling. + let mut t: Table = Table::new(PAGE + 3); + let keys: Vec = (0..PAGE + 3) + .map(|i| t.insert(i).expect("within the ceiling")) + .collect(); + assert_eq!(t.remaining(), 0); + assert!(t.insert(0).is_none(), "the ceiling refuses one more"); + assert!( + keys.iter().all(|k| (*k as u32 as usize) < PAGE + 3), + "no key addresses a slot past the ceiling" + ); + t.remove(keys[0]).expect("live"); + assert_eq!(t.remaining(), 1); + assert!(t.insert(99).is_some(), "a freed slot is offered again"); + } + #[test] + fn growth_builds_one_page_at_a_time() { + let mut t: Table = Table::new(1_000_000); + assert_eq!(t.pages.len(), 1, "page zero comes with the table"); + t.insert(0).expect("capacity"); + assert_eq!(t.pages.len(), 1, "one page serves the first insert"); + for i in 1..PAGE { + t.insert(i).expect("capacity"); + } + assert_eq!( + t.pages.len(), + 1, + "the page is filled before the next is built" + ); + t.insert(PAGE).expect("capacity"); + assert_eq!(t.pages.len(), 2); + } + #[test] + fn a_retired_slot_is_not_counted_as_remaining_capacity() { + let mut t: Table = Table::new(2); + let key = t.insert(0).expect("capacity"); + assert_eq!(t.remaining(), 1); + // Drive that slot's generation to saturation: it retires instead of + // returning to the free list, so the ceiling permanently loses a slot. + let i = key as u32 as usize; + t.slot_mut(i).expect("built").generation = u32::MAX; + let saturated = (u64::from(u32::MAX) << 32) | i as u64; + assert_eq!(t.remove(saturated), Some(0)); + assert_eq!(t.remaining(), 1, "the retired slot stays spent"); + assert!(t.insert(1).is_some(), "the other slot is still available"); + assert_eq!(t.remaining(), 0); + assert!( + t.insert(2).is_none(), + "a retired slot is never offered again" + ); + } } diff --git a/crates/turnloop/src/timer/heap.rs b/crates/turnloop/src/timer/heap.rs index 9d99f26..9eb6f89 100644 --- a/crates/turnloop/src/timer/heap.rs +++ b/crates/turnloop/src/timer/heap.rs @@ -1,15 +1,21 @@ use super::{Entry, Instant}; -/// Preallocated four-ary heap with an index for immediate O(log n) cancellation. -/// Keys encode the slot index in their low 32 bits; one deadline per slot. +use crate::slots::Slots; +/// Four-ary heap with an index for immediate O(log n) cancellation. Keys encode +/// the slot index in their low 32 bits; one deadline per slot. +/// +/// `capacity` is a ceiling, not a preallocation: the position index is built in +/// pages as slots are used, and the heap itself grows with the timers actually +/// armed, so a loop that arms none costs nothing for a large ceiling. pub struct Heap { heap: Vec, - positions: Vec, + /// Each armed slot's position in `heap`; vacant means that slot has no timer. + positions: Slots, } impl Heap { pub fn new(capacity: usize) -> Self { Self { - heap: Vec::with_capacity(capacity), - positions: vec![usize::MAX; capacity], + heap: Vec::with_capacity(crate::slots::PAGE.min(capacity)), + positions: Slots::new(capacity), } } pub fn len(&self) -> usize { @@ -23,23 +29,23 @@ impl Heap { } pub fn insert(&mut self, id: u64, at: Instant) { let i = id as u32 as usize; - assert_eq!(self.positions[i], usize::MAX, "one timer per slot"); - self.positions[i] = self.heap.len(); + assert!(self.positions[i].is_none(), "one timer per slot"); + self.positions[i] = Some(self.heap.len()); self.heap.push(Entry { id, at }); self.up(self.heap.len() - 1); } pub fn cancel(&mut self, id: u64) -> bool { let i = id as u32 as usize; - let Some(&p) = self.positions.get(i) else { + let Some(&Some(p)) = self.positions.get(i) else { return false; }; - if p == usize::MAX || self.heap[p].id != id { + if self.heap[p].id != id { return false; } self.heap.swap_remove(p); - self.positions[i] = usize::MAX; + self.positions[i] = None; if p < self.heap.len() { - self.positions[self.heap[p].id as u32 as usize] = p; + self.positions[self.heap[p].id as u32 as usize] = Some(p); if p > 0 && self.heap[p] < self.heap[(p - 1) / 4] { self.up(p); } else { @@ -58,8 +64,8 @@ impl Heap { } fn swap(&mut self, a: usize, b: usize) { self.heap.swap(a, b); - self.positions[self.heap[a].id as u32 as usize] = a; - self.positions[self.heap[b].id as u32 as usize] = b; + self.positions[self.heap[a].id as u32 as usize] = Some(a); + self.positions[self.heap[b].id as u32 as usize] = Some(b); } fn up(&mut self, mut p: usize) { while p > 0 { diff --git a/crates/turnloop/src/types.rs b/crates/turnloop/src/types.rs index 7ffe411..696e40c 100644 --- a/crates/turnloop/src/types.rs +++ b/crates/turnloop/src/types.rs @@ -16,7 +16,7 @@ macro_rules! id { pub(crate) key: u64, } impl $name { - /// Slot index, for preallocated backend operation/resource storage. + /// Slot index, for paged backend operation/resource storage. pub fn index(self) -> usize { self.key as u32 as usize } @@ -166,8 +166,18 @@ impl Timeout { /// Fixed loop capacities and shared blocking-pool configuration. pub struct Config { /// Maximum simultaneously allocated handles, including closed results awaiting delivery. + /// + /// A ceiling, not a reservation: slot storage is built in pages as the loop's + /// high-water mark rises, so an idle loop costs the same whatever this is and + /// a large value is affordable for a loop per agent. Reaching it refuses an + /// accept at submission, before the kernel is asked for a connection, so + /// pending connections wait in the listener's backlog rather than being + /// accepted and destroyed. pub max_handles: usize, /// Maximum outstanding operations, including terminal results awaiting delivery. + /// + /// A ceiling paged like [`Config::max_handles`], except for the Windows + /// `OVERLAPPED` slab, which stays contiguous. pub max_operations: usize, /// Native event budget and per-source completion reserve. pub events_per_turn: usize,