Compare commits

...
2 Commits
Author SHA1 Message Date
Claude 393cdd01f4 feat(select): fd arms in select + timed fd waits (RFC 008 phase 1)
FdArm composes fd readiness with channel arms on one wait epoch.
Selectable grows fallible sel_register and an eager-cleanup hook;
losing/stop-unwound/timed-out fd arms are unregistered (waiters entry
+ kernel ONESHOT) so the fd is never poisoned. try_select /
try_select_timeout surface registration errors (EBADF, EMFILE,
AlreadyExists) instead of RFC 008's permanently-ready lean, which
would busy-loop a healthy-but-unregistrable fd; select/select_timeout
stay infallible for channel-only arms. Adds wait_readable_timeout /
wait_writable_timeout as one-arm selects.

Known benign race (pre-existing, slightly widened): a queued FdReady
racing the cleanup DEL can spuriously wake a fresh waiter on that fd;
absorbed by select's defensive re-loop. Fixable by epoch-stamping
completions.
2026-06-11 11:28:58 +00:00
Claude 4e4b617559 docs(roadmap): restructure around the rq shootout verdict
Shootout (6d9f369 harness, 24-core sweep) landed in RFC 005's World 3:
queue variants indistinguishable in rq_runtime; per-wake latency is the
ceiling. Record the queue-topology closure (rq-mutex frozen as default,
benchmark-driven reopening only), define v0.9 as the wake-path latency
cycle (RFC 005 slot -> bench -> RFC 004 spinning workers -> bench +
interaction pass), and add per-switch cost to Later pending a profiling
spike + RFC.

Completed cycles compacted to the two trailing minors (v0.7, v0.8);
earlier history lives in git.
2026-06-10 21:58:40 +00:00
5 changed files with 792 additions and 326 deletions
+98 -298
View File
@@ -1,331 +1,131 @@
# smarm — Roadmap
## v0.4 — Actor primitives ✅ DONE
## Shipped (compacted — full cycle plans and deviation records live in git history)
### 1. Cooperative cancellation ✅ (`a8ddb4a`)
Stop flag on `Actor` behind `Arc<AtomicBool>`. Observation points:
`maybe_preempt`/`check!()` + the wakeup side of `park_current`/`yield_now`.
Sentinel = `StopSentinel` (zero-size), caught in trampoline → `Outcome::Stopped`.
`join()` on a stopped actor returns `Ok(())`.
Cycles before v0.7 (v0.4 actor primitives, v0.5 runtime decomposition &
pluggable run queue, v0.6 actor ergonomics): see `git log ROADMAP.md`.
Known gap (by design): a tight no-alloc loop without `check!()` cannot be
stopped, same as preemption.
### v0.7 — select, on epoch-stamped consuming wakes ✅
A 24-bit park-epoch packed into the slot word gives every wait an identity:
registrations carry `(pid, epoch)`, every successful wake **consumes** the
epoch, stale wakes die at one failed CAS. Subsumed the per-primitive wait
seqs (channel, mutex, timer); the only wildcard wake left is `request_stop`
(terminal). On top: `select`/`select_timeout` — ready-index wait over many
receivers, priority order, no cancellation pass. Loom theorems re-proved on
the new word. Notable deviations: `retire_wait` for the no-park exit path
(plan missed it); the Drop-guard deregistration was superseded by relaxed
single-receiver asserts; a closed arm is ready *forever* (documented gotcha).
Commits `4913835``00128f3`, `a0a93b6`.
### 2. one_for_all / rest_for_one + ordered shutdown ✅ (`351dc9c`)
`Strategy::{OneForOne,OneForAll,RestForOne}` via `.strategy()`. Triggering
child's `Restart` policy decides whether anything restarts; strategy decides
which siblings are cycled. Survivors stopped in reverse start order, awaited on
the existing `supervisor_channel` funnel, restarted in start order.
`Signal::Stopped(pid)` + `DownReason::Stopped` added (distinct from Exit).
Also fixed a latent bug (`e80334b`): orphaned timers were gating shutdown.
`live == 0` is now the shutdown condition; timer heap cleared on exit.
### 3. Links + trap_exit ✅ (`6581484`)
`Slot.links: Vec<Pid>` (bidirectional). On finalize: abnormal death →
`request_stop(peer)` unless peer traps, in which case deliver `ExitSignal`
message. Normal exit does not propagate. `trap_exit()` returns a dedicated
`Receiver<ExitSignal>` (distinct from monitor `Down` channel).
### 4. Selective receive ✅ (`03f3875`)
`Receiver::recv_match(pred)` + `try_recv_match`. Scans `VecDeque` front-to-back,
removes+returns first match, parks and re-scans on new arrivals. `pred` is
`Fn(&T) -> bool` (not `FnMut`; runs under the channel lock — keep it cheap).
### 5. Grab-bag ✅
- **demonitor** (`5181037`): `monitor()``Monitor { id, target, rx }`.
`demonitor(&m) -> Option<MonitorId>`. Dropped `Sender` released post-lock
(non-reentrant shared mutex discipline).
- **gen_server** (`55221e9`): `GenServer` trait, `call`/`cast` over one
`Envelope` inbox. `terminate` hook via drop guard. No `handle_info`, no call
timeout (deferred — needs cross-channel mailbox merge or per-recv deadline).
- **Named registry**: `register`/`whereis`/`send_by_name`**not yet done**.
HashMap in `SharedState`. No dependencies; can land any time.
- **arm-port branch**: aarch64 context-switch backend extracted into
`src/arch/`. Untested on hardware. Merge only after on-device validation.
### v0.8 — gen_server: handle_info / handle_down + io fd hygiene ✅
Spent `select` on the server loop: static info arms (`type Info`,
`ServerBuilder::with_info`) and dynamic monitor forwarding
(`ServerCtx`/`Watcher` + a control arm), priority downs → control → infos →
inbox. Closed the v0.2 fd hole: a drop guard in `wait_fd` DELs the kernel
registration on unwind (the leak was worse than documented — a stale waiters
entry permanently poisoned the fd). Deviation: the plain-inbox fast path
narrowed; servers holding a `Watcher` select forever.
Commits `e5d1b3b`, `24b95c9`, `f6969e5`.
---
## v0.5 — Runtime decomposition & run-path scaling ✅ DONE
## Decision record — queue topology 🔒 CLOSED (2026-06-10)
Goal: dismantle the single `Mutex<SharedState>` along the run path so the
runtime scales to dozens of cores, with crash isolation hardened so a torn
write or a stray cancellation can never poison shared runtime state.
The run-queue shootout (harness `6d9f369`, 24-core sweep 124 schedulers,
report `bench_report_rq_shootout.html`) landed in **RFC 005's World 3**: the
three queue variants are within 1015% of each other in `rq_runtime` at every
scheduler count ≥ 4, on all three workloads. Striped's real micro-bench
dominance (24.8 M ops/s at 24t, 4.3× over mutex) does not propagate to the
runtime because the queue is not the hot path in any measured workload. USL
fits put σ at 5.57.6 (super-serial) for ping-pong: the ceiling is per-wake
latency through the park/unpark protocol, not queue push/pop.
Guiding rules established this cycle:
- **Lock count that matters is hot-path locks.** Cold lifecycle paths may keep a
small per-slot lock; the run path targets zero locks.
- **Lock ordering: io-before-shared, slot locks are leaves** (never hold two
slot locks at once).
- **No unwind inside a runtime critical section.** Stop sentinel only fires at
lock-free observation points.
- **Never hold a thread-local guard across a possible switch point.** An actor
can be preempted at any allocation and resume on a different OS thread.
`with_runtime`, `with_shared`, and every `RawMutex` guard disable preemption
for this reason.
### Phase 1 — State decomposition ✅ (`3c7e26b`)
`next_monitor_id``AtomicU64`. `timers` → own `Mutex<Timers>`. `io` → own
`Mutex<Option<IoThread>>`. `pending_closures` → folded into `Slot::pending_closure`.
`check_cancelled()` gated behind `PREEMPTION_ENABLED` (poison fix).
`SharedState` now holds only: `slots`, `free_list`, `run_queue`, `root_pid`.
### Phase 2 — Slot table split ✅ (`5e0c9d4`)
Fixed slab `Box<[Slot]>`. Generation packed into state word: one
`AtomicU64 = (gen << 32) | state` — gen check is atomic with every transition.
Per-slot CAS state machine `Vacant/Queued/Running/RunningNotified/Parked/Done`.
Per-slot `RawMutex` for cold collections. Free list → `RawMutex<Vec<u32>>`.
`live_actors` atomic; termination = io_out==0 && queue empty && live==0.
### Phase 3 — Pluggable run queue ✅ (`1b3b618`)
Compile-time selected via feature flags. Variants:
- `rq-mutex``Mutex<VecDeque>`, baseline.
- `rq-mpmc` — hand-rolled Vyukov bounded MPMC ring.
- `rq-striped` — M Vyukov rings, fetch-add ticket distribution.
All variants compile in every build; feature only picks the alias. Bounded rings
are sound via slab cap + at-most-once-enqueued invariant.
Queue ops require preemption disabled (a producer suspended mid-publish stalls
every consumer behind its cell — livelock).
### Phase 4 — Bench harness ✅ (`6d9f369`)
`benches/rq_micro.rs`: raw structure sweep. `benches/rq_runtime.rs`:
yield-storm, ping-pong-pairs, spawn-storm. `scripts/bench_rq.sh`: rebuilds per
feature, aggregates to `bench_results/summary.csv`.
### Phase 5 — Safety hardening & model checking ✅ (`039703d`)
State machine extracted to `src/slot_state.rs` for loom checking. Loom on the
rings via `src/sync_shim.rs`. `NoPreempt`/no-unwind/TLS-guard audit complete.
Invariant-assertion sweep: every `StateWord` transition self-asserts its
precondition.
Consequences:
- **`rq-mutex` stays the default** — simplest correct, no capacity
constraints, locking model already integrated with the preemption-disabled
queue-op invariant.
- **Feature plumbing stays as is.** All three variants keep compiling in
every build; `rq-mpmc`/`rq-striped` remain selectable for benching.
- **Reopening is benchmark-driven only.** The report documents the
conditional upgrade paths if a future workload qualifies: mpmc for
message-passing-dominant loads at N ≤ 8 (1.3× lower σ, 0.16 µs ping-pong
at 1t); striped for high-contention balanced push/pop at N ≥ 16 — neither
is a scheduler workload as measured.
- **Effort redirects to the wake path**: RFC 005 (billed as a latency patch,
per its own World 3 framing), RFC 004, and eventually per-switch cost.
---
## v0.6Actor ergonomics ✅ DONE
## v0.9Wake-path latency
### 1. Channel mutex migration ✅ (`8ff6cf4`)
`channel::Inner<T>` migrated from `std::sync::Mutex` to `RawMutex`, closing
the cross-thread pthread-release hole (guard held with preemption enabled →
timeslice switch → unlock from a different OS thread, UB) and removing poison.
Goal: attack per-wake latency — the measured ceiling — with the two specified
mechanisms, each benched against a clean baseline before the next lands.
Order: RFC 005 first (the slot is measured against the just-benched idle
policy), then RFC 004 (measured against the slot-accepted baseline), then an
interaction pass.
The strict leaf rule did not survive contact: finalize clones the
supervisor/trap senders and `monitor()` clones the Down sender, all under a
cold lock, and those senders *live in the slot* — the nesting is structural.
The check generalized to two classes: **Leaf → Channel**, at most one of
each, both violation directions debug-asserted at the acquisition site.
Channel critical sections call only the lock-free unpark protocol, so the
order is acyclic. `recv_match`'s user predicate still runs under the channel
lock — cheap/pure/channel-free, as documented.
Full specs: `rfc_005-wake-slot.md`, `rfc_004-tunable-scheduler-idle-policy.md`
(artefacts.kalsbeek.dev).
### 2. Named registry ✅ (`2c7cf0b`)
`register` / `whereis` / `unregister` + the inverse `name_of(pid)`: a
**bimap** (two HashMaps in exact inverse, debug-asserted) under one Leaf
`RawMutex` in `RuntimeInner` — the planned "HashMap in SharedState" predated
the v0.5 decomposition. Erlang semantics: one name per pid, one pid per
name, NameTaken only against a *live* holder. Cleanup is lazy: liveness
rides the generation-checked slot word (pids are never reused), dead
bindings behave as absent and are pruned on contact — no `finalize_actor`
hook, no Slot field, no reset-in-three-places exposure.
### 1. Wake slot (RFC 005)
Per-scheduler, thread-local, capacity-one wake cache, checked before the
shared queue. Runtime-selected via `Config { wake_slot: bool }`, default off
until accepted (one binary benches both arms).
- **Push policy:** slot-eligible iff the wake originates from actor context
(`current_pid().is_some()`). Scheduler-context wakes (timer/IO drain) and
spawns always go shared.
- **Displacement:** newest wake takes the slot, occupant pushed shared (Go
semantics); the branch swap is the fallback if benches look pathological.
- **Pop order:** slot, then shared. Slot-popped actors inherit the waker's
remaining timeslice — a handoff chain is bounded by one slice, so the
shared queue is consulted at least once per slice per scheduler (the
starvation bound, zero new counters).
- **Invariant care:** the slot push replaces `run_queue.push` at the tail of
the `Parked → Queued` CAS; at-most-once-enqueued holds verbatim (pid in
slot ⊕ shared queue). Stall blast radius grows by exactly one actor.
- Counters: `slot_hits`, `slot_displacements`.
`send_by_name` was dropped from scope deliberately: smarm has no per-pid
mailbox, so there is nothing generic to send *to*. The registry yields a
`Pid` for monitor/link/request_stop and for routing user-owned channels.
### 2. Slot shootout
Extend `rq_runtime`/`bench_rq.sh` with the slot on/off dimension (cheap —
it's a Config knob, not a feature rebuild). Same sweep as the rq shootout.
- **ping-pong-pairs** — target metric; expect the win, single- and
multi-scheduler.
- **yield-storm** — regression guard; yields never touch the slot, any delta
is pop-path overhead.
- **spawn-storm** — neutrality check; spawns bypass the slot by policy.
Acceptance flips the default on and re-baselines for RFC 004.
### 3. gen_server: call timeout ✅ (`134ff52`, `90b7040`)
Landed as two primitives, *without* the sketched monitor machinery:
### 3. Spinning workers (RFC 004)
Bounded spin-before-park for idle schedulers, killing the ~100 µs
`thread::sleep` worst case on cross-thread handoff. Two Config knobs:
`spin_budget_cycles` (0 recovers today's behaviour) and `max_spinners`
(default N/2); one new atomic `n_spinning`. No run-queue changes — lands on
the frozen `rq-mutex` substrate, independent of the slot mechanically.
- `Receiver::recv_timeout(Duration)` — the per-`recv` deadline, on the
existing `timer::Reason::WaitTimeout`/`TimerTarget` machinery exactly as
`Mutex::lock_timeout` uses it (per-wait seq; stale heap entries no-op on
seq mismatch; message-first race resolution; `Disconnected` distinct from
`Timeout`).
- `ServerRef::call_timeout` = send + `recv_timeout` on the reply channel.
Server death is already observable there (reply sender drops as the server
unwinds), so monitor + demonitor-on-timeout had nothing left to guarantee:
no registration exists, so nothing can leak. Timed-out requests are still
handled; the late reply's send fails harmlessly (Erlang semantics).
(`handle_info` still deferred — needs cross-channel mailbox merge; see Later.)
---
## v0.7 — select, on epoch-stamped consuming wakes ✅ DONE
Goal: cross-channel wait (`select`) without surrendering the precise-wake
invariant. Decision record: the alternative ("wakes are hints", loop-and-
recheck at every park site) was evaluated and rejected — it distributes the
soundness obligation across every present and future park site and turns
every stale wake into a wasted reschedule. Instead the slot protocol itself
gains *wait identity*: a park-epoch packed into the spare bits of the slot
word, so that a stale unpark dies at one failed CAS in the waker.
Protocol (the consuming-wake rules):
- Word becomes `(gen << 32) | (epoch << 8) | state` — 24-bit epoch in the
previously unused bits; every transition still one CAS, gen check still
atomic with it.
- `begin_wait` (actor-side, before registering with any waker) bumps the
epoch and returns it; registrations carry `(pid, epoch)`.
- Every successful wake transition **consumes** the epoch (bumps it):
`Parked(e) → Queued(e+1)`, `Running(e) → RunningNotified(e+1)`. A second
unpark stamped with the same epoch can never land — at most one wake per
wait, by construction. This is what makes select's *loser* arms harmless
without a cancellation pass: their stale registrations fire one futile CAS
on next send and self-clean.
- `unpark_at(pid, epoch)` = epoch-matched wake. Bare `unpark(pid)` remains as
the wildcard and is reserved for **terminal** wakes (`request_stop`), which
never return control to the code that parked.
- One-shot park sites (`Mutex::lock_timeout`, `sleep`, `block_on_io`,
`wait_fd`) stay exactly as written: precise wakes are preserved, so a wake
still *means* something.
### 1. Slot-word epoch + consuming unparks ✅ (`4913835`)
`slot_state.rs`: repack, `begin_wait`, epoch-matched/consuming `unpark`,
epoch-preserving claim/yield/park transitions. Loom: existing theorems
re-proved on the new word + a new stale-epoch theorem (an unpark stamped with
a consumed epoch can never enqueue *or* notify). Plumbing only — no caller
stamps yet; behaviour identical.
### 2. Stamp every registration-based wake; retire per-primitive wait seqs ✅ (`400854a`)
The epoch IS the wait identity, so `channel::Inner.{cur_wait,next_wait_seq}`
and `mutex`'s per-wait `seq` are subsumed: registrations become
`(pid, epoch)`, `TimerTarget::on_timeout(pid, epoch)`, wakers call
`unpark_at`. Sleep timers, io completions (`Blocking` + `FdReady`) and join
waiters stamped likewise. End-state invariant, auditable in one sentence:
**the only wildcard wake is `request_stop`.**
### 3. `select` — ready-index wait over multiple receivers ✅ (`00128f3`)
`select(&[&dyn Selectable]) -> usize`: register self in each arm's
`parked_receiver` (sequentially — the park protocol makes cross-arm atomicity
unnecessary), park once, wake, scan, return the first ready index; caller
`try_recv`s it. Sound as ready-index *because channels are single-receiver*:
nobody can steal between ready and recv. A closed arm counts as ready (its
`try_recv` reports the closure). Arms are scanned in order — priority order,
documented, like BEAM receive clauses; no fairness promise. No cancel pass
(see protocol above). Drop-guard deregistration on the unwind path only, to
keep tests' single-parked-receiver asserts honest.
### 4. `select_timeout` ✅ (`a0a93b6`)
A `WaitTimeout` timer stamped with the select's epoch, whose target is a
stateless `unpark_at`. The woken selector classifies the wake from state
alone: some arm ready → that arm; none ready → with precise wakes the timer
is the only remaining stamped waker → `None`. No flag, no shared waiter
struct, no registration to leak. The timer entry expires as a stale-epoch
no-op when an arm wins, per the no-cancellation convention in `timer.rs`.
### 5. Docs + integration proof ✅ (this commit)
README module table, gotchas rewrite (the "No `select`" bullet becomes the
consuming-wake invariant), and an integration test of the motivating
pattern: a server selecting over its inbox and a monitor `Down` channel
(`tests/select.rs::motivating_pattern_inbox_plus_monitor_down`).
Unblocked but deliberately out of scope: `gen_server::handle_info` (now an
API-design question, not a runtime question — see Later).
### Deviations from the plan
- **The no-park exit needed a protocol piece the plan missed.** A `select`
returning an arm ready at *registration* time leaves earlier arms holding
live-epoch registrations — a sender could still set RunningNotified and
fault the actor's next one-shot park. Landed as `scheduler::retire_wait`
(bump the epoch, eat a landed notification via the new
`StateWord::clear_notify`, re-observe the stop flag — in that order) plus
the loom theorem `retire_eats_late_arm_notification`. The parked path
needed nothing: the winning wake's consume already retires the epoch.
- **No Drop guard.** The planned unwind-path deregistration was superseded:
the single-receiver debug_asserts relax to "none, or my own pid", so a
leftover own-registration (consumed epoch — inert by theorem) is simply
overwritten by the receiver's next wait on that channel. Less machinery,
and the asserts still catch the real misuse (two actors on one channel).
- **A closed arm is ready *forever*** — callers must drop it from the arm
set once observed, or priority order starves every higher-indexed arm
(documented on `select`; found the hard way by a livelocking test).
---
## v0.8 — gen_server: handle_info / handle_down + io fd hygiene ✅ DONE
Goal: spend the `select` primitive on the server loop — out-of-band messages
(`handle_info`) and monitor `Down`s (`handle_down`) dispatched alongside
calls and casts — and close the v0.2 fd-hygiene hole in `io.rs`.
Design decisions (settled up front):
- **Info channels are static.** `type Info: Send + 'static` on the trait; a
`Vec<Receiver<G::Info>>` handed over at start. Users enum their own Info.
Dynamic info subscription is deliberately out: revisit against a consumer.
- **Down forwarding is dynamic, because monitors are.** You monitor a worker
you just spawned inside a handler, so down arms cannot ride the static
list. Mechanism: the loop owns a control arm (`Receiver<Monitor>`);
`init` grows a `&ServerCtx` parameter (breaking, no-op default body) whose
clonable `Watcher` the state stores and calls `watch(monitor)` from any
handler thereafter. Handler signatures otherwise untouched.
- **Priority order: downs → control → infos → inbox.** System messages
preempt; a hot inbox cannot starve a death notice. Within each class,
declaration order (select's documented priority semantics).
- **Closed arms drop silently** (the closed-arm-is-ready-forever gotcha).
Closed inbox still means graceful shutdown, unchanged.
### 1. handle_info — static info arms in the server loop ✅ (`e5d1b3b`)
`type Info` + `handle_info` (no-op default) on `GenServer`; `ServerBuilder`
(`Server::new(state).with_info(rx).under(sup).start()`) so start variants
don't multiply; `start`/`start_under` stay as wrappers. Loop becomes
`select` over `[infos…, inbox]`, dispatching by arm index.
### 2. handle_down — ServerCtx::watcher + control arm ✅ (`24b95c9`)
`handle_down` (no-op default), `ServerCtx`/`Watcher`, the control arm, and
the loop's live `Vec<Monitor>`. A `Down` retires its arm (monitors are
one-shot). Integration test: the motivating worker-pool pattern — a server
that spawns workers in a handler, watches them, and reacts to their deaths.
### 3. io fd hygiene — DEL on the unwind path ✅ (`f6969e5`)
A stopped actor unwinding out of `wait_fd`'s park leaks its `waiters` entry
and the kernel registration; the stale entry then fails every future
`wait_*` on that fd with `AlreadyExists` (the defensive bare-DEL in
`epoll_register` sits behind the `contains_key` check, so it never runs).
Fix where the invariant breaks: a drop guard in `wait_fd`, armed after a
successful register, disarmed on normal wake; on unwind it removes the
waiters entry iff it is still `(me, epoch)` and only then DELs the fd
(an entry consumed by a racing `FdReady` means the fd may already be
re-registered by someone else — leave it alone).
### Deviations from the plan
- **The plain-inbox fast path narrowed.** It now also requires the control
arm closed, so every server pays one select on its first iteration (the
ctx drops after `init`; a state that didn't clone the Watcher closes the
arm there and then). Servers that DO hold a Watcher select forever — the
price of dynamic watch.
- **The fd leak was worse than the v0.2 caveat admitted.** Not "at most one
stale wakeup": the stale `waiters` entry failed every future `wait_*` on
that fd with AlreadyExists, permanently (the defensive DEL sat behind the
`contains_key` check). The regression test exhibits it as a hang.
### 4. RFC 004 bench + interaction pass
Re-run the sweep with spinning enabled, slot on and off. The known
interaction: once idle pickup latency ≪ slice, the slot's latency trade
(occupant waits out the waker's slice while other schedulers idle) may stop
paying — RFC 005's documented escape hatch is Go's "spinners exist → push
shared instead" check. In scope only if the data demands it.
---
## Later
### Tunable scheduler idle policy (RFC 004)
Today idle schedulers fall back to `thread::sleep(100µs)`, putting ~100µs
worst-case latency on every cross-thread actor handoff. RFC 004 specifies a
bounded spinning-worker policy (Go/Tokio model): idle schedulers spin on a
cheap atomic for a configurable budget before parking, so a burst of work is
picked up in cache-coherency time rather than after a kernel reschedule.
Two new `Config` knobs: `spin_budget_cycles` (RDTSC cycles to spin; `0`
recovers today's behaviour) and `max_spinners` (cap on concurrent spinners;
default `N/2`). Single new atomic `n_spinning`. No change to the run-queue
data structure — lands on the current `Mutex<VecDeque>` and is independent of
any future queue topology work.
Full spec: `rfc_004-tunable-scheduler-idle-policy.md` (artefacts.kalsbeek.dev).
### Per-switch cost (context shims, epoch protocol)
The shootout's residual: per-wake latency is 0.160.18 µs at N=1 and
0.81.2 µs at N=8+, dominated by the context-switch shims and the epoch
protocol, not the queue. On current evidence this is the larger constant —
"the whole game" alongside the v0.9 work — but there is no spec yet. Needs a
profiling spike (where do the cycles actually go per park/unpark round-trip)
and then an RFC before it can be scheduled.
### Unbounded / configurable-bounded actor count
Fixed slab with a loud assert (`Config::max_actors(n)`, default 16 384).
Revisit with a segmented slab (array of `AtomicPtr<Segment>`, doubling segment
sizes, append-only) once the cap is actually hit. Do not let it calcify.
### arm-port validation & merge
`arm-port` branch carries an AAPCS64 context-switch backend, never run on
hardware. Build + run full test suite on an aarch64 device; check
+153 -26
View File
@@ -361,7 +361,7 @@ impl<T: Send + 'static> crate::timer::TimerTarget for RawMutex<Inner<T>> {
// select — ready-index wait over multiple receivers
// ---------------------------------------------------------------------------
mod sealed {
pub(crate) mod sealed {
pub trait Sealed {}
}
impl<T> sealed::Sealed for Receiver<T> {}
@@ -371,28 +371,46 @@ impl<T> sealed::Sealed for Receiver<T> {}
///
/// Contract (all under the arm's own lock): `sel_register` checks-or-
/// registers atomically — if the arm is ready it does NOT register and
/// returns `false`; otherwise it publishes `(pid, epoch)` where its wakers
/// will find it. "Ready" means a receive would not park: a message is
/// queued, or the arm is closed.
/// returns `Ok(false)`; otherwise it publishes `(pid, epoch)` where its
/// wakers will find it and returns `Ok(true)`. "Ready" means a receive
/// would not park: a message is queued, or the arm is closed. `Err` means
/// the arm could not register at all (only fd arms can fail; channel
/// registration is infallible) — the wait must be retired and earlier
/// eager-cleanup arms unregistered.
pub trait Selectable: sealed::Sealed {
#[doc(hidden)]
fn sel_register(&self, pid: Pid, epoch: u32) -> bool;
fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result<bool>;
#[doc(hidden)]
fn sel_ready(&self) -> bool;
/// Remove this arm's `(pid, epoch)` registration if — and only if — it
/// is still in place. Default no-op: a losing channel arm's stale
/// registration is inert (its wakers die at the epoch CAS; the next
/// wait overwrites the slot). Fd arms override this: their staleness
/// poisons the fd (waiters entry + kernel-side ONESHOT registration)
/// and needs an eager cleanup pass.
#[doc(hidden)]
fn sel_unregister(&self, _pid: Pid, _epoch: u32) {}
/// Whether this arm requires the eager cleanup pass at all. Gates the
/// post-wake `sel_unregister` sweep so channel-only selects keep
/// today's zero-cancellation hot path.
#[doc(hidden)]
fn sel_eager_cleanup(&self) -> bool {
false
}
}
impl<T> Selectable for Receiver<T> {
fn sel_register(&self, pid: Pid, epoch: u32) -> bool {
fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result<bool> {
let mut g = self.inner.lock();
if !g.queue.is_empty() || g.senders == 0 {
return false;
return Ok(false);
}
debug_assert!(
g.parked_receiver.is_none_or(|(p, _)| p == pid),
"channel has more than one receiver"
);
g.parked_receiver = Some((pid, epoch));
true
Ok(true)
}
fn sel_ready(&self) -> bool {
@@ -429,30 +447,95 @@ impl<T> Selectable for Receiver<T> {
/// self-clean at their wakers' failed CAS, or get overwritten by this
/// receiver's next wait on that channel.
///
/// Panics if `arms` is empty, or when called outside an actor.
/// Panics if `arms` is empty, when called outside an actor, or if an fd
/// arm fails to register (EBADF, EMFILE, a second waiter on one fd —
/// see [`try_select`] for the fallible form; channel-only selects cannot
/// fail).
pub fn select(arms: &[&dyn Selectable]) -> usize {
try_select(arms).expect("select(): fd arm failed to register (use try_select)")
}
/// [`select`], fallible: `Err` when an arm fails to register (only fd
/// arms can — EBADF, EMFILE on the epoll set, or a second waiter on an
/// fd that already has one). On `Err` the wait is fully retired and no
/// registration is left behind: every arm registered before the failing
/// one has been unregistered.
pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result<usize> {
assert!(!arms.is_empty(), "select() on an empty arm list");
let me = crate::actor::current_pid().expect("select() called outside an actor");
loop {
let epoch = crate::scheduler::begin_wait();
if let Some(i) = register_arms(me, epoch, arms) {
return i;
if let Some(i) = register_arms(me, epoch, arms)? {
return Ok(i);
}
// Stale fd registrations are not harmless (a losing fd arm's
// waiters entry poisons the fd with AlreadyExists and its
// kernel-side ONESHOT registration can fire arbitrarily late), so
// selects containing fd arms run an eager cleanup pass after the
// park — including when a terminal stop unwinds out of it, via
// the guard. Channel-only selects skip all of it: `eager` is
// false, the guard is disarmed, and the loser-arm self-cleaning
// story is unchanged.
let eager = arms.iter().any(|a| a.sel_eager_cleanup());
let mut guard = UnregisterGuard { arms, me, epoch, armed: eager };
crate::scheduler::park_current();
if eager {
unregister_arms(arms, me, epoch);
}
guard.armed = false;
drop(guard);
// Woken precisely: an arm's send (message) or last-sender drop
// (closure) consumed our epoch, and both leave their arm ready —
// return the first one, in priority order (which may be a
// different, higher-priority arm than the one that woke us; its
// message stays queued and re-reports ready on the next call).
// Fd arms classify by a fresh zero-timeout poll, so they too are
// a pure function of state — independent of the registration the
// cleanup pass just removed.
for (i, arm) in arms.iter().enumerate() {
if arm.sel_ready() {
return i;
return Ok(i);
}
}
// Unreachable by protocol (a stop wake unwinds out of
// park_current). Defensive: re-open the wait and re-register —
// stale own-registrations are overwritten.
// stale own-registrations are overwritten (channels) or were
// removed by the cleanup pass above (fds).
}
}
/// Eager-cleanup sweep: remove every fd arm's registration that is still
/// ours. No-op per channel arm (one virtual call); one io-lock visit per
/// fd arm.
fn unregister_arms(arms: &[&dyn Selectable], me: Pid, epoch: u32) {
for arm in arms {
if arm.sel_eager_cleanup() {
arm.sel_unregister(me, epoch);
}
}
}
/// Stop-unwind twin of the explicit cleanup pass: a terminal stop unwinds
/// out of `park_current`, and a registered fd arm must not outlive its
/// actor (the generalization of `wait_fd`'s `Dereg`). Disarmed on the
/// normal path after the explicit pass runs; never armed when no fd arm
/// registered, keeping the channel-only path guard-free in effect.
struct UnregisterGuard<'a> {
arms: &'a [&'a dyn Selectable],
me: Pid,
epoch: u32,
armed: bool,
}
impl Drop for UnregisterGuard<'_> {
fn drop(&mut self) {
if self.armed {
unregister_arms(self.arms, self.me, self.epoch);
}
}
}
@@ -462,19 +545,35 @@ pub fn select(arms: &[&dyn Selectable]) -> usize {
/// right after its registration wakes the caller through the protocol (the
/// prep-to-park window is closed by RunningNotified).
///
/// `Some(i)` = arm `i` was ready, the pass stopped, and the wait has been
/// RETIRED (no park may follow): earlier arms hold live-epoch
/// registrations, so the epoch is bumped, a landed notification eaten, and
/// a pending stop re-observed — without which a stale arm wake could fault
/// a later one-shot park. `None` = every arm registered; the caller parks.
fn register_arms(me: Pid, epoch: u32, arms: &[&dyn Selectable]) -> Option<usize> {
/// `Ok(Some(i))` = arm `i` was ready, the pass stopped, and the wait has
/// been RETIRED (no park may follow): earlier arms hold live-epoch
/// registrations, so earlier *fd* arms are unregistered eagerly, then the
/// epoch is bumped, a landed notification eaten, and a pending stop
/// re-observed — without which a stale arm wake could fault a later
/// one-shot park. `Err` = an arm failed to register; identical unwind
/// (earlier fd arms unregistered, wait retired). `Ok(None)` = every arm
/// registered; the caller parks.
fn register_arms(
me: Pid,
epoch: u32,
arms: &[&dyn Selectable],
) -> std::io::Result<Option<usize>> {
for (i, arm) in arms.iter().enumerate() {
if !arm.sel_register(me, epoch) {
let registered = match arm.sel_register(me, epoch) {
Ok(r) => r,
Err(e) => {
unregister_arms(&arms[..i], me, epoch);
crate::scheduler::retire_wait();
return Some(i);
return Err(e);
}
};
if !registered {
unregister_arms(&arms[..i], me, epoch);
crate::scheduler::retire_wait();
return Ok(Some(i));
}
}
None
Ok(None)
}
/// The [`select_timeout`] timer target: stateless, because precise wakes
@@ -506,16 +605,29 @@ impl crate::timer::TimerTarget for SelectTimeout {
/// `Duration::ZERO` is a valid timeout: it parks until the immediately-due
/// timer is drained, then reports `None` unless an arm was already ready.
///
/// Panics if `arms` is empty, or when called outside an actor.
/// Panics if `arms` is empty, when called outside an actor, or if an fd
/// arm fails to register (see [`try_select_timeout`] for the fallible
/// form; channel-only selects cannot fail).
pub fn select_timeout(
arms: &[&dyn Selectable],
timeout: std::time::Duration,
) -> Option<usize> {
try_select_timeout(arms, timeout)
.expect("select_timeout(): fd arm failed to register (use try_select_timeout)")
}
/// [`select_timeout`], fallible: `Err` when an arm fails to register
/// (only fd arms can). On `Err` the wait is fully retired and no
/// registration — arm-side or kernel-side — is left behind.
pub fn try_select_timeout(
arms: &[&dyn Selectable],
timeout: std::time::Duration,
) -> std::io::Result<Option<usize>> {
assert!(!arms.is_empty(), "select_timeout() on an empty arm list");
let me = crate::actor::current_pid().expect("select_timeout() called outside an actor");
let epoch = crate::scheduler::begin_wait();
if let Some(i) = register_arms(me, epoch, arms) {
return Some(i); // ready now: the timer was never armed
if let Some(i) = register_arms(me, epoch, arms)? {
return Ok(Some(i)); // ready now: the timer was never armed
}
// Arm the timer after the registration pass, outside every Channel
@@ -524,7 +636,22 @@ pub fn select_timeout(
let target: std::sync::Arc<dyn crate::timer::TimerTarget> = std::sync::Arc::new(SelectTimeout);
crate::scheduler::insert_wait_timer(deadline, me, target, epoch);
// Same eager-cleanup story as `try_select`: the timer arm needs none
// (stateless, stale entries die at the epoch CAS), channel arms need
// none, fd arms do — and a timer win in particular leaves every fd
// arm's registration behind, which without this pass would poison
// those fds until a kernel event happened to fire.
let eager = arms.iter().any(|a| a.sel_eager_cleanup());
let mut guard = UnregisterGuard { arms, me, epoch, armed: eager };
crate::scheduler::park_current();
if eager {
unregister_arms(arms, me, epoch);
}
guard.armed = false;
drop(guard);
// Woken precisely: an arm (ready below) or the timer (nothing ready).
arms.iter().position(|arm| arm.sel_ready())
Ok(arms.iter().position(|arm| arm.sel_ready()))
}
+6 -2
View File
@@ -45,7 +45,10 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator;
// Public API re-exports
// ---------------------------------------------------------------------------
pub use channel::{channel, select, select_timeout, Receiver, RecvError, RecvTimeoutError, Selectable, Sender};
pub use channel::{
channel, select, select_timeout, try_select, try_select_timeout, Receiver, RecvError,
RecvTimeoutError, Selectable, Sender,
};
pub use gen_server::{CallError, CallTimeoutError, CastError, GenServer, ServerBuilder, ServerCtx, ServerRef, Watcher};
pub use link::{link, trap_exit, unlink, ExitSignal};
pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId};
@@ -55,7 +58,8 @@ pub use registry::{name_of, register, unregister, whereis, RegisterError};
pub use runtime::{init, Config, Runtime};
pub use scheduler::{
block_on_io, request_stop, run, self_pid, sleep, spawn, spawn_under, wait_readable,
wait_writable, yield_now, JoinError, JoinHandle,
wait_readable_timeout, wait_writable, wait_writable_timeout, yield_now, FdArm, JoinError,
JoinHandle,
};
pub use supervisor::{ChildSpec, OneForOne, Restart, Signal, Strategy};
+132
View File
@@ -432,6 +432,138 @@ fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::R
Ok(())
}
// ---------------------------------------------------------------------------
// FdArm — fd readiness as a select arm (RFC 008)
// ---------------------------------------------------------------------------
/// An fd-readiness arm for [`crate::select`] / [`crate::select_timeout`]:
/// ready when the fd is readable (resp. writable), composable with channel
/// receivers on one wait epoch. Phase-1 rules apply: one waiter per fd at a
/// time, one direction per arm (duplex on a single fd needs `dup`; epoll
/// registrations key on the open file description, so dup'd fds register
/// independently).
pub struct FdArm {
fd: std::os::fd::RawFd,
readable: bool,
writable: bool,
}
impl FdArm {
pub fn readable(fd: std::os::fd::RawFd) -> Self {
FdArm { fd, readable: true, writable: false }
}
pub fn writable(fd: std::os::fd::RawFd) -> Self {
FdArm { fd, readable: false, writable: true }
}
}
impl crate::channel::sealed::Sealed for FdArm {}
impl crate::channel::Selectable for FdArm {
/// Ready-now check is a zero-timeout `poll(2)`; if the requested events
/// are pending the wait is retired without registering (`Ok(false)`,
/// the channel-arm contract). Otherwise register with the io thread —
/// every failure surfaces as `Err` (EBADF including a closed-fd
/// POLLNVAL, EMFILE on the epoll set, AlreadyExists for a second
/// waiter on the fd): the fallible-out, nothing-left-behind rule, in
/// deviation from RFC 008's permanently-ready lean, which would spin a
/// consumer whose fd is healthy but unregistrable (EMFILE).
fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result<bool> {
if poll_events(self.fd, self.readable, self.writable)? {
return Ok(false);
}
with_runtime(|inner| {
let mut io = inner.io.lock().unwrap();
io.as_mut()
.expect("io thread not started")
.epoll_register(self.fd, pid, epoch, self.readable, self.writable)
})?;
Ok(true)
}
/// Classification is the same zero-timeout poll: a pure function of fd
/// state, independent of the registration the cleanup pass removed. An
/// error here (EBADF: fd closed mid-wait) reports READY — the
/// consumer's read/write surfaces the errno; a dead arm is an event,
/// not a hang.
fn sel_ready(&self) -> bool {
poll_events(self.fd, self.readable, self.writable).unwrap_or(true)
}
/// `wait_fd`'s `Dereg` compare, verbatim: remove the waiters entry and
/// kernel-side registration iff the entry is still `(pid, epoch)`-ours.
/// A `FdReady` racing the wake may have consumed it (it removes + DELs
/// under the io lock), after which the fd may even carry ANOTHER
/// actor's fresh registration; in that case touch nothing.
fn sel_unregister(&self, pid: Pid, epoch: u32) {
with_runtime(|inner| {
let mut io = inner.io.lock().unwrap();
if let Some(io) = io.as_mut() {
if io.waiters.get(&self.fd) == Some(&(pid, epoch)) {
io.waiters.remove(&self.fd);
io.epoll_deregister(self.fd);
}
}
});
}
fn sel_eager_cleanup(&self) -> bool {
true
}
}
/// Zero-timeout `poll(2)`: are any of the requested events (or ERR/HUP,
/// which make the consumer's read/write fail loudly rather than park
/// forever) pending on `fd`? POLLNVAL maps to `Err(EBADF)`.
fn poll_events(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::Result<bool> {
let mut events: libc::c_short = 0;
if readable {
events |= libc::POLLIN;
}
if writable {
events |= libc::POLLOUT;
}
let mut pfd = libc::pollfd { fd, events, revents: 0 };
loop {
let r = unsafe { libc::poll(&mut pfd, 1, 0) };
if r < 0 {
let e = std::io::Error::last_os_error();
if e.kind() == std::io::ErrorKind::Interrupted {
continue;
}
return Err(e);
}
if r == 0 {
return Ok(false);
}
if pfd.revents & libc::POLLNVAL != 0 {
return Err(std::io::Error::from_raw_os_error(libc::EBADF));
}
return Ok(pfd.revents & (events | libc::POLLERR | libc::POLLHUP) != 0);
}
}
/// Wait until `fd` is readable or `timeout` elapses: `Ok(true)` = ready,
/// `Ok(false)` = timed out. A one-arm [`crate::try_select_timeout`].
pub fn wait_readable_timeout(
fd: std::os::fd::RawFd,
timeout: std::time::Duration,
) -> std::io::Result<bool> {
let arm = FdArm::readable(fd);
Ok(crate::channel::try_select_timeout(&[&arm], timeout)?.is_some())
}
/// Wait until `fd` is writable or `timeout` elapses: `Ok(true)` = ready,
/// `Ok(false)` = timed out.
pub fn wait_writable_timeout(
fd: std::os::fd::RawFd,
timeout: std::time::Duration,
) -> std::io::Result<bool> {
let arm = FdArm::writable(fd);
Ok(crate::channel::try_select_timeout(&[&arm], timeout)?.is_some())
}
pub fn read(fd: std::os::fd::RawFd, buf: &mut [u8]) -> std::io::Result<usize> {
wait_readable(fd)?;
let n = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) };
+403
View File
@@ -0,0 +1,403 @@
//! RFC 008 — fd arms in select. Beyond the functional cases, the
//! *_stays_usable tests are the soundness probes for the one asymmetry the
//! RFC must close: a losing CHANNEL arm's stale registration is inert, but
//! a losing FD arm's registration (waiters entry + kernel ONESHOT) poisons
//! the fd with AlreadyExists until the eager cleanup pass removes it. Every
//! "loser" scenario therefore re-waits on the same fd afterwards and must
//! succeed — pre-cleanup, each of those re-waits errors or hangs.
//!
//! House pattern: actor panics are trampoline-caught and `run` returns
//! normally, so every test funnels its result into an outcome flag asserted
//! OUTSIDE `run` — an in-actor assertion alone passes vacuously.
use smarm::{
channel, run, select, select_timeout, spawn, try_select, wait_readable,
wait_readable_timeout, wait_writable_timeout, yield_now, FdArm,
};
use std::os::fd::RawFd;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
// ---------------------------------------------------------------------------
// Pipe helper (as in io_epoll.rs)
// ---------------------------------------------------------------------------
struct Pipe {
read: RawFd,
write: RawFd,
}
impl Pipe {
fn new() -> Self {
let mut fds: [libc::c_int; 2] = [0; 2];
let r = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
assert_eq!(r, 0, "pipe2 failed");
Pipe { read: fds[0], write: fds[1] }
}
}
impl Drop for Pipe {
fn drop(&mut self) {
unsafe {
libc::close(self.read);
libc::close(self.write);
}
}
}
fn raw_write(fd: RawFd, buf: &[u8]) -> isize {
unsafe { libc::write(fd, buf.as_ptr() as *const _, buf.len()) }
}
fn raw_read(fd: RawFd, buf: &mut [u8]) -> isize {
unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) }
}
fn flag() -> (Arc<AtomicBool>, Arc<AtomicBool>) {
let f = Arc::new(AtomicBool::new(false));
(f.clone(), f)
}
// ---------------------------------------------------------------------------
// Ready-now: data already pending retires the wait without parking.
// ---------------------------------------------------------------------------
#[test]
fn fd_arm_ready_now_returns_without_parking() {
let (ok, ok2) = flag();
run(move || {
let p = Pipe::new();
assert_eq!(raw_write(p.write, b"x"), 1);
let (_tx, rx) = channel::<i64>();
let fd_arm = FdArm::readable(p.read);
// fd arm at index 1: the ready-now path must also work for a
// non-first arm (and clean nothing — channel arms are inert).
let i = select(&[&rx, &fd_arm]);
assert_eq!(i, 1);
let mut buf = [0u8; 1];
assert_eq!(raw_read(p.read, &mut buf), 1);
ok2.store(true, Ordering::SeqCst);
});
assert!(ok.load(Ordering::SeqCst));
}
// ---------------------------------------------------------------------------
// Park-then-wake: fd arm wins against an idle channel arm.
// ---------------------------------------------------------------------------
#[test]
fn fd_arm_parks_until_data_then_wins() {
let got = Arc::new(AtomicU32::new(0));
let got2 = got.clone();
run(move || {
let p = Pipe::new();
let (rfd, wfd) = (p.read, p.write);
let (_tx_keepalive, rx) = channel::<i64>();
let h = spawn(move || {
let fd_arm = FdArm::readable(rfd);
let i = select(&[&fd_arm, &rx]);
assert_eq!(i, 0);
let mut buf = [0u8; 1];
assert_eq!(raw_read(rfd, &mut buf), 1);
got2.store(buf[0] as u32, Ordering::SeqCst);
});
yield_now(); // let it park
assert_eq!(raw_write(wfd, b"y"), 1);
let _ = h.join();
});
assert_eq!(got.load(Ordering::SeqCst), b'y' as u32);
}
// ---------------------------------------------------------------------------
// THE asymmetry probe: channel arm wins, losing fd arm must be cleaned —
// the same actor (and the io thread) must be able to wait that fd again.
// ---------------------------------------------------------------------------
#[test]
fn losing_fd_arm_is_cleaned_up_and_fd_stays_usable() {
let got = Arc::new(AtomicU32::new(0));
let got2 = got.clone();
run(move || {
let p = Pipe::new();
let (rfd, wfd) = (p.read, p.write);
let (tx, rx) = channel::<i64>();
let h = spawn(move || {
let fd_arm = FdArm::readable(rfd);
// Channel wins; the fd arm registered and lost.
let i = select(&[&fd_arm, &rx]);
assert_eq!(i, 1);
assert_eq!(rx.try_recv().unwrap(), Some(7));
// Pre-cleanup this wait_readable fails AlreadyExists (the
// waiters entry is stale-ours) — the cleanup pass must have
// removed it.
wait_readable(rfd).unwrap();
let mut buf = [0u8; 1];
assert_eq!(raw_read(rfd, &mut buf), 1);
got2.store(buf[0] as u32, Ordering::SeqCst);
});
yield_now(); // let it park in the select
tx.send(7).unwrap();
yield_now(); // let it reach the second wait
assert_eq!(raw_write(wfd, b"z"), 1);
let _ = h.join();
});
assert_eq!(got.load(Ordering::SeqCst), b'z' as u32);
}
// ---------------------------------------------------------------------------
// Ready-now on a LATER arm must unregister an earlier fd arm (the
// register_arms prefix-cleanup path: no park ever happens).
// ---------------------------------------------------------------------------
#[test]
fn ready_now_later_arm_cleans_earlier_fd_arm() {
let (ok, ok2) = flag();
run(move || {
let p = Pipe::new();
let (rfd, wfd) = (p.read, p.write);
let (tx, rx) = channel::<i64>();
tx.send(1).unwrap(); // arm 1 ready before the select
let fd_arm = FdArm::readable(rfd);
let i = select(&[&fd_arm, &rx]);
assert_eq!(i, 1);
assert_eq!(rx.try_recv().unwrap(), Some(1));
// The fd arm registered (idle pipe), then arm 1 retired the wait.
// Its registration must have been removed in the same pass.
assert_eq!(raw_write(wfd, b"a"), 1);
wait_readable(rfd).unwrap();
let mut buf = [0u8; 1];
assert_eq!(raw_read(rfd, &mut buf), 1);
ok2.store(true, Ordering::SeqCst);
});
assert!(ok.load(Ordering::SeqCst));
}
// ---------------------------------------------------------------------------
// Two fd arms in one select (phase-1: distinct fds, nothing special).
// ---------------------------------------------------------------------------
#[test]
fn two_fd_arms_second_fires_first_stays_usable() {
let (ok, ok2) = flag();
run(move || {
let pa = Pipe::new();
let pb = Pipe::new();
let (rfd_a, wfd_a) = (pa.read, pa.write);
let (rfd_b, wfd_b) = (pb.read, pb.write);
let h = spawn(move || {
let a = FdArm::readable(rfd_a);
let b = FdArm::readable(rfd_b);
let i = select(&[&a, &b]);
assert_eq!(i, 1);
let mut buf = [0u8; 1];
assert_eq!(raw_read(rfd_b, &mut buf), 1);
// Arm a lost; its fd must be immediately re-waitable.
assert_eq!(raw_write(wfd_a, b"q"), 1);
wait_readable(rfd_a).unwrap();
assert_eq!(raw_read(rfd_a, &mut buf), 1);
assert_eq!(buf[0], b'q');
ok2.store(true, Ordering::SeqCst);
});
yield_now();
assert_eq!(raw_write(wfd_b, b"b"), 1);
let _ = h.join();
});
assert!(ok.load(Ordering::SeqCst));
}
// ---------------------------------------------------------------------------
// select_timeout: timer wins over an idle fd arm; the fd is left clean.
// ---------------------------------------------------------------------------
#[test]
fn select_timeout_timer_beats_idle_fd_arm_and_fd_stays_usable() {
let (ok, ok2) = flag();
run(move || {
let p = Pipe::new();
let (rfd, wfd) = (p.read, p.write);
let fd_arm = FdArm::readable(rfd);
let start = Instant::now();
let r = select_timeout(&[&fd_arm], Duration::from_millis(30));
assert!(r.is_none(), "idle fd must time out");
assert!(start.elapsed() >= Duration::from_millis(30));
// Timer win is exactly the case where the fd arm's registration is
// left behind without an eager pass.
assert_eq!(raw_write(wfd, b"c"), 1);
wait_readable(rfd).unwrap();
let mut buf = [0u8; 1];
assert_eq!(raw_read(rfd, &mut buf), 1);
ok2.store(true, Ordering::SeqCst);
});
assert!(ok.load(Ordering::SeqCst));
}
// ---------------------------------------------------------------------------
// Derived wrappers.
// ---------------------------------------------------------------------------
#[test]
fn wait_readable_timeout_times_out_then_succeeds_with_data() {
let (ok, ok2) = flag();
run(move || {
let p = Pipe::new();
let (rfd, wfd) = (p.read, p.write);
let start = Instant::now();
assert_eq!(wait_readable_timeout(rfd, Duration::from_millis(30)).unwrap(), false);
assert!(start.elapsed() >= Duration::from_millis(30));
// Timed-out wait must leave the fd clean; ready path returns true.
assert_eq!(raw_write(wfd, b"d"), 1);
assert_eq!(wait_readable_timeout(rfd, Duration::from_secs(5)).unwrap(), true);
let mut buf = [0u8; 1];
assert_eq!(raw_read(rfd, &mut buf), 1);
ok2.store(true, Ordering::SeqCst);
});
assert!(ok.load(Ordering::SeqCst));
}
#[test]
fn wait_readable_timeout_wakes_on_late_data() {
let got = Arc::new(AtomicU32::new(0));
let got2 = got.clone();
run(move || {
let p = Pipe::new();
let (rfd, wfd) = (p.read, p.write);
let h = spawn(move || {
assert_eq!(wait_readable_timeout(rfd, Duration::from_secs(5)).unwrap(), true);
let mut buf = [0u8; 1];
assert_eq!(raw_read(rfd, &mut buf), 1);
got2.store(buf[0] as u32, Ordering::SeqCst);
});
yield_now();
assert_eq!(raw_write(wfd, b"e"), 1);
let _ = h.join();
});
assert_eq!(got.load(Ordering::SeqCst), b'e' as u32);
}
#[test]
fn wait_writable_timeout_ready_now_on_empty_pipe() {
let (ok, ok2) = flag();
run(move || {
let p = Pipe::new();
// An empty pipe's write end is writable: ready-now path, no park.
assert_eq!(wait_writable_timeout(p.write, Duration::from_secs(5)).unwrap(), true);
ok2.store(true, Ordering::SeqCst);
});
assert!(ok.load(Ordering::SeqCst));
}
// ---------------------------------------------------------------------------
// Error surface: registration failure is an Err from try_select, with the
// wait retired (the actor can immediately wait on something else).
// ---------------------------------------------------------------------------
#[test]
fn try_select_surfaces_registration_error_and_retires_the_wait() {
let (ok, ok2) = flag();
run(move || {
let bad: RawFd = {
let p = Pipe::new();
p.read
}; // both ends closed by Drop: EBADF on registration
let fd_arm = FdArm::readable(bad);
let err = try_select(&[&fd_arm]).unwrap_err();
// EBADF, whether the pre-poll or epoll_ctl ADD reports it.
assert_eq!(err.raw_os_error(), Some(libc::EBADF));
// The wait was retired: a normal select right after works.
let (tx, rx) = channel::<i64>();
tx.send(9).unwrap();
let i = select(&[&rx]);
assert_eq!(i, 0);
assert_eq!(rx.try_recv().unwrap(), Some(9));
ok2.store(true, Ordering::SeqCst);
});
assert!(ok.load(Ordering::SeqCst));
}
// ---------------------------------------------------------------------------
// Stop-unwind: an actor stopped while parked in an fd-arm select must not
// poison the fd (the UnregisterGuard generalization of wait_fd's Dereg).
// Mirrors io_epoll.rs::stopped_waiter_does_not_poison_the_fd.
// ---------------------------------------------------------------------------
#[test]
fn stopped_selector_does_not_poison_the_fd() {
let seen = Arc::new(AtomicU32::new(0));
let seen_outer = seen.clone();
run(move || {
let p = Pipe::new();
let (rfd, wfd) = (p.read, p.write);
let (_tx_keepalive, rx) = channel::<i64>();
let h = spawn(move || {
let fd_arm = FdArm::readable(rfd);
select(&[&fd_arm, &rx]);
unreachable!("neither arm ever fires while this actor lives");
});
yield_now(); // let it reach the park
smarm::request_stop(h.pid());
let _ = h.join(); // Ok(()): stopped, not panicked
// Second waiter on the SAME fd must register and be woken.
let seen2 = seen.clone();
let h2 = spawn(move || {
wait_readable(rfd).unwrap();
let mut buf = [0u8; 1];
assert_eq!(raw_read(rfd, &mut buf), 1);
seen2.store(buf[0] as u32, Ordering::SeqCst);
});
yield_now();
assert_eq!(raw_write(wfd, b"x"), 1);
let _ = h2.join();
});
assert_eq!(seen_outer.load(Ordering::SeqCst), b'x' as u32);
}
// ---------------------------------------------------------------------------
// Phase-1 misuse: a second waiter on an fd that already has one is an Err
// (AlreadyExists), not a hang — and does not disturb the first waiter.
// ---------------------------------------------------------------------------
#[test]
fn second_waiter_on_same_fd_errs_without_disturbing_the_first() {
let got = Arc::new(AtomicU32::new(0));
let got2 = got.clone();
let (ok, ok2) = flag();
run(move || {
let p = Pipe::new();
let (rfd, wfd) = (p.read, p.write);
let h = spawn(move || {
wait_readable(rfd).unwrap();
let mut buf = [0u8; 1];
assert_eq!(raw_read(rfd, &mut buf), 1);
got2.store(buf[0] as u32, Ordering::SeqCst);
});
yield_now(); // first waiter parked
let fd_arm = FdArm::readable(rfd);
let err = try_select(&[&fd_arm]).unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::AlreadyExists);
// First waiter still wakes normally.
assert_eq!(raw_write(wfd, b"w"), 1);
let _ = h.join();
ok2.store(true, Ordering::SeqCst);
});
assert_eq!(got.load(Ordering::SeqCst), b'w' as u32);
assert!(ok.load(Ordering::SeqCst));
}