From 9c3274cff4ddb3ab59983c20b8d5fb46771cb232 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:42:34 +0700 Subject: [PATCH 01/17] Add all-core runtime installation --- Cargo.toml | 8 ++++---- src/runtime.rs | 47 ++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 50 insertions(+), 5 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 2e6e6e8..4e86551 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "nagoya" -version = "0.1.16" -edition = "2021" +version = "0.1.17" +edition = "2024" rust-version = "1.85" license = "MIT OR Apache-2.0" description = "An async runtime that does not need an operating system" @@ -32,7 +32,7 @@ syscall-counters = ["reactor"] [dependencies] -ps-st3 = { version = "^0.6", default-features = false, features = ["fanout"] } +ps-st3 = { path = "../ps-st3", version = "^0.6", default-features = false, features = ["fanout"] } spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"] } # The task itself: one allocation holding the future, its state machine, its # waker vtable and its output, with `Runnable::into_raw` handing the pointer @@ -56,7 +56,7 @@ futures-io = { version = "0.3", default-features = false, optional = true } futures-util = { version = "0.3", default-features = false, optional = true } [dev-dependencies] -ps-st3 = { version = "^0.6", features = ["fanout", "host"] } +ps-st3 = { path = "../ps-st3", version = "^0.6", features = ["fanout", "host"] } # The comparison arm. Only the scheduler is used; nothing here needs its I/O. tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util"] } # The other comparison arm. Not an async runtime: a data-parallel pool, so it diff --git a/src/runtime.rs b/src/runtime.rs index 02b01fd..1eb7925 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -203,6 +203,39 @@ impl Builder { self } + /// Use every logical CPU available to this process, including performance + /// and efficiency cores. This sets the worker count, not OS CPU affinity. + /// + /// The default still uses `shared_threads()`, which prefers Apple's + /// `hw.perflevel0.logicalcpu`. Call this after any `workers` override. + /// Failure to detect a count or fit the ps-st3 bitmap is reported instead + /// of silently configuring fewer workers. + pub fn all_cores(mut self) -> std::io::Result { + let workers = std::thread::available_parallelism()?.get(); + if workers > usize::BITS as usize { + return Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "all-core worker count exceeds the ps-st3 bitmap capacity", + )); + } + self.workers = workers; + Ok(self) + } + + /// Install this configuration as the process's shared pool. + /// + /// Call before any `nagoya::spawn` or `background()` use. Concurrent + /// initializers create exactly one runtime; every loser gets an error. + /// Libraries using the shared entry points then use this same pool. + pub fn install_background(&self) -> Result<&'static Runtime, BackgroundStarted> { + let mut installed = false; + let runtime = SHARED.get_or_init(|| { + installed = true; + self.build() + }); + if installed { Ok(runtime) } else { Err(BackgroundStarted) } + } + /// The idle policy. [`Tuning::default`] unless set. pub fn tuning(mut self, tuning: Tuning) -> Self { self.tuning = tuning; @@ -238,6 +271,18 @@ impl Builder { } } +/// The shared pool was initialized before configuration could be installed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct BackgroundStarted; + +impl core::fmt::Display for BackgroundStarted { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.write_str("nagoya shared runtime already started") + } +} + +impl std::error::Error for BackgroundStarted {} + /// Threads for the shared pool. /// /// The machine's parallelism, except that on a machine with more than one @@ -306,7 +351,7 @@ fn sysctl_usize(name: &core::ffi::CStr) -> Option { } #[cfg(target_vendor = "apple")] -extern "C" { +unsafe extern "C" { fn sysctlbyname( name: *const core::ffi::c_char, oldp: *mut core::ffi::c_void, From 98f69d956521a1aecada5de26a74f8f64c5dd5ae Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:42:44 +0700 Subject: [PATCH 02/17] Add typed exclusive owner turns --- src/lib.rs | 10 ++++ src/owner.rs | 151 +++++++++++++++++++++++++++++++++++++++++++++++++++ src/task.rs | 18 ++++++ 3 files changed, 179 insertions(+) create mode 100644 src/owner.rs diff --git a/src/lib.rs b/src/lib.rs index cc0ee0f..0830092 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -74,6 +74,7 @@ pub mod guide {} #[cfg(feature = "reactor")] pub mod net; mod par; +pub mod owner; // Child processes, at the path tokio gives them. They register on a reactor of // their own that starts on first use, rather than one the caller passes, so a // child's futures finish under any executor. Unix only, because the reactor is: @@ -298,6 +299,15 @@ impl Executor { &self.pool } + /// The executing worker for this pool, when called from a pool task. + #[must_use] + pub fn current_worker(&self) -> Option { + match self.context.as_deref() { + Some(context) => context.current_worker(&self.pool), + None => crate::task::current_worker(&self.pool), + }.filter(|id| *id < self.pool.workers()) + } + /// Run a future to completion somewhere on the pool. pub fn spawn(&self, future: F) -> JoinHandle where diff --git a/src/owner.rs b/src/owner.rs new file mode 100644 index 0000000..5edff76 --- /dev/null +++ b/src/owner.rs @@ -0,0 +1,151 @@ +//! Typed exclusive turns with admission before queue submission. + +use alloc::sync::Arc; +use core::cell::UnsafeCell; +use core::sync::atomic::{AtomicBool, Ordering}; +use crate::{Executor, JoinHandle}; + +/// Caller-owned CPU admission policy, shared by every worker and helper. +pub trait CpuBudget: Send + Sync + 'static { + /// A reservation refunded on cancellation and reconciled after execution. + type Permit: CpuPermit + Send + 'static; + /// Reserve credits for this stable key before placing work on the pool. + fn reserve(&self, key: &Arc, credits: u64) -> Option; +} + +/// A CPU reservation's worker-local execution boundary. +/// +/// Implementations must release/refund on Drop, including unwinding. Both +/// methods run on the same worker, with no suspension or migration between. +pub trait CpuPermit { + /// Begin measuring this worker's CPU consumption. + fn enter(&mut self); + /// Reconcile actual consumption and release the in-flight reservation. + fn finish(&mut self); +} + +struct State { + admitted: AtomicBool, + value: UnsafeCell>, +} + +// SAFETY: a successful Acquire admission grants the only access to value. +// It is moved into Lease, then returned before Release clears admission. +// T must actually be Send; this does not assert Send for an agent's fields. +unsafe impl Sync for State {} + +struct Lease { + state: Arc>, + value: Option, +} + +impl Drop for Lease { + fn drop(&mut self) { + // SAFETY: this lease is the sole successful admission, including + // cancellation before execution and unwinding during a turn. + unsafe { *self.state.value.get() = self.value.take(); } + self.state.admitted.store(false, Ordering::Release); + } +} + +/// A cloneable admission handle for one owned, movable value. +/// +/// Clones share the gate, key and budget, never the mutable value. T needs +/// Send, not Sync. Turn callbacks cannot await or retain a borrow of T. +/// A native caller must end worker-bound frames/guards before returning. +pub struct Owner { + state: Arc>, + key: Arc, + budget: Arc, +} + +impl Clone for Owner { + fn clone(&self) -> Self { + Self { + state: Arc::clone(&self.state), + key: Arc::clone(&self.key), + budget: Arc::clone(&self.budget), + } + } +} + +/// Why a turn was not submitted. Its callback and captured input are returned. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DispatchDenied { + /// Another queued or executing turn already owns this value. + Busy, + /// The stable key has insufficient credits or in-flight capacity. + Budget, + /// The preferred initial worker is outside this executor's pool. + Worker, +} + +/// A rejected callback, retaining all of its owned input for a reactive retry. +pub struct Rejected { + /// The reason no job was enqueued. + pub reason: DispatchDenied, + /// The original callback with its original captures. + pub turn: F, +} + +impl> Owner { + /// Move a value into a stable owner and bind its key and credit authority. + #[must_use] + pub fn new(value: T, key: Arc, budget: Arc) -> Self { + Self { + state: Arc::new(State { + admitted: AtomicBool::new(false), + value: UnsafeCell::new(Some(value)), + }), + key, + budget, + } + } + + /// The same typed key is used by every turn, independent of worker choice. + #[must_use] + pub fn key(&self) -> &Arc { &self.key } + + /// Dispatch one complete synchronous turn on any worker. + pub fn dispatch(&self, executor: &Executor, credits: u64, turn: F) + -> Result, Rejected> + where F: FnOnce(&mut T) -> R + Send + 'static, R: Send + 'static { + self.dispatch_to(executor, None, credits, turn) + } + + /// Dispatch with an optional initial queue preference, never affinity. + /// + /// Work stealing may move the queued turn. The next turn can prefer a + /// different worker only after this one has returned ownership. + pub fn dispatch_to(&self, executor: &Executor, worker: Option, + credits: u64, turn: F) -> Result, Rejected> + where F: FnOnce(&mut T) -> R + Send + 'static, R: Send + 'static { + if worker.is_some_and(|id| id >= executor.pool().workers()) { + return Err(Rejected { reason: DispatchDenied::Worker, turn }); + } + if self.state.admitted.compare_exchange(false, true, + Ordering::Acquire, Ordering::Relaxed).is_err() { + return Err(Rejected { reason: DispatchDenied::Busy, turn }); + } + // SAFETY: the successful admission owns this slot until Lease drops. + let value = unsafe { (*self.state.value.get()).take() } + .expect("ExclusiveOwnerDuplicateAdmission"); + let mut lease = Lease { state: Arc::clone(&self.state), value: Some(value) }; + let Some(mut permit) = self.budget.reserve(&self.key, credits) else { + drop(lease); + return Err(Rejected { reason: DispatchDenied::Budget, turn }); + }; + let future = async move { + permit.enter(); + let output = turn(lease.value.as_mut().expect("exclusive turn value")); + permit.finish(); + drop(permit); + // Publish T before the join becomes ready, so the next dispatch + // can immediately acquire it on a different worker. + drop(lease); + output + }; + Ok(JoinHandle { task: Some(crate::task::spawn_on_at(future, + Arc::clone(executor.pool()), executor.context.clone(), worker).fallible()) }) + } +} diff --git a/src/task.rs b/src/task.rs index e046d07..98f2d71 100644 --- a/src/task.rs +++ b/src/task.rs @@ -124,12 +124,30 @@ pub(crate) fn spawn_on( where F: Future + Send + 'static, F::Output: Send + 'static, +{ + spawn_on_at(future, pool, context, None) +} + +/// The ordinary task path with a preference for its first queue only. +pub(crate) fn spawn_on_at(future: F, pool: Arc, + context: Option>, first_worker: Option) -> Task +where F: Future + Send + 'static, F::Output: Send + 'static, { // Called for the first poll and for every wake after it. The `Runnable` is // a single heap allocation that already holds the future and its state, so // turning it into a job is `into_raw`: no box, no closure, nothing new. let weak_pool = Arc::downgrade(&pool); + let first = core::sync::atomic::AtomicBool::new(true); let schedule = move |runnable: Runnable| { + if first.swap(false, core::sync::atomic::Ordering::Relaxed) { + if let Some(worker) = first_worker { + if let Some(pool) = weak_pool.upgrade() { + assert!(worker < pool.workers(), "invalid initial worker"); + pool.submit_local(worker, into_job(runnable)); + } + return; + } + } let mut runnable = Some(runnable); let mut local = |owner: &Pool, worker: usize| { if core::ptr::eq(owner, weak_pool.as_ptr()) && worker < owner.workers() { From 2e137f6ab44a36e7684813cf2dd40abdd99d9fec Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:42:49 +0700 Subject: [PATCH 03/17] Add shared CPU credit accounting --- src/credits.rs | 153 +++++++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 1 + 2 files changed, 154 insertions(+) create mode 100644 src/credits.rs diff --git a/src/credits.rs b/src/credits.rs new file mode 100644 index 0000000..67bfc13 --- /dev/null +++ b/src/credits.rs @@ -0,0 +1,153 @@ +//! Shared CPU credit authority with cancellation refunds and overshoot debt. + +use alloc::sync::Arc; +use core::num::{NonZeroU64, NonZeroUsize}; +use core::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; +use crate::owner::{CpuBudget, CpuPermit, DispatchDenied, Rejected}; +use crate::sync::{Notified, Notify}; +use crate::{Executor, JoinHandle}; + +/// One stable accounting key, shared across its agents and helper jobs. +/// +/// Migration and host reentry reuse this same Arc. Replenishment is a policy +/// event, never a side effect of admission or worker selection. +pub struct CreditAccount { + key: K, + capacity: i64, + balance: AtomicI64, + in_flight: AtomicUsize, + parallelism: usize, + changed: Notify, +} + +impl CreditAccount { + /// Start with capacity CPU nanoseconds and an admitted-job cap. + /// Returns None if capacity exceeds the signed debt representation. + #[must_use] + pub fn new(key: K, capacity: NonZeroU64, parallelism: NonZeroUsize) -> Option { + let capacity = i64::try_from(capacity.get()).ok()?; + Some(Self { + key, capacity, balance: AtomicI64::new(capacity), + in_flight: AtomicUsize::new(0), parallelism: parallelism.get(), + changed: Notify::new(), + }) + } + + /// Immutable policy identity, such as a schemeful site and incarnation. + #[must_use] + pub fn key(&self) -> &K { &self.key } + + /// Spendable CPU nanoseconds; a negative value is retained overshoot debt. + #[must_use] + pub fn balance(&self) -> i64 { self.balance.load(Ordering::Acquire) } + + /// Number of jobs reserved or executing across all workers. + #[must_use] + pub fn in_flight(&self) -> usize { self.in_flight.load(Ordering::Acquire) } + + /// Add policy-approved CPU nanoseconds at a retained refill/class event. + /// Credit first pays debt and is capped at this account's capacity. + pub fn replenish(&self, credits: u64) { + let credits = i64::try_from(credits).unwrap_or(i64::MAX); + self.adjust(credits); + self.changed.notify_waiters(); + } + + /// Register before rechecking admission, then await a refund/refill event. + /// Use Notified::enable for register/recheck. Never scan accounts on a tick. + pub fn changed(&self) -> Notified<'_> { self.changed.notified() } + + fn adjust(&self, credits: i64) { + let _ = self.balance.fetch_update(Ordering::AcqRel, Ordering::Acquire, + |balance| Some(balance.saturating_add(credits).min(self.capacity))); + } +} + +/// CPU admission using a host's monotonic current-worker CPU nanosecond clock. +/// +/// The supplied function must measure CPU, not wall/guest time. A turn never +/// suspends between its two reads, so per-worker clock origins may differ. +/// This hook supplies accounting, not native preemption: callers must bound +/// turns at real safepoints and react to retained debt by throttling or killing. +pub struct CpuCredits { + clock: fn() -> u64, +} + +impl CpuCredits { + /// Bind the host's worker CPU clock without installing an OS service. + #[must_use] + pub const fn new(clock: fn() -> u64) -> Self { Self { clock } } + + /// Admit an attributed helper on the same executor, measuring its CPU. + /// The rejected closure retains its typed owned/Arc input. + pub fn dispatch(&self, executor: &Executor, + key: &Arc>, credits: u64, work: F) + -> Result, Rejected> + where K: Send + Sync + 'static, F: FnOnce() -> R + Send + 'static, + R: Send + 'static { + let Some(mut permit) = self.reserve(key, credits) else { + return Err(Rejected { reason: DispatchDenied::Budget, turn: work }); + }; + Ok(executor.spawn(async move { + permit.enter(); + let output = work(); + permit.finish(); + output + })) + } +} + +/// A reservation carrying its original account across queueing and execution. +pub struct CpuReservation { + account: Arc>, + reserved: i64, + clock: fn() -> u64, + started: Option, + done: bool, +} + +impl CpuBudget> for CpuCredits { + type Permit = CpuReservation; + + fn reserve(&self, key: &Arc>, credits: u64) + -> Option { + let reserved = i64::try_from(credits).ok().filter(|value| *value > 0)?; + key.in_flight.fetch_update(Ordering::AcqRel, Ordering::Acquire, + |count| (count < key.parallelism).then_some(count + 1)).ok()?; + if key.balance.fetch_update(Ordering::AcqRel, Ordering::Acquire, + |balance| (balance >= reserved).then(|| balance - reserved)).is_err() { + key.in_flight.fetch_sub(1, Ordering::AcqRel); + key.changed.notify_waiters(); + return None; + } + Some(CpuReservation { + account: Arc::clone(key), reserved, clock: self.clock, + started: None, done: false, + }) + } +} + +impl CpuReservation { + fn settle(&mut self) { + if self.done { return; } + let actual = self.started.map_or(0, |start| { + i64::try_from((self.clock)().saturating_sub(start)).unwrap_or(i64::MAX) + }); + self.account.adjust(self.reserved.saturating_sub(actual)); + self.account.in_flight.fetch_sub(1, Ordering::AcqRel); + self.done = true; + self.account.changed.notify_waiters(); + } +} + +impl CpuPermit for CpuReservation { + fn enter(&mut self) { + assert!(!self.done && self.started.is_none(), "CPU reservation entered twice"); + self.started = Some((self.clock)()); + } + fn finish(&mut self) { self.settle(); } +} + +impl Drop for CpuReservation { + fn drop(&mut self) { self.settle(); } +} diff --git a/src/lib.rs b/src/lib.rs index 0830092..62a7b96 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -61,6 +61,7 @@ use st3::fanout::Pool; mod block_on; pub mod io; +pub mod credits; /// Tutorial and callsite reference, with compiled examples. #[cfg(feature = "std")] #[doc = include_str!(concat!(env!("OUT_DIR"), "/guide-examples.md"))] From 173c530b4f79cb26cb5b0e95e99fa0e0f50136ab Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:42:56 +0700 Subject: [PATCH 04/17] Bind timers to the shared readiness host --- src/reactor/driver.rs | 21 ++++++++------ src/runtime.rs | 47 ++++++++++++++++++++++++++++++ src/time.rs | 67 ++++++++++++++++++++++++++++++++++++++----- 3 files changed, 119 insertions(+), 16 deletions(-) diff --git a/src/reactor/driver.rs b/src/reactor/driver.rs index 2a8ddef..7798eaf 100644 --- a/src/reactor/driver.rs +++ b/src/reactor/driver.rs @@ -670,20 +670,23 @@ fn run(shared: &Arc) { /// arrives: there is no deadline to wake for. fn service_timers(shared: &Arc) -> (Option, bool) { let deadline = shared.next_deadline.load(Ordering::Acquire); + if deadline == NO_DEADLINE { + return (None, false); + } let now = crate::now_ns(); - - // Not due yet: wait exactly until it is, and do not touch the wheel. - if deadline != NO_DEADLINE && deadline > now { + if deadline > now { return (Some(deadline - now), false); } - // Either a deadline has arrived or a timer was armed and the cache was - // invalidated. Both mean the wheel has work to report. + // Clear before draining, then merge rather than overwrite. An arm before + // the swap is in the heap; an arm after it lowers the cache and wakes the + // kernel wait. An arm racing with fire_due's return must survive publication. + shared.next_deadline.swap(NO_DEADLINE, Ordering::AcqRel); let (next, woke) = crate::time::fire_due(now); - shared - .next_deadline - .store(next.unwrap_or(NO_DEADLINE), Ordering::Release); - (next.map(|deadline| deadline.saturating_sub(now)), woke) + shared.next_deadline.fetch_min(next.unwrap_or(NO_DEADLINE), Ordering::AcqRel); + let cached = shared.next_deadline.load(Ordering::Acquire); + let wait = (cached != NO_DEADLINE).then(|| cached.saturating_sub(now)); + (wait, woke) } /// Wake the tasks named by `events`. diff --git a/src/runtime.rs b/src/runtime.rs index 1eb7925..31abe3c 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -154,6 +154,26 @@ impl Runtime { &self.executor } + /// The process's shared readiness and timer host, owned by nagoya. + /// + /// Initialize immediately after installing the shared pool, before any + /// pending timer is polled. One accounted OS driver waits for readiness + /// and deadlines; all task and page execution remains on pool workers. + /// Pass this Handle into sockets and services instead of starting reactors. + #[cfg(feature = "reactor")] + pub fn readiness(&self) -> Result { + let host = READINESS.get_or_init(|| { + let reactor = crate::reactor::Reactor::start().map_err(ReadinessError::Reactor)?; + crate::time::set_timer_reactor(reactor.handle()) + .map_err(|_| ReadinessError::TimerDriverStarted)?; + Ok(reactor) + }); + match host { + Ok(reactor) => Ok(reactor.handle()), + Err(error) => Err(*error), + } + } + /// The pool underneath, for the APIs that take one directly. /// /// [`crate::par_for`] takes an `Arc` because it schedules closures @@ -164,6 +184,33 @@ impl Runtime { } } +/// A failure to initialize the one process readiness/timer host. +#[cfg(feature = "reactor")] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReadinessError { + /// The OS refused the reactor or its driver thread. + Reactor(crate::reactor::Errno), + /// A prior timer selected a different backend; initialize readiness first. + TimerDriverStarted, +} + +#[cfg(feature = "reactor")] +impl core::fmt::Display for ReadinessError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::Reactor(error) => write!(f, "nagoya readiness host: {error}"), + Self::TimerDriverStarted => f.write_str("nagoya timer driver already selected"), + } + } +} + +#[cfg(feature = "reactor")] +impl std::error::Error for ReadinessError {} + +#[cfg(feature = "reactor")] +static READINESS: std::sync::OnceLock> = + std::sync::OnceLock::new(); + /// A [`Runtime`] described before it is started, from [`Runtime::builder`]. /// /// Exists for the settings that [`Runtime::with_tuning`] has no argument for. diff --git a/src/time.rs b/src/time.rs index 2181258..6367a2c 100644 --- a/src/time.rs +++ b/src/time.rs @@ -18,7 +18,10 @@ //! //! # What drives it //! -//! With `std`, a thread, started the first time anything sleeps. Never before: +//! With `std`, a bound reactor, or a thread started on first timer admission. +//! `Runtime::readiness` binds the shared reactor before the first timer; +//! an already-selected timer backend is rejected instead of doubled. +//! The dedicated-thread default remains for existing callers. Never before: //! a runtime that is only ever polled should not own a thread it does not use. //! //! Without `std` there is no thread and no clock, so the consumer supplies @@ -370,7 +373,7 @@ impl Future for Sleep { this.slot = Some(slot); if earliest { #[cfg(feature = "std")] - driver::nudge(); + driver::nudge(this.deadline); } Poll::Pending } @@ -455,6 +458,33 @@ impl Future for Timeout { } } +/// The timer backend was selected by an earlier timer or reactor binding. +#[cfg(feature = "reactor")] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TimerDriverStarted; + +#[cfg(feature = "reactor")] +impl core::fmt::Display for TimerDriverStarted { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.write_str("nagoya timer driver already selected") + } +} + +#[cfg(feature = "reactor")] +impl std::error::Error for TimerDriverStarted {} + +/// Route all future timer arms to this retained reactor, without a timer thread. +/// +/// Call before the first pending Sleep is polled. The reactor must outlive all +/// armed timers and must have exactly one servicing loop. Runtime::readiness +/// supplies that process-lifetime ownership. Clock reads alone do not select +/// the backend. This function never replaces an already-selected driver. +#[cfg(feature = "reactor")] +pub fn set_timer_reactor(handle: crate::reactor::Handle) + -> Result<(), TimerDriverStarted> { + driver::bind_reactor(handle) +} + #[cfg(feature = "std")] mod driver { use super::{now_ns, poll, set_clock}; @@ -465,6 +495,20 @@ mod driver { static NUDGE: OnceLock<(Mutex, Condvar)> = OnceLock::new(); static THREAD: OnceLock<()> = OnceLock::new(); + enum Host { + Dedicated, + #[cfg(feature = "reactor")] + Reactor(crate::reactor::Handle), + } + + static HOST: OnceLock = OnceLock::new(); + + #[cfg(feature = "reactor")] + pub(super) fn bind_reactor(handle: crate::reactor::Handle) + -> Result<(), super::TimerDriverStarted> { + HOST.set(Host::Reactor(handle)).map_err(|_| super::TimerDriverStarted) + } + fn std_clock() -> u64 { let origin = ORIGIN.get_or_init(Instant::now); u64::try_from(origin.elapsed().as_nanos()).unwrap_or(u64::MAX) @@ -482,11 +526,20 @@ mod driver { /// Wake the driver because a nearer deadline than it is waiting for now /// exists. Starts the driver if this is the first timer in the process. - pub(super) fn nudge() { - start(); - let (lock, signal) = nudge_pair(); - *lock.lock().expect("the nudge lock") = true; - signal.notify_one(); + pub(super) fn nudge(deadline: u64) { + match HOST.get_or_init(|| Host::Dedicated) { + Host::Dedicated => { + let _ = deadline; + start(); + let (lock, signal) = nudge_pair(); + *lock.lock().expect("the nudge lock") = true; + signal.notify_one(); + } + #[cfg(feature = "reactor")] + Host::Reactor(handle) => { + handle.timer_armed(deadline).expect("nagoya timer reactor wake"); + } + } } fn start() { From 37a0d7da52e3272165964d605aec58e59f019a98 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:43:15 +0700 Subject: [PATCH 05/17] Add retained readiness reasons --- docs/agent-pool.md | 14 ++++++ src/lib.rs | 1 + src/readiness.rs | 109 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 124 insertions(+) create mode 100644 docs/agent-pool.md create mode 100644 src/readiness.rs diff --git a/docs/agent-pool.md b/docs/agent-pool.md new file mode 100644 index 0000000..c2f771a --- /dev/null +++ b/docs/agent-pool.md @@ -0,0 +1,14 @@ +# One process, one injected execution domain + +Source baseline: nagoya master `4cae808eccec503107bd8a9769f385e0ddbdbfdd`, +version 0.1.16. The owner's 2026-10-14 ruling overrides contradictory adaptive +compilation language in the coordinator documents. Suspected attack means +throttle or kill that agent. These changes introduce no compile tier. + +Start once, before libraries submit work or poll a pending timer: + +```rust +let builder = nagoya::runtime::Runtime::builder().all_cores()?; +let runtime = builder.install_background()?; +let readiness = runtime.readiness()?; +let executor = runtime.executor().clone_handle(); diff --git a/src/lib.rs b/src/lib.rs index 62a7b96..af04ac6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -100,6 +100,7 @@ pub mod runtime; pub mod sync; pub mod task; pub mod time; +pub mod readiness; mod yield_now; /// Spawn onto the shared pool, as `tokio::spawn` does. diff --git a/src/readiness.rs b/src/readiness.rs new file mode 100644 index 0000000..2beebed --- /dev/null +++ b/src/readiness.rs @@ -0,0 +1,109 @@ +//! Retained, coalesced wake reasons for one logical scheduling owner. + +use alloc::sync::Arc; +use core::num::NonZeroU64; +use core::sync::atomic::{AtomicU64, Ordering}; +use crate::sync::Notify; +use crate::{Executor, JoinHandle}; + +struct Signals { + reasons: AtomicU64, + notify: Notify, +} + +/// The unique consumer of a logical owner's ready reasons. +/// +/// Reasons carry no guest payload. Owned values and immutable typed Arcs +/// remain in the caller's bounded ingress/continuation records. +pub struct ReadySource { + signals: Arc, +} + +/// A cloneable producer that can wake the consumer from any completion. +#[derive(Clone)] +pub struct ReadyHandle { + signals: Arc, +} + +impl Default for ReadySource { + fn default() -> Self { Self::new() } +} + +impl ReadySource { + /// Make a single-consumer source with no pending reasons. + #[must_use] + pub fn new() -> Self { + Self { signals: Arc::new(Signals { + reasons: AtomicU64::new(0), notify: Notify::new(), + }) } + } + + /// Give a timer, readiness callback or typed completion an owned producer. + #[must_use] + pub fn handle(&self) -> ReadyHandle { + ReadyHandle { signals: Arc::clone(&self.signals) } + } + + /// Consume retained reasons, or register and suspend until a real event. + /// + /// Registration precedes the atomic recheck, so a wake racing with idle + /// suspension is retained. There is no periodic wake or account scan. + pub async fn wait(&mut self) -> u64 { + loop { + let notified = self.signals.notify.notified(); + let mut notified = core::pin::pin!(notified); + notified.as_mut().enable(); + let reasons = self.signals.reasons.swap(0, Ordering::AcqRel); + if reasons != 0 { return reasons; } + notified.await; + } + } +} + +/// Rounding exceeded the monotonic clock's usable deadline range. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DeadlineOverflow; + +impl core::fmt::Display for DeadlineOverflow { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.write_str("nagoya coarse deadline overflow") + } +} + +#[cfg(feature = "std")] +impl std::error::Error for DeadlineOverflow {} + +/// Round up to a configured coarse delivery quantum. Never round earlier. +/// u64::MAX is reserved by the reactor for an unarmed deadline. +pub fn coarse_deadline(deadline: u64, quantum: NonZeroU64) -> Result { + let remainder = deadline % quantum.get(); + let extra = if remainder == 0 { 0 } else { quantum.get() - remainder }; + deadline.checked_add(extra).filter(|value| *value != u64::MAX) + .ok_or(DeadlineOverflow) +} + +impl ReadyHandle { + /// Retain reasons before waking. Concurrent signals coalesce into bits. + /// Zero has no effect. Each bit means recheck that obligation's real state. + pub fn wake(&self, reasons: u64) { + if reasons != 0 && self.signals.reasons.fetch_or(reasons, Ordering::AcqRel) == 0 { + self.signals.notify.notify_one(); + } + } + + /// Arm one retained coarse deadline on this executor and its timer host. + /// + /// For last-view sleep deadlines and staggered service-worker admission. + /// The caller retains the JoinHandle and cancels it when eligibility or + /// the due event changes. It must initialize Runtime::readiness first. + /// This is delivery only; it does not grant CPU or service-worker admission. + pub fn wake_at(&self, executor: &Executor, deadline: u64, + quantum: NonZeroU64, reasons: u64) -> Result, DeadlineOverflow> { + let deadline = coarse_deadline(deadline, quantum)?; + let handle = self.clone(); + Ok(executor.spawn(async move { + crate::sleep_until(deadline).await; + handle.wake(reasons); + })) + } +} From 2cf6800711ab31956c1a5bba503bf7d1dd32ba55 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:43:25 +0700 Subject: [PATCH 06/17] Add the agent pool functional witness --- Cargo.toml | 4 + docs/agent-pool.md | 161 ++++++++++++++++++++++++ examples/agent_pool_witness.rs | 210 +++++++++++++++++++++++++++++++ scripts/agent-pool-functional.ts | 35 ++++++ 4 files changed, 410 insertions(+) create mode 100644 examples/agent_pool_witness.rs create mode 100644 scripts/agent-pool-functional.ts diff --git a/Cargo.toml b/Cargo.toml index 4e86551..35c6ac4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -187,3 +187,7 @@ required-features = ["reactor"] [[example]] name = "syscall_cost" required-features = ["reactor"] + +[[example]] +name = "agent_pool_witness" +required-features = ["syscall-counters"] diff --git a/docs/agent-pool.md b/docs/agent-pool.md index c2f771a..f9424a0 100644 --- a/docs/agent-pool.md +++ b/docs/agent-pool.md @@ -12,3 +12,164 @@ let builder = nagoya::runtime::Runtime::builder().all_cores()?; let runtime = builder.install_background()?; let readiness = runtime.readiness()?; let executor = runtime.executor().clone_handle(); +``` + +Enable the `reactor` feature. Inject this executor, its `Arc` and this +readiness handle throughout chuzz. `nagoya::spawn` and `background()` find the +configured shared pool. Do not call `Runtime::new`, `Builder::build`, +`Reactor::start`, `Reactor::sharded` or resolver-owned blocking thread helpers +from application/library components in this topology. Existing APIs remain +available for other embeddings. Platform main does only OS-required work. + +`Builder::all_cores` is the documented core-selection knob. It uses the +process's available logical CPU count, including performance and efficiency +cores, and rejects a count above ps-st3's `usize::BITS` bitmap capacity. It +does not pin OS threads or guarantee that an OS schedules one on each core. +An embedding needing affinity must supply that separately inside nagoya. +Today's unchanged default is `Builder::new -> shared_threads` and +`background -> Runtime::new(shared_threads())` in `src/runtime.rs`. +`shared_threads` tries `performance_cores` first. On Apple that reads +`hw.perflevel0.logicalcpu`; only its fallback uses `available_parallelism`. +An explicit `workers` call after `all_cores` overrides the count. + +Nagoya owns N pool worker threads plus one explicitly accounted OS +readiness/deadline driver. The driver waits in kqueue/epoll and runs no agent +code. No application thread, secondary executor or per-agent runtime is +required. The timer backend must be bound before first timer admission; +late binding is a named error and never silently creates two timer drivers. + +# Exclusive turns and typed ownership + +`Owner` requires the actual value T to be Send, without requiring Sync. +Only admission handles clone. They retain the same typed key and authority +Arcs, and share one atomic admission gate. Successful admission Acquire-takes +an owned T into a private lease. `dispatch` reserves CPU credits before the +async task is scheduled. The worker executes one synchronous callback with +exclusive `&mut T`, reconciles CPU, then puts T back before Release-clearing +the gate. The join becomes ready only after that publication. Queued work, +running work, cancellation and unwinding all retain or return the one lease. +Busy or budget-denied submission returns the original callback and captures. + +The slot uses an audited `UnsafeCell>` and a conditional Sync +implementation for T: Send. This synchronizes movement of an already-Send +value; it does not assert Send for any DOM, realm, native frame or platform +field. Agent fields containing Rc or thread-affine objects must be refactored +before using this primitive. There is no Mutex or dyn Any in owner transport. +Immutable inputs, outputs, source and code products can be owned typed Arcs. + +`dispatch_to` is an optional initial queue preference for deterministic +placement work, not lifetime affinity. Stealing can move queued work. End all +worker-specific borrows, native frames, code-reader pins and publication +guards before returning a turn. A bounded native suspension must retain roots +and owned continuation/code records, then acquire new worker guards on entry. +This primitive does not make arbitrary native execution resumable. + +# CPU fairness and concurrency at every level + +`CpuBudget` and `CpuPermit` are policy hooks. The supplied `CpuCredits` +implementation uses `Arc>` as the stable accounting key. +Give agents from one site and all of their parse, compile, style, layout, +paint, decode, network and marking helpers that same account. Its atomic +credit balance and in-flight cap apply across every worker. Migration, +reentry, new registrations and agent construction must reuse the account; +they must never construct fresh site credit. Retain policy accounts across +worker-instance teardown while the policy key remains alive. + +Each reservation removes CPU nanoseconds before enqueueing. Entry/exit read +the host's current-thread CPU clock, with no await between them. Waiting and +queue residence are not charged as CPU. Cancellation before entry refunds +everything; unused reservations are refunded; overshoot becomes negative +debt. Replenishment pays debt first and caps spendable credit. The host must +provide a monotonic CPU clock, never guest time or elapsed wall time. + +`CpuCredits::dispatch` admits attributed synchronous helpers on the same +executor. `par_for` already uses the same Arc, but does not itself call +the new CPU hook. Reserve each bounded chunk before submitting it, or run +chunks through attributed helper dispatch. Do not reserve once around an +unbounded parallel range. Agent turns remain serial while independent agents +and immutable helper DAG inputs can execute concurrently from the first +integration. Chuzz supplies the DAG, revision validation and deficit/deadline +ordering, foreground dependency priority, control/progress capacity and +site-to-agent selection. Admission hooks alone do not implement that policy. + +Register `CreditAccount::changed()` with `Notified::enable`, recheck the +blocked dispatch and await only if still denied. Refunds and explicit refill +events wake waiters. Refill is one retained due deadline or a class/input +event, never a scan. A background site has a shared in-flight limit; trusted +foreground/control policy may use separate reserved authority. Native +safepoints must bound CPU overshoot and termination. Cooperative dispatch +cannot preempt a monolithic callback or a stuck OS call. + +# Timer and readiness delivery + +`Runtime::readiness` starts one existing `Reactor::start` and retains it for +the process. `time::set_timer_reactor` binds its Handle before timer backend +selection. First pending `Sleep::poll` still registers in the existing indexed +timer heap; earliest-arm notification calls `Handle::timer_armed`. The driver +caches deadlines, services only due timers and blocks in `Poller::wait` until +an I/O edge, the actual deadline or an arm/shutdown notification. Publication +uses swap-before-drain and fetch_min-after-drain so a racing earlier deadline +is retained. A cancelled earliest timer may cause one stale deadline wake, +but no periodic inspection. + +`ReadySource` is one movable consumer; cloneable `ReadyHandle`s publish reason +bits before notification. Register/recheck retains wakes racing with idle +suspension. Bits coalesce eligibility changes, not distinct events or data. +Keep ordered inputs in bounded typed owned records or correctly owned ps-spsc +endpoints. `ReadyHandle::wake_at` rounds delivery up to a caller-provided +coarse quantum and arms one Sleep task on the same executor. Retain and cancel +its join when a deadline changes. No tick, interval or readiness probe runs. + +For automatic tab sleep, chuzz records each window's five-minute last-view +deadline, arms the next relevant eligibility deadline and rechecks the whole +window agent on wake. An active same-origin popup keeps its shared agent live. +Only after exclusive turn exit, bounded teardown and helper-reader drain may +the caller revoke the incarnation and release heap/cage state. Sleeping tab +records and service-worker registrations retain policy and bounded metadata, +not an eager agent or heap. A delivery bit alone never authorizes discard. + +For service workers, real readiness/completion events or admitted scheduled +pings retain a broker event. Compute a coarse hash-staggered next wake from +the event due time and retained site minimum gap, then arm one deadline. +Use one site authority to atomically admit across all registrations. Trusted +events can be immediate; normal batching and throttled 900-second gaps are +chuzz policy. New events joining a wake do not reset its CPU account. Root +waitUntil obligations until completion or a named cap failure, then discard +idle worker state. Foreground fetch fallback/deadlines remain explicit. +All exposed guest clocks remain coarse and frozen across fairness continuations; +the scheduler and credit hooks use trusted host clocks. + +Sockets already use `Handle::register`, owned `Registration`, +`take_readable_or_park`, `take_writable_or_park`, `poll_readable`, +`poll_writable`, and generation-qualified readiness. Reuse these, draining +actual I/O until EWOULDBLOCK before parking; do not periodically check a file +descriptor. Immutable typed completions can wake `ReadyHandle` after their +owned payload and generation are published. + +# Existing seams and release impact + +Built on `Runtime::builder`, `Builder::workers`, `Builder::build`, +`Runtime::executor`, `Executor::clone_handle`, `Executor::spawn`, +`Executor::pool`, `WorkerContext::current_worker`, private +`task::current_worker`, `task::spawn_on`, `async_task::Builder::spawn`, +`Runnable::schedule/into_raw/from_raw/run`, `JoinHandle`, +ps-st3 `Pool::submit_local/submit_job`, `Job::from_raw`, `Act`, +`Reactor::start/handle`, `Handle::timer_armed/wake/register`, +`Poller::wait/wake`, `sleep_until`, `Sleep`, private +`time::arm/fire_due`, `Notify::notified/notify_one/notify_waiters` and +`Notified::enable`. Task allocations and wakers are reused across waits; +one new owner-turn task is allocated per admitted turn. No new scheduler, +timer heap, general channel or thread pool is introduced. + +The published crate is 0.1.16. These APIs are additive and legacy defaults +remain; no minor bump is required for these signatures. Edition 2024 is a +deliberate toolchain change, with the existing Rust 1.85 floor supporting it. +The owner must use a 0.2 minor bump if later work changes defaults, removes +existing constructors or changes a published contract. Path plus caret +ps-st3 dependencies use the sibling in-house checkout locally and retain a +registry version for publication. No lockfile is supplied. + +The example witness and named production mutants are supplied for separate +owner-run functional acceptance. No build, unit test, benchmark or functional +execution was performed in the read-only lane. SEARCH checks establish patch +applicability, not compiler acceptance or concurrency correctness. diff --git a/examples/agent_pool_witness.rs b/examples/agent_pool_witness.rs new file mode 100644 index 0000000..64a70bc --- /dev/null +++ b/examples/agent_pool_witness.rs @@ -0,0 +1,210 @@ +//! Functional CLI acceptance for the injected pool and owner contracts. + +use nagoya::credits::{CpuCredits, CreditAccount}; +use nagoya::owner::{DispatchDenied, Owner}; +use nagoya::readiness::ReadySource; +use nagoya::runtime::Runtime; +use std::cell::Cell; +use std::io::{Read, Write}; +use std::num::{NonZeroU64, NonZeroUsize}; +use std::os::fd::AsRawFd; +use std::os::unix::net::UnixStream; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::task::Poll; + +fn fail(name: &str) -> ! { + eprintln!("FAIL {name}"); + std::process::exit(1) +} + +fn require(condition: bool, name: &str) { + if !condition { fail(name); } +} + +fn thread_cpu_ns() -> u64 { + let mut value = libc::timespec { tv_sec: 0, tv_nsec: 0 }; + // SAFETY: value is a writable timespec; this reads the current worker. + let result = unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &mut value) }; + require(result == 0, "WorkerCpuClockUnavailable"); + (value.tv_sec as u64).saturating_mul(1_000_000_000).saturating_add(value.tv_nsec as u64) +} + +struct Agent { + turns: Cell, + active: Arc, + input: Arc, +} + +impl Agent { + fn turn(&mut self) -> u64 { + require(self.active.fetch_add(1, Ordering::AcqRel) == 0, + "ExclusiveOwnerDuplicateAdmission"); + let mut digest = 0xcbf29ce484222325_u64; + for byte in self.input.bytes() { + digest = (digest ^ u64::from(byte)).wrapping_mul(0x100000001b3); + } + self.turns.set(self.turns.get() + 1); + self.active.fetch_sub(1, Ordering::AcqRel); + std::hint::black_box(digest) + } +} + +type Account = CreditAccount>; +type AgentOwner = Owner; + +fn owner(input: Arc) -> (AgentOwner, Arc) { + let account = Arc::new(CreditAccount::new(Arc::from("https://witness.invalid"), + NonZeroU64::new(1_000_000_000).unwrap(), NonZeroUsize::new(2).unwrap()).unwrap()); + let value = Agent { turns: Cell::new(0), active: Arc::new(AtomicUsize::new(0)), input }; + (Owner::new(value, account.clone(), Arc::new(CpuCredits::new(thread_cpu_ns))), account) +} + +fn exclusive(runtime: &Runtime, input: Arc) { + let (owner, _) = owner(input); + let (mut reader, mut writer) = UnixStream::pair().unwrap(); + let first = owner.dispatch(runtime.executor(), 1_000_000, move |agent| { + let mut delivery = [0_u8; 1]; + // One bounded witness rendezvous, released by CLI coordinator below. + // The ownership gate remains held while queued or executing. + reader.read_exact(&mut delivery).unwrap(); + require(delivery == [7], "OwnerInputDelivery"); + agent.turn(); + agent.turns.get() + }).unwrap_or_else(|_| fail("OwnerFirstAdmission")); + let second = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + owner.dispatch(runtime.executor(), 1_000_000, |agent| agent.turn()) + })); + writer.write_all(&[7]).unwrap(); + require(nagoya::block_on(first) == Some(1), "OwnerTurnOutput"); + match second { + Ok(Err(rejected)) if rejected.reason == DispatchDenied::Busy => {} + Ok(Ok(handle)) => { + let _ = nagoya::block_on(handle); + fail("ExclusiveOwnerDuplicateAdmission"); + } + _ => fail("ExclusiveOwnerDuplicateAdmission"), + } + let next = owner.dispatch(runtime.executor(), 1_000_000, |agent| { + agent.turn(); + agent.turns.get() + }).unwrap_or_else(|_| fail("OwnerHandoffAdmission")); + require(nagoya::block_on(next) == Some(2), "OwnerHandoffOutput"); + println!("PASS exclusive: one admission, owned state retained through two turns"); +} + +fn migration(runtime: &Runtime, input: Arc) { + let width = runtime.pool().workers(); + require(width >= 2, "MigrationRequiresTwoWorkers"); + let (owner, account) = owner(input); + let observer = runtime.executor().clone_handle(); + let first = owner.dispatch_to(runtime.executor(), Some(0), 1_000_000, move |agent| { + agent.turn(); + observer.current_worker().unwrap() + }).unwrap_or_else(|_| fail("MigrationFirstAdmission")); + let first_worker = nagoya::block_on(first).unwrap(); + let mut migrated = None; + for attempt in 0..64 { + let worker = (first_worker + 1 + attempt) % width; + let observer = runtime.executor().clone_handle(); + let next = owner.dispatch_to(runtime.executor(), Some(worker), 1_000_000, move |agent| { + agent.turn(); + observer.current_worker().unwrap() + }).unwrap_or_else(|_| fail("MigrationTurnAdmission")); + let actual = nagoya::block_on(next).unwrap(); + if actual != first_worker { migrated = Some(actual); break; } + } + let second_worker = migrated.unwrap_or_else(|| fail("MigrationNotObserved")); + require(Arc::ptr_eq(owner.key(), &account), "MigrationChangedOwnerKey"); + let remaining = account.balance(); + require(remaining < 1_000_000_000, "CpuConsumptionNotObserved"); + let denied_credits = if remaining > 0 { remaining as u64 + 1 } else { 1 }; + let rejected = owner.dispatch_to(runtime.executor(), Some(first_worker), denied_credits, + |agent| agent.turn()); + match rejected { + Err(rejected) if rejected.reason == DispatchDenied::Budget => {} + Ok(handle) => { + let _ = nagoya::block_on(handle); + fail("BudgetResetByMigration"); + } + _ => fail("BudgetResetByMigration"), + } + let helper = CpuCredits::new(thread_cpu_ns).dispatch(runtime.executor(), &account, + denied_credits, || 5_u64); + match helper { + Err(rejected) if rejected.reason == DispatchDenied::Budget => {} + Ok(handle) => { let _ = nagoya::block_on(handle); fail("HelperBudgetEscapedOwnerKey"); } + _ => fail("HelperBudgetEscapedOwnerKey"), + } + println!("PASS migration: worker {first_worker} -> {second_worker}, debt and helper cap retained"); +} + +fn timer(runtime: &Runtime) { + let mut source = ReadySource::new(); + let producer = source.handle(); + let start = nagoya::now_ns(); + let waits = nagoya::reactor::counters::WAIT.load(Ordering::Acquire); + let delivery = producer.wake_at(runtime.executor(), start + 500_000_000, + NonZeroU64::new(50_000_000).unwrap(), 4).unwrap(); + let receive = runtime.spawn(async move { source.wait().await }); + require(nagoya::block_on(receive) == Some(4), "TimerReadyReasonLost"); + require(nagoya::now_ns() >= start + 500_000_000, "TimerDeliveredEarly"); + require(nagoya::block_on(delivery).is_some(), "TimerDeliveryCancelled"); + let waits = nagoya::reactor::counters::WAIT.load(Ordering::Acquire) - waits; + // A quiet one-shot deadline has only arm/expiry waits. A planted 1 ms + // inspection interval produces hundreds of kernel waits before delivery. + require(waits <= 16, "TimerFiresByPolling"); + println!("PASS timer: coarse deadline delivered with {waits} event/deadline waits"); +} + +fn readiness(runtime: &Runtime, input: Arc) { + let (mut reader, mut writer) = UnixStream::pair().unwrap(); + reader.set_nonblocking(true).unwrap(); + let registration = runtime.readiness().unwrap().register(reader.as_raw_fd(), + nagoya::reactor::Interest::READABLE).unwrap(); + let expected = input.len(); + let received = runtime.spawn(async move { + let mut output = Vec::with_capacity(expected); + let mut chunk = [0_u8; 4096]; + while output.len() < expected { + let count = futures_util::future::poll_fn(|cx| { + match reader.read(&mut chunk) { + Ok(count) => Poll::Ready(count), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + registration.poll_readable(cx.waker()); + Poll::Pending + } + Err(_) => fail("ReadinessReadFailure"), + } + }).await; + require(count != 0, "ReadinessUnexpectedEof"); + output.extend_from_slice(&chunk[..count]); + } + output + }); + let payload = input.clone(); + let sent = runtime.spawn(async move { + nagoya::sleep(std::time::Duration::from_millis(25)).await; + writer.write_all(payload.as_bytes()).unwrap(); + }); + let output = nagoya::block_on(received).unwrap(); + require(output == input.as_bytes(), "ReadinessInputMismatch"); + require(nagoya::block_on(sent).is_some(), "ReadinessWriterCancelled"); + println!("PASS readiness: delivered {}", String::from_utf8(output).unwrap()); +} + +fn main() { + let mut args = std::env::args().skip(1); + let mode = args.next().unwrap_or_else(|| fail("MissingWitnessMode")); + let input: Arc = Arc::from(args.next().unwrap_or_else(|| fail("MissingWitnessInput"))); + require(!input.is_empty() && input.len() <= 4096, "WitnessInputSize"); + let runtime = Runtime::builder().all_cores().unwrap().install_background().unwrap(); + runtime.readiness().unwrap(); + match mode.as_str() { + "exclusive" => exclusive(runtime, input), + "migration" => migration(runtime, input), + "timer" => timer(runtime), + "readiness" => readiness(runtime, input), + _ => fail("UnknownWitnessMode"), + } +} diff --git a/scripts/agent-pool-functional.ts b/scripts/agent-pool-functional.ts new file mode 100644 index 0000000..0b0ef50 --- /dev/null +++ b/scripts/agent-pool-functional.ts @@ -0,0 +1,35 @@ +const [baseline, twoOwners, budgetReset, timerPolling] = Bun.argv.slice(2); +if (![baseline, twoOwners, budgetReset, timerPolling].every(Boolean)) { + throw new Error("Usage: bun functional-witness.ts BASELINE TWO_OWNERS BUDGET_RESET TIMER_POLLING"); +} +async function invoke(binary: string, mode: string): Promise<{ code: number; output: string }> { + const child = Bun.spawn([binary, mode, "owned tab input: wake and account"], { + stdout: "pipe", stderr: "pipe", + }); + let timedOut = false; + const deadline = setTimeout(() => { timedOut = true; child.kill(); }, 15_000); + const [stdout, stderr, code] = await Promise.all([ + new Response(child.stdout).text(), new Response(child.stderr).text(), child.exited, + ]); + clearTimeout(deadline); + if (timedOut) throw new Error(`${mode}: functional entry point did not complete`); + return { code, output: stdout + stderr }; +} +for (const mode of ["exclusive", "migration", "timer", "readiness"]) { + const result = await invoke(baseline, mode); + if (result.code !== 0 || !result.output.includes(`PASS ${mode}:`)) { + throw new Error(`Baseline ${mode} failed: ${result.output}`); + } + console.log(result.output.trim()); +} +for (const [binary, mode, name] of [ + [twoOwners, "exclusive", "ExclusiveOwnerDuplicateAdmission"], + [budgetReset, "migration", "BudgetResetByMigration"], + [timerPolling, "timer", "TimerFiresByPolling"], +]) { + const result = await invoke(binary, mode); + if (result.code === 0 || !result.output.includes(`FAIL ${name}`)) { + throw new Error(`${name}: planted production defect was not detected: ${result.output}`); + } + console.log(`PASS planted defect detected: ${name}`); +} From 0ea4ed89834db550b89622ef38212798aa021ae0 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:43:33 +0700 Subject: [PATCH 07/17] Add reactive credit waits --- docs/agent-pool.md | 9 ++++++--- src/credits.rs | 40 ++++++++++++++++++++++++++++++++++++---- 2 files changed, 42 insertions(+), 7 deletions(-) diff --git a/docs/agent-pool.md b/docs/agent-pool.md index f9424a0..508bd34 100644 --- a/docs/agent-pool.md +++ b/docs/agent-pool.md @@ -92,9 +92,12 @@ integration. Chuzz supplies the DAG, revision validation and deficit/deadline ordering, foreground dependency priority, control/progress capacity and site-to-agent selection. Admission hooks alone do not implement that policy. -Register `CreditAccount::changed()` with `Notified::enable`, recheck the -blocked dispatch and await only if still denied. Refunds and explicit refill -events wake waiters. Refill is one retained due deadline or a class/input +After denial, call `CreditAccount::wait_for` with the retained request size. +It registers then rechecks balance and in-flight state without attempting a +reservation or emitting its own rollback notification. After it returns, retry +the retained callback; another real admission may win first. `changed` remains +a lower-level notification hook for register/recheck of accounting state. +Refunds and explicit refill events wake waiters. Refill is one retained due deadline or a class/input event, never a scan. A background site has a shared in-flight limit; trusted foreground/control policy may use separate reserved authority. Native safepoints must bound CPU overshoot and termination. Cooperative dispatch diff --git a/src/credits.rs b/src/credits.rs index 67bfc13..1318796 100644 --- a/src/credits.rs +++ b/src/credits.rs @@ -53,19 +53,51 @@ impl CreditAccount { self.changed.notify_waiters(); } - /// Register before rechecking admission, then await a refund/refill event. - /// Use Notified::enable for register/recheck. Never scan accounts on a tick. + /// Register before inspecting balance and in-flight state, then await a + /// refund/refill event. Prefer wait_for after a rejected reservation: + /// attempting reservation here can itself emit a rollback notification. pub fn changed(&self) -> Notified<'_> { self.changed.notified() } + /// Suspend until this request could be admitted, without reserving or + /// emitting notifications. The caller then retries its retained callback. + /// A racing admission can win first, so repeat this wait after rejection. + pub async fn wait_for(&self, credits: NonZeroU64) -> Result<(), CreditRequestTooLarge> { + let credits = i64::try_from(credits.get()).map_err(|_| CreditRequestTooLarge)?; + if credits > self.capacity { return Err(CreditRequestTooLarge); } + loop { + let notified = self.changed.notified(); + let mut notified = core::pin::pin!(notified); + notified.as_mut().enable(); + if self.balance.load(Ordering::Acquire) >= credits && + self.in_flight.load(Ordering::Acquire) < self.parallelism { + return Ok(()); + } + notified.await; + } + } + fn adjust(&self, credits: i64) { let _ = self.balance.fetch_update(Ordering::AcqRel, Ordering::Acquire, |balance| Some(balance.saturating_add(credits).min(self.capacity))); } } +/// A request exceeds the account's maximum spendable credit. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CreditRequestTooLarge; + +impl core::fmt::Display for CreditRequestTooLarge { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.write_str("CPU credit request exceeds account capacity") + } +} + +#[cfg(feature = "std")] +impl std::error::Error for CreditRequestTooLarge {} + /// CPU admission using a host's monotonic current-worker CPU nanosecond clock. /// -/// The supplied function must measure CPU, not wall/guest time. A turn never +/// The supplied function must never panic and must measure CPU, not wall/guest time. A turn never /// suspends between its two reads, so per-worker clock origins may differ. /// This hook supplies accounting, not native preemption: callers must bound /// turns at real safepoints and react to retained debt by throttling or killing. @@ -113,7 +145,7 @@ impl CpuBudget> for CpuCredits { -> Option { let reserved = i64::try_from(credits).ok().filter(|value| *value > 0)?; key.in_flight.fetch_update(Ordering::AcqRel, Ordering::Acquire, - |count| (count < key.parallelism).then_some(count + 1)).ok()?; + |count| (count < key.parallelism).then(|| count + 1)).ok()?; if key.balance.fetch_update(Ordering::AcqRel, Ordering::Acquire, |balance| (balance >= reserved).then(|| balance - reserved)).is_err() { key.in_flight.fetch_sub(1, Ordering::AcqRel); From d5243a216e10e525482a4780bbfe0d2c2a145adb Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:46:45 +0700 Subject: [PATCH 08/17] Format agent pool code and gate reactor wake routing --- examples/agent_pool_witness.rs | 190 +++++++++++++++++++++++---------- src/credits.rs | 114 ++++++++++++++------ src/owner.rs | 82 ++++++++++---- src/reactor/driver.rs | 4 +- src/readiness.rs | 44 +++++--- src/runtime.rs | 7 +- src/task.rs | 22 ++-- src/time.rs | 25 +++-- 8 files changed, 348 insertions(+), 140 deletions(-) diff --git a/examples/agent_pool_witness.rs b/examples/agent_pool_witness.rs index 64a70bc..8bdd385 100644 --- a/examples/agent_pool_witness.rs +++ b/examples/agent_pool_witness.rs @@ -19,15 +19,22 @@ fn fail(name: &str) -> ! { } fn require(condition: bool, name: &str) { - if !condition { fail(name); } + if !condition { + fail(name); + } } fn thread_cpu_ns() -> u64 { - let mut value = libc::timespec { tv_sec: 0, tv_nsec: 0 }; + let mut value = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; // SAFETY: value is a writable timespec; this reads the current worker. let result = unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &mut value) }; require(result == 0, "WorkerCpuClockUnavailable"); - (value.tv_sec as u64).saturating_mul(1_000_000_000).saturating_add(value.tv_nsec as u64) + (value.tv_sec as u64) + .saturating_mul(1_000_000_000) + .saturating_add(value.tv_nsec as u64) } struct Agent { @@ -38,8 +45,10 @@ struct Agent { impl Agent { fn turn(&mut self) -> u64 { - require(self.active.fetch_add(1, Ordering::AcqRel) == 0, - "ExclusiveOwnerDuplicateAdmission"); + require( + self.active.fetch_add(1, Ordering::AcqRel) == 0, + "ExclusiveOwnerDuplicateAdmission", + ); let mut digest = 0xcbf29ce484222325_u64; for byte in self.input.bytes() { digest = (digest ^ u64::from(byte)).wrapping_mul(0x100000001b3); @@ -54,24 +63,43 @@ type Account = CreditAccount>; type AgentOwner = Owner; fn owner(input: Arc) -> (AgentOwner, Arc) { - let account = Arc::new(CreditAccount::new(Arc::from("https://witness.invalid"), - NonZeroU64::new(1_000_000_000).unwrap(), NonZeroUsize::new(2).unwrap()).unwrap()); - let value = Agent { turns: Cell::new(0), active: Arc::new(AtomicUsize::new(0)), input }; - (Owner::new(value, account.clone(), Arc::new(CpuCredits::new(thread_cpu_ns))), account) + let account = Arc::new( + CreditAccount::new( + Arc::from("https://witness.invalid"), + NonZeroU64::new(1_000_000_000).unwrap(), + NonZeroUsize::new(2).unwrap(), + ) + .unwrap(), + ); + let value = Agent { + turns: Cell::new(0), + active: Arc::new(AtomicUsize::new(0)), + input, + }; + ( + Owner::new( + value, + account.clone(), + Arc::new(CpuCredits::new(thread_cpu_ns)), + ), + account, + ) } fn exclusive(runtime: &Runtime, input: Arc) { let (owner, _) = owner(input); let (mut reader, mut writer) = UnixStream::pair().unwrap(); - let first = owner.dispatch(runtime.executor(), 1_000_000, move |agent| { - let mut delivery = [0_u8; 1]; - // One bounded witness rendezvous, released by CLI coordinator below. - // The ownership gate remains held while queued or executing. - reader.read_exact(&mut delivery).unwrap(); - require(delivery == [7], "OwnerInputDelivery"); - agent.turn(); - agent.turns.get() - }).unwrap_or_else(|_| fail("OwnerFirstAdmission")); + let first = owner + .dispatch(runtime.executor(), 1_000_000, move |agent| { + let mut delivery = [0_u8; 1]; + // One bounded witness rendezvous, released by CLI coordinator below. + // The ownership gate remains held while queued or executing. + reader.read_exact(&mut delivery).unwrap(); + require(delivery == [7], "OwnerInputDelivery"); + agent.turn(); + agent.turns.get() + }) + .unwrap_or_else(|_| fail("OwnerFirstAdmission")); let second = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { owner.dispatch(runtime.executor(), 1_000_000, |agent| agent.turn()) })); @@ -85,10 +113,12 @@ fn exclusive(runtime: &Runtime, input: Arc) { } _ => fail("ExclusiveOwnerDuplicateAdmission"), } - let next = owner.dispatch(runtime.executor(), 1_000_000, |agent| { - agent.turn(); - agent.turns.get() - }).unwrap_or_else(|_| fail("OwnerHandoffAdmission")); + let next = owner + .dispatch(runtime.executor(), 1_000_000, |agent| { + agent.turn(); + agent.turns.get() + }) + .unwrap_or_else(|_| fail("OwnerHandoffAdmission")); require(nagoya::block_on(next) == Some(2), "OwnerHandoffOutput"); println!("PASS exclusive: one admission, owned state retained through two turns"); } @@ -98,29 +128,47 @@ fn migration(runtime: &Runtime, input: Arc) { require(width >= 2, "MigrationRequiresTwoWorkers"); let (owner, account) = owner(input); let observer = runtime.executor().clone_handle(); - let first = owner.dispatch_to(runtime.executor(), Some(0), 1_000_000, move |agent| { - agent.turn(); - observer.current_worker().unwrap() - }).unwrap_or_else(|_| fail("MigrationFirstAdmission")); + let first = owner + .dispatch_to(runtime.executor(), Some(0), 1_000_000, move |agent| { + agent.turn(); + observer.current_worker().unwrap() + }) + .unwrap_or_else(|_| fail("MigrationFirstAdmission")); let first_worker = nagoya::block_on(first).unwrap(); let mut migrated = None; for attempt in 0..64 { let worker = (first_worker + 1 + attempt) % width; let observer = runtime.executor().clone_handle(); - let next = owner.dispatch_to(runtime.executor(), Some(worker), 1_000_000, move |agent| { - agent.turn(); - observer.current_worker().unwrap() - }).unwrap_or_else(|_| fail("MigrationTurnAdmission")); + let next = owner + .dispatch_to(runtime.executor(), Some(worker), 1_000_000, move |agent| { + agent.turn(); + observer.current_worker().unwrap() + }) + .unwrap_or_else(|_| fail("MigrationTurnAdmission")); let actual = nagoya::block_on(next).unwrap(); - if actual != first_worker { migrated = Some(actual); break; } + if actual != first_worker { + migrated = Some(actual); + break; + } } let second_worker = migrated.unwrap_or_else(|| fail("MigrationNotObserved")); - require(Arc::ptr_eq(owner.key(), &account), "MigrationChangedOwnerKey"); + require( + Arc::ptr_eq(owner.key(), &account), + "MigrationChangedOwnerKey", + ); let remaining = account.balance(); require(remaining < 1_000_000_000, "CpuConsumptionNotObserved"); - let denied_credits = if remaining > 0 { remaining as u64 + 1 } else { 1 }; - let rejected = owner.dispatch_to(runtime.executor(), Some(first_worker), denied_credits, - |agent| agent.turn()); + let denied_credits = if remaining > 0 { + remaining as u64 + 1 + } else { + 1 + }; + let rejected = owner.dispatch_to( + runtime.executor(), + Some(first_worker), + denied_credits, + |agent| agent.turn(), + ); match rejected { Err(rejected) if rejected.reason == DispatchDenied::Budget => {} Ok(handle) => { @@ -129,14 +177,23 @@ fn migration(runtime: &Runtime, input: Arc) { } _ => fail("BudgetResetByMigration"), } - let helper = CpuCredits::new(thread_cpu_ns).dispatch(runtime.executor(), &account, - denied_credits, || 5_u64); + let helper = CpuCredits::new(thread_cpu_ns).dispatch( + runtime.executor(), + &account, + denied_credits, + || 5_u64, + ); match helper { Err(rejected) if rejected.reason == DispatchDenied::Budget => {} - Ok(handle) => { let _ = nagoya::block_on(handle); fail("HelperBudgetEscapedOwnerKey"); } + Ok(handle) => { + let _ = nagoya::block_on(handle); + fail("HelperBudgetEscapedOwnerKey"); + } _ => fail("HelperBudgetEscapedOwnerKey"), } - println!("PASS migration: worker {first_worker} -> {second_worker}, debt and helper cap retained"); + println!( + "PASS migration: worker {first_worker} -> {second_worker}, debt and helper cap retained" + ); } fn timer(runtime: &Runtime) { @@ -144,12 +201,24 @@ fn timer(runtime: &Runtime) { let producer = source.handle(); let start = nagoya::now_ns(); let waits = nagoya::reactor::counters::WAIT.load(Ordering::Acquire); - let delivery = producer.wake_at(runtime.executor(), start + 500_000_000, - NonZeroU64::new(50_000_000).unwrap(), 4).unwrap(); + let delivery = producer + .wake_at( + runtime.executor(), + start + 500_000_000, + NonZeroU64::new(50_000_000).unwrap(), + 4, + ) + .unwrap(); let receive = runtime.spawn(async move { source.wait().await }); require(nagoya::block_on(receive) == Some(4), "TimerReadyReasonLost"); - require(nagoya::now_ns() >= start + 500_000_000, "TimerDeliveredEarly"); - require(nagoya::block_on(delivery).is_some(), "TimerDeliveryCancelled"); + require( + nagoya::now_ns() >= start + 500_000_000, + "TimerDeliveredEarly", + ); + require( + nagoya::block_on(delivery).is_some(), + "TimerDeliveryCancelled", + ); let waits = nagoya::reactor::counters::WAIT.load(Ordering::Acquire) - waits; // A quiet one-shot deadline has only arm/expiry waits. A planted 1 ms // inspection interval produces hundreds of kernel waits before delivery. @@ -160,23 +229,25 @@ fn timer(runtime: &Runtime) { fn readiness(runtime: &Runtime, input: Arc) { let (mut reader, mut writer) = UnixStream::pair().unwrap(); reader.set_nonblocking(true).unwrap(); - let registration = runtime.readiness().unwrap().register(reader.as_raw_fd(), - nagoya::reactor::Interest::READABLE).unwrap(); + let registration = runtime + .readiness() + .unwrap() + .register(reader.as_raw_fd(), nagoya::reactor::Interest::READABLE) + .unwrap(); let expected = input.len(); let received = runtime.spawn(async move { let mut output = Vec::with_capacity(expected); let mut chunk = [0_u8; 4096]; while output.len() < expected { - let count = futures_util::future::poll_fn(|cx| { - match reader.read(&mut chunk) { - Ok(count) => Poll::Ready(count), - Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { - registration.poll_readable(cx.waker()); - Poll::Pending - } - Err(_) => fail("ReadinessReadFailure"), + let count = futures_util::future::poll_fn(|cx| match reader.read(&mut chunk) { + Ok(count) => Poll::Ready(count), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + registration.poll_readable(cx.waker()); + Poll::Pending } - }).await; + Err(_) => fail("ReadinessReadFailure"), + }) + .await; require(count != 0, "ReadinessUnexpectedEof"); output.extend_from_slice(&chunk[..count]); } @@ -190,7 +261,10 @@ fn readiness(runtime: &Runtime, input: Arc) { let output = nagoya::block_on(received).unwrap(); require(output == input.as_bytes(), "ReadinessInputMismatch"); require(nagoya::block_on(sent).is_some(), "ReadinessWriterCancelled"); - println!("PASS readiness: delivered {}", String::from_utf8(output).unwrap()); + println!( + "PASS readiness: delivered {}", + String::from_utf8(output).unwrap() + ); } fn main() { @@ -198,7 +272,11 @@ fn main() { let mode = args.next().unwrap_or_else(|| fail("MissingWitnessMode")); let input: Arc = Arc::from(args.next().unwrap_or_else(|| fail("MissingWitnessInput"))); require(!input.is_empty() && input.len() <= 4096, "WitnessInputSize"); - let runtime = Runtime::builder().all_cores().unwrap().install_background().unwrap(); + let runtime = Runtime::builder() + .all_cores() + .unwrap() + .install_background() + .unwrap(); runtime.readiness().unwrap(); match mode.as_str() { "exclusive" => exclusive(runtime, input), diff --git a/src/credits.rs b/src/credits.rs index 1318796..3f00de0 100644 --- a/src/credits.rs +++ b/src/credits.rs @@ -1,11 +1,11 @@ //! Shared CPU credit authority with cancellation refunds and overshoot debt. -use alloc::sync::Arc; -use core::num::{NonZeroU64, NonZeroUsize}; -use core::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; use crate::owner::{CpuBudget, CpuPermit, DispatchDenied, Rejected}; use crate::sync::{Notified, Notify}; use crate::{Executor, JoinHandle}; +use alloc::sync::Arc; +use core::num::{NonZeroU64, NonZeroUsize}; +use core::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; /// One stable accounting key, shared across its agents and helper jobs. /// @@ -27,23 +27,32 @@ impl CreditAccount { pub fn new(key: K, capacity: NonZeroU64, parallelism: NonZeroUsize) -> Option { let capacity = i64::try_from(capacity.get()).ok()?; Some(Self { - key, capacity, balance: AtomicI64::new(capacity), - in_flight: AtomicUsize::new(0), parallelism: parallelism.get(), + key, + capacity, + balance: AtomicI64::new(capacity), + in_flight: AtomicUsize::new(0), + parallelism: parallelism.get(), changed: Notify::new(), }) } /// Immutable policy identity, such as a schemeful site and incarnation. #[must_use] - pub fn key(&self) -> &K { &self.key } + pub fn key(&self) -> &K { + &self.key + } /// Spendable CPU nanoseconds; a negative value is retained overshoot debt. #[must_use] - pub fn balance(&self) -> i64 { self.balance.load(Ordering::Acquire) } + pub fn balance(&self) -> i64 { + self.balance.load(Ordering::Acquire) + } /// Number of jobs reserved or executing across all workers. #[must_use] - pub fn in_flight(&self) -> usize { self.in_flight.load(Ordering::Acquire) } + pub fn in_flight(&self) -> usize { + self.in_flight.load(Ordering::Acquire) + } /// Add policy-approved CPU nanoseconds at a retained refill/class event. /// Credit first pays debt and is capped at this account's capacity. @@ -56,20 +65,25 @@ impl CreditAccount { /// Register before inspecting balance and in-flight state, then await a /// refund/refill event. Prefer wait_for after a rejected reservation: /// attempting reservation here can itself emit a rollback notification. - pub fn changed(&self) -> Notified<'_> { self.changed.notified() } + pub fn changed(&self) -> Notified<'_> { + self.changed.notified() + } /// Suspend until this request could be admitted, without reserving or /// emitting notifications. The caller then retries its retained callback. /// A racing admission can win first, so repeat this wait after rejection. pub async fn wait_for(&self, credits: NonZeroU64) -> Result<(), CreditRequestTooLarge> { let credits = i64::try_from(credits.get()).map_err(|_| CreditRequestTooLarge)?; - if credits > self.capacity { return Err(CreditRequestTooLarge); } + if credits > self.capacity { + return Err(CreditRequestTooLarge); + } loop { let notified = self.changed.notified(); let mut notified = core::pin::pin!(notified); notified.as_mut().enable(); - if self.balance.load(Ordering::Acquire) >= credits && - self.in_flight.load(Ordering::Acquire) < self.parallelism { + if self.balance.load(Ordering::Acquire) >= credits + && self.in_flight.load(Ordering::Acquire) < self.parallelism + { return Ok(()); } notified.await; @@ -77,8 +91,11 @@ impl CreditAccount { } fn adjust(&self, credits: i64) { - let _ = self.balance.fetch_update(Ordering::AcqRel, Ordering::Acquire, - |balance| Some(balance.saturating_add(credits).min(self.capacity))); + let _ = self + .balance + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |balance| { + Some(balance.saturating_add(credits).min(self.capacity)) + }); } } @@ -108,17 +125,29 @@ pub struct CpuCredits { impl CpuCredits { /// Bind the host's worker CPU clock without installing an OS service. #[must_use] - pub const fn new(clock: fn() -> u64) -> Self { Self { clock } } + pub const fn new(clock: fn() -> u64) -> Self { + Self { clock } + } /// Admit an attributed helper on the same executor, measuring its CPU. /// The rejected closure retains its typed owned/Arc input. - pub fn dispatch(&self, executor: &Executor, - key: &Arc>, credits: u64, work: F) - -> Result, Rejected> - where K: Send + Sync + 'static, F: FnOnce() -> R + Send + 'static, - R: Send + 'static { + pub fn dispatch( + &self, + executor: &Executor, + key: &Arc>, + credits: u64, + work: F, + ) -> Result, Rejected> + where + K: Send + Sync + 'static, + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, + { let Some(mut permit) = self.reserve(key, credits) else { - return Err(Rejected { reason: DispatchDenied::Budget, turn: work }); + return Err(Rejected { + reason: DispatchDenied::Budget, + turn: work, + }); }; Ok(executor.spawn(async move { permit.enter(); @@ -141,27 +170,39 @@ pub struct CpuReservation { impl CpuBudget> for CpuCredits { type Permit = CpuReservation; - fn reserve(&self, key: &Arc>, credits: u64) - -> Option { + fn reserve(&self, key: &Arc>, credits: u64) -> Option { let reserved = i64::try_from(credits).ok().filter(|value| *value > 0)?; - key.in_flight.fetch_update(Ordering::AcqRel, Ordering::Acquire, - |count| (count < key.parallelism).then(|| count + 1)).ok()?; - if key.balance.fetch_update(Ordering::AcqRel, Ordering::Acquire, - |balance| (balance >= reserved).then(|| balance - reserved)).is_err() { + key.in_flight + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + (count < key.parallelism).then(|| count + 1) + }) + .ok()?; + if key + .balance + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |balance| { + (balance >= reserved).then(|| balance - reserved) + }) + .is_err() + { key.in_flight.fetch_sub(1, Ordering::AcqRel); key.changed.notify_waiters(); return None; } Some(CpuReservation { - account: Arc::clone(key), reserved, clock: self.clock, - started: None, done: false, + account: Arc::clone(key), + reserved, + clock: self.clock, + started: None, + done: false, }) } } impl CpuReservation { fn settle(&mut self) { - if self.done { return; } + if self.done { + return; + } let actual = self.started.map_or(0, |start| { i64::try_from((self.clock)().saturating_sub(start)).unwrap_or(i64::MAX) }); @@ -174,12 +215,19 @@ impl CpuReservation { impl CpuPermit for CpuReservation { fn enter(&mut self) { - assert!(!self.done && self.started.is_none(), "CPU reservation entered twice"); + assert!( + !self.done && self.started.is_none(), + "CPU reservation entered twice" + ); self.started = Some((self.clock)()); } - fn finish(&mut self) { self.settle(); } + fn finish(&mut self) { + self.settle(); + } } impl Drop for CpuReservation { - fn drop(&mut self) { self.settle(); } + fn drop(&mut self) { + self.settle(); + } } diff --git a/src/owner.rs b/src/owner.rs index 5edff76..4418fa4 100644 --- a/src/owner.rs +++ b/src/owner.rs @@ -1,9 +1,9 @@ //! Typed exclusive turns with admission before queue submission. +use crate::{Executor, JoinHandle}; use alloc::sync::Arc; use core::cell::UnsafeCell; use core::sync::atomic::{AtomicBool, Ordering}; -use crate::{Executor, JoinHandle}; /// Caller-owned CPU admission policy, shared by every worker and helper. pub trait CpuBudget: Send + Sync + 'static { @@ -43,7 +43,9 @@ impl Drop for Lease { fn drop(&mut self) { // SAFETY: this lease is the sole successful admission, including // cancellation before execution and unwinding during a turn. - unsafe { *self.state.value.get() = self.value.take(); } + unsafe { + *self.state.value.get() = self.value.take(); + } self.state.admitted.store(false, Ordering::Release); } } @@ -104,12 +106,21 @@ impl> Owner &Arc { &self.key } + pub fn key(&self) -> &Arc { + &self.key + } /// Dispatch one complete synchronous turn on any worker. - pub fn dispatch(&self, executor: &Executor, credits: u64, turn: F) - -> Result, Rejected> - where F: FnOnce(&mut T) -> R + Send + 'static, R: Send + 'static { + pub fn dispatch( + &self, + executor: &Executor, + credits: u64, + turn: F, + ) -> Result, Rejected> + where + F: FnOnce(&mut T) -> R + Send + 'static, + R: Send + 'static, + { self.dispatch_to(executor, None, credits, turn) } @@ -117,23 +128,47 @@ impl> Owner(&self, executor: &Executor, worker: Option, - credits: u64, turn: F) -> Result, Rejected> - where F: FnOnce(&mut T) -> R + Send + 'static, R: Send + 'static { + pub fn dispatch_to( + &self, + executor: &Executor, + worker: Option, + credits: u64, + turn: F, + ) -> Result, Rejected> + where + F: FnOnce(&mut T) -> R + Send + 'static, + R: Send + 'static, + { if worker.is_some_and(|id| id >= executor.pool().workers()) { - return Err(Rejected { reason: DispatchDenied::Worker, turn }); + return Err(Rejected { + reason: DispatchDenied::Worker, + turn, + }); } - if self.state.admitted.compare_exchange(false, true, - Ordering::Acquire, Ordering::Relaxed).is_err() { - return Err(Rejected { reason: DispatchDenied::Busy, turn }); + if self + .state + .admitted + .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed) + .is_err() + { + return Err(Rejected { + reason: DispatchDenied::Busy, + turn, + }); } // SAFETY: the successful admission owns this slot until Lease drops. - let value = unsafe { (*self.state.value.get()).take() } - .expect("ExclusiveOwnerDuplicateAdmission"); - let mut lease = Lease { state: Arc::clone(&self.state), value: Some(value) }; + let value = + unsafe { (*self.state.value.get()).take() }.expect("ExclusiveOwnerDuplicateAdmission"); + let mut lease = Lease { + state: Arc::clone(&self.state), + value: Some(value), + }; let Some(mut permit) = self.budget.reserve(&self.key, credits) else { drop(lease); - return Err(Rejected { reason: DispatchDenied::Budget, turn }); + return Err(Rejected { + reason: DispatchDenied::Budget, + turn, + }); }; let future = async move { permit.enter(); @@ -145,7 +180,16 @@ impl> Owner) -> (Option, bool) { // kernel wait. An arm racing with fire_due's return must survive publication. shared.next_deadline.swap(NO_DEADLINE, Ordering::AcqRel); let (next, woke) = crate::time::fire_due(now); - shared.next_deadline.fetch_min(next.unwrap_or(NO_DEADLINE), Ordering::AcqRel); + shared + .next_deadline + .fetch_min(next.unwrap_or(NO_DEADLINE), Ordering::AcqRel); let cached = shared.next_deadline.load(Ordering::Acquire); let wait = (cached != NO_DEADLINE).then(|| cached.saturating_sub(now)); (wait, woke) diff --git a/src/readiness.rs b/src/readiness.rs index 2beebed..3f7239c 100644 --- a/src/readiness.rs +++ b/src/readiness.rs @@ -1,10 +1,10 @@ //! Retained, coalesced wake reasons for one logical scheduling owner. +use crate::sync::Notify; +use crate::{Executor, JoinHandle}; use alloc::sync::Arc; use core::num::NonZeroU64; use core::sync::atomic::{AtomicU64, Ordering}; -use crate::sync::Notify; -use crate::{Executor, JoinHandle}; struct Signals { reasons: AtomicU64, @@ -26,22 +26,29 @@ pub struct ReadyHandle { } impl Default for ReadySource { - fn default() -> Self { Self::new() } + fn default() -> Self { + Self::new() + } } impl ReadySource { /// Make a single-consumer source with no pending reasons. #[must_use] pub fn new() -> Self { - Self { signals: Arc::new(Signals { - reasons: AtomicU64::new(0), notify: Notify::new(), - }) } + Self { + signals: Arc::new(Signals { + reasons: AtomicU64::new(0), + notify: Notify::new(), + }), + } } /// Give a timer, readiness callback or typed completion an owned producer. #[must_use] pub fn handle(&self) -> ReadyHandle { - ReadyHandle { signals: Arc::clone(&self.signals) } + ReadyHandle { + signals: Arc::clone(&self.signals), + } } /// Consume retained reasons, or register and suspend until a real event. @@ -54,7 +61,9 @@ impl ReadySource { let mut notified = core::pin::pin!(notified); notified.as_mut().enable(); let reasons = self.signals.reasons.swap(0, Ordering::AcqRel); - if reasons != 0 { return reasons; } + if reasons != 0 { + return reasons; + } notified.await; } } @@ -77,8 +86,14 @@ impl std::error::Error for DeadlineOverflow {} /// u64::MAX is reserved by the reactor for an unarmed deadline. pub fn coarse_deadline(deadline: u64, quantum: NonZeroU64) -> Result { let remainder = deadline % quantum.get(); - let extra = if remainder == 0 { 0 } else { quantum.get() - remainder }; - deadline.checked_add(extra).filter(|value| *value != u64::MAX) + let extra = if remainder == 0 { + 0 + } else { + quantum.get() - remainder + }; + deadline + .checked_add(extra) + .filter(|value| *value != u64::MAX) .ok_or(DeadlineOverflow) } @@ -97,8 +112,13 @@ impl ReadyHandle { /// The caller retains the JoinHandle and cancels it when eligibility or /// the due event changes. It must initialize Runtime::readiness first. /// This is delivery only; it does not grant CPU or service-worker admission. - pub fn wake_at(&self, executor: &Executor, deadline: u64, - quantum: NonZeroU64, reasons: u64) -> Result, DeadlineOverflow> { + pub fn wake_at( + &self, + executor: &Executor, + deadline: u64, + quantum: NonZeroU64, + reasons: u64, + ) -> Result, DeadlineOverflow> { let deadline = coarse_deadline(deadline, quantum)?; let handle = self.clone(); Ok(executor.spawn(async move { diff --git a/src/runtime.rs b/src/runtime.rs index 31abe3c..e8d738f 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -280,7 +280,11 @@ impl Builder { installed = true; self.build() }); - if installed { Ok(runtime) } else { Err(BackgroundStarted) } + if installed { + Ok(runtime) + } else { + Err(BackgroundStarted) + } } /// The idle policy. [`Tuning::default`] unless set. @@ -435,6 +439,7 @@ fn supported_default_threads(available: usize) -> usize { /// Spreads descriptors over the pool by index while keeping each one on a /// stable worker, so a task finds its own cache lines without every /// connection sharing a queue. +#[cfg(feature = "reactor")] pub(crate) fn route_reactor_wake(index: u64) -> Option { // Only route onto a pool that is already running. Asking `background()` // for its width *starts* it, one thread per core, and the reactor is not diff --git a/src/task.rs b/src/task.rs index 98f2d71..c21cd5e 100644 --- a/src/task.rs +++ b/src/task.rs @@ -37,7 +37,7 @@ std::thread_local! { /// measured 38 percent faster at eight connections and two and a half times /// slower at ten thousand, which is the wrong trade. Routing by descriptor /// spreads them and keeps each one settled. -#[cfg(feature = "std")] +#[cfg(feature = "reactor")] pub(crate) fn route_wakes(worker: usize) -> RouteGuard { let previous = ROUTE.with(|route| route.replace(Some(worker))); RouteGuard { @@ -46,13 +46,13 @@ pub(crate) fn route_wakes(worker: usize) -> RouteGuard { } } -#[cfg(feature = "std")] +#[cfg(feature = "reactor")] pub(crate) struct RouteGuard { previous: Option, _thread_bound: core::marker::PhantomData>, } -#[cfg(feature = "std")] +#[cfg(feature = "reactor")] impl Drop for RouteGuard { fn drop(&mut self) { ROUTE.with(|route| route.set(self.previous.take())); @@ -129,9 +129,15 @@ where } /// The ordinary task path with a preference for its first queue only. -pub(crate) fn spawn_on_at(future: F, pool: Arc, - context: Option>, first_worker: Option) -> Task -where F: Future + Send + 'static, F::Output: Send + 'static, +pub(crate) fn spawn_on_at( + future: F, + pool: Arc, + context: Option>, + first_worker: Option, +) -> Task +where + F: Future + Send + 'static, + F::Output: Send + 'static, { // Called for the first poll and for every wake after it. The `Runnable` is // a single heap allocation that already holds the future and its state, so @@ -333,6 +339,6 @@ mod tests { /// `spawn_local` either; the non-`Send` equivalent is a /// [`TaskSet`](crate::reactor::TaskSet) driven by /// [`block_on_with`](crate::reactor::block_on_with). -pub use crate::yield_now::{yield_now, YieldNow}; +pub use crate::yield_now::{YieldNow, yield_now}; #[cfg(feature = "std")] -pub use crate::{spawn, JoinHandle}; +pub use crate::{JoinHandle, spawn}; diff --git a/src/time.rs b/src/time.rs index 6367a2c..a393bab 100644 --- a/src/time.rs +++ b/src/time.rs @@ -480,8 +480,7 @@ impl std::error::Error for TimerDriverStarted {} /// supplies that process-lifetime ownership. Clock reads alone do not select /// the backend. This function never replaces an already-selected driver. #[cfg(feature = "reactor")] -pub fn set_timer_reactor(handle: crate::reactor::Handle) - -> Result<(), TimerDriverStarted> { +pub fn set_timer_reactor(handle: crate::reactor::Handle) -> Result<(), TimerDriverStarted> { driver::bind_reactor(handle) } @@ -504,9 +503,11 @@ mod driver { static HOST: OnceLock = OnceLock::new(); #[cfg(feature = "reactor")] - pub(super) fn bind_reactor(handle: crate::reactor::Handle) - -> Result<(), super::TimerDriverStarted> { - HOST.set(Host::Reactor(handle)).map_err(|_| super::TimerDriverStarted) + pub(super) fn bind_reactor( + handle: crate::reactor::Handle, + ) -> Result<(), super::TimerDriverStarted> { + HOST.set(Host::Reactor(handle)) + .map_err(|_| super::TimerDriverStarted) } fn std_clock() -> u64 { @@ -537,7 +538,9 @@ mod driver { } #[cfg(feature = "reactor")] Host::Reactor(handle) => { - handle.timer_armed(deadline).expect("nagoya timer reactor wake"); + handle + .timer_armed(deadline) + .expect("nagoya timer reactor wake"); } } } @@ -585,7 +588,7 @@ mod driver { #[cfg(all(test, feature = "std"))] mod tests { - use super::{now_ns, sleep, timeout, Elapsed}; + use super::{Elapsed, now_ns, sleep, timeout}; use crate::block_on; use core::time::Duration; use std::time::Instant; @@ -652,9 +655,11 @@ mod tests { use std::task::Waker; let waker = Waker::noop(); let mut sleep = super::sleep_until(u64::MAX); - assert!(Pin::new(&mut sleep) - .poll(&mut Context::from_waker(waker)) - .is_pending()); + assert!( + Pin::new(&mut sleep) + .poll(&mut Context::from_waker(waker)) + .is_pending() + ); let weak = Arc::downgrade(sleep.slot.as_ref().unwrap()); drop(sleep); assert!( From 212c9f754add230f5bc4e597dcce41ff4ec50fb5 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:47:57 +0700 Subject: [PATCH 09/17] Use explicit unsafe blocks in sockets --- src/reactor/socket.rs | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/reactor/socket.rs b/src/reactor/socket.rs index 8019f26..4d2b545 100644 --- a/src/reactor/socket.rs +++ b/src/reactor/socket.rs @@ -708,7 +708,8 @@ impl TcpSocket { /// `fd` must be an open, non-blocking, connected TCP socket that nothing /// else will close. pub unsafe fn from_raw(fd: i32) -> Self { - Self(Fd::from_raw(fd)) + // SAFETY: the caller guarantees the descriptor is open and uniquely owned. + Self(unsafe { Fd::from_raw(fd) }) } /// The descriptor, for registering with the poller. @@ -804,7 +805,9 @@ impl TcpSocket { pub unsafe fn recv(&self, pointer: *mut u8, len: usize) -> Result { #[cfg(feature = "syscall-counters")] crate::reactor::counters::RECV.fetch_add(1, core::sync::atomic::Ordering::Relaxed); - let read = libc::recv(self.raw(), pointer.cast(), len, 0); + // SAFETY: the caller guarantees a writable buffer of `len` bytes, and + // this socket owns a live descriptor. + let read = unsafe { libc::recv(self.raw(), pointer.cast(), len, 0) }; if read < 0 { let error = crate::reactor::error::last(); #[cfg(feature = "syscall-counters")] @@ -911,7 +914,8 @@ impl UdpSocket { /// `fd` must be an open, non-blocking datagram socket that nothing else /// will close. pub unsafe fn from_raw(fd: i32) -> Self { - Self(Fd::from_raw(fd)) + // SAFETY: the caller guarantees the descriptor is open and uniquely owned. + Self(unsafe { Fd::from_raw(fd) }) } /// The descriptor, for registering with the poller. From 956cf401f607977e38c69cf24ccd61cdca6478ec Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:47:57 +0700 Subject: [PATCH 10/17] Return reactor read result directly --- tests/reactor_unix.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/reactor_unix.rs b/tests/reactor_unix.rs index 737bcd5..4da25fa 100644 --- a/tests/reactor_unix.rs +++ b/tests/reactor_unix.rs @@ -253,8 +253,7 @@ fn read_exact_names_a_stream_that_ends_early() { let mut length = [0u8; 4]; stream.read_exact(&mut length).await.expect("length"); let mut payload = vec![0u8; u32::from_be_bytes(length) as usize]; - let outcome = stream.read_exact(&mut payload).await; - outcome + stream.read_exact(&mut payload).await }); server.join().expect("server"); From 8b133201db0ebfb2b3f214b5b0534ddfda455c6c Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 12:56:33 +0700 Subject: [PATCH 11/17] Exclude repository workflows from the crate --- Cargo.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/Cargo.toml b/Cargo.toml index 35c6ac4..5bf0c3b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,7 @@ license = "MIT OR Apache-2.0" description = "An async runtime that does not need an operating system" repository = "https://github.com/pathscale/nagoya" readme = "README.md" +exclude = [".github/**"] keywords = ["async", "runtime", "executor", "no-std", "work-stealing"] categories = ["asynchronous", "concurrency", "no-std"] From 7960993c5269f28e58d9c40227ab9adaa4e00a5c Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 13:22:13 +0700 Subject: [PATCH 12/17] Use the published ps-st3 --- Cargo.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 5bf0c3b..f3cf8de 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,7 @@ syscall-counters = ["reactor"] [dependencies] -ps-st3 = { path = "../ps-st3", version = "^0.6", default-features = false, features = ["fanout"] } +ps-st3 = { version = "^0.6", default-features = false, features = ["fanout"] } spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"] } # The task itself: one allocation holding the future, its state machine, its # waker vtable and its output, with `Runnable::into_raw` handing the pointer @@ -57,7 +57,7 @@ futures-io = { version = "0.3", default-features = false, optional = true } futures-util = { version = "0.3", default-features = false, optional = true } [dev-dependencies] -ps-st3 = { path = "../ps-st3", version = "^0.6", features = ["fanout", "host"] } +ps-st3 = { version = "^0.6", features = ["fanout", "host"] } # The comparison arm. Only the scheduler is used; nothing here needs its I/O. tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util"] } # The other comparison arm. Not an async runtime: a data-parallel pool, so it From 9201727d34a1415b18fcf68a722bbffa699ef5be Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 13:44:04 +0700 Subject: [PATCH 13/17] ci: never persist the GitHub token in the checkout's Git config --- .github/workflows/ci.yml | 4 ++++ .github/workflows/publish.yml | 2 ++ 2 files changed, 6 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 42eccbc..0d9d0b3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,6 +14,8 @@ jobs: rust: [stable, '1.85.0'] steps: - uses: actions/checkout@v4 + with: + persist-credentials: false - uses: dtolnay/rust-toolchain@master with: toolchain: ${{ matrix.rust }} @@ -29,6 +31,8 @@ jobs: timeout-minutes: 2 steps: - uses: actions/checkout@v4 + with: + persist-credentials: false - uses: dtolnay/rust-toolchain@stable with: components: rustfmt, clippy diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 0b79c99..841f14d 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -20,6 +20,8 @@ jobs: runs-on: ubicloud-standard-2 steps: - uses: actions/checkout@v4 + with: + persist-credentials: false - uses: dtolnay/rust-toolchain@stable # The registry is the only authority on what is already released. From 3a4ea1eb74694eccc07daf2381911d92f58417fb Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 13:44:04 +0700 Subject: [PATCH 14/17] Format examples, sources and tests --- examples/counter_floor.rs | 2 +- examples/nagoya_only.rs | 12 +++++++----- examples/profile_me.rs | 2 +- examples/profile_tokio.rs | 2 +- examples/read_shape.rs | 4 ++-- examples/submit_only.rs | 2 +- examples/three_pools.rs | 2 +- examples/uds_scale.rs | 34 ++++++++++++++++++++-------------- examples/uds_vs_tokio.rs | 4 ++-- examples/wake_latency.rs | 2 +- examples/where_the_time_is.rs | 2 +- src/io.rs | 6 +++--- src/lib.rs | 15 ++++++++------- src/net.rs | 6 +++--- src/reactor.rs | 6 +++--- src/reactor/poller.rs | 4 ++-- src/resolve.rs | 2 +- src/signal.rs | 10 +++++----- tests/executor.rs | 14 ++++++++------ tests/reactor_local.rs | 4 ++-- tests/reactor_local_timeout.rs | 4 ++-- tests/reactor_net.rs | 2 +- tests/reactor_unix.rs | 2 +- tests/reactor_write_counts.rs | 2 +- tests/sync_regressions.rs | 2 +- 25 files changed, 79 insertions(+), 68 deletions(-) diff --git a/examples/counter_floor.rs b/examples/counter_floor.rs index 26308df..02aba81 100644 --- a/examples/counter_floor.rs +++ b/examples/counter_floor.rs @@ -8,8 +8,8 @@ //! //! No pool, no tasks: threads and the counter alone. -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::thread; use std::time::Instant; diff --git a/examples/nagoya_only.rs b/examples/nagoya_only.rs index 1b3138c..3ab6062 100644 --- a/examples/nagoya_only.rs +++ b/examples/nagoya_only.rs @@ -53,7 +53,7 @@ fn main() { fn run(join: bool) -> f64 { use nagoya::io::Stream as _; - use nagoya::reactor::{block_on_with, Addr, Reactor, TaskSet, TcpStream}; + use nagoya::reactor::{Addr, Reactor, TaskSet, TcpStream, block_on_with}; let path = std::env::temp_dir().join(format!( "nagoya-only-{}-{}.sock", @@ -74,10 +74,12 @@ fn run(join: bool) -> f64 { let reactor = Reactor::local().expect("reactor"); let handle = reactor.handle(); let sockets: Vec<_> = (0..CONNECTIONS) - .map(|_| loop { - match nagoya::reactor::socket::TcpSocket::connect(addr) { - Ok(socket) => break socket, - Err(_) => std::thread::yield_now(), + .map(|_| { + loop { + match nagoya::reactor::socket::TcpSocket::connect(addr) { + Ok(socket) => break socket, + Err(_) => std::thread::yield_now(), + } } }) .collect(); diff --git a/examples/profile_me.rs b/examples/profile_me.rs index 1ec8ae6..ed23c19 100644 --- a/examples/profile_me.rs +++ b/examples/profile_me.rs @@ -3,8 +3,8 @@ //! Batched rather than one huge spawn: the spawn loop outruns the workers, so a //! single batch of the size this needs would hold millions of live tasks and //! measure the allocator instead of the scheduler. -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::thread; use std::time::{Duration, Instant}; diff --git a/examples/profile_tokio.rs b/examples/profile_tokio.rs index 832b31f..aec6b6a 100644 --- a/examples/profile_tokio.rs +++ b/examples/profile_tokio.rs @@ -1,6 +1,6 @@ //! The same batched first-poll workload on tokio, so the two profiles compare. -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Duration, Instant}; fn main() { diff --git a/examples/read_shape.rs b/examples/read_shape.rs index 3fdca2c..a70ef10 100644 --- a/examples/read_shape.rs +++ b/examples/read_shape.rs @@ -44,7 +44,7 @@ fn main() { /// Run the client end here and the server in a child, so the counters see one /// end only. fn client(read_buffer: usize) -> (f64, f64, f64) { - use nagoya::reactor::{block_on_with, Addr, Reactor, TcpStream}; + use nagoya::reactor::{Addr, Reactor, TcpStream, block_on_with}; let path = std::env::temp_dir().join(format!( "nagoya-readshape-{}-{read_buffer}.sock", @@ -98,7 +98,7 @@ fn client(read_buffer: usize) -> (f64, f64, f64) { fn serve(path: &str) { use nagoya::io::Stream as _; use nagoya::reactor::socket::TcpListener as RawListener; - use nagoya::reactor::{block_on_with, Addr, Reactor, TcpListener}; + use nagoya::reactor::{Addr, Reactor, TcpListener, block_on_with}; let addr = Addr::path(path.as_bytes()).expect("path fits"); let listener = RawListener::bind(addr, 128).expect("bind"); diff --git a/examples/submit_only.rs b/examples/submit_only.rs index afaa289..bac7b0f 100644 --- a/examples/submit_only.rs +++ b/examples/submit_only.rs @@ -7,8 +7,8 @@ //! //! No workers are started. The pool fills up and nothing runs. -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Instant; use st3::fanout::{Job, Pool, StdHost}; diff --git a/examples/three_pools.rs b/examples/three_pools.rs index 89888b3..61ef1d0 100644 --- a/examples/three_pools.rs +++ b/examples/three_pools.rs @@ -15,8 +15,8 @@ //! CPU is user + system across every thread, so an arm that finishes quickly by //! burning eight cores cannot hide it. -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Condvar, Mutex}; use std::thread; use std::time::Instant; diff --git a/examples/uds_scale.rs b/examples/uds_scale.rs index 9a040d7..f08c470 100644 --- a/examples/uds_scale.rs +++ b/examples/uds_scale.rs @@ -237,10 +237,12 @@ fn start_child(path: &Path, mode: &str, connections: usize) -> std::process::Chi /// Connect `count` blocking sockets, retrying while the child finishes binding. fn connect_all(path: &Path, count: usize) -> Vec { (0..count) - .map(|_| loop { - match std::os::unix::net::UnixStream::connect(path) { - Ok(stream) => break stream, - Err(_) => std::thread::yield_now(), + .map(|_| { + loop { + match std::os::unix::net::UnixStream::connect(path) { + Ok(stream) => break stream, + Err(_) => std::thread::yield_now(), + } } }) .collect() @@ -251,7 +253,7 @@ fn connect_all(path: &Path, count: usize) -> Vec /// nagoya driving `connections` echo clients on one thread, one reactor. fn nagoya_echo(connections: usize) -> f64 { use nagoya::io::Stream as _; - use nagoya::reactor::{block_on_with, Addr, Reactor, TaskSet, TcpStream}; + use nagoya::reactor::{Addr, Reactor, TaskSet, TcpStream, block_on_with}; let path = SocketPath::new("nagoya-echo"); let mut child = start_child(path.as_path(), "echo", connections); @@ -262,10 +264,12 @@ fn nagoya_echo(connections: usize) -> f64 { // Connected before timing, so the measured region is messages only. let sockets: Vec<_> = (0..connections) - .map(|_| loop { - match nagoya::reactor::socket::TcpSocket::connect(addr) { - Ok(socket) => break socket, - Err(_) => std::thread::yield_now(), + .map(|_| { + loop { + match nagoya::reactor::socket::TcpSocket::connect(addr) { + Ok(socket) => break socket, + Err(_) => std::thread::yield_now(), + } } }) .collect(); @@ -303,17 +307,19 @@ fn nagoya_echo(connections: usize) -> f64 { /// nagoya streaming `VOLUME` bytes and counting them in. fn nagoya_throughput(chunk_size: usize, connections: usize) -> f64 { - use nagoya::reactor::{block_on_with, Addr, Reactor, TaskSet, TcpStream}; + use nagoya::reactor::{Addr, Reactor, TaskSet, TcpStream, block_on_with}; let path = SocketPath::new("nagoya-sink"); let mut child = start_child(path.as_path(), "sink", connections); let addr = Addr::path(path.as_path().as_os_str().as_bytes()).expect("path fits"); let sockets: Vec<_> = (0..connections) - .map(|_| loop { - match nagoya::reactor::socket::TcpSocket::connect(addr) { - Ok(socket) => break socket, - Err(_) => std::thread::yield_now(), + .map(|_| { + loop { + match nagoya::reactor::socket::TcpSocket::connect(addr) { + Ok(socket) => break socket, + Err(_) => std::thread::yield_now(), + } } }) .collect(); diff --git a/examples/uds_vs_tokio.rs b/examples/uds_vs_tokio.rs index 5fb95fa..bad01bb 100644 --- a/examples/uds_vs_tokio.rs +++ b/examples/uds_vs_tokio.rs @@ -241,7 +241,7 @@ impl Drop for SocketPath { /// nagoya over AF_UNIX, on a reactor with no thread of its own. fn nagoya_unix() -> Vec { use nagoya::reactor::socket::TcpListener as RawListener; - use nagoya::reactor::{block_on_with, Addr, Reactor, TcpListener, TcpStream}; + use nagoya::reactor::{Addr, Reactor, TcpListener, TcpStream, block_on_with}; let path = SocketPath::new("nagoya-uds"); let addr = Addr::path(path.as_path().as_os_str().as_bytes()).expect("path fits"); @@ -273,7 +273,7 @@ fn nagoya_unix() -> Vec { /// The same nagoya code over loopback TCP, for the family delta. fn nagoya_tcp() -> Vec { use nagoya::reactor::socket::TcpListener as RawListener; - use nagoya::reactor::{block_on_with, Addr, Reactor, TcpListener, TcpStream}; + use nagoya::reactor::{Addr, Reactor, TcpListener, TcpStream, block_on_with}; let listener = RawListener::bind(Addr::localhost(0), 128).expect("bind"); let addr = listener.local_addr().expect("addr"); diff --git a/examples/wake_latency.rs b/examples/wake_latency.rs index 33d57e9..e2d4e6a 100644 --- a/examples/wake_latency.rs +++ b/examples/wake_latency.rs @@ -13,8 +13,8 @@ //! //! One task at a time, submitted to an idle pool, timed to completion. -use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Condvar, Mutex}; use std::thread; use std::time::Instant; diff --git a/examples/where_the_time_is.rs b/examples/where_the_time_is.rs index d1c993c..2ac80ed 100644 --- a/examples/where_the_time_is.rs +++ b/examples/where_the_time_is.rs @@ -10,8 +10,8 @@ //! If arm 1 is close to arm 2, the executor is not the problem and no amount of //! tuning it will help. -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Condvar, Mutex}; use std::thread; use std::time::Instant; diff --git a/src/io.rs b/src/io.rs index 28a8a5b..38e4886 100644 --- a/src/io.rs +++ b/src/io.rs @@ -447,8 +447,8 @@ impl From for std::io::SeekFrom { #[cfg(feature = "std")] pub use host::{ - append, create, create_dir_all, metadata, open, open_or_create, read, remove_dir_all, - remove_file, rename, write, Compat, HostFile, + Compat, HostFile, append, create, create_dir_all, metadata, open, open_or_create, read, + remove_dir_all, remove_file, rename, write, }; #[cfg(feature = "std")] @@ -822,7 +822,7 @@ mod portable { #[cfg(all(test, feature = "std"))] mod tests { use super::{ - create, metadata, open, remove_file, ErrorKind, File, Read, Seek, SeekFrom, Write, + ErrorKind, File, Read, Seek, SeekFrom, Write, create, metadata, open, remove_file, }; use crate::block_on; diff --git a/src/lib.rs b/src/lib.rs index af04ac6..0261580 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -60,8 +60,8 @@ use core::task::{Context, Poll}; use st3::fanout::Pool; mod block_on; -pub mod io; pub mod credits; +pub mod io; /// Tutorial and callsite reference, with compiled examples. #[cfg(feature = "std")] #[doc = include_str!(concat!(env!("OUT_DIR"), "/guide-examples.md"))] @@ -74,8 +74,8 @@ pub mod guide {} // that is the path already-published consumers compiled against. #[cfg(feature = "reactor")] pub mod net; -mod par; pub mod owner; +mod par; // Child processes, at the path tokio gives them. They register on a reactor of // their own that starts on first use, rather than one the caller passes, so a // child's futures finish under any executor. Unix only, because the reactor is: @@ -97,10 +97,10 @@ pub mod runtime; // The four async primitives a consumer was otherwise taking from `tokio::sync`. // They are runtime-agnostic there too, which is exactly why the dependency was // easy to acquire and hard to notice: see the module comment. +pub mod readiness; pub mod sync; pub mod task; pub mod time; -pub mod readiness; mod yield_now; /// Spawn onto the shared pool, as `tokio::spawn` does. @@ -119,7 +119,7 @@ where } pub use block_on::{block_on, block_on_with_host}; -pub use par::{par_for, Cancel, ParFor}; +pub use par::{Cancel, ParFor, par_for}; /// The idle policy a pool runs with, from the queues underneath. /// /// Re-exported because selecting one is a nagoya-level decision and every @@ -128,9 +128,9 @@ pub use par::{par_for, Cancel, ParFor}; /// contains it transitively. pub use st3::fanout::Tuning; pub use time::{ - now_ns, poll as poll_timers, set_clock, sleep, sleep_until, timeout, Elapsed, Sleep, Timeout, + Elapsed, Sleep, Timeout, now_ns, poll as poll_timers, set_clock, sleep, sleep_until, timeout, }; -pub use yield_now::{yield_now, YieldNow}; +pub use yield_now::{YieldNow, yield_now}; /// A handle to a spawned task's output. /// @@ -307,7 +307,8 @@ impl Executor { match self.context.as_deref() { Some(context) => context.current_worker(&self.pool), None => crate::task::current_worker(&self.pool), - }.filter(|id| *id < self.pool.workers()) + } + .filter(|id| *id < self.pool.workers()) } /// Run a future to completion somewhere on the pool. diff --git a/src/net.rs b/src/net.rs index abe87fb..ec5dcb5 100644 --- a/src/net.rs +++ b/src/net.rs @@ -46,7 +46,7 @@ use crate::reactor::socket::{ /// calling thread, because this crate has no blocking pool to hand it to. /// Resolve before entering the reactor, or the lookup stalls every socket on /// it. -pub use crate::resolve::{connect_any, resolve as lookup_host, resolve, ResolveError}; +pub use crate::resolve::{ResolveError, connect_any, resolve as lookup_host, resolve}; /// A connection that yields to the executor instead of blocking. #[derive(Debug)] @@ -441,7 +441,7 @@ impl core::future::Future for WriteAllVectored<'_> { // peer has gone. Reported as a broken pipe, which is what the // next write would have produced anyway. Poll::Ready(Ok(0)) => { - return Poll::Ready(Err(crate::reactor::error::codes::BROKEN_PIPE)) + return Poll::Ready(Err(crate::reactor::error::codes::BROKEN_PIPE)); } Poll::Ready(Ok(n)) => this.written += n, Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), @@ -487,7 +487,7 @@ impl core::future::Future for WriteAll<'_> { // peer has gone. Reported as a broken pipe, which is what the // next write would have produced anyway. Poll::Ready(Ok(0)) => { - return Poll::Ready(Err(crate::reactor::error::codes::BROKEN_PIPE)) + return Poll::Ready(Err(crate::reactor::error::codes::BROKEN_PIPE)); } Poll::Ready(Ok(n)) => this.written += n, Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), diff --git a/src/reactor.rs b/src/reactor.rs index a9d03a0..93c4e52 100644 --- a/src/reactor.rs +++ b/src/reactor.rs @@ -77,10 +77,10 @@ pub use crate::net::{TcpListener, TcpStream, UdpSocket}; // a module is in the type namespace and a function in the value namespace. // Keeping the function called `resolve` is what makes `nagoya::reactor:: // resolve(host, port)` still compile for the callers that already wrote it. -pub use crate::resolve::{connect_any, resolve, ResolveError}; +pub use crate::resolve::{ResolveError, connect_any, resolve}; pub use crate::signal::{Signal, SignalKind}; pub use driver::{Handle, Reactor, Registration, Sharded}; pub use error::{Errno, Result}; -pub use local::{block_on, block_on_with, TaskSet}; +pub use local::{TaskSet, block_on, block_on_with}; pub use poller::{Event, Interest, Poller}; -pub use socket::{Addr, UnixPath, UNIX_PATH_CAPACITY}; +pub use socket::{Addr, UNIX_PATH_CAPACITY, UnixPath}; diff --git a/src/reactor/poller.rs b/src/reactor/poller.rs index a3d13fd..a7eea22 100644 --- a/src/reactor/poller.rs +++ b/src/reactor/poller.rs @@ -155,7 +155,7 @@ fn check(value: i32) -> Result { ))] mod sys { use super::super::error::Result; - use super::{check, Event, Interest}; + use super::{Event, Interest, check}; use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; /// The identifier used for the user event that implements `wake`. @@ -342,7 +342,7 @@ mod sys { #[cfg(target_os = "linux")] mod sys { use super::super::error::Result; - use super::{check, Event, Interest}; + use super::{Event, Interest, check}; use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; /// Token reserved for the eventfd that implements `wake`. diff --git a/src/resolve.rs b/src/resolve.rs index 141e6c6..368af53 100644 --- a/src/resolve.rs +++ b/src/resolve.rs @@ -13,9 +13,9 @@ use std::ffi::CString; use crate::net::TcpStream; +use crate::reactor::Handle; use crate::reactor::error::{Errno, Result}; use crate::reactor::socket::Addr; -use crate::reactor::Handle; /// Why [`resolve`] produced no addresses. #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/src/signal.rs b/src/signal.rs index d98d283..380ac2f 100644 --- a/src/signal.rs +++ b/src/signal.rs @@ -324,7 +324,7 @@ fn restore_mask(sig: i32, was_blocked: bool) -> Result<()> { #[cfg(target_os = "linux")] mod platform { - use super::{block_one, check, restore_mask, Errno, Result, SignalKind}; + use super::{Errno, Result, SignalKind, block_one, check, restore_mask}; use std::sync::atomic::{AtomicI32, Ordering}; /// One `signalfd` per signal number, open for the process. @@ -411,9 +411,9 @@ mod platform { #[cfg(not(target_os = "linux"))] mod platform { - use super::{block_one, check, restore_mask, unblock_one, Errno, Result, SignalKind}; - use std::sync::atomic::{AtomicI32, Ordering}; + use super::{Errno, Result, SignalKind, block_one, check, restore_mask, unblock_one}; use std::sync::Mutex; + use std::sync::atomic::{AtomicI32, Ordering}; struct Ends { read: AtomicI32, @@ -625,13 +625,13 @@ mod platform { #[cfg(test)] mod tests { use std::future::Future; - use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Mutex; + use std::sync::atomic::{AtomicBool, Ordering}; use std::task::{Context, Poll, Waker}; use std::time::Duration; use super::{Signal, SignalKind}; - use crate::reactor::{block_on_with, Reactor}; + use crate::reactor::{Reactor, block_on_with}; /// Disposition is process-wide. Two tests installing handlers at once /// would restore each other's. diff --git a/tests/executor.rs b/tests/executor.rs index 0fc7071..37a025d 100644 --- a/tests/executor.rs +++ b/tests/executor.rs @@ -2,12 +2,12 @@ use std::future::Future; use std::pin::Pin; -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::task::{Context, Poll}; use std::thread; -use nagoya::{block_on, Executor}; +use nagoya::{Executor, block_on}; use st3::fanout::{Pool, StdHost}; /// A pool with real threads behind it, and a guard that stops them. @@ -289,10 +289,12 @@ fn a_reversed_empty_range_finishes_without_starting_workers() { let pool = Pool::new(1, 256, Arc::new(StdHost::new(1))); let mut loop_ = Box::pin(nagoya::par_for(pool, 10..5, |_| panic!("empty range ran"))); let waker = std::task::Waker::noop(); - assert!(loop_ - .as_mut() - .poll(&mut Context::from_waker(waker)) - .is_ready()); + assert!( + loop_ + .as_mut() + .poll(&mut Context::from_waker(waker)) + .is_ready() + ); } #[test] diff --git a/tests/reactor_local.rs b/tests/reactor_local.rs index 74a3b33..e1283eb 100644 --- a/tests/reactor_local.rs +++ b/tests/reactor_local.rs @@ -4,10 +4,10 @@ //! exports, so they are integration tests and belong beside the other ones //! rather than at the bottom of the module they exercise. -use nagoya::reactor::{block_on_with, Reactor}; +use nagoya::reactor::{Reactor, block_on_with}; -use nagoya::reactor::socket::Addr; use nagoya::reactor::TcpListener; +use nagoya::reactor::socket::Addr; #[test] fn drives_a_connection_on_one_thread() { diff --git a/tests/reactor_local_timeout.rs b/tests/reactor_local_timeout.rs index 51b3ae9..695c98d 100644 --- a/tests/reactor_local_timeout.rs +++ b/tests/reactor_local_timeout.rs @@ -20,8 +20,8 @@ use std::cell::Cell; use std::sync::mpsc; use std::time::Duration; -use nagoya::reactor::{block_on_with, Reactor}; -use nagoya::{now_ns, set_clock, timeout, Elapsed}; +use nagoya::reactor::{Reactor, block_on_with}; +use nagoya::{Elapsed, now_ns, set_clock, timeout}; /// Where the clock starts. Any non-zero reading will do. const START: u64 = 1_000_000_000; diff --git a/tests/reactor_net.rs b/tests/reactor_net.rs index 17b51db..ae4bfa8 100644 --- a/tests/reactor_net.rs +++ b/tests/reactor_net.rs @@ -368,8 +368,8 @@ fn tcp_sees_eof_after_a_short_read() { /// so this is where that gets answered rather than reasoned about. #[test] fn a_write_of_the_whole_send_buffer_fills_it() { - use nagoya::reactor::socket::TcpSocket; use nagoya::reactor::Addr; + use nagoya::reactor::socket::TcpSocket; use std::os::unix::ffi::OsStrExt as _; let path = std::env::temp_dir().join(format!("nagoya-fill-{}.sock", std::process::id())); diff --git a/tests/reactor_unix.rs b/tests/reactor_unix.rs index 4da25fa..a9bb8ae 100644 --- a/tests/reactor_unix.rs +++ b/tests/reactor_unix.rs @@ -17,7 +17,7 @@ use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU32, Ordering}; use nagoya::io::{Stream, StreamError}; -use nagoya::reactor::{block_on_with, Addr, Reactor, TcpListener, TcpStream, UnixPath}; +use nagoya::reactor::{Addr, Reactor, TcpListener, TcpStream, UnixPath, block_on_with}; /// A socket path that removes itself. /// diff --git a/tests/reactor_write_counts.rs b/tests/reactor_write_counts.rs index 88f2699..227315f 100644 --- a/tests/reactor_write_counts.rs +++ b/tests/reactor_write_counts.rs @@ -8,7 +8,7 @@ #![cfg(feature = "syscall-counters")] -use nagoya::reactor::{block_on_with, counters, Addr, Reactor, TcpStream}; +use nagoya::reactor::{Addr, Reactor, TcpStream, block_on_with, counters}; use std::io::Read as _; use std::os::unix::ffi::OsStrExt as _; diff --git a/tests/sync_regressions.rs b/tests/sync_regressions.rs index a5a6d84..bead8d6 100644 --- a/tests/sync_regressions.rs +++ b/tests/sync_regressions.rs @@ -1,8 +1,8 @@ //! Deterministic polling tests: no executor scheduling or sleeps are involved. use std::future::Future; use std::pin::Pin; -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::task::{Context, Poll, Wake, Waker}; use nagoya::sync::{Notify, RwLock, Semaphore}; From e076cbb1cb1fbfb00fed113681c821b988e37876 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 13:49:37 +0700 Subject: [PATCH 15/17] Ask the timer wheel when no deadline is cached --- src/reactor/driver.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/reactor/driver.rs b/src/reactor/driver.rs index 8ac9a94..6b569b5 100644 --- a/src/reactor/driver.rs +++ b/src/reactor/driver.rs @@ -670,11 +670,10 @@ fn run(shared: &Arc) { /// arrives: there is no deadline to wake for. fn service_timers(shared: &Arc) -> (Option, bool) { let deadline = shared.next_deadline.load(Ordering::Acquire); - if deadline == NO_DEADLINE { - return (None, false); - } let now = crate::now_ns(); - if deadline > now { + // No cached deadline still asks the wheel: a timer armed through another + // reactor's handle (a local reactor) never lowered this cache. + if deadline != NO_DEADLINE && deadline > now { return (Some(deadline - now), false); } From a438670375d264d48e77c8496088f9707cb39cea Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 14:50:58 +0700 Subject: [PATCH 16/17] Update credit atomics with compare-exchange loops --- src/credits.rs | 55 +++++++++++++++++++++++++++++++++++--------------- 1 file changed, 39 insertions(+), 16 deletions(-) diff --git a/src/credits.rs b/src/credits.rs index 3f00de0..9a42aa5 100644 --- a/src/credits.rs +++ b/src/credits.rs @@ -7,6 +7,34 @@ use alloc::sync::Arc; use core::num::{NonZeroU64, NonZeroUsize}; use core::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; +/// Compare-exchange update for `AtomicI64`; `fetch_update` is deprecated on +/// current stable and its replacement is newer than the crate MSRV. +fn update_i64(cell: &AtomicI64, mut f: impl FnMut(i64) -> Option) -> Result { + let mut current = cell.load(Ordering::Acquire); + loop { + let next = f(current).ok_or(current)?; + match cell.compare_exchange_weak(current, next, Ordering::AcqRel, Ordering::Acquire) { + Ok(previous) => return Ok(previous), + Err(observed) => current = observed, + } + } +} + +/// The `AtomicUsize` form of [`update_i64`]. +fn update_usize( + cell: &AtomicUsize, + mut f: impl FnMut(usize) -> Option, +) -> Result { + let mut current = cell.load(Ordering::Acquire); + loop { + let next = f(current).ok_or(current)?; + match cell.compare_exchange_weak(current, next, Ordering::AcqRel, Ordering::Acquire) { + Ok(previous) => return Ok(previous), + Err(observed) => current = observed, + } + } +} + /// One stable accounting key, shared across its agents and helper jobs. /// /// Migration and host reentry reuse this same Arc. Replenishment is a policy @@ -91,11 +119,9 @@ impl CreditAccount { } fn adjust(&self, credits: i64) { - let _ = self - .balance - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |balance| { - Some(balance.saturating_add(credits).min(self.capacity)) - }); + let _ = update_i64(&self.balance, |balance| { + Some(balance.saturating_add(credits).min(self.capacity)) + }); } } @@ -172,17 +198,14 @@ impl CpuBudget> for CpuCredits { fn reserve(&self, key: &Arc>, credits: u64) -> Option { let reserved = i64::try_from(credits).ok().filter(|value| *value > 0)?; - key.in_flight - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { - (count < key.parallelism).then(|| count + 1) - }) - .ok()?; - if key - .balance - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |balance| { - (balance >= reserved).then(|| balance - reserved) - }) - .is_err() + update_usize(&key.in_flight, |count| { + (count < key.parallelism).then(|| count + 1) + }) + .ok()?; + if update_i64(&key.balance, |balance| { + (balance >= reserved).then(|| balance - reserved) + }) + .is_err() { key.in_flight.fetch_sub(1, Ordering::AcqRel); key.changed.notify_waiters(); From ea98d1593d88baadef14c53c7642bf1396ef4efb Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 14:50:58 +0700 Subject: [PATCH 17/17] Release 0.1.18 --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index f3cf8de..24cc61f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nagoya" -version = "0.1.17" +version = "0.1.18" edition = "2024" rust-version = "1.85" license = "MIT OR Apache-2.0"