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. diff --git a/Cargo.toml b/Cargo.toml index 2e6e6e8..24cc61f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,13 @@ [package] name = "nagoya" -version = "0.1.16" -edition = "2021" +version = "0.1.18" +edition = "2024" rust-version = "1.85" 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"] @@ -187,3 +188,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 new file mode 100644 index 0000000..508bd34 --- /dev/null +++ b/docs/agent-pool.md @@ -0,0 +1,178 @@ +# 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(); +``` + +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. + +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 +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..8bdd385 --- /dev/null +++ b/examples/agent_pool_witness.rs @@ -0,0 +1,288 @@ +//! 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/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/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}`); +} diff --git a/src/credits.rs b/src/credits.rs new file mode 100644 index 0000000..9a42aa5 --- /dev/null +++ b/src/credits.rs @@ -0,0 +1,256 @@ +//! Shared CPU credit authority with cancellation refunds and overshoot debt. + +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}; + +/// 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 +/// 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 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 _ = update_i64(&self.balance, |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 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. +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)?; + 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(); + 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/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 cc0ee0f..0261580 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -60,6 +60,7 @@ use core::task::{Context, Poll}; use st3::fanout::Pool; mod block_on; +pub mod credits; pub mod io; /// Tutorial and callsite reference, with compiled examples. #[cfg(feature = "std")] @@ -73,6 +74,7 @@ pub mod guide {} // that is the path already-published consumers compiled against. #[cfg(feature = "reactor")] pub mod net; +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 @@ -95,6 +97,7 @@ 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; @@ -116,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 @@ -125,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. /// @@ -298,6 +301,16 @@ 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/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/owner.rs b/src/owner.rs new file mode 100644 index 0000000..4418fa4 --- /dev/null +++ b/src/owner.rs @@ -0,0 +1,195 @@ +//! 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}; + +/// 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/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/driver.rs b/src/reactor/driver.rs index 2a8ddef..6b569b5 100644 --- a/src/reactor/driver.rs +++ b/src/reactor/driver.rs @@ -671,19 +671,23 @@ fn run(shared: &Arc) { fn service_timers(shared: &Arc) -> (Option, bool) { let deadline = shared.next_deadline.load(Ordering::Acquire); let now = crate::now_ns(); - - // Not due yet: wait exactly until it is, and do not touch the wheel. + // 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); } - // 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) + .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/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/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. diff --git a/src/readiness.rs b/src/readiness.rs new file mode 100644 index 0000000..3f7239c --- /dev/null +++ b/src/readiness.rs @@ -0,0 +1,129 @@ +//! 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}; + +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); + })) + } +} 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/runtime.rs b/src/runtime.rs index 02b01fd..e8d738f 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. @@ -203,6 +250,43 @@ 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 +322,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 +402,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, @@ -343,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/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/src/task.rs b/src/task.rs index e046d07..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())); @@ -121,6 +121,20 @@ pub(crate) fn spawn_on( pool: Arc, context: Option>, ) -> Task +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, @@ -129,7 +143,17 @@ where // 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() { @@ -315,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 2181258..a393bab 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,32 @@ 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 +494,22 @@ 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 +527,22 @@ 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() { @@ -532,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; @@ -599,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!( 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 737bcd5..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. /// @@ -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"); 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};