From dc7e95ddda0e20a74dbb3c14d8a38ffaef7fe665 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 11:47:29 +0200 Subject: [PATCH 1/3] Page the slot tables so a capacity ceiling costs nothing idle `Config::max_handles` read like a ceiling and behaved like a preallocation: `Table::new` built every slot and the whole free list up front, and so did the timer index and every backend array addressed by a slot index. A loop cost what it might ever hold, which forced the ceiling to stay small, and a small ceiling refused connections. Slots now live in fixed-size pages built on demand: page zero with the loop, another when the high-water mark crosses into it. Substituting pages for a flat vector is safe because 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. An idle loop's cost is now identical from 1K to 1M handles: 784 allocations and 4,999 KB either way, against 22,843/15.1 MB and 2,117,947/1,323 MB before. On an HTTP fixture holding 5,000 connections at `max_handles: 16384`, peak RSS falls from 56.3 MB to 12.3 MB, and loop construction from 52 MB to 6.2 MB. Reaching the ceiling is now backpressure rather than refusal. An operation whose completion creates a handle reserves its slot when it is submitted, and `new_handle` will not spend a slot another operation is holding, 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, before the kernel is asked, and the pending connection stays in the listener's backlog. Measured at `max_handles: 512`: before, two accepts completed with `ResourceLimit` after the kernel had already handed the connections over, and they were destroyed; after, one submission is refused and no accept completes with an error. Two structures stay contiguous and sized by their ceiling, and say why: the Windows OVERLAPPED slab, whose reverse map from a completion packet's pointer to an operation index needs one allocation, and the lock-free cross-thread result rings, whose capacity is also their backpressure bound. `pooled_buffers` is the remaining per-loop cost that scales with configuration, tracked in #43. Every existing allocation gate passes unmodified, which is the evidence that steady-state behaviour is unchanged. New gates pin that a loop built with a 1024x larger ceiling makes the same allocations and requests the same bytes, and that a loop pinned 500 handles above page zero turns 1000 times without allocating; both weigh bytes as well as counting, because one oversized reservation is a single allocation, which is how the cost used to hide. New contract tests cover handles surviving growth and the accept reservation on every backend. Closes #75 --- DESIGN.md | 40 +- crates/turnloop-contract/src/lib.rs | 279 +++++++++++++ crates/turnloop-contract/tests/allocations.rs | 112 +++++- crates/turnloop-contract/tests/wasi.rs | 4 + crates/turnloop-contract/tests/windows.rs | 3 + crates/turnloop/Cargo.toml | 2 +- crates/turnloop/src/backend/files.rs | 83 ++-- crates/turnloop/src/backend/fsevents.rs | 4 +- crates/turnloop/src/backend/iocp/mod.rs | 22 +- crates/turnloop/src/backend/iocp/services.rs | 5 +- crates/turnloop/src/backend/iocp/watch.rs | 17 +- crates/turnloop/src/backend/services.rs | 25 +- crates/turnloop/src/backend/unix.rs | 13 +- crates/turnloop/src/backend/wasi_fs.rs | 25 +- crates/turnloop/src/backend/wasi_p2.rs | 17 +- crates/turnloop/src/backend/wasi_p3.rs | 17 +- crates/turnloop/src/backend/watch.rs | 17 +- crates/turnloop/src/backend/web.rs | 9 +- crates/turnloop/src/driver.rs | 115 +++++- crates/turnloop/src/fs/service.rs | 104 +++-- crates/turnloop/src/lib.rs | 1 + crates/turnloop/src/slots.rs | 367 ++++++++++++++++++ crates/turnloop/src/table.rs | 159 +++++++- crates/turnloop/src/timer/heap.rs | 32 +- crates/turnloop/src/types.rs | 12 +- 25 files changed, 1288 insertions(+), 196 deletions(-) create mode 100644 crates/turnloop/src/slots.rs 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..ddf5613 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, @@ -395,13 +396,13 @@ unsafe impl Backend for WasiP2 { .and_then(|n| n.checked_add(1)) .ok_or(Error::new(ErrorKind::ResourceLimit))?; 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)), handles: Vec::with_capacity(polls), owners: Vec::with_capacity(polls), - lookups: (0..config.max_operations).map(|_| None).collect(), + lookups: Slots::new(config.max_operations), indices: Vec::with_capacity(polls), poll_storage: vec![0; polls], scratch: vec![0; 16400], 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..5ced4b3 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,11 +161,15 @@ 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, @@ -178,7 +191,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 { @@ -219,6 +242,7 @@ impl Driver { job_cancel: None, external_wait: false, fs: false, + reserved_handles: 0, native, previous, next: None, @@ -254,6 +278,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 +339,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 +809,32 @@ 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; if let Err(e) = self.backend.submit(Request { op, handle: h, @@ -992,7 +1037,42 @@ 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; + } + } + 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 +1451,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, From 5da24139caa0afcde569e3696634fb4357155b9f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 11:52:59 +0200 Subject: [PATCH 2/3] Grow WASI 0.2's poll batch with its pollables, not its ceiling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit wasi_p2 sized four poll-batch arrays at max_handles * 2 + max_operations + 1 and reserved all of it up front, which the new ceiling gate caught: a 1024x larger ceiling reserved 80,261,208 bytes against 4,837,464, or 72 bytes per handle of ceiling. Three of them are scratch cleared and refilled per poll, so they reserve a page and grow with the pollables actually subscribed. The fourth, poll_storage, is the canonical return arena that wasi:io/poll lowers its result into, and the arena asserts rather than reallocating, so it is resized to one u32 per subscribed pollable before the call — the exact bound the import can produce. The worst-case count is still computed, because an overflow there is a configuration this backend cannot serve. --- crates/turnloop/src/backend/wasi_p2.rs | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/crates/turnloop/src/backend/wasi_p2.rs b/crates/turnloop/src/backend/wasi_p2.rs index ddf5613..2770c03 100644 --- a/crates/turnloop/src/backend/wasi_p2.rs +++ b/crates/turnloop/src/backend/wasi_p2.rs @@ -389,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: 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(polls), - owners: Vec::with_capacity(polls), + handles: Vec::with_capacity(batch), + owners: Vec::with_capacity(batch), lookups: Slots::new(config.max_operations), - indices: Vec::with_capacity(polls), - poll_storage: vec![0; polls], + indices: Vec::with_capacity(batch), + poll_storage: vec![0; batch], scratch: vec![0; 16400], files: super::wasi_fs::Files::new(config, pool.clone()), pool, @@ -705,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` From 7493e0be8c7c1107308e3ab0883b734b38f6eb7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 16 Sep 2026 12:01:52 +0200 Subject: [PATCH 3/3] Assert that promised handle slots are a subset of free ones The accept reservation is only worth something if the two counters cannot drift: a slot promised to an armed accept has to still be free when the connection arrives. Every path that moves either side already keeps that -- a reserving submission checks before promising, new_handle refuses to spend a promised slot, and a delivery releases its promise before taking the slot -- but drift would be silent until a connection was destroyed for it, which is the failure this mechanism exists to remove. Checked in debug at the three points that move a counter. --- crates/turnloop/src/driver.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/crates/turnloop/src/driver.rs b/crates/turnloop/src/driver.rs index 5ced4b3..45303f4 100644 --- a/crates/turnloop/src/driver.rs +++ b/crates/turnloop/src/driver.rs @@ -175,6 +175,21 @@ impl Driver { _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, @@ -218,6 +233,7 @@ impl Driver { if matches!(kind, Kind::Socket | Kind::File) { self.refs += 1; } + self.assert_reservations(); Ok(Handle { owner: self.owner, key, @@ -835,6 +851,7 @@ impl Driver { 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, @@ -1067,6 +1084,7 @@ impl Driver { self.reserved_handles -= held; } } + self.assert_reservations(); attached } /// Register an owning transport on this loop; failure drops the rejected transport.