From 4384f615c9a0601eb597dd25bec98f7d988bd52a Mon Sep 17 00:00:00 2001 From: CBoYXD <135316445+CBoYXD@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:44:22 +0300 Subject: [PATCH 1/7] fix: fence the block_on park against a wake that races it A cross-thread wake stores the root future's ready flag (or pushes the task onto the remote queue) and then loads the runtime's `waiting` flag; the park stores `waiting` and then loads ready. With release/acquire only, both stores can still sit in their store buffers when the loads run, so the waker skips the interrupt and the park sleeps through the wake. The loop thread recovers at its next park deadline, but the coordination thread parks with no timer and never wakes again: callbacks queued before a run_forever (a run_until_complete task's first step) are then never forwarded while loop-local work keeps running. The greenlet repro of #115 hung this way in about 1 of 300-450 runs under Docker (epoll), on v0.1.64 as well. A SeqCst fence now separates store and load on the park side and on both notifiers (root waker and remote task queue). Same-thread wakes skip it. Two stress tests ping-pong 100k wakes against the park on the mio driver and fail when progress stops for 5 s; without the fences the root-waker test stalls in 2 of 3 runs. --- src/vibeio/executor.rs | 110 +++++++++++++++++++++++++++++++++++++---- src/vibeio/task.rs | 5 +- 2 files changed, 105 insertions(+), 10 deletions(-) diff --git a/src/vibeio/executor.rs b/src/vibeio/executor.rs index 5a013e6..a7b0645 100644 --- a/src/vibeio/executor.rs +++ b/src/vibeio/executor.rs @@ -29,7 +29,7 @@ use std::{ rc::Rc, sync::{ Arc, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, Ordering, fence}, }, task::{Context, Poll, Wake, Waker}, }; @@ -282,14 +282,21 @@ impl BlockOnNotify { fn notify(&self) { self.ready.store(true, Ordering::Release); - if std::thread::current().id() != self.thread_id - && self.remote_wake.waiting.load(Ordering::Acquire) - && !self - .remote_wake - .interrupt_pending - .swap(true, Ordering::AcqRel) - { - self.remote_wake.interruptor.interrupt(); + if std::thread::current().id() != self.thread_id { + // Pairs with the fence in `drive` after it publishes `waiting`: + // either the park sees `ready`, or this sees `waiting` and + // interrupts it. Release/acquire alone lets both loads read the + // stale value (store-load reordering), and the park then sleeps + // through this wake with no deadline. + fence(Ordering::SeqCst); + if self.remote_wake.waiting.load(Ordering::Acquire) + && !self + .remote_wake + .interrupt_pending + .swap(true, Ordering::AcqRel) + { + self.remote_wake.interruptor.interrupt(); + } } } @@ -939,6 +946,11 @@ impl Runtime { .interrupt_pending .store(false, Ordering::Release); inner.remote_wake.waiting.store(true, Ordering::Release); + // Pairs with the fences in `BlockOnNotify::notify` and + // `TaskWake::enqueue_if_needed`: a wake that published before + // this point is seen below, and one that publishes after it + // sees `waiting` and interrupts the wait. + fence(Ordering::SeqCst); if root_notify.is_ready() || inner.should_skip_wait() { inner.stop_waiting(); @@ -1786,6 +1798,86 @@ mod tests { assert_eq!(value, 11); } + #[test] + fn block_on_never_misses_a_wake_racing_with_the_park() { + // A cross-thread wake stores the root's ready flag and then loads the + // park's `waiting` flag; the park stores `waiting` and then loads + // ready. Each side's store can still sit in its store buffer when it + // loads, so without a full fence both can read the stale value: the + // waker skips the interrupt and the park sleeps with no deadline. + assert_cross_thread_wakes_never_missed(false); + } + + #[test] + fn spawned_task_never_misses_a_wake_racing_with_the_park() { + // The same handshake through the remote task queue. + assert_cross_thread_wakes_never_missed(true); + } + + fn assert_cross_thread_wakes_never_missed(spawned: bool) { + use std::{ + sync::{Mutex, atomic::AtomicUsize, mpsc}, + time::Duration, + }; + + const ROUNDS: usize = 100_000; + let woken = Arc::new(AtomicUsize::new(0)); + let armed = Arc::new(AtomicUsize::new(0)); + let slot: Arc>> = Arc::new(Mutex::new(None)); + let (done_tx, done_rx) = mpsc::channel(); + let progress = woken.clone(); + { + let (woken, armed, slot) = (woken.clone(), armed.clone(), slot.clone()); + std::thread::spawn(move || { + for round in 1..=ROUNDS { + while armed.load(Ordering::Acquire) < round { + std::hint::spin_loop(); + } + let waker = slot.lock().unwrap().take().unwrap(); + woken.store(round, Ordering::Release); + waker.wake(); + } + }); + } + std::thread::spawn(move || { + let runtime = Runtime::new(AnyDriver::new_mio().unwrap()); + for round in 1..=ROUNDS { + let (woken, armed, slot) = (woken.clone(), armed.clone(), slot.clone()); + let mut published = false; + let wait = std::future::poll_fn(move |cx| { + if woken.load(Ordering::Acquire) >= round { + return Poll::Ready(()); + } + if !published { + published = true; + *slot.lock().unwrap() = Some(cx.waker().clone()); + armed.store(round, Ordering::Release); + } + Poll::Pending + }); + if spawned { + runtime.block_on(runtime.spawn(wait)); + } else { + runtime.block_on(wait); + } + } + done_tx.send(()).unwrap(); + }); + // Fail on a stall, not on a slow machine: a missed wake stops all + // progress, while every completed round only takes microseconds. + let mut last = 0; + while done_rx.recv_timeout(Duration::from_secs(5)).is_err() { + let now = progress.load(Ordering::Acquire); + assert!( + now != last, + "{} parked through the wake of round {}", + if spawned { "spawned task" } else { "block_on" }, + now + 1 + ); + last = now; + } + } + #[test] fn spawned_task_resumes_after_remote_wake() { let runtime = Runtime::new(AnyDriver::new_mock()); diff --git a/src/vibeio/task.rs b/src/vibeio/task.rs index 6f24d35..579db61 100644 --- a/src/vibeio/task.rs +++ b/src/vibeio/task.rs @@ -5,7 +5,7 @@ use std::{ rc::Rc, sync::{ Arc, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, Ordering, fence}, }, task::Waker, }; @@ -84,6 +84,9 @@ impl TaskWake { if !was_queued { remote.queue.push(wake.clone()); } + // Pairs with the fence after `drive` publishes `waiting`: either the + // park sees this task in the queue, or this sees `waiting`. + fence(Ordering::SeqCst); if remote.waiting.load(Ordering::Acquire) && !remote.interrupt_pending.swap(true, Ordering::AcqRel) { From 98e14e2a356272982752c5fd64027f6a3f8da763 Mon Sep 17 00:00:00 2001 From: CBoYXD <135316445+CBoYXD@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:44:41 +0300 Subject: [PATCH 2/7] perf: enqueue off-thread call_soon_threadsafe directly into the active run Every off-thread handle used to travel through the command channel to the coordination thread, which then pushed it into the run's queue and woke the loop thread: two thread wakeups per handle where one is enough. Behind a server that resumes requests with call_soon_threadsafe from its own I/O threads (Granian), that is two wakeups per request start. While a run is active and no earlier handle is still on the channel, the handle now joins the run's pending queue directly (one wake). A counter of handles still on the channel keeps each producer's order: once one is queued there (before EnterRun, between runs), later handles follow it. finish_run now puts the run's queue (older work) ahead of commands received since the last dispatch. Microbenchmark on the v0.1.64 base (Docker epoll, --cpus 2): cross-thread ping-pong CPU 11.4 -> 7.8 us/op, task ping-pong 10.4 -> 6.9 us/op (uvloop 5.9 / 6.9). --- src/engine/dispatcher.rs | 13 ++++++++++++- src/engine/loop_core.rs | 37 +++++++++++++++++++++++++++++++----- tests/test_fast_callbacks.py | 32 +++++++++++++++++++++++++++++++ 3 files changed, 76 insertions(+), 6 deletions(-) diff --git a/src/engine/dispatcher.rs b/src/engine/dispatcher.rs index eadb064..5e7ae7a 100644 --- a/src/engine/dispatcher.rs +++ b/src/engine/dispatcher.rs @@ -130,6 +130,9 @@ struct RuntimeDispatcher { core: Arc, command_rx: Receiver, ready_batch: VecDeque, + /// Channel handles in `ready_batch`, released from + /// `LoopCore::channel_handles` once they reach an active run. + channel_handles: usize, active_run: Option, #[cfg(unix)] signal_tasks: HashMap, @@ -218,6 +221,7 @@ pub fn run_runtime_thread(core: Arc, command_rx: Receiver core: Arc::clone(&core), command_rx, ready_batch: VecDeque::new(), + channel_handles: 0, active_run: None, #[cfg(unix)] signal_tasks: HashMap::new(), @@ -296,6 +300,7 @@ impl RuntimeDispatcher { self.ready_batch.push_back(ReadyItem::Callback(callback)); } LoopCommand::ScheduleReadyHandle(handle) => { + self.channel_handles += 1; self.ready_batch .push_back(ReadyItem::HandleCallback(handle)); } @@ -558,6 +563,8 @@ impl RuntimeDispatcher { .expect("poisoned pending ready queue"); pending.extend(self.ready_batch.drain(..)); drop(pending); + self.core + .channel_handles_forwarded(std::mem::take(&mut self.channel_handles)); // Wake the parked loop thread so runtime-dispatched work (fd // watchers) is observed; `signal_ready` coalesces redundant wakes. self.core.signal_ready(); @@ -569,8 +576,12 @@ impl RuntimeDispatcher { return; }; + // The run's queue holds older work (its unexecuted callbacks, then + // direct off-thread handles) than commands received since the last + // dispatch, so it goes first. if let Ok(mut pending) = active_run.pending_ready.lock() { - self.ready_batch.extend(pending.drain(..)); + pending.extend(self.ready_batch.drain(..)); + std::mem::swap(&mut self.ready_batch, &mut *pending); } { diff --git a/src/engine/loop_core.rs b/src/engine/loop_core.rs index 16da2cb..abf8fb4 100644 --- a/src/engine/loop_core.rs +++ b/src/engine/loop_core.rs @@ -391,6 +391,11 @@ pub struct LoopCore { pending_timers: Mutex>, pending_timers_dirty: AtomicBool, cancelled_timers: Arc, + /// Handles sent through the command channel that the dispatcher has not + /// yet moved into an active run's queue. While it is zero, an off-thread + /// handle may join the active queue directly without overtaking an + /// earlier handle from the same producer. + channel_handles: AtomicUsize, // Wakes the loop thread when a producer enqueues a ready item. Held in an // Shared with producers; the park future borrows it under `py.detach`. wake: Arc, @@ -425,6 +430,7 @@ impl LoopCore { pending_timers: Mutex::new(BinaryHeap::new()), pending_timers_dirty: AtomicBool::new(false), cancelled_timers: Arc::new(AtomicUsize::new(0)), + channel_handles: AtomicUsize::new(0), wake: Arc::new(LoopWake::new()), }); @@ -485,15 +491,36 @@ impl LoopCore { Ok(()) => return Ok(()), Err(item) => item, }; + // Off the loop thread (call_soon_threadsafe from a server's I/O + // threads): join the active run's queue directly, one wake instead of + // a dispatcher-thread hop plus a second wake. Only while no handle is + // still on the command channel: one queued before EnterRun, or sent + // between runs, would otherwise be overtaken. + let item = if self.channel_handles.load(Ordering::Acquire) == 0 { + match self.try_enqueue_active_ready(item) { + Ok(()) => return Ok(()), + Err(item) => item, + } + } else { + item + }; let ReadyItem::HandleCallback(handle) = item else { unreachable!("ready handle enqueue preserves item kind") }; - // Keep every off-thread callback on the runtime command queue. During - // run_forever startup, callbacks queued before EnterRun already live - // there; sending later callbacks directly to pending_ready can overtake - // that older batch before the dispatcher processes EnterRun. The local - // loop-thread path above remains direct and allocation-free. + self.channel_handles.fetch_add(1, Ordering::AcqRel); self.send_remote_command(LoopCommand::ScheduleReadyHandle(handle)) + .inspect_err(|_| { + self.channel_handles.fetch_sub(1, Ordering::AcqRel); + }) + } + + /// The dispatcher moved `count` channel handles into the active run's + /// queue; later handles may bypass the channel once none remain. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))] + pub(crate) fn channel_handles_forwarded(&self, count: usize) { + if count != 0 { + self.channel_handles.fetch_sub(count, Ordering::AcqRel); + } } #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))] diff --git a/tests/test_fast_callbacks.py b/tests/test_fast_callbacks.py index 4aa265a..6725eac 100644 --- a/tests/test_fast_callbacks.py +++ b/tests/test_fast_callbacks.py @@ -5,6 +5,7 @@ import gc import sys import threading +import time import weakref from typing import Any, cast @@ -214,6 +215,37 @@ def producer(): thread.join() loop.close() + def test_threadsafe_producer_order_survives_run_restarts(self): + # A producer that keeps posting while runs stop and restart sends some + # callbacks through the dispatcher (between runs) and some straight to + # the active run; neither path may overtake the other. + loop = rsloop.new_event_loop() + events = [] + total = 20000 + + def record(value): + events.append(value) + if len(events) % 97 == 0: + loop.stop() + + def producer(): + for value in range(total): + loop.call_soon_threadsafe(record, value) + if value % 64 == 0: + time.sleep(0) # let runs start and stop while posting + loop.call_soon_threadsafe(loop.stop) + + thread = threading.Thread(target=producer) + try: + thread.start() + while len(events) < total: + loop.run_forever() + thread.join() + assert events == list(range(total)) + finally: + thread.join() + loop.close() + @pytest.mark.parametrize("factory", [asyncio.new_event_loop, rsloop.new_event_loop]) @pytest.mark.parametrize("threadsafe", [False, True]) From ad4ee252cc9779bd44a305e7d9157511090b0f57 Mon Sep 17 00:00:00 2001 From: CBoYXD <135316445+CBoYXD@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:44:52 +0300 Subject: [PATCH 3/7] perf: park instead of spinning for cross-thread wakes by default The loop thread spun for up to 50 us, with the GIL released, waiting for a cross-thread wake before parking. Behind Granian those wakes come from Python threads (call_soon_threadsafe in its future watcher) that still hold the GIL when the flag flips, so the spinning loop thread queued for the GIL behind them, paying up to the 5 ms switch interval per reattach. Request starts and completions then reached the loop in bursts. With a small SQLAlchemy pool (4 + 2 per worker at 8 concurrent requests) the pool kept discarding returned connections into a full idle queue and opening new ones: 33-55 new Postgres sessions per 1k requests vs 3-4 on uvloop, the whole CPU-per-request gap at c16. RSLOOP_WAKE_SPIN_US now defaults to 0 (park at once) and remains an opt-in for workloads whose wakes come from GIL-free threads. Microbenchmarks on the v0.1.64 base (Docker epoll, --cpus 2, medians of 3): Granian-shaped cross-thread request CPU 54.6 -> 41.1 us/req (uvloop 50.0), cross-thread ping-pong 14.1 -> 6.5 us/op (uvloop 7.6), and a 4 + 2 pool churn repro 598 -> 2,593 req/s with 61.3 -> 0.2 connects per 1k (uvloop 2,166, 0.1). v0.1.65 keeps the spin: the same repro opens 60-74 connections per 1k there. --- docs/configuration.md | 1 + src/engine/loop_core.rs | 30 ++++++++++++++++++------------ 2 files changed, 19 insertions(+), 12 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 4c1cd84..15995c4 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -9,6 +9,7 @@ environment later is not a supported way to retune an active process. | `RSLOOP_TRANSPORT_STATS` | Boolean; disabled | Enables transport counters. Trimmed, case-sensitive values `1`, `true`, `yes`, `on` enable it; all other values disable it. | | `RSLOOP_MAX_WRITE_BUFFER_BYTES` | Positive integer bytes; `67108864` (64 MiB) | Caps buffered writes. Missing, invalid, zero, or out-of-range values use the default. | | `RSLOOP_MAX_PENDING_TLS_HANDSHAKES` | Positive integer count; `256` | Caps pending server TLS handshakes. Missing, invalid, zero, or out-of-range values use the default. | +| `RSLOOP_WAKE_SPIN_US` | Unsigned integer microseconds; `0` | The loop thread busy-waits this long for a cross-thread wake before parking. Zero (the default) parks at once; values above `1000` clamp to `1000`; invalid values use `0`. Spinning helps only when wakes come from GIL-free threads: a spinning loop competes for the GIL with the Python threads that wake it. | | `RSLOOP_READER_SPIN_US` | Unsigned integer microseconds; `30` | Blocking reader workers retry nonblocking reads before polling. Zero disables spinning; values above `1000` clamp to `1000`; invalid values use `30`. | For example, enable counters for a local script on Unix: diff --git a/src/engine/loop_core.rs b/src/engine/loop_core.rs index abf8fb4..98cc0cf 100644 --- a/src/engine/loop_core.rs +++ b/src/engine/loop_core.rs @@ -110,11 +110,18 @@ impl<'a> WaitForWake<'a> { } } -/// Bounded busy-wait before parking in `block_on`. Cross-thread wakeups (reader -/// worker threads, the transitional runtime thread) otherwise pay the full -/// `driver.wait` park + interrupt round-trip, which dominates request/response -/// ping-pong latency and inflates its variance. Env-tunable via -/// `RSLOOP_WAKE_SPIN_US` (0 disables). +/// Optional bounded busy-wait before parking in `block_on`, off by default. +/// +/// A spin saves the `driver.wait` park + interrupt round-trip when the waker is +/// a GIL-free Rust thread, but most cross-thread wakes come from Python threads +/// (`call_soon_threadsafe` from a server's I/O threads) that still hold the +/// GIL when the flag flips. The spinning loop thread then queues for the GIL +/// behind those threads, each reattach costing up to the switch interval, and +/// request starts and completions arrive in bursts: behind Granian with a +/// small SQLAlchemy pool that made the pool discard and reopen connections +/// (~55 new Postgres sessions per 1k requests vs 0 parked). A parked loop +/// wakes once the producer has let go of the GIL and drains everything queued. +/// `RSLOOP_WAKE_SPIN_US` opts back in (microseconds, at most 1000). #[cfg_attr(feature = "profile", hotpath::measure)] fn wake_spin_window() -> Duration { static WINDOW: std::sync::OnceLock = std::sync::OnceLock::new(); @@ -122,7 +129,7 @@ fn wake_spin_window() -> Duration { let micros = std::env::var("RSLOOP_WAKE_SPIN_US") .ok() .and_then(|value| value.trim().parse::().ok()) - .unwrap_or(50); + .unwrap_or(0); Duration::from_micros(micros.min(1_000)) }) } @@ -1106,12 +1113,11 @@ impl LoopCore { } } - // Wait for the next wakeup with the GIL released. First spin - // briefly to catch an imminent cross-thread wake - // (reader worker / runtime thread) in user space — this - // keeps request/response ping-pong latency low and - // tight. On spin timeout (or after too many consecutive - // catches, to avoid starving this loop's own runtime + // Wait for the next wakeup with the GIL released. When + // `RSLOOP_WAKE_SPIN_US` enables it, first spin briefly to catch an + // imminent cross-thread wake in user space (see + // `wake_spin_window`). On spin timeout (or after too many + // consecutive catches, to avoid starving this loop's own runtime // tasks) park by driving the runtime: its `driver.wait` runs here // on the loop thread and is interrupted by a // cross-thread wake. Keep the `!Send` runtime lookup From 585250de733f63b8f77f255a442e0fa9e17ea06d Mon Sep 17 00:00:00 2001 From: CBoYXD <135316445+CBoYXD@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:45:40 +0300 Subject: [PATCH 4/7] perf: run one callback generation per turn and poll I/O without the GIL run_forever drained each turn to quiescence: callbacks scheduled during the turn joined it (every 64 items, up to 1024), and the reactor was polled only once nothing was left. Under load, database responses piled up between polls and completed in one burst, so more connections came back to a SQLAlchemy pool in one turn than its idle queue holds before any woken waiter ran; the overflow was closed and reopened (a Litestar API at c64: ~20 new Postgres sessions per 1k requests vs 0.1-3.7 on uvloop, ~690 full-queue check-ins in 20 s). A turn now runs the callbacks queued when it started, as asyncio's _run_once does; newer ones wait for the next turn, after another non-blocking poll. That poll runs with the GIL released, as uvloop's does: back-to-back turns would otherwise hold the GIL for the switch interval while call_soon_threadsafe producers wait. A stop() whose Stop item is still queued keeps the turn going until it runs, so stop keeps taking effect in the turn that requested it. The loop thread schedules into its own lock-free queue and other threads into the run's pending queue, so a generation has to be merged from both. Off-thread items now carry the loop thread's enqueue count, read when they are queued, and the drain merges them behind exactly the local items scheduled before them: call_soon and call_soon_threadsafe callbacks run in the order they were scheduled, as on asyncio (before, 50 alternating call_soon / thread-posted call_soon_threadsafe calls from one callback did not interleave in call order). Marks are stored once per change, the loop thread keeps a single-writer counter, and work queued before a run goes ahead of the run's own. On the v0.1.64 base, the same API at c64: 1,618 -> 1,936 rps (1.3-3.0 new sessions per 1k; uvloop 1,746). Microbenchmark cost against v0.1.65: a self-rescheduling call_soon chain 0.20 -> 0.51 us/op and a cancelled timer 1.52 -> 2.21 us/op (uvloop 0.79 and 2.61), now one turn per step as in asyncio and uvloop; benches/compare_event_loops.py is level. New tests: a call_soon chain no longer delays a readable protocol by hundreds of callbacks (513 before, 1 after), and three ordering tests compare call_soon / call_soon_threadsafe order with asyncio. --- src/engine/commands.rs | 74 +++++++++- src/engine/dispatcher.rs | 28 +++- src/engine/loop_core.rs | 256 +++++++++++++++++++++++++++-------- tests/test_fast_callbacks.py | 87 ++++++++++++ tests/test_local_sockets.py | 44 ++++++ 5 files changed, 421 insertions(+), 68 deletions(-) diff --git a/src/engine/commands.rs b/src/engine/commands.rs index b932ca1..9f2e1c4 100644 --- a/src/engine/commands.rs +++ b/src/engine/commands.rs @@ -83,6 +83,75 @@ pub struct ServerAcceptStart { pub task_guard: ServerAcceptTaskGuard, } +/// Off-thread ready items queued for the active run. +/// +/// The loop thread schedules into its own queue without a lock, so the two +/// sources only meet when the drain starts the next generation. Each item +/// here records `LoopCore::local_enqueues` as read when it was queued: it was +/// scheduled after that many loop-thread enqueues, and the drain merges it +/// behind exactly those (see `merge_ready_generation`). Marks are stored once +/// per change, as `(index, mark)` pairs, because a burst of off-thread work +/// between two loop-thread enqueues shares one. +pub struct PendingReady { + pub(crate) items: VecDeque, + pub(crate) marks: Vec<(usize, u64)>, +} + +impl Default for PendingReady { + #[cfg_attr( + feature = "profile", + hotpath::measure(impl_type = "") + )] + fn default() -> Self { + Self { + items: VecDeque::new(), + marks: Vec::new(), + } + } +} + +impl PendingReady { + /// Queues `item`, scheduled after `local_mark` loop-thread enqueues. + #[inline] + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "PendingReady"))] + pub(crate) fn push(&mut self, item: T, local_mark: u64) { + self.mark(local_mark); + self.items.push_back(item); + } + + /// Queues `items` in order, all scheduled after `local_mark` loop-thread + /// enqueues. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "PendingReady"))] + pub(crate) fn extend(&mut self, items: impl IntoIterator, local_mark: u64) { + let (len, marks) = (self.items.len(), self.marks.len()); + self.mark(local_mark); + self.items.extend(items); + if self.items.len() == len { + self.marks.truncate(marks); + } + } + + /// Takes every item out of scheduling order tracking, once the run that + /// could merge it with loop-thread work has ended. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "PendingReady"))] + pub(crate) fn take_items(&mut self) -> VecDeque { + self.marks.clear(); + std::mem::take(&mut self.items) + } + + #[inline] + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "PendingReady"))] + fn mark(&mut self, local_mark: u64) { + if self + .marks + .last() + .is_none_or(|&(_, mark)| mark != local_mark) + { + self.marks.push((self.items.len(), local_mark)); + } + } +} + /// Control-plane messages consumed by the dedicated runtime dispatcher. pub enum LoopCommand { /// Enqueues a callback whose lifetime is shared independently of Python. @@ -121,7 +190,10 @@ pub enum LoopRunCommand { /// Installs the ready queue used by the active loop-thread run. EnterRun { /// Queue shared between producers and the loop-thread drain. - pending_ready: Arc>>, + pending_ready: Arc>, + /// `LoopCore::local_enqueues` before the run could enqueue anything: + /// work queued before the run goes ahead of all of the run's own. + local_mark: u64, }, /// Clears the active run and acknowledges dispatcher cleanup. FinishRun { diff --git a/src/engine/dispatcher.rs b/src/engine/dispatcher.rs index 5e7ae7a..e341807 100644 --- a/src/engine/dispatcher.rs +++ b/src/engine/dispatcher.rs @@ -27,7 +27,7 @@ use super::commands::ConnectCompletion; use super::{ commands::{ AcceptedConnection, FutureCompletion, LoopCommand, LoopFutureCommand, LoopIoCommand, - LoopRunCommand, LoopSignalCommand, LoopTransportCommand, ReadyItem, + LoopRunCommand, LoopSignalCommand, LoopTransportCommand, PendingReady, ReadyItem, }, loop_core::LoopCore, }; @@ -146,7 +146,7 @@ struct RuntimeDispatcher { /// Queue shared with the Python loop thread while `run_forever` is active. struct ActiveRun { - pending_ready: Arc>>, + pending_ready: Arc>, } #[cfg(unix)] @@ -337,9 +337,12 @@ impl RuntimeDispatcher { stream, }))); } - LoopCommand::Run(LoopRunCommand::EnterRun { pending_ready }) => { + LoopCommand::Run(LoopRunCommand::EnterRun { + pending_ready, + local_mark, + }) => { self.active_run = Some(ActiveRun { pending_ready }); - self.dispatch_ready_batch(); + self.forward_ready_batch(local_mark); } LoopCommand::Run(LoopRunCommand::FinishRun { done_tx }) => { self.finish_run(); @@ -549,6 +552,16 @@ impl RuntimeDispatcher { #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "RuntimeDispatcher"))] fn dispatch_ready_batch(&mut self) { + // Commands received mid-run are merged behind every loop-thread + // enqueue so far; their producers called in before forwarding, so this + // only errs towards running them later, as a busy queue would. + self.forward_ready_batch(self.core.local_enqueue_mark()); + } + + /// Moves received commands into the active run, scheduled after + /// `local_mark` loop-thread enqueues. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "RuntimeDispatcher"))] + fn forward_ready_batch(&mut self, local_mark: u64) { let Some(active_run) = self.active_run.as_ref() else { return; }; @@ -561,7 +574,7 @@ impl RuntimeDispatcher { .pending_ready .lock() .expect("poisoned pending ready queue"); - pending.extend(self.ready_batch.drain(..)); + pending.extend(self.ready_batch.drain(..), local_mark); drop(pending); self.core .channel_handles_forwarded(std::mem::take(&mut self.channel_handles)); @@ -580,8 +593,9 @@ impl RuntimeDispatcher { // direct off-thread handles) than commands received since the last // dispatch, so it goes first. if let Ok(mut pending) = active_run.pending_ready.lock() { - pending.extend(self.ready_batch.drain(..)); - std::mem::swap(&mut self.ready_batch, &mut *pending); + let mut queued = pending.take_items(); + queued.append(&mut self.ready_batch); + self.ready_batch = queued; } { diff --git a/src/engine/loop_core.rs b/src/engine/loop_core.rs index 98cc0cf..fb3b5ca 100644 --- a/src/engine/loop_core.rs +++ b/src/engine/loop_core.rs @@ -33,7 +33,7 @@ use super::{ callbacks::{CallbackArgs, CallbackId, CallbackKind, ReadyCallback, TimerCallback}, commands::{ AcceptedConnection, FutureCompletion, LoopCommand, LoopFutureCommand, LoopIoCommand, - LoopRunCommand, LoopTransportCommand, ReadyItem, ServerAcceptStart, + LoopRunCommand, LoopTransportCommand, PendingReady, ReadyItem, ServerAcceptStart, }, dispatcher::run_runtime_thread, timer_entry::{TimerEntry, TimerQueue}, @@ -168,7 +168,6 @@ impl Future for WaitForWake<'_> { } } -const READY_DRAIN_SLICE: usize = 64; const MAX_READY_ITEMS_PER_TURN: usize = 1024; const SIGNAL_POLL_INTERVAL: Duration = Duration::from_millis(50); const RUN_FINISH_TIMEOUT: Duration = Duration::from_secs(5); @@ -323,7 +322,7 @@ pub struct FdWatch { } struct ActiveReadyDispatch { - pending_ready: Arc>>, + pending_ready: Arc>, } /// Mutable lifecycle and Python-facing configuration guarded by `LoopCore`. @@ -403,6 +402,15 @@ pub struct LoopCore { /// handle may join the active queue directly without overtaking an /// earlier handle from the same producer. channel_handles: AtomicUsize, + /// A `stop()` whose queued Stop item has not run yet. Until it runs, the + /// ready drain keeps taking newly queued work into the current turn, so a + /// stop takes effect at the end of the turn that requested it. + stop_requested: AtomicBool, + /// Items the loop thread has queued for itself (its lock-free local ready + /// queue). Only that thread writes it; off-thread producers stamp their + /// items with it so the drain can restore scheduling order between the + /// two queues. + local_enqueues: AtomicU64, // Wakes the loop thread when a producer enqueues a ready item. Held in an // Shared with producers; the park future borrows it under `py.detach`. wake: Arc, @@ -438,6 +446,8 @@ impl LoopCore { pending_timers_dirty: AtomicBool::new(false), cancelled_timers: Arc::new(AtomicUsize::new(0)), channel_handles: AtomicUsize::new(0), + stop_requested: AtomicBool::new(false), + local_enqueues: AtomicU64::new(0), wake: Arc::new(LoopWake::new()), }); @@ -777,7 +787,7 @@ impl LoopCore { state.stopping = false; } - let pending_ready = Arc::new(Mutex::new(VecDeque::new())); + let pending_ready = Arc::new(Mutex::new(PendingReady::default())); self.wake.ready_pending.store(false, Ordering::Release); { let mut active_dispatch = self @@ -790,6 +800,7 @@ impl LoopCore { } self.send_command(LoopCommand::Run(LoopRunCommand::EnterRun { pending_ready: Arc::clone(&pending_ready), + local_mark: self.local_enqueue_mark(), })) .map_err(|err| pyo3::exceptions::PyRuntimeError::new_err(err.to_string()))?; @@ -821,6 +832,9 @@ impl LoopCore { // TLS must point to heap storage, not a queue header on that stack. let mut local_ready = Box::new(VecDeque::new()); self.install_local_ready_queue(&mut *local_ready); + // Loop-thread enqueues already merged into a generation; the items in + // `local_ready` follow them in order. + let mut local_taken = self.local_enqueues.load(Ordering::Relaxed); let mut pending_signal_error: Option = None; let mut ready_batch = VecDeque::new(); @@ -836,62 +850,46 @@ impl LoopCore { #[cfg(all(not(Py_LIMITED_API), not(Py_GIL_DISABLED)))] Python::attach(|_| {}); self.set_ready_drain_active(true); + let queued = local_ready.len(); local_timers.collect(&mut local_ready); + self.note_local_enqueues(local_ready.len() - queued); let mut ready_error = None; let mut deferred_fd_rearms = Vec::new(); - let mut processed_since_refill = 0_usize; let mut processed_this_turn = 0_usize; loop { - if ready_batch.is_empty() || processed_since_refill >= READY_DRAIN_SLICE { + if ready_batch.is_empty() { + // One generation per turn, as in asyncio's `_run_once`: + // callbacks scheduled while this batch runs wait for the + // next turn, after the reactor has been polled again. + // Running each turn to quiescence instead let completions + // pile up between polls, then land in one burst: behind a + // SQLAlchemy pool, more connections came back in one turn + // than its idle queue holds before any woken waiter ran, + // and the overflow was closed and reopened. A pending + // stop() still ends the turn that requested it. + if processed_this_turn != 0 && !self.stop_requested.load(Ordering::Acquire) { + break; + } // Every cross-thread producer raises `wake_pending` after // pushing, so the pending queue only needs to be locked - // when the flag is set; a hot chain of locally scheduled - // callbacks otherwise skips the mutex entirely. + // when the flag is set. Both queues form the next + // generation in the order their items were scheduled. if self.wake.ready_pending.load(Ordering::Acquire) { let mut pending = pending_ready.lock().expect("poisoned pending ready queue"); - if !pending.is_empty() { - if ready_batch.is_empty() { - std::mem::swap(&mut ready_batch, pending.deref_mut()); - } else { - // A refill triggered by READY_DRAIN_SLICE - // leaves - // older items in `ready_batch`. Newly queued - // items have to - // go behind them: asyncio orders callbacks - // by when they were scheduled, so putting the - // fresh arrivals first would run a producer's - // later `call_soon_threadsafe` before its - // earlier - // one. Rare with the GIL, because a producer - // only - // gets to enqueue while this thread is parked; - // constant on a free-threaded interpreter, - // where - // producers append all through the drain. - ready_batch.append(pending.deref_mut()); - } - } - if pending.is_empty() { - self.wake.ready_pending.store(false, Ordering::Release); - } - } - - // Prioritize cross-thread wakeups such as signals and - // transport connection_lost - // notifications so they cannot be starved by a - // hot stream of locally-scheduled callbacks. - if !local_ready.is_empty() { - if ready_batch.is_empty() { - std::mem::swap(&mut ready_batch, &mut *local_ready); - } else { - ready_batch.extend(local_ready.drain(..)); - } + merge_ready_generation( + &mut ready_batch, + &mut local_ready, + &mut local_taken, + pending.deref_mut(), + ); + self.wake.ready_pending.store(false, Ordering::Release); + } else if !local_ready.is_empty() { + local_taken += local_ready.len() as u64; + std::mem::swap(&mut ready_batch, &mut *local_ready); } - processed_since_refill = 0; - if ready_batch.is_empty() { break; } @@ -902,6 +900,7 @@ impl LoopCore { .expect("ready batch was checked as non-empty"); match item { ReadyItem::Stop => { + self.stop_requested.store(false, Ordering::Release); self.state.lock().expect("poisoned loop state").stopping = true; } ReadyItem::Callback(callback) => { @@ -1052,7 +1051,6 @@ impl LoopCore { } } - processed_since_refill += 1; processed_this_turn += 1; if processed_this_turn >= MAX_READY_ITEMS_PER_TURN { // stop() must not truncate callbacks already queued for @@ -1085,7 +1083,7 @@ impl LoopCore { if pending_signal_error.is_none() && let Err(err) = py.check_signals() { - let _ = self.send_command(LoopCommand::RequestStop); + let _ = self.schedule_stop(); pending_signal_error = Some(err); continue; } @@ -1102,8 +1100,14 @@ impl LoopCore { // worker-thread ping-pong on its cheaper spin path. As before, // hot Python callback chains must also service local I/O. if !ready_batch.is_empty() || !local_ready.is_empty() || spin_cooldown > 0 { - LOOP_RUNTIMES.with(|runtimes| { - runtimes.borrow()[&loop_runtime_key].poll_once(); + // Poll with the GIL released, as uvloop does around every + // poll: back-to-back turns otherwise hold it for the whole + // switch interval while threads that feed the loop + // (call_soon_threadsafe producers) wait for it. + py.detach(|| { + LOOP_RUNTIMES.with(|runtimes| { + runtimes.borrow()[&loop_runtime_key].poll_once(); + }); }); if !ready_batch.is_empty() || !local_ready.is_empty() @@ -1182,14 +1186,20 @@ impl LoopCore { // or KeyboardInterrupt) breaks out of the drain loop mid-batch: // FinishRun moves pending_ready back into the runtime's ready_batch. if !ready_batch.is_empty() || !local_ready.is_empty() { - // We want to *prepend* the scheduled items to preserve order (even - // if it's not strictly guaranteed). so rebuild and replace the - // pending Deque + // The unexecuted batch is the oldest work; the next generation + // (both queues, merged in scheduling order) follows it, and items + // queued off-thread from here on go behind both. let mut leftover = std::mem::take(&mut ready_batch); - leftover.extend(local_ready.drain(..)); let mut pending = pending_ready.lock().expect("poisoned pending ready queue"); - leftover.append(pending.deref_mut()); - *pending = leftover; + let mut queued = VecDeque::new(); + merge_ready_generation( + &mut queued, + &mut local_ready, + &mut local_taken, + pending.deref_mut(), + ); + leftover.append(&mut queued); + pending.items = leftover; } self.active_ready_dispatch @@ -1243,6 +1253,7 @@ impl LoopCore { /// stop semantics before `run_forever` returns. #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))] pub fn schedule_stop(&self) -> Result<(), LoopCoreError> { + self.stop_requested.store(true, Ordering::Release); self.send_command(LoopCommand::RequestStop) } @@ -1891,6 +1902,7 @@ impl LoopCore { // to it across this call. TLS is cleared before the allocation // drops. unsafe { (*ready).push_back(item) }; + self.note_local_enqueues(1); if !tls.drain_active.get() { // I/O futures are polled with Python detached, but still on // this thread. Wake the park future without locking the @@ -1912,14 +1924,85 @@ impl LoopCore { return Err(item); }; - dispatch + let mut pending = dispatch .pending_ready .lock() - .expect("poisoned pending ready queue") - .push_back(item); + .expect("poisoned pending ready queue"); + // Read under the queue lock so marks never decrease along the queue. + pending.push(item, self.local_enqueue_mark()); + drop(pending); self.wake.signal(); Ok(()) } + + /// Counts items the loop thread queued for itself. + #[inline] + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))] + fn note_local_enqueues(&self, count: usize) { + // Single writer (the loop thread): a plain load/store pair, no + // read-modify-write. Readers that are ordered after an enqueue by + // other synchronization (a thread start, a lock, a queue) see it by + // coherence, so relaxed accesses suffice. + let enqueues = self.local_enqueues.load(Ordering::Relaxed); + self.local_enqueues + .store(enqueues + count as u64, Ordering::Relaxed); + } + + /// Mark for an item queued off the loop thread now: it follows every + /// loop-thread enqueue counted so far. Call with the pending queue locked. + #[inline] + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))] + pub(crate) fn local_enqueue_mark(&self) -> u64 { + self.local_enqueues.load(Ordering::Relaxed) + } +} + +/// Starts the next ready generation in scheduling order. +/// +/// `local` holds the loop thread's own enqueues, numbered from `local_taken`; +/// an item in `pending` stamped with mark `m` was queued off-thread after the +/// first `m` loop-thread enqueues, so it goes behind the local items numbered +/// below `m` and ahead of the rest. Each queue keeps its own order. Taking the +/// off-thread queue wholesale ahead of the local one instead let a later +/// `call_soon_threadsafe` overtake an earlier `call_soon` whenever both waited +/// for the same generation. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn merge_ready_generation( + batch: &mut VecDeque, + local: &mut VecDeque, + local_taken: &mut u64, + pending: &mut PendingReady, +) { + debug_assert!(batch.is_empty(), "a generation starts from an empty batch"); + let PendingReady { items, marks } = pending; + if items.is_empty() { + *local_taken += local.len() as u64; + std::mem::swap(batch, local); + marks.clear(); + return; + } + if local.is_empty() { + std::mem::swap(batch, items); + marks.clear(); + return; + } + batch.reserve(items.len() + local.len()); + for (index, &(start, mark)) in marks.iter().enumerate() { + // Earlier segments were drained from the front: the last one is + // whatever remains. + let len = marks + .get(index + 1) + .map_or(items.len(), |next| next.0 - start); + let due = usize::try_from(mark.saturating_sub(*local_taken)) + .unwrap_or(usize::MAX) + .min(local.len()); + batch.extend(local.drain(..due)); + *local_taken += due as u64; + batch.extend(items.drain(..len)); + } + marks.clear(); + *local_taken += local.len() as u64; + batch.append(local); } #[cfg(test)] @@ -2166,3 +2249,56 @@ mod wake_tests { )); } } + +#[cfg(test)] +mod generation_order_tests { + use super::*; + + fn merged( + local: &[u32], + local_taken: u64, + pending: &[(u32, u64)], + ) -> (Vec, u64, PendingReady) { + let mut queue = PendingReady::default(); + for &(item, mark) in pending { + queue.push(item, mark); + } + let mut local = local.iter().copied().collect::>(); + let mut taken = local_taken; + let mut batch = VecDeque::new(); + merge_ready_generation(&mut batch, &mut local, &mut taken, &mut queue); + assert!(local.is_empty()); + (batch.into_iter().collect(), taken, queue) + } + + #[test] + fn off_thread_items_follow_the_local_items_queued_before_them() { + // Local items 10, 11, 12 are enqueues #5, #6, #7. + let (order, taken, rest) = merged(&[10, 11, 12], 5, &[(1, 5), (2, 6), (3, 6), (4, 8)]); + assert_eq!(order, [1, 10, 2, 3, 11, 12, 4]); + assert_eq!(taken, 8); + assert!(rest.items.is_empty() && rest.marks.is_empty()); + } + + #[test] + fn stale_marks_and_single_sources_keep_queue_order() { + assert_eq!(merged(&[10, 11], 7, &[(1, 3), (2, 7)]).0, [1, 2, 10, 11]); + assert_eq!(merged(&[10, 11], 7, &[]).0, [10, 11]); + assert_eq!(merged(&[10, 11], 7, &[]).1, 9); + assert_eq!(merged(&[], 7, &[(1, 9), (2, 9)]).0, [1, 2]); + assert_eq!(merged(&[], 7, &[(1, 9)]).1, 7); + } + + #[test] + fn pending_marks_are_stored_once_per_change() { + let mut queue = PendingReady::default(); + queue.push(1_u32, 4); + queue.extend([2, 3], 4); + queue.extend([], 9); + queue.extend([4, 5], 6); + queue.push(6, 6); + assert_eq!(queue.marks, [(0, 4), (3, 6)]); + assert_eq!(queue.take_items(), [1, 2, 3, 4, 5, 6]); + assert!(queue.marks.is_empty()); + } +} diff --git a/tests/test_fast_callbacks.py b/tests/test_fast_callbacks.py index 6725eac..0cbee51 100644 --- a/tests/test_fast_callbacks.py +++ b/tests/test_fast_callbacks.py @@ -247,6 +247,93 @@ def producer(): loop.close() +def _post_threadsafe(loop, callback, *args): + # Thread start and join order this call after everything the caller + # scheduled before and before everything it schedules after. + thread = threading.Thread(target=loop.call_soon_threadsafe, args=(callback, *args)) + thread.start() + thread.join(timeout=5) + assert not thread.is_alive() + + +@pytest.mark.parametrize("factory", [asyncio.new_event_loop, rsloop.new_event_loop]) +@pytest.mark.parametrize("filler", [0, 5, 130]) +def test_earlier_call_soon_runs_before_later_threadsafe_call(factory, filler): + # Both callbacks wait for the same next generation: the loop thread's + # call_soon was queued first, so it must run first. + loop = factory() + events = [] + + def record(value): + events.append(value) + if len(events) == 2: + loop.stop() + + try: + loop.call_soon(loop.call_soon, record, "older") + for index in range(filler + 1): + if index == filler * 3 // 4: + loop.call_soon(_post_threadsafe, loop, record, "newer") + else: + loop.call_soon(lambda: None) + loop.call_later(5, loop.stop) + loop.run_forever() + assert events == ["older", "newer"] + finally: + loop.close() + + +@pytest.mark.parametrize("factory", [asyncio.new_event_loop, rsloop.new_event_loop]) +def test_call_soon_and_threadsafe_calls_interleave_in_scheduling_order(factory): + loop = factory() + events = [] + rounds = 50 + + def schedule(): + for index in range(rounds): + loop.call_soon(events.append, ("local", index)) + _post_threadsafe(loop, events.append, ("thread", index)) + loop.call_soon(loop.stop) + + try: + loop.call_soon(schedule) + loop.call_later(5, loop.stop) + loop.run_forever() + expected = [ + (source, index) for index in range(rounds) for source in ("local", "thread") + ] + assert events == expected + finally: + loop.close() + + +@pytest.mark.parametrize("factory", [asyncio.new_event_loop, rsloop.new_event_loop]) +def test_threadsafe_call_runs_before_later_call_soon(factory): + # The reverse direction: a thread posts, then lets a running callback go + # on to schedule from the loop thread. + loop = factory() + events = [] + posted = threading.Event() + + def poster(): + loop.call_soon_threadsafe(events.append, "older") + posted.set() + + def schedule(): + threading.Thread(target=poster).start() + assert posted.wait(timeout=5) + loop.call_soon(events.append, "newer") + loop.call_soon(loop.stop) + + try: + loop.call_soon(schedule) + loop.call_later(5, loop.stop) + loop.run_forever() + assert events == ["older", "newer"] + finally: + loop.close() + + @pytest.mark.parametrize("factory", [asyncio.new_event_loop, rsloop.new_event_loop]) @pytest.mark.parametrize("threadsafe", [False, True]) def test_empty_callback_contexts_keep_snapshots_tokens_and_explicit_identity( diff --git a/tests/test_local_sockets.py b/tests/test_local_sockets.py index 78bddbd..ea3b36f 100644 --- a/tests/test_local_sockets.py +++ b/tests/test_local_sockets.py @@ -303,3 +303,47 @@ def test_loop_close_releases_socket_wait_cancelled_before_native_poll(): finally: loop.close() sender.close() + + +def test_tcp_data_is_delivered_between_call_soon_generations(): + # Like asyncio's _run_once, a turn runs the callbacks that were ready when + # it started and then polls I/O again, so a callback chain that keeps + # rescheduling itself cannot hold back a readable protocol for a whole + # budget of callbacks (MAX_READY_ITEMS_PER_TURN); hundreds before the fix. + loop = rsloop.new_event_loop() + listener = socket.create_server(("127.0.0.1", 0)) + chained = [0] + seen_at: list[int] = [] + + class Client(asyncio.Protocol): + def data_received(self, data: bytes) -> None: + seen_at.append(chained[0]) + + try: + transport, _ = loop.run_until_complete( + loop.create_connection(Client, *listener.getsockname()) + ) + peer, _ = listener.accept() + + def spin() -> None: + chained[0] += 1 + if chained[0] == 5: + peer.send(b"x") + if not seen_at and chained[0] < 10_000: + loop.call_soon(spin) + + async def until_seen() -> None: + while not seen_at and chained[0] < 10_000: + await asyncio.sleep(0) + + try: + loop.call_soon(spin) + loop.run_until_complete(asyncio.wait_for(until_seen(), 10)) + transport.close() + loop.run_until_complete(asyncio.sleep(0)) + finally: + peer.close() + finally: + listener.close() + loop.close() + assert seen_at and seen_at[0] - 5 < 32, seen_at From 2c95fa2878ff39b8a4a065e7ab8a86f0d51afeeb Mon Sep 17 00:00:00 2001 From: CBoYXD <135316445+CBoYXD@users.noreply.github.com> Date: Wed, 14 Oct 2026 19:02:45 +0300 Subject: [PATCH 5/7] perf: resolve host names natively, without the GIL loop.getaddrinfo() of a host name goes through run_in_executor: a blocking worker takes the GIL to call socket.getaddrinfo and takes it again after the lookup. On a busy loop each acquisition waits for the loop thread to drop the GIL. uvloop resolves in libuv's pool without it. Services that connect by name (a NATS or Postgres client per new connection) pay this on every connect. - resolver.rs: on Unix, an exact loop without a custom default executor, with the standard library's socket.getaddrinfo installed and plain host and port arguments (an ASCII name the IDNA codec passes unchanged or bytes; an int port in range, str or bytes), calls getaddrinfo(3) on the blocking pool without the GIL. The loop thread raises the socket.getaddrinfo audit event through PySys_Audit, as _socket does (a replaced sys.audit skips no hook), and builds the same rows (enum members, glibc's inet_ntop text) or the same gaierror / OSError (EAI_SYSTEM as PyErr_SetFromErrno raises it, errno 0 included). Every other shape keeps the executor path. - A replaced or wrapped socket.getaddrinfo still receives every lookup, as on asyncio: the standard function is recognised by what a functools.wraps wrapper cannot copy (its globals are the socket module's namespace, its code is named getaddrinfo, it has no __wrapped__). A replaced _socket.getaddrinfo, the delegate it calls, keeps the executor path too. Both are checked again on every lookup. - blocking.rs: at most min(32, CPUs + 4) lookups run at once, the size of asyncio's default executor (CPUs from the affinity mask, as os.process_cpu_count counts them); further lookups wait in order, so a burst of 1,000 lookups no longer starts hundreds of threads. run_in_executor jobs never wait for a lookup slot. - engine: a LoopWork ready item runs the loop-thread half of such work, so the worker never takes the GIL. Tests: tests/test_native_addrinfo.py compares 25 argument shapes with socket.getaddrinfo (rows, enum types and errors, including the shapes left to CPython), 64 concurrent lookups, cancelled lookups, a lookup created before run_until_complete, the audit event (also with sys.audit replaced), a functools.wraps wrapper installed before the first lookup, a replaced _socket.getaddrinfo before and after native lookups, and the thread cap under a burst. Rust tests cover the bounded slots, glibc's IPv6 text, the IDNA fast path, the cap formula and the EAI_SYSTEM mapping. --- docs/how-it-works.md | 9 + src/bindings/loop_api.rs | 2 + src/bindings/loop_api/executor.rs | 4 + src/bindings/loop_api/resolver.rs | 643 ++++++++++++++++++++++++++++++ src/blocking.rs | 223 +++++++++++ src/engine/commands.rs | 10 + src/engine/dispatcher.rs | 4 + src/engine/loop_core.rs | 12 + src/engine/mod.rs | 2 + tests/test_native_addrinfo.py | 372 +++++++++++++++++ 10 files changed, 1281 insertions(+) create mode 100644 src/bindings/loop_api/resolver.rs create mode 100644 tests/test_native_addrinfo.py diff --git a/docs/how-it-works.md b/docs/how-it-works.md index 76a417b..6909b15 100644 --- a/docs/how-it-works.md +++ b/docs/how-it-works.md @@ -83,6 +83,15 @@ Each loop combines a coordination runtime with a loop-thread I/O runtime: workers because `vibeio` does not expose arbitrary raw-descriptor registration - some transport paths still fall back to helper threads, especially TLS I/O, TLS server accept, and parts of the legacy transport write path +- on Unix, `getaddrinfo()` of plain host and port arguments calls the system + resolver on the blocking pool without the GIL and builds the `socket.getaddrinfo` + result on the loop thread. At most `min(32, CPUs + 4)` lookups run at once + per process, the size of asyncio's default executor (CPUs are those the + process may run on, not a container's CPU quota); further lookups wait + their turn instead of starting more threads. Loop subclasses, a custom + default executor, a replaced or wrapped `socket.getaddrinfo` or + `_socket.getaddrinfo` (checked on every lookup) and arguments that need CPython's own conversion (IDNA, + unusual port types) keep the executor path Python tasks and callbacks still execute on the Python side. The coordination and loop-thread I/O runtimes use the embedded `src/vibeio/` implementation. diff --git a/src/bindings/loop_api.rs b/src/bindings/loop_api.rs index ad6733b..42378cc 100644 --- a/src/bindings/loop_api.rs +++ b/src/bindings/loop_api.rs @@ -27,6 +27,8 @@ mod pre_exec; mod process_handles; mod process_spawn; mod process_stdio; +#[cfg(unix)] +mod resolver; mod servers; mod signals; mod sock_ops; diff --git a/src/bindings/loop_api/executor.rs b/src/bindings/loop_api/executor.rs index 55939d7..739ca4e 100644 --- a/src/bindings/loop_api/executor.rs +++ b/src/bindings/loop_api/executor.rs @@ -99,6 +99,10 @@ pub(super) fn getaddrinfo<'py>( if let Some(result) = numeric_addrinfo(&slf, py, &request)? { return Ok(result); } + #[cfg(unix)] + if let Some(result) = super::resolver::getaddrinfo(&slf, py, &request)? { + return Ok(result); + } let socket = py.import("socket")?; let host = request.host.unwrap_or_else(|| py.None()); let port = request.port.unwrap_or_else(|| py.None()); diff --git a/src/bindings/loop_api/resolver.rs b/src/bindings/loop_api/resolver.rs new file mode 100644 index 0000000..9a8b648 --- /dev/null +++ b/src/bindings/loop_api/resolver.rs @@ -0,0 +1,643 @@ +//! Native `loop.getaddrinfo()` for exact loops. +//! +//! Through `run_in_executor`, `socket.getaddrinfo` needs the GIL on the +//! worker before and after the lookup, so every resolution of a host name +//! queues behind the loop thread twice. uvloop resolves in libuv's pool +//! without the GIL. Here the common argument shapes are encoded on the loop +//! thread as CPython's `_socket.getaddrinfo` encodes them, `getaddrinfo(3)` +//! runs on a blocking-pool worker without the GIL, and the loop thread builds +//! the rows or the error `socket.getaddrinfo` would produce. Every other shape +//! keeps the Python path. + +use std::{ + ffi::{CStr, CString, c_char, c_int}, + net::Ipv4Addr, + ptr, +}; + +use pyo3::{ + exceptions::PyValueError, + prelude::*, + sync::PyOnceLock, + types::{PyBytes, PyInt, PyList, PyModule, PyString, PyTuple}, +}; + +use super::{PyLoop, executor::AddrInfoRequest}; +use crate::engine::{LoopCommand, LoopWork}; + +/// Arguments already encoded for `getaddrinfo(3)`. +struct Query { + host: Option, + port: Option, + family: i32, + sock_type: i32, + proto: i32, + flags: i32, +} + +enum Address { + V4(String, u16), + V6(String, u16, u32, u32), + /// Any other family: CPython reports `(family, sa_data)`. + Other(i32, Vec), +} + +struct Row { + family: i32, + sock_type: i32, + proto: i32, + canonname: Vec, + /// None when the entry carries no address (`ai_addrlen == 0`). + address: Option
, +} + +enum Outcome { + Rows(Vec), + /// A `getaddrinfo` error code other than `EAI_SYSTEM`. + Gai(i32), + /// `EAI_SYSTEM` with its `errno`. + Os(i32), +} + +/// Resolves `request` natively when the loop and the arguments allow it. +/// Returns None to keep the Python path: subclasses and custom default +/// executors stay authoritative, and unusual arguments keep CPython's +/// conversion and errors. +#[cfg_attr(feature = "profile", hotpath::measure)] +pub(super) fn getaddrinfo<'py>( + slf: &Py, + py: Python<'py>, + request: &AddrInfoRequest, +) -> PyResult>> { + if !slf.bind(py).is_exact_instance_of::() { + return Ok(None); + } + { + let loop_ref = slf.borrow(py); + let state = loop_ref.core.state.lock().expect("poisoned loop state"); + if state.default_executor.is_some() || state.executor_shutdown_called { + return Ok(None); + } + } + if !stdlib_getaddrinfo_in_place(py)? { + return Ok(None); + } + let host = request.host.as_ref().map(|host| host.bind(py)); + let port = request.port.as_ref().map(|port| port.bind(py)); + let (Some(host_arg), Some(port_arg)) = ( + encode_optional(host, encode_host), + encode_optional(port, encode_port), + ) else { + return Ok(None); + }; + let query = Query { + host: host_arg, + port: port_arg, + family: request.family, + sock_type: request.sock_type, + proto: request.proto, + flags: request.flags, + }; + + let future = super::tasks::create_future(slf.clone_ref(py), py)?; + let none = py.None().into_bound(py); + let audited = audit(py, host.unwrap_or(&none), port.unwrap_or(&none), request); + match audited { + Ok(_) => { + let core = slf.borrow(py).core.clone(); + let pending = future.clone_ref(py); + crate::blocking::spawn_bounded(move || { + let outcome = query.resolve(); + let work = LoopWork(Box::new(move |py| complete(py, &pending, outcome))); + // A closed loop drops the work; nobody awaits its future any + // more. + let _ = core.send_command(LoopCommand::LoopWork(Box::new(work))); + }); + } + Err(err) => { + future.call_method1(py, "set_exception", (err.into_value(py),))?; + } + } + Ok(Some(future.into_bound(py))) +} + +unsafe extern "C" { + // CPython's C audit API (pyo3-ffi does not bind it). + fn PySys_Audit(event: *const c_char, format: *const c_char, ...) -> c_int; +} + +/// Raises the audit event `_socket.getaddrinfo` raises before resolving, +/// through the C API as CPython does, so replacing the Python-level +/// `sys.audit` cannot skip a registered hook. A hook's error refuses the +/// lookup. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn audit( + py: Python<'_>, + host: &Bound<'_, PyAny>, + port: &Bound<'_, PyAny>, + request: &AddrInfoRequest, +) -> PyResult<()> { + // SAFETY: the GIL is held, the format matches the arguments (two + // borrowed objects and three C ints, as in CPython's + // socket_getaddrinfo), and a -1 return leaves the hook's error set. + let code = unsafe { + PySys_Audit( + c"socket.getaddrinfo".as_ptr(), + c"OOiii".as_ptr(), + host.as_ptr(), + port.as_ptr(), + request.family, + request.sock_type, + request.proto, + ) + }; + if code < 0 { + return Err(PyErr::fetch(py)); + } + Ok(()) +} + +/// Whether `socket.getaddrinfo` is still the standard library's function +/// and the `_socket.getaddrinfo` it delegates to is still CPython's, +/// checked on every lookup. The Python path calls whatever is installed +/// there, as asyncio does, so a test double or wrapper at either level keeps +/// receiving every name lookup. `functools.wraps` copies a function's name, +/// module and `__dict__`, so the function is recognised by what a wrapper +/// cannot copy: a plain function whose globals are the `socket` module's +/// namespace, whose code is named `getaddrinfo`, and which wraps nothing. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn stdlib_getaddrinfo_in_place(py: Python<'_>) -> PyResult { + static STDLIB: PyOnceLock> = PyOnceLock::new(); + let socket = py.import("socket")?; + let installed = socket.getattr("getaddrinfo")?; + let is_stdlib = STDLIB + .get(py) + .is_some_and(|stdlib| installed.is(stdlib.bind(py))) + || { + let is_stdlib = installed + .get_type() + .is(&py.import("types")?.getattr("FunctionType")?) + && installed + .getattr("__globals__")? + .is(&socket.getattr("__dict__")?) + && installed + .getattr("__code__")? + .getattr("co_name")? + .eq("getaddrinfo")? + && !installed.hasattr("__wrapped__")?; + if is_stdlib { + // A reloaded module's new function is recognised again above. + let _ = STDLIB.set(py, installed.unbind()); + } + is_stdlib + }; + Ok(is_stdlib && builtin_delegate_in_place(py, &socket)?) +} + +/// Whether the `_socket.getaddrinfo` that `socket.getaddrinfo` calls (the +/// `socket` module's `_socket` global) is CPython's builtin: a builtin +/// function of the `_socket` extension module named `getaddrinfo`. A +/// replaced delegate keeps the Python path. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn builtin_delegate_in_place(py: Python<'_>, socket: &Bound<'_, PyModule>) -> PyResult { + static BUILTIN: PyOnceLock> = PyOnceLock::new(); + let delegate = socket.getattr("_socket")?.getattr("getaddrinfo")?; + if BUILTIN + .get(py) + .is_some_and(|builtin| delegate.is(builtin.bind(py))) + { + return Ok(true); + } + let is_builtin = delegate + .get_type() + .is(&py.import("types")?.getattr("BuiltinFunctionType")?) + && delegate.getattr("__self__")?.is(&py.import("_socket")?) + && delegate.getattr("__name__")?.eq("getaddrinfo")?; + if is_builtin { + let _ = BUILTIN.set(py, delegate.unbind()); + } + Ok(is_builtin) +} + +/// `Some(None)` for a None argument, None when `encode` declines the value. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn encode_optional( + value: Option<&Bound<'_, PyAny>>, + encode: fn(&Bound<'_, PyAny>) -> Option, +) -> Option> { + match value { + None => Some(None), + Some(value) if value.is_none() => Some(None), + Some(value) => encode(value).map(Some), + } +} + +/// A `str` host in the IDNA codec's ASCII fast path (encoded unchanged), or +/// `bytes`. Other strings need the codec itself. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn encode_host(host: &Bound<'_, PyAny>) -> Option { + if let Ok(text) = host.cast_exact::() { + let name = text.to_str().ok()?; + return idna_ascii_unchanged(name) + .then(|| CString::new(name).ok()) + .flatten(); + } + let bytes = host.cast_exact::().ok()?; + CString::new(bytes.as_bytes()).ok() +} + +/// Whether `encodings.idna` returns `name.encode("ascii")` for `name`: ASCII, +/// no empty label before the last one, and no label of 64 bytes or more. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn idna_ascii_unchanged(name: &str) -> bool { + if name.is_empty() || !name.is_ascii() { + return false; + } + let labels: Vec<&str> = name.split('.').collect(); + let last = labels.len() - 1; + labels + .iter() + .enumerate() + .all(|(index, label)| label.len() < 64 && (index == last || !label.is_empty())) +} + +/// An `int` port in range (formatted as CPython formats it), or a `str` / +/// `bytes` service. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn encode_port(port: &Bound<'_, PyAny>) -> Option { + if let Ok(number) = port.cast_exact::() { + return CString::new(number.extract::().ok()?.to_string()).ok(); + } + if let Ok(text) = port.cast_exact::() { + return CString::new(text.to_str().ok()?).ok(); + } + let bytes = port.cast_exact::().ok()?; + CString::new(bytes.as_bytes()).ok() +} + +impl Query { + /// Calls `getaddrinfo(3)`; runs on a pool worker without the GIL. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "Query"))] + fn resolve(&self) -> Outcome { + // SAFETY: an all-zero addrinfo is the documented empty hints value. + let mut hints: libc::addrinfo = unsafe { std::mem::zeroed() }; + hints.ai_family = self.family; + hints.ai_socktype = self.sock_type; + hints.ai_protocol = self.proto; + hints.ai_flags = self.flags; + let mut list: *mut libc::addrinfo = ptr::null_mut(); + // SAFETY: host and port are NUL-terminated or null, hints is + // initialized, and list receives a result freed below. + let code = unsafe { + libc::getaddrinfo( + self.host.as_ref().map_or(ptr::null(), |host| host.as_ptr()), + self.port.as_ref().map_or(ptr::null(), |port| port.as_ptr()), + &raw const hints, + &raw mut list, + ) + }; + if code == libc::EAI_SYSTEM { + return Outcome::Os(std::io::Error::last_os_error().raw_os_error().unwrap_or(0)); + } + if code != 0 { + return Outcome::Gai(code); + } + let mut rows = Vec::new(); + let mut entry = list; + while !entry.is_null() { + // SAFETY: entry is a node of the list getaddrinfo returned. + let info = unsafe { &*entry }; + rows.push(Row { + family: info.ai_family, + sock_type: info.ai_socktype, + proto: info.ai_protocol, + canonname: if info.ai_canonname.is_null() { + Vec::new() + } else { + // SAFETY: a non-null ai_canonname is NUL-terminated. + unsafe { CStr::from_ptr(info.ai_canonname) } + .to_bytes() + .to_vec() + }, + address: address(info), + }); + entry = info.ai_next; + } + // SAFETY: list came from a successful getaddrinfo and is freed once. + unsafe { libc::freeaddrinfo(list) }; + Outcome::Rows(rows) + } +} + +/// The sockaddr of one entry, decoded as CPython's `makesockaddr` does. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn address(info: &libc::addrinfo) -> Option
{ + if info.ai_addr.is_null() || info.ai_addrlen == 0 { + return None; + } + let length = info.ai_addrlen as usize; + // SAFETY: ai_addr points to ai_addrlen bytes; each family is read only + // when it is at least as long as its sockaddr type. + unsafe { + let family = i32::from((*info.ai_addr).sa_family); + if family == libc::AF_INET && length >= size_of::() { + let v4 = &*info.ai_addr.cast::(); + let ip = Ipv4Addr::from(u32::from_be(v4.sin_addr.s_addr)); + return Some(Address::V4(ip.to_string(), u16::from_be(v4.sin_port))); + } + if family == libc::AF_INET6 && length >= size_of::() { + let v6 = &*info.ai_addr.cast::(); + return Some(Address::V6( + ntop6(v6.sin6_addr.s6_addr), + u16::from_be(v6.sin6_port), + u32::from_be(v6.sin6_flowinfo), + v6.sin6_scope_id, + )); + } + Some(Address::Other( + family, + (*info.ai_addr) + .sa_data + .iter() + .map(|byte| *byte as u8) + .collect(), + )) + } +} + +/// glibc's `inet_ntop(AF_INET6)`, which CPython uses for IPv6 addresses. +/// Rust's own formatting differs for IPv4-compatible addresses (`::1.2.3.4`). +#[cfg_attr(feature = "profile", hotpath::measure)] +fn ntop6(octets: [u8; 16]) -> String { + use std::fmt::Write; + + let words: [u16; 8] = + std::array::from_fn(|index| u16::from_be_bytes([octets[2 * index], octets[2 * index + 1]])); + // The first longest run of at least two zero words becomes "::". + // (first word, length) of a run of zero words. + let mut best: Option<(usize, usize)> = None; + let mut current: Option<(usize, usize)> = None; + for (index, word) in words.iter().enumerate() { + if *word == 0 { + current = Some(current.map_or((index, 1), |(base, len)| (base, len + 1))); + } else if let Some(run) = current.take() + && best.is_none_or(|(_, len)| run.1 > len) + { + best = Some(run); + } + } + if let Some(run) = current + && best.is_none_or(|(_, len)| run.1 > len) + { + best = Some(run); + } + let best = best.filter(|(_, len)| *len >= 2); + let mut text = String::with_capacity(46); + for (index, word) in words.iter().enumerate() { + if let Some((base, len)) = best + && (base..base + len).contains(&index) + { + if index == base { + text.push(':'); + } + continue; + } + if index != 0 { + text.push(':'); + } + if index == 6 + && let Some((0, len)) = best + && (len == 6 || (len == 5 && words[5] == 0xffff)) + { + let _ = write!( + text, + "{}", + Ipv4Addr::new(octets[12], octets[13], octets[14], octets[15]) + ); + return text; + } + let _ = write!(text, "{word:x}"); + } + if best.is_some_and(|(base, len)| base + len == 8) { + text.push(':'); + } + text +} + +/// Settles the future on the loop thread with what `socket.getaddrinfo` +/// would have returned or raised. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn complete(py: Python<'_>, future: &Py, outcome: Outcome) -> PyResult<()> { + let future = future.bind(py); + if crate::python_names::call_method0(py, future, crate::python_names::done(py))? + .bind(py) + .extract::()? + { + return Ok(()); + } + let settled = match outcome { + Outcome::Rows(rows) => rows_to_list(py, rows), + Outcome::Gai(code) => Err(gaierror(py, code)), + Outcome::Os(errno) => Err(os_error(py, errno)), + }; + match settled { + Ok(list) => crate::python_names::call_method1( + py, + future, + crate::python_names::set_result(py), + list.as_any(), + ), + Err(err) => crate::python_names::call_method1( + py, + future, + crate::python_names::set_exception(py), + err.into_value(py).bind(py).as_any(), + ), + }?; + Ok(()) +} + +#[cfg_attr(feature = "profile", hotpath::measure)] +fn rows_to_list(py: Python<'_>, rows: Vec) -> PyResult> { + let socket = py.import("socket")?; + let families = socket.getattr("AddressFamily")?; + let kinds = socket.getattr("SocketKind")?; + let list = PyList::empty(py); + for row in rows { + let canonname = match std::str::from_utf8(&row.canonname) { + Ok(text) => PyString::new(py, text).into_any(), + // CPython decodes strictly; let the codec raise its own error. + Err(_) => PyBytes::new(py, &row.canonname).call_method1("decode", ("utf-8",))?, + }; + let address = match row.address { + None => py.None().into_bound(py), + Some(Address::V4(host, port)) => (host, port).into_pyobject(py)?.into_any(), + Some(Address::V6(host, port, flowinfo, scope_id)) => (host, port, flowinfo, scope_id) + .into_pyobject(py)? + .into_any(), + Some(Address::Other(family, data)) => (family, PyBytes::new(py, &data)) + .into_pyobject(py)? + .into_any(), + }; + list.append(PyTuple::new( + py, + [ + int_enum(&families, row.family)?, + int_enum(&kinds, row.sock_type)?, + row.proto.into_pyobject(py)?.into_any(), + canonname, + address, + ], + )?)?; + } + Ok(list) +} + +/// `socket._intenum_converter`: the enum member, or the plain int. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn int_enum<'py>(class: &Bound<'py, PyAny>, value: i32) -> PyResult> { + let py = class.py(); + match class.call1((value,)) { + Ok(member) => Ok(member), + Err(err) if err.is_instance_of::(py) => { + Ok(value.into_pyobject(py)?.into_any()) + } + Err(err) => Err(err), + } +} + +/// `socket.gaierror(code, gai_strerror(code))`, as CPython raises it. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn gaierror(py: Python<'_>, code: i32) -> PyErr { + // SAFETY: gai_strerror returns a static NUL-terminated message. + let message = unsafe { CStr::from_ptr(libc::gai_strerror(code)) } + .to_string_lossy() + .into_owned(); + match py + .import("socket") + .and_then(|socket| socket.getattr("gaierror")) + .and_then(|class| class.call1((code, message))) + { + Ok(error) => PyErr::from_value(error), + Err(err) => err, + } +} + +/// The `OSError` `PyErr_SetFromErrno` raises for `EAI_SYSTEM`: +/// `OSError(errno, os.strerror(errno))`, which picks the errno's subclass, or +/// `OSError(0, "Error")` when `errno` was not set. +#[cfg_attr(feature = "profile", hotpath::measure)] +fn os_error(py: Python<'_>, errno: i32) -> PyErr { + let message = if errno == 0 { + Ok(PyString::new(py, "Error").into_any()) + } else { + py.import("os") + .and_then(|os| os.call_method1("strerror", (errno,))) + }; + match message.and_then(|message| { + py.get_type::() + .call1((errno, message)) + }) { + Ok(error) => PyErr::from_value(error), + Err(err) => err, + } +} + +#[cfg(test)] +mod tests { + use pyo3::types::IntoPyDict; + + use super::*; + + fn v6(text: &str) -> String { + ntop6(text.parse::().unwrap().octets()) + } + + #[test] + fn ipv6_text_matches_glibc_inet_ntop() { + // Expected values are glibc's (socket.inet_ntop on Linux). + for (address, expected) in [ + ("::", "::"), + ("::1", "::1"), + ("::1.2.3.4", "::1.2.3.4"), + ("::ffff:1.2.3.4", "::ffff:1.2.3.4"), + ("::ffff:0:0", "::ffff:0.0.0.0"), + ("::0.0.0.1", "::1"), + ("1::", "1::"), + ("fe80::1", "fe80::1"), + ("2001:db8::ff00:42:8329", "2001:db8::ff00:42:8329"), + ("2001:db8:0:0:1:0:0:1", "2001:db8::1:0:0:1"), + ("2001:0:0:1:0:0:0:1", "2001:0:0:1::1"), + ("1:0:2:0:3:0:4:0", "1:0:2:0:3:0:4:0"), + ("1:2:3:4:5:6:7:8", "1:2:3:4:5:6:7:8"), + ("0:0:1::", "0:0:1::"), + ("::1:0:0:0:0", "0:0:0:1::"), + ("64:ff9b::1.2.3.4", "64:ff9b::102:304"), + ] { + assert_eq!(v6(address), expected, "{address}"); + } + } + + #[cfg(target_os = "linux")] + #[test] + fn eai_system_errors_match_pyerr_set_from_errno() { + // socket.getaddrinfo raises EAI_SYSTEM through PyErr_SetFromErrno. + crate::initialize_python_for_tests(); + Python::attach(|py| { + let args = |err: &PyErr| { + err.value(py) + .getattr("args") + .and_then(|args| args.repr()) + .unwrap() + .to_string() + }; + for errno in [0, libc::ENOENT, libc::EAGAIN, libc::ENOMEM, libc::EMFILE] { + // SAFETY: errno belongs to this thread and is read at once by + // PyErr_SetFromErrno, whose error PyErr::fetch takes back. + let expected = unsafe { + *libc::__errno_location() = errno; + pyo3::ffi::PyErr_SetFromErrno(pyo3::ffi::PyExc_OSError); + PyErr::fetch(py) + }; + let actual = os_error(py, errno); + assert!( + actual.get_type(py).is(expected.get_type(py)), + "{errno}: {actual} vs {expected}" + ); + assert_eq!(args(&actual), args(&expected), "{errno}"); + } + }); + } + + #[cfg(target_os = "linux")] + #[test] + fn lookups_run_as_many_at_once_as_asyncios_default_executor() { + crate::initialize_python_for_tests(); + let expected = Python::attach(|py| { + // ThreadPoolExecutor's default max_workers on Python 3.13+, whose + // os.process_cpu_count() counts the affinity mask. + py.eval( + c"min(32, (os.process_cpu_count() if hasattr(os, 'process_cpu_count') else len(os.sched_getaffinity(0))) + 4)", + Some(&[("os", py.import("os").unwrap())].into_py_dict(py).unwrap()), + None, + ) + .unwrap() + .extract::() + .unwrap() + }); + assert_eq!(crate::blocking::bounded_limit(), expected); + } + + #[test] + fn idna_fast_path_matches_the_codec() { + assert!(idna_ascii_unchanged("localhost")); + assert!(idna_ascii_unchanged("localhost.")); + assert!(idna_ascii_unchanged("a.b.c")); + assert!(idna_ascii_unchanged(&"x".repeat(63))); + assert!(!idna_ascii_unchanged(&"x".repeat(64))); + assert!(!idna_ascii_unchanged("")); + assert!(!idna_ascii_unchanged("a..b")); + assert!(!idna_ascii_unchanged(".a")); + assert!(!idna_ascii_unchanged("ünï.example")); + } +} diff --git a/src/blocking.rs b/src/blocking.rs index 28eb54e..4573eb8 100644 --- a/src/blocking.rs +++ b/src/blocking.rs @@ -1,4 +1,15 @@ //! Adapter for blocking operations that cannot run on an async reactor thread. +//! +//! On Unix, [`spawn_bounded`] jobs (name resolution) run on the same pool as +//! [`run`], but at most `min(32, CPUs + 4)` at a time; the rest wait in +//! submission order, so a burst of lookups does not start a thread per lookup. + +#[cfg(unix)] +use std::{ + collections::VecDeque, + panic::{self, AssertUnwindSafe}, + sync::{LazyLock, Mutex, MutexGuard}, +}; /// Run blocking work on the shared pool. Panic isolation retains the previous /// dropped-worker error contract without spawning a thread for every call. @@ -12,3 +23,215 @@ where .await .map_err(|_| "blocking worker dropped".to_owned()) } + +#[cfg(unix)] +type Job = Box; + +/// The slots of [`spawn_bounded`]. +#[cfg(unix)] +struct Bounded { + /// `min(32, CPUs + 4)`, the size of asyncio's default executor (see + /// [`usable_cpus`]). + limit: usize, + state: Mutex, +} + +#[cfg(unix)] +#[derive(Default)] +struct BoundedState { + /// Jobs holding a slot. + running: usize, + /// Jobs waiting for a slot. A worker that finishes a bounded job takes + /// the next of these itself. + waiting: VecDeque, +} + +#[cfg(unix)] +static BOUNDED: LazyLock = LazyLock::new(|| Bounded { + limit: usable_cpus().saturating_add(4).min(32), + state: Mutex::new(BoundedState::default()), +}); + +#[cfg(unix)] +impl Bounded { + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "Bounded"))] + fn lock(&self) -> MutexGuard<'_, BoundedState> { + // Jobs run outside the lock, so a poisoned lock still has a consistent + // state. + self.state.lock().unwrap_or_else(|error| error.into_inner()) + } + + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "Bounded"))] + fn submit(&'static self, job: Job) { + let mut state = self.lock(); + if state.running >= self.limit { + state.waiting.push_back(job); + return; + } + state.running += 1; + drop(state); + smol::unblock(move || self.run_slot(job)).detach(); + } + + /// Runs `job`, then every job that waits for a slot when it finishes. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "Bounded"))] + fn run_slot(&self, mut job: Job) { + loop { + // The panic hook has reported a panic; the slot still passes on. + let _ = panic::catch_unwind(AssertUnwindSafe(job)); + let mut state = self.lock(); + match state.waiting.pop_front() { + Some(next) => job = next, + None => { + state.running -= 1; + return; + } + } + } + } +} + +/// The CPUs this process may run on, as `os.process_cpu_count()` counts +/// them for asyncio's default executor: the affinity mask, not a cgroup CPU +/// quota. Lookups mostly wait on the network, so a container's `--cpus` +/// should not shrink how many run at once. +#[cfg(unix)] +#[cfg_attr(feature = "profile", hotpath::measure)] +fn usable_cpus() -> usize { + #[cfg(target_os = "linux")] + { + // SAFETY: an all-zero cpu_set_t is an empty set, which + // sched_getaffinity fills for the calling thread. + let mut set: libc::cpu_set_t = unsafe { std::mem::zeroed() }; + // SAFETY: set is a valid cpu_set_t of the size passed. + if unsafe { libc::sched_getaffinity(0, size_of::(), &raw mut set) } == 0 { + // SAFETY: set was filled by a successful sched_getaffinity. + let count = unsafe { libc::CPU_COUNT(&set) }; + if let Ok(count @ 1..) = usize::try_from(count) { + return count; + } + } + } + std::thread::available_parallelism().map_or(1, std::num::NonZero::get) +} + +/// Runs one blocking closure on the shared pool, without waiting for it, with +/// at most `min(32, CPUs + 4)` of these running at once: beyond that, the +/// closure waits in submission order for one of them to finish. +#[cfg(unix)] +#[cfg_attr(feature = "profile", hotpath::measure)] +pub fn spawn_bounded(task: impl FnOnce() + Send + 'static) { + BOUNDED.submit(Box::new(task)); +} + +/// How many [`spawn_bounded`] jobs run at once in this process. +#[cfg(all(unix, test))] +pub(crate) fn bounded_limit() -> usize { + BOUNDED.limit +} + +#[cfg(all(unix, test))] +mod tests { + use std::{ + pin::pin, + sync::{Arc, atomic::Ordering, mpsc}, + task::{Context, Poll, Wake, Waker}, + thread, + time::Duration, + }; + + use super::*; + + struct Unpark(thread::Thread); + + impl Wake for Unpark { + fn wake(self: Arc) { + self.0.unpark(); + } + } + + fn block_on(future: impl Future) -> T { + let waker = Waker::from(Arc::new(Unpark(thread::current()))); + let mut context = Context::from_waker(&waker); + let mut future = pin!(future); + loop { + if let Poll::Ready(value) = future.as_mut().poll(&mut context) { + return value; + } + thread::park(); + } + } + + #[test] + fn bounded_jobs_run_at_most_the_limit_at_once_and_all_finish() { + use std::sync::atomic::AtomicUsize; + + let limit = bounded_limit(); + let jobs = 4 * limit + 3; + let running = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let (done, finished) = mpsc::channel(); + for index in 0..jobs { + let (running, peak, done) = (running.clone(), peak.clone(), done.clone()); + spawn_bounded(move || { + let now = running.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(now, Ordering::SeqCst); + thread::sleep(Duration::from_millis(5)); + running.fetch_sub(1, Ordering::SeqCst); + let _ = done.send(index); + }); + } + let mut seen: Vec<_> = (0..jobs) + .map(|_| finished.recv_timeout(Duration::from_secs(30)).unwrap()) + .collect(); + seen.sort_unstable(); + assert_eq!(seen, (0..jobs).collect::>()); + // Other tests' bounded jobs may hold slots too, so the peak can stay + // below the limit, never above it. + let peak = peak.load(Ordering::SeqCst); + assert!((1..=limit).contains(&peak), "{peak} of {limit}"); + } + + #[test] + fn blocked_bounded_jobs_do_not_delay_other_jobs() { + let limit = bounded_limit(); + let (release, blocked) = mpsc::channel::<()>(); + let blocked = Arc::new(Mutex::new(blocked)); + let (done, finished) = mpsc::channel(); + // Every bounded slot waits, and one more job queues behind them. + for _ in 0..=limit { + let (blocked, done) = (blocked.clone(), done.clone()); + spawn_bounded(move || { + let _ = blocked + .lock() + .unwrap() + .recv_timeout(Duration::from_secs(30)); + let _ = done.send(()); + }); + } + // A blocked job finishes only once released (or after its timeout, + // which only a stall reaches), so `run` returning first shows it did + // not wait for a bounded slot. + assert_eq!(block_on(run("test", || 7)).unwrap(), 7); + assert_eq!(finished.try_recv(), Err(mpsc::TryRecvError::Empty)); + for _ in 0..=limit { + release.send(()).unwrap(); + } + for _ in 0..=limit { + finished.recv_timeout(Duration::from_secs(30)).unwrap(); + } + } + + #[test] + fn a_panicking_bounded_job_passes_its_slot_on() { + let limit = bounded_limit(); + for _ in 0..2 * limit { + spawn_bounded(|| panic!("bounded job panics")); + } + let (done, finished) = mpsc::channel(); + spawn_bounded(move || { + let _ = done.send(9); + }); + assert_eq!(finished.recv_timeout(Duration::from_secs(30)), Ok(9)); + } +} diff --git a/src/engine/commands.rs b/src/engine/commands.rs index 9f2e1c4..8f17291 100644 --- a/src/engine/commands.rs +++ b/src/engine/commands.rs @@ -42,6 +42,8 @@ pub enum ReadyItem { // batch instead of one contended handoff per completion. #[cfg(unix)] ConnectCompleted(Box), + #[cfg(unix)] + LoopWork(Box), Stop, } @@ -50,6 +52,11 @@ pub enum ReadyItem { // notifications remain allocation-free at the queue boundary. const _: () = assert!(std::mem::size_of::() <= 2 * std::mem::size_of::()); +/// The loop-thread half of work finished off the loop thread: it builds the +/// Python result and settles a future, so the worker never takes the GIL. +#[cfg(unix)] +pub struct LoopWork(pub Box) -> PyResult<()> + Send>); + pub struct FutureCompletion { pub future: Py, pub value: Py, @@ -178,6 +185,9 @@ pub enum LoopCommand { /// Error reported while waiting for writability, or zero. wait_errno: i32, }, + /// Runs the loop-thread half of off-thread work. + #[cfg(unix)] + LoopWork(Box), /// Requests that the active `run_forever` call return after ready work /// drains. RequestStop, diff --git a/src/engine/dispatcher.rs b/src/engine/dispatcher.rs index e341807..130d00d 100644 --- a/src/engine/dispatcher.rs +++ b/src/engine/dispatcher.rs @@ -536,6 +536,10 @@ impl RuntimeDispatcher { wait_errno, }))); } + #[cfg(unix)] + LoopCommand::LoopWork(work) => { + self.ready_batch.push_back(ReadyItem::LoopWork(work)); + } LoopCommand::RequestStop => { self.ready_batch.push_back(ReadyItem::Stop); } diff --git a/src/engine/loop_core.rs b/src/engine/loop_core.rs index fb3b5ca..8064cfd 100644 --- a/src/engine/loop_core.rs +++ b/src/engine/loop_core.rs @@ -1049,6 +1049,8 @@ impl LoopCore { } = *payload; self.resolve_connect_completed(py, future, fd, wait_errno)?; } + #[cfg(unix)] + ReadyItem::LoopWork(work) => (work.0)(py)?, } processed_this_turn += 1; @@ -1701,6 +1703,8 @@ impl LoopCore { ReadyItem::Callback(callback) => LoopCommand::ScheduleReady(callback), ReadyItem::HandleCallback(handle) => LoopCommand::ScheduleReadyHandle(handle), ReadyItem::Stop => LoopCommand::RequestStop, + #[cfg(unix)] + ReadyItem::LoopWork(work) => LoopCommand::LoopWork(work), ReadyItem::FutureSetResult(payload) => { let FutureCompletion { future, value } = *payload; LoopCommand::Future(LoopFutureCommand::SetResult { future, value }) @@ -1875,6 +1879,14 @@ impl LoopCore { } _ => unreachable!("local connect completion enqueue preserves item kind"), }), + #[cfg(unix)] + LoopCommand::LoopWork(work) => self + .try_enqueue_local_ready(ReadyItem::LoopWork(work)) + .or_else(|item| self.try_enqueue_active_ready(item)) + .map_err(|item| match item { + ReadyItem::LoopWork(work) => LoopCommand::LoopWork(work), + _ => unreachable!("local loop work enqueue preserves item kind"), + }), LoopCommand::RequestStop => self .try_enqueue_local_ready(ReadyItem::Stop) .or_else(|item| self.try_enqueue_active_ready(item)) diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 8faef5a..d4fe167 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -8,6 +8,8 @@ mod timer_entry; pub(crate) use callbacks::{CallbackArgs, CallbackKind}; pub use callbacks::{PyHandle, PyTimerHandle, ReadyCallback}; +#[cfg(unix)] +pub use commands::LoopWork; pub use commands::{ LoopCommand, LoopFutureCommand, LoopIoCommand, LoopRunCommand, LoopSignalCommand, LoopTransportCommand, diff --git a/tests/test_native_addrinfo.py b/tests/test_native_addrinfo.py new file mode 100644 index 0000000..b23f6ac --- /dev/null +++ b/tests/test_native_addrinfo.py @@ -0,0 +1,372 @@ +"""Native name resolution: same rows and errors as socket.getaddrinfo, without the GIL.""" + +import _socket +import asyncio +import os +import socket +import subprocess +import sys +import textwrap +from typing import Any, cast + +import pytest +import rsloop + +pytestmark = pytest.mark.skipif( + sys.platform == "win32", reason="the native resolver is Unix-only" +) + +STREAM = {"type": socket.SOCK_STREAM} + + +def _resolve(host, port, **kwargs): + async def main(): + return await asyncio.get_running_loop().getaddrinfo(host, port, **kwargs) + + return rsloop.run(main()) + + +def _outcome(call): + try: + return "ok", call() + except Exception as error: # noqa: BLE001 - compared below + return type(error), error.args + + +@pytest.mark.parametrize( + "host,port,kwargs", + [ + ("localhost", 80, STREAM), + ("localhost", "80", STREAM), + ("localhost", b"80", STREAM), + ("localhost", "http", STREAM), + (b"localhost", 80, STREAM), + ("localhost.", 80, STREAM), + ("localhost", None, {}), + ("localhost", 80, {}), + ("localhost", 0, {"type": socket.SOCK_DGRAM}), + ("localhost", 65535, {"family": socket.AF_INET, **STREAM}), + ("localhost", 80, {"flags": socket.AI_CANONNAME, **STREAM}), + (None, 80, {"flags": socket.AI_PASSIVE}), + (None, 80, STREAM), + ("::1%lo", 80, STREAM), + ("localhost", 80, {"proto": socket.IPPROTO_UDP, **STREAM}), + # Errors: gaierror codes and messages as socket.getaddrinfo raises them. + ("localhost", "no-such-service-rsloop", STREAM), + ("localhost", 80, {"flags": socket.AI_NUMERICHOST, **STREAM}), + (None, None, {}), + # Shapes the native path leaves to CPython's own conversion. + ("a..b", 80, STREAM), + ("x" * 64, 80, STREAM), + ("localhost", -1, STREAM), + ("localhost", 2**70, STREAM), + ("localhost", 1.5, STREAM), + ("localhost", True, STREAM), + ("local\x00host", 80, STREAM), + ], +) +def test_results_and_errors_match_socket_getaddrinfo(host, port, kwargs): + if host == "::1%lo" and not socket.has_ipv6: + pytest.skip("IPv6 is unavailable") + expected = _outcome(lambda: socket.getaddrinfo(host, port, **kwargs)) + actual = _outcome(lambda: _resolve(host, port, **kwargs)) + assert actual == expected + if expected[0] == "ok": + # The enum members, not only equal ints. + assert [tuple(map(type, row)) for row in actual[1]] == [ + tuple(map(type, row)) for row in expected[1] + ] + + +def test_lookups_run_concurrently_and_keep_their_own_results(): + async def main(): + loop = asyncio.get_running_loop() + ports = list(range(1000, 1064)) + results = await asyncio.gather( + *(loop.getaddrinfo("localhost", port, **STREAM) for port in ports) + ) + for port, rows in zip(ports, results, strict=True): + assert rows == socket.getaddrinfo("localhost", port, **STREAM) + + rsloop.run(main()) + + +def test_a_cancelled_lookup_leaves_the_loop_running(): + async def main(): + loop = asyncio.get_running_loop() + # rsloop returns a Future (the asyncio stubs say coroutine). + getaddrinfo = cast(Any, loop.getaddrinfo) + lookups = [getaddrinfo("localhost", 80, **STREAM) for _ in range(32)] + for lookup in lookups: + lookup.cancel() + # Completions of the cancelled lookups arrive while this one waits. + assert await loop.getaddrinfo("localhost", 81, **STREAM) + await asyncio.sleep(0.05) + return [lookup.cancelled() for lookup in lookups] + + assert rsloop.run(main()) == [True] * 32 + + +def test_a_lookup_started_before_run_completes_in_the_next_run(): + loop = rsloop.new_event_loop() + try: + lookup = loop.getaddrinfo("localhost", 80, **STREAM) + rows = loop.run_until_complete(lookup) + assert rows == socket.getaddrinfo("localhost", 80, **STREAM) + finally: + loop.close() + + +def _run_script(source): + result = subprocess.run( + [sys.executable, "-c", textwrap.dedent(source)], + capture_output=True, + text=True, + timeout=60, + check=False, + ) + assert result.returncode == 0, result.stdout + result.stderr + return result.stdout + + +def test_audit_event_is_raised_on_the_loop_thread_and_can_refuse(): + # Audit hooks cannot be removed, so this runs in its own interpreter. + output = _run_script( + """ + import asyncio, socket, sys, threading + import rsloop + + events = [] + refuse = [] + + def hook(event, args): + if event == "socket.getaddrinfo": + events.append((args, threading.get_ident())) + if refuse: + raise PermissionError("refused by audit hook") + + sys.addaudithook(hook) + + async def main(): + loop = asyncio.get_running_loop() + await loop.getaddrinfo("localhost", 80, type=socket.SOCK_STREAM) + refuse.append(True) + try: + await loop.getaddrinfo("localhost", "http", type=socket.SOCK_STREAM) + except PermissionError as error: + print("refused:", error) + + loop_thread = threading.get_ident() + rsloop.run(main()) + print(events[0][0]) + # Native: raised by the loop thread, not by an executor worker. + print("loop thread:", events[0][1] == loop_thread) + print(events[1][0]) + """ + ) + assert output.splitlines() == [ + "refused: refused by audit hook", + "('localhost', 80, 0, 1, 0)", + "loop thread: True", + "('localhost', 'http', 0, 1, 0)", + ] + + +def test_native_resolver_uses_c_audit_even_if_python_sys_audit_is_replaced(): + # CPython raises the event through PySys_Audit, so a replaced sys.audit + # cannot skip a registered hook. + output = _run_script( + """ + import asyncio, socket, sys + import rsloop + + def hook(event, args): + if event == "socket.getaddrinfo": + raise PermissionError("audit refusal") + + sys.addaudithook(hook) + sys.audit = lambda *args: None + + async def main(): + loop = asyncio.get_running_loop() + return await loop.getaddrinfo("localhost", 80, type=socket.SOCK_STREAM) + + try: + rsloop.run(main()) + except PermissionError as error: + print("refused:", error) + """ + ) + assert output.splitlines() == ["refused: audit refusal"] + + +def test_native_resolver_respects_low_level_socket_override(monkeypatch): + # socket.getaddrinfo delegates to _socket.getaddrinfo; a replaced + # delegate receives the lookup, as on asyncio. + expected = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.25", 80))] + calls = [] + + def lookup(*args): + calls.append(args) + return expected + + monkeypatch.setattr(_socket, "getaddrinfo", lookup) + assert _resolve("localhost", 80, **STREAM) == expected + assert len(calls) == 1 + + +def test_a_low_level_override_receives_every_lookup(): + # Installed before this interpreter's first lookup, removed and installed + # again: native lookups (audited on the loop thread) only while the + # builtin delegate is in place. + output = _run_script( + """ + import _socket, asyncio, socket, sys, threading + import rsloop + + builtin = _socket.getaddrinfo + calls = [] + native = {} + + def override(host, port, *args): + calls.append(port) + return builtin(host, port, *args) + + def hook(event, args): + if event == "socket.getaddrinfo" and args[1] not in native: + native[args[1]] = threading.get_ident() == loop_thread + + sys.addaudithook(hook) + loop_thread = threading.get_ident() + + def lookup(port): + async def main(): + loop = asyncio.get_running_loop() + await loop.getaddrinfo("localhost", port, type=socket.SOCK_STREAM) + + rsloop.run(main()) + + _socket.getaddrinfo = override + lookup(1) + _socket.getaddrinfo = builtin + lookup(2) + _socket.getaddrinfo = override + lookup(3) + print(calls) + print(native) + """ + ) + assert output.splitlines() == [ + "[1, 3]", + "{1: False, 2: True, 3: False}", + ] + + +def test_a_functools_wraps_wrapper_receives_every_lookup(): + # A wrapper installed before this interpreter's first lookup copies the + # stdlib function's name and module; it must still see every call. The + # audit event shows which path ran: native raises it on the loop thread. + output = _run_script( + """ + import asyncio, functools, socket, sys, threading + import rsloop + + original = socket.getaddrinfo + calls = [] + native = {} + + @functools.wraps(original) + def wrapper(*args, **kwargs): + calls.append(args[1]) + return original(*args, **kwargs) + + def hook(event, args): + if event == "socket.getaddrinfo": + native[args[1]] = threading.get_ident() == loop_thread + + sys.addaudithook(hook) + loop_thread = threading.get_ident() + + def lookup(port): + async def main(): + loop = asyncio.get_running_loop() + await loop.getaddrinfo("localhost", port, type=socket.SOCK_STREAM) + + rsloop.run(main()) + + socket.getaddrinfo = wrapper + lookup(1) + lookup(2) + socket.getaddrinfo = original + lookup(3) + socket.getaddrinfo = wrapper + lookup(4) + print(calls) + print(native) + """ + ) + assert output.splitlines() == [ + "[1, 2, 4]", + "{1: False, 2: False, 3: True, 4: False}", + ] + + +@pytest.mark.skipif( + not sys.platform.startswith("linux"), reason="reads thread names in /proc" +) +def test_a_lookup_burst_starts_a_bounded_number_of_threads(): + # At most min(32, CPUs + 4) lookups run at once, as on asyncio's default + # executor (blocking::tests checks that exact bound). This counts smol's + # blocking-pool threads ("blocking-N"): the pool may start a thread while + # one that just gave up its slot is still on its way back to idle, so + # allow up to twice the cap. Without the cap a burst grows the pool to its + # 500-thread limit. + output = _run_script( + """ + import asyncio, os, socket, threading, time + import rsloop + + def workers(): + count = 0 + for task in os.listdir("/proc/self/task"): + try: + with open(f"/proc/self/task/{task}/comm") as comm: + count += comm.read().startswith("blocking-") + except OSError: + pass + return count + + async def main(): + loop = asyncio.get_running_loop() + peak = [workers()] + stop = threading.Event() + + def sample(): + while not stop.is_set(): + peak[0] = max(peak[0], workers()) + time.sleep(0.0005) + + sampler = threading.Thread(target=sample) + sampler.start() + ports = [1000 + index for index in range(1000)] + try: + results = await asyncio.gather( + *(loop.getaddrinfo("localhost", port, type=socket.SOCK_STREAM) + for port in ports) + ) + finally: + stop.set() + sampler.join() + for port, rows in zip(ports, results, strict=True): + assert rows == socket.getaddrinfo("localhost", port, type=socket.SOCK_STREAM) + print(peak[0]) + + rsloop.run(main()) + """ + ) + cpus = ( + os.process_cpu_count() + if hasattr(os, "process_cpu_count") + else len(os.sched_getaffinity(0)) + ) + assert 1 <= int(output) <= 2 * min(32, (cpus or 1) + 4), output From d5a9ef9a2b6619050494a802c373cf66695a8242 Mon Sep 17 00:00:00 2001 From: CBoYXD <135316445+CBoYXD@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:42:51 +0300 Subject: [PATCH 6/7] perf: add the benchmark tools and the analysis behind this series perf/ANALYSIS.md records how the gap behind Granian was found (per-thread profiles, perf and strace, a standalone SQLAlchemy pool repro, a pool trace inside the service), what each commit of this series changes and its tests, and microbenchmark and pool-repro measurements against v0.1.65 and uvloop, the full-service comparison against v0.1.65 (not a quiet run, so only the "ahead" cells are differences) and the quiet-gated run on the v0.1.64 base. perf/bench/ holds the tools those measurements used: - harness.py: the full-service harness (quiet-gated, interleaved and rotated arms, CPU and new Postgres sessions per 1k requests); it needs an application stack, described by RSPERF_* variables, which also name the endpoints and test-account login of the API under test. - quiet.py: CPU used outside the benchmark's containers over a window. - compare.py: per-row medians, round ranges and deltas for any arms; a measurement with non-2xx responses or transport errors is counted as failed and left out; oha's end-of-run "aborted due to deadline" entries are not failures (tests/tooling/test_perf_compare.py). - micro.py, pool_churn.py: standalone microbenchmarks and the pool repro. - sitecustomize.py: an opt-in SQLAlchemy pool trace for the service workers. --- perf/ANALYSIS.md | 534 +++++++++++++++++++++ perf/bench/compare.py | 163 +++++++ perf/bench/harness.py | 742 +++++++++++++++++++++++++++++ perf/bench/micro.py | 396 +++++++++++++++ perf/bench/pool_churn.py | 164 +++++++ perf/bench/quiet.py | 146 ++++++ perf/bench/sitecustomize.py | 70 +++ tests/tooling/test_perf_compare.py | 125 +++++ 8 files changed, 2340 insertions(+) create mode 100644 perf/ANALYSIS.md create mode 100755 perf/bench/compare.py create mode 100755 perf/bench/harness.py create mode 100644 perf/bench/micro.py create mode 100644 perf/bench/pool_churn.py create mode 100755 perf/bench/quiet.py create mode 100644 perf/bench/sitecustomize.py create mode 100644 tests/tooling/test_perf_compare.py diff --git a/perf/ANALYSIS.md b/perf/ANALYSIS.md new file mode 100644 index 0000000..2602a08 --- /dev/null +++ b/perf/ANALYSIS.md @@ -0,0 +1,534 @@ +# rsloop behind Granian: cross-thread wakes, turn shape and name resolution + +This branch is based on upstream v0.1.65 (`37a3d3f`). It changes four +things in the loop and fixes one lost-wake race: + +| commit | change | +|---|---| +| `fix: fence the block_on park against a wake that races it` | a SeqCst fence between the park's store of `waiting` and its load of the ready flag, and on both notifier sides | +| `perf: enqueue off-thread call_soon_threadsafe directly into the active run` | one wake per off-thread handle instead of a hop through the coordination thread | +| `perf: park instead of spinning for cross-thread wakes by default` | `RSLOOP_WAKE_SPIN_US` defaults to 0 (was 50) | +| `perf: run one callback generation per turn and poll I/O without the GIL` | asyncio's `_run_once` turn shape, with `call_soon` / `call_soon_threadsafe` kept in scheduling order | +| `perf: resolve host names natively, without the GIL` | `loop.getaddrinfo()` calls `getaddrinfo(3)` on the blocking pool without the GIL, at most `min(32, CPUs + 4)` at once | + +The investigation started on v0.1.64, where a Litestar web service behind +Granian, with an async SQLAlchemy/asyncpg pool, ran up to 44% slower on rsloop than on uvloop. Sections 2–4 keep +that investigation because it explains each change; every code path it +names is still present in v0.1.65. +[Section 1](#1-results-against-v0165-and-uvloop) compares this branch with +v0.1.65 and uvloop. + +## Summary + +- **v0.1.65 still has the defect.** In the pool repro (a small SQLAlchemy pool, + 8 Granian-style driver threads posting through `call_soon_threadsafe`) + v0.1.65 opens 60–74 new Postgres connections per 1k requests and runs at + 439 req/s, against uvloop's 0.0–0.1 and 1,747 req/s. This branch opens + 0.8–1.8 per 1k and runs at 1,427 req/s, at 652 µs of CPU per request vs + v0.1.65's 2,152 and uvloop's 552. +- **Root cause:** the loop thread spun for up to 50 µs, with the GIL + released, waiting for a cross-thread wake. Behind Granian those wakes come + from Python threads that still hold the GIL, so the spinning loop thread + queued for the GIL behind them, and request starts and completions reached + the loop in bursts. A small SQLAlchemy pool then kept discarding returned + connections into a full idle queue and opening new ones (a SCRAM login + each). +- **Second cause:** each turn ran to quiescence before polling I/O, so under + saturation database responses completed in bursts larger than the pool's + idle queue. +- **Third cost:** every `getaddrinfo` took the GIL twice on a blocking + worker; a burst of 1,000 lookups grows v0.1.65's pool to its 500-thread + limit. +- **Microbenchmarks against v0.1.65** (all three arms interleaved, under + load): cross-thread ping-pong 15.3 → 9.9 µs CPU per op (uvloop 10.6), + Granian-shaped requests 62.4 → 51.8 µs (uvloop 64.1), single-connection + TCP 17.7 → 10.8 µs (uvloop 11.7), `getaddrinfo` 99.5 → 33.7 µs (uvloop + 47.8). Costs: a self-rescheduling `call_soon` chain 0.20 → 0.51 µs and a + cancelled timer 1.52 → 2.21 µs, both still below uvloop (0.79, 2.61). + Upstream's `benches/compare_event_loops.py` (callbacks, tasks, TCP) is + level. +- **A lost-wake race** in the runtime park, present in v0.1.64 and v0.1.65, + is fixed; its new test stalls on v0.1.65 in 2 of 3 runs. +- **Full-service comparison against v0.1.65** (uvloop, v0.1.65, this branch; + five endpoints, c16 and c64, 3 rotated rounds): at c16 the branch is ahead + of v0.1.65 on four of five rows (+17.5% to +64.9%) with 3–8 new Postgres + sessions per 1k requests against v0.1.65's 27–53, and within noise of + uvloop on all five. The machine was not quiet, so only the "ahead" cells + are differences; see + [the run](#full-service-run-against-v0165). + +## Setup + +- **The service.** A Litestar web service behind Granian (2 workers), + backed by Postgres (SQLAlchemy async + asyncpg), Redis and NATS. Its + endpoints are authenticated JSON reads, named here by workload (two session + reads, an aggregate read, a count read, a list read, and a health probe + that opens a NATS connection); the routes themselves stay with the + application. +- **The measured stack.** Litestar 2.23.0, Granian 2.7.5, SQLAlchemy 2.0.50 + (async engine, `QueuePool`) over asyncpg 0.31.0, redis-py 5.3.1 and + nats-py 2.15.0; uvloop 0.22.1 is the reference loop and oha the load + generator. What each showed: + - Granian hands every request to the loop with `call_soon_threadsafe` from + its own threads, which still hold the GIL when the wake flag flips: the + wake pattern behind the root cause. + - SQLAlchemy's `QueuePool` turned the resulting bursts into connection + churn, and every new asyncpg connection is a SCRAM login + (`_generate_salted_password`, `libcrypto`): that is where the extra CPU + went. + - nats-py and asyncpg connect by host name, so each connect goes through + `loop.getaddrinfo` (the health probe opens a NATS connection per + request). + - No effect of Litestar's or redis-py's own was isolated; every arm ran + the same application code. +- **The machine.** One desktop machine with an i7-13650HX (6 hyper-threaded + P-cores, CPUs 0–11, and 8 E-cores, CPUs 12–19) and 31 GiB, with other work + running on it. +- **Harness** (`perf/bench/harness.py`). + - Granian 2.7.5 with 2 workers, CPython 3.14.5, the API container capped + at `--cpus 2`, oha in its own `--cpus 2` container on the same Docker + network. + - 15 s warm-up, then 60 s at c16 and at c64 per endpoint. + - Arms are interleaved and rotated per round; all arms use the same image + and application code, and only the event loop differs. A rsloop build is + selected by putting its unpacked wheel first on `PYTHONPATH`; every + start row records which modules the arm imported and the sha256 of the + native module. + - In Docker the loop runs on **epoll** (the default seccomp profile + refuses `io_uring_setup`). + - The harness also reports CPU per 1k requests (the API container's + cgroup) and new Postgres sessions per 1k requests + (`pg_stat_database.sessions`), the counter that exposed the root cause. +- **Quiet gate** (`perf/bench/quiet.py`). Before every row the harness + waits until CPU used outside the benchmark's containers stays at or below + 2.0 cores over a 30 s probe, and measures that foreign CPU over the row + itself; a row above 2.5 cores is discarded and measured again. +- **Builds.** Release wheels (`maturin build --release --locked`) of + `37a3d3f` and of this branch, built from `git archive` in a + `python:3.14.5-slim` builder image with the repository's toolchain. + +## 1. Results against v0.1.65 and uvloop + +### Pool repro (`perf/bench/pool_churn.py`) + +A small-pool database shape on one loop: SQLAlchemy async engine with +`pool_size=4, max_overflow=2` against Postgres, 8 driver threads each +posting a request through `loop.call_soon_threadsafe` and blocking on its +reply, 2 queries per request, 10 s per run. Docker epoll, `--cpus 1`, 3 +runs per arm. Measured with other builds running on the machine (load1 +13–25); the connect count is a mechanism count and does not depend on that +load. + +| arm | req/s median (min–max) | CPU µs/req | new connections /1k | post → task start µs | reply → driver resume µs | +|---|---|---|---|---|---| +| uvloop 0.22.1 | 1,747 (1,627–2,260) | 552 | 0.0 (0.0–0.1) | 254 | 286 | +| rsloop 0.1.65 | 439 (272–479) | 2,152 | **65.9** (59.9–74.2) | 2,359 | 1,769 | +| this branch | 1,427 (1,412–2,135) | 652 | 1.4 (0.8–1.8) | 395 | 309 | + +### Microbenchmarks (`perf/bench/micro.py`) + +Docker epoll, `--cpus 2`, 3 rounds with the arm order rotated, medians; CPU +µs per op, wall µs in parentheses. Measured under load (load1 10.8–20.6), so +read differences of a few percent as noise. + +| bench | uvloop 0.22.1 | rsloop 0.1.65 | this branch | +|---|---|---|---| +| `xthread_pingpong` (thread posts `call_soon_threadsafe`, waits for the callback) | 10.55 (9.25) | 15.32 (9.12) | **9.89** (7.95) | +| `xthread_task_pingpong` (thread starts a task, the task answers through a queue) | 11.26 (11.27) | 13.09 (8.27) | **9.27** (9.01) | +| `granian_like` (8 requests in flight: thread post → task with 2 DB round trips → thread-safe reply) | 64.05 (87.04) | 62.39 (84.48) | **51.82** (81.09) | +| `soon_chain` (self-rescheduling `call_soon`) | 0.79 | **0.20** | 0.51 | +| `timers_cancel` (`asyncio.timeout` around a resolved await) | 2.61 | **1.52** | 2.21 | +| `tcp_pingpong` (one connection) | 11.70 (17.17) | 17.68 (22.70) | **10.84** (16.21) | +| `tcp_pingpong_c8` (8 connections) | 8.48 (10.38) | **7.53** (9.31) | 8.00 (9.69) | +| `resolve_name` (`getaddrinfo("localhost")`, one at a time) | 47.76 (47.66) | 99.47 (79.89) | **33.67** (33.03) | +| `resolve_name_c16` (16 at once) | **36.53** (14.74) | 108.60 (51.50) | 39.73 (16.34) | +| `executor_noop` (`run_in_executor(None, int)`) | **22.84** (20.31) | 37.51 (20.00) | 23.34 (22.04) | +| `connect_by_name` (`open_connection("localhost", ...)`, one round trip, close) | **269.9** (373.9) | 385.7 (482.0) | 284.6 (397.6) | + +- `soon_chain` and `timers_cancel` are the cost of the turn shape: a + callback that reschedules itself now waits for the next turn and its + non-blocking reactor poll, as on asyncio and uvloop. +- `executor_noop` gains although `run_in_executor` itself is unchanged: on + v0.1.65 the parked-loop wake path costs more CPU (the spin). + +Upstream's own `benches/compare_event_loops.py` (pinned to one CPU, 5 +measured runs after 1 warm-up, 1,000,000 chained callbacks, 200,000 tasks in +batches of 5,000, 20,000 TCP round trips of 1,024 bytes; median seconds, two +passes per build, interleaved): + +| workload | rsloop 0.1.65 | this branch | +|---|---|---| +| callbacks | 0.171 / 0.168 | 0.177 / 0.163 | +| tasks | 0.267 / 0.289 | 0.275 / 0.262 | +| tcp_streams | 0.202 / 0.206 | 0.208 / 0.209 | + +All three are level within their pass-to-pass spread. + +### Name-resolution bursts + +1,000 concurrent `loop.getaddrinfo("localhost", port)` on the host (20 +CPUs, cap 24), peak threads of the blocking pool, 2 runs each: v0.1.65 500 +threads (smol's limit) in 114–123 ms; this branch 24 threads in 29–30 ms +(25 in 1 of 20 further runs: the pool can start a thread while one that +just released its slot is not yet idle). Pinned to 2 CPUs (cap 6), the +branch peaks at 6. + +### Full-service run against v0.1.65 + +Three arms (uvloop, v0.1.65 `37a3d3f`, this branch), five endpoints, c16 and +c64, 15 s warm-up and 60 s per row, 3 rounds with the arm order rotated, the +API container pinned to two physical P-cores and oha and Postgres/Redis/NATS to +others. Median rps with the min–max over the 3 rounds; "ahead" means the +branch's range lies entirely above the other arm's, "overlap" that the two +ranges overlap: + +| endpoint | c | uvloop | v0.1.65 | branch | branch vs v0.1.65 | branch vs uvloop | new PG sessions /1k uvloop / v0.1.65 / branch | +|---|---|---|---|---|---|---|---| +| session read 1 | 16 | 1,499 (1,420–1,703) | 1,200 (997–1,374) | 1,424 (1,316–1,650) | +18.7% (overlap) | −5.0% (overlap) | 3.4 / 27.7 / 3.3 | +| session read 1 | 64 | 1,443 (1,194–1,476) | 1,101 (1,033–1,120) | 1,628 (1,469–1,654) | +47.8% (ahead) | +12.8% (overlap) | 3.0 / 27.9 / 4.8 | +| aggregate read | 16 | 949 (821–1,019) | 792 (789–822) | 930 (917–968) | +17.5% (ahead) | −1.9% (overlap) | 5.4 / 28.0 / 7.5 | +| aggregate read | 64 | 814 (786–971) | 913 (796–923) | 939 (909–966) | +2.8% (overlap) | +15.4% (overlap) | 4.6 / 11.0 / 5.7 | +| session read 2 | 16 | 1,650 (1,595–1,723) | 1,078 (1,025–1,083) | 1,779 (1,700–1,970) | +64.9% (ahead) | +7.8% (overlap) | 5.1 / 53.4 / 2.8 | +| session read 2 | 64 | 1,765 (1,717–1,823) | 1,818 (1,787–1,952) | 1,925 (1,726–1,927) | +5.9% (overlap) | +9.1% (overlap) | 2.5 / 2.3 / 2.3 | +| count read | 16 | 1,335 (1,319–1,605) | 997 (977–1,044) | 1,401 (1,400–1,507) | +40.4% (ahead) | +5.0% (overlap) | 10.7 / 33.4 / 3.7 | +| count read | 64 | 1,443 (1,224–1,598) | 1,503 (1,464–1,525) | 1,450 (1,291–1,509) | −3.5% (overlap) | +0.5% (overlap) | 3.0 / 2.4 / 2.5 | +| list read | 16 | 1,263 (1,162–1,280) | 922 (915–1,058) | 1,194 (1,091–1,228) | +29.5% (ahead) | −5.5% (overlap) | 8.7 / 26.9 / 4.0 | +| list read | 64 | 1,287 (1,152–1,304) | 1,210 (1,079–1,246) | 1,175 (1,137–1,257) | −2.9% (overlap) | −8.7% (overlap) | 3.4 / 3.6 / 3.8 | + +(Session reads 1 and 2 are light GETs on different routes, aggregate read a +GET with a DB aggregate, count read a GET with a small DB count, list read a +GET with a small DB list.) + +This run was not quiet: other builds were running on the 20-CPU host (load1 +3.1–9.7, 1.5–5.9 cores of CPU outside the benchmark), and 77 of the 90 rows +were above the 2.5-core quiet threshold, so only the "ahead" cells are +differences. At c16 the branch uses 14–39% less CPU per request than +v0.1.65; at c64 four of the five rows are within noise of v0.1.65. + +### Full-service run on the v0.1.64 base + +The last quiet-gated run of all five endpoints measured the same +changes on the v0.1.64 base (before later fixes to name resolution +and before v0.1.65's rework of `blocking::run`, which the branch now uses): uvloop, +rsloop 0.1.64 and the fork, 3 rotated rounds, 60 s per row, every row with +≤ 2.0 foreign cores (192 clean rows, none discarded). Median rps: + +| endpoint | c | uvloop | rsloop 0.1.64 | fork | fork vs uvloop | new PG sessions /1k uvloop / 0.1.64 / fork | +|---|---|---|---|---|---|---| +| session read 1 | 16 | 1,787 | 1,636 | 1,799 | +0.7% | 2.9 / 17.9 / 2.9 | +| session read 1 | 64 | 1,668 | 1,525 | 1,827 | +9.5% | 2.7 / 17.0 / 5.1 | +| aggregate read | 16 | 1,069 | 940 | 1,070 | +0.1% | 6.5 / 33.5 / 8.1 | +| aggregate read | 64 | 1,029 | 944 | 1,052 | +2.3% | 4.3 / 27.5 / 4.2 | +| session read 2 | 16 | 2,208 | 1,172 | 2,222 | +0.6% | 2.9 / 55.7 / 1.4 | +| session read 2 | 64 | 2,190 | 2,348 | 2,172 | −0.8% | 1.4 / 2.7 / 2.0 | +| count read | 16 | 1,771 | 1,138 | 1,689 | −4.6% (ranges overlap) | 3.2 / 46.3 / 2.8 | +| count read | 64 | 1,745 | 1,821 | 1,744 | −0.1% | 2.2 / 4.9 / 2.5 | +| list read | 16 | 1,349 | 1,057 | 1,445 | +7.2% | 9.2 / 40.1 / 4.0 | +| list read | 64 | 1,405 | 1,510 | 1,416 | +0.8% | 2.8 / 3.2 / 3.1 | + +On the v0.1.64 base the c16 rows rsloop lost by 22–47% were level with +uvloop, at uvloop's CPU and connection churn per request; the session read 1 +and aggregate read c64 rows were ahead. v0.1.64 beat both on the session +read 2, count read and list read c64 rows, where its spin pays off +in a saturated container (see [Remaining differences](#6-remaining-differences)). +To repeat the run against v0.1.65: +`harness.py api 1` with +`BENCH_ARMS=uvloop,fork:,fork:`, three rounds +(`BENCH_ROUND_BASE=0..2`), then `compare.py`. + +## 2. How the gap was found (v0.1.64) + +Baseline on v0.1.64 (3 rounds, median rps; load1 2.2–6.2): + +| endpoint | c | uvloop | rsloop 0.1.64 | Δ | CPU-ms/1k uvloop / rsloop | +|---|---|---|---|---|---| +| session read 2 | 16 | 2,057 | 1,150 | −44.1% | 972 / 1,740 | +| session read 2 | 64 | 1,965 | 2,150 | +9.4% | 1,018 / 931 | +| count read | 16 | 1,543 | 1,093 | −29.1% | 1,295 / 1,830 | +| count read | 64 | 1,568 | 1,626 | +3.7% | 1,276 / 1,231 | +| list read | 16 | 1,281 | 1,022 | −20.2% | 1,559 / 1,957 | +| list read | 64 | 1,289 | 1,369 | +6.2% | 1,551 / 1,462 | +| session read 1 | 16 | 1,689 | 1,888 | +11.7% | 1,182 / 1,051 | +| aggregate read | 16 | 1,003 | 881 | −12.2% | 1,984 / 2,256 | + +At c16, rps × CPU-ms per request ≈ 2,000 ms/s for both loops: the container +is pinned at its 2-CPU quota, so the throughput gap is a CPU-per-request +gap (about 535 µs per request on the count read). + +**Where the CPU went** (count read c16; per-thread +CPU from `/proc`, `perf record -g`, `strace -c`): + +- The extra CPU was on the Python loop threads themselves. The coordination + thread, rsloop's helper threads and Granian's tokio threads together + accounted for less than 10 µs per request of the difference. rsloop + context-switched less than uvloop (1.9 vs 4.1 per request): the busy-wait + replaced sleeping. +- `libcrypto` held 8.0% of rsloop's cycles and 0.02% of uvloop's, and + asyncpg's protocol module 6.4% vs 1.9%; the call graphs put SCRAM's + `_generate_salted_password` at 8.6–9.6% of rsloop's samples. Postgres + authenticates only when a connection is opened. +- The new-session counter confirmed it: 33–40 new Postgres sessions per 1k + requests on rsloop, 4 on uvloop. `strace` showed the footprint of the + reconnects (1,378 `close`, 1,652 `epoll_ctl`, 275 new threads in 10 s, + against 6, 12 and 0 on uvloop). + +**The mechanism, isolated** (`perf/bench/pool_churn.py`): the same small-pool +database shape on one loop without Granian or the application. A SQLAlchemy async +engine with `pool_size=4, max_overflow=2` against Postgres, 8 driver threads +each posting a request through `loop.call_soon_threadsafe` (as Granian +does) and blocking on its reply, 2 queries per request. Docker epoll, +`--cpus 1`, medians of 3: + +| arm | req/s | CPU µs/req | connects /1k | post→task start µs | reply→driver resume µs | +|---|---|---|---|---|---| +| uvloop | 2,166 | 434 | 0.1 | 222 | 225 | +| rsloop 0.1.64 | 598 | 1,661 | 61.3 | 1,688 | 1,448 | +| + direct off-thread enqueue | 641 | 1,534 | 55.0 | 1,724 | 1,495 | +| + park instead of spin | **2,593** | **377** | **0.2** | **39** | **52** | + +The pool trace showed connections coming back while the idle queue already +held 3–4 with no waiter (at least 343 check-ins discarded into a full queue +in 6 s, 265 connections created into an empty one), while the remaining +requests waited 1.4–1.7 ms between their driver thread and the loop. + +Hypotheses tested on that repro: + +| hypothesis | test | verdict | +|---|---|---| +| Every off-thread `call_soon_threadsafe` takes two wakeups (channel → coordination thread → run queue → loop thread). | cross-thread ping-pong microbenchmark; per-thread CPU | Confirmed, minor: 11.4 → 7.8 µs CPU per cross-thread callback; does not touch the churn (61 → 55 /1k). | +| The 50 µs loop-thread spin before parking hurts when wakes come from Python threads that hold the GIL. | `RSLOOP_WAKE_SPIN_US=0` on the same build | Confirmed, root cause: churn 55 → 0.2 /1k, post→start 1.7 ms → 39 µs, 4× req/s. | +| The spin starves the reactor (up to 64 consecutive caught spins without a poll). | variant polling the reactor after every caught spin | Refuted: still 55–82 connects /1k. | +| The spin burns the CFS quota and the cgroup is throttled. | same repro at `--cpus 2` and `--cpus 3` | Refuted: identical churn without a binding quota. | +| GIL convoy: the spinner sees the wake while its producer still holds the GIL and queues behind the driver threads (`take_gil` waits up to the 5 ms switch interval). A parked loop wakes after the producer let go and drains everything queued. | latency split above; no-spin variant | Consistent with every measurement. | +| Timer heap, ready-queue or transport costs. | timer, `call_soon` and TCP microbenchmarks | Refuted: rsloop was faster than uvloop on timers and `call_soon`, level with 8 connections. | +| Each turn runs to quiescence before polling the reactor, so I/O completions arrive in bursts larger than a pool's idle queue. | pool trace inside the service (below) | Confirmed, second cause. | + +**The second mechanism: completions in bursts.** Once the spin was +gone, some c64 rows (session read 1 among them) still opened 14–20 new +sessions per 1k requests (uvloop 0.1–3.7). An opt-in trace of the pool inside the service +(`perf/bench/sitecustomize.py`, `RSPERF_TRACE_POOL=1`) counted, in 20 s of +session read 1 at c64, about 690 check-ins into a full idle queue with 5–20 +getters waiting, against at most 3 on uvloop. A `run_forever` turn took in +every callback it scheduled (refill every 64 items, up to 1024) and polled +the reactor only when nothing was left, so a whole chain of response → +request step → check-in → next response ran inside one turn. asyncio's +`_run_once` runs only what was ready when the iteration started and polls +I/O again before anything newer. + +**The third cost: name resolution.** `loop.getaddrinfo()` of a host name +goes through `run_in_executor`: a worker takes the GIL to call +`socket.getaddrinfo` and again after the lookup, each time behind a busy +loop thread. Every nats-py connect and every new Postgres session resolves a +name. On v0.1.64 that also started a thread per lookup (v0.1.65 reuses +`smol`'s blocking threads); it cost the health probe ~250 µs of CPU +per request. + +## 3. The changes + +### Fence the park against a racing wake + +- **Symptom.** In Docker (epoll), the #115 greenlet repro hung in about 1 of + 300–450 runs on every build, v0.1.64 included (2 of 440). +- **What was stuck.** A core dump of a hung run showed the loop thread + cycling through its bounded parks and the coordination thread in + `epoll_wait` with no timeout. The first step of a + `run_until_complete` task, queued before `run_forever`, never ran: work + queued before a run reaches it only through the coordination thread. +- **Root cause.** The waker stores the root's ready flag (or pushes onto the + remote task queue), then loads `waiting`; the park stores `waiting`, then + loads the ready flag. With release/acquire only, store-load reordering lets + both loads read the old value: the waker skips the interrupt and the park + sleeps through the wake. v0.1.65 restructured `BlockOnNotify` around + `RemoteWakeContext` but kept the same handshake. +- **Fix.** A SeqCst fence between store and load on the park side and on + both notifier sides; same-thread wakes skip it. +- **Tests.** `block_on_never_misses_a_wake_racing_with_the_park` and + `spawned_task_never_misses_a_wake_racing_with_the_park` ping-pong 100,000 + wakes against the park on the mio driver and fail when progress stops for + 5 s. On v0.1.65 without the fences the first stalls in 2 of 3 runs (at + rounds 20,765 and 79,563); with them both pass. On v0.1.64 the #115 repro + then ran 2,200 times in Docker without a hang (5–7 expected at the old + rate). + +### Off-thread `call_soon_threadsafe` joins the active run directly + +`LoopCore::schedule_ready_handle`: while a run is active and no earlier +handle is still on the command channel, an off-thread handle goes straight +into the run's pending queue and wakes the loop once. A counter of handles +in flight on the channel (released by the dispatcher when it forwards them) +keeps each producer's FIFO order across run start and end, and `finish_run` +puts the run's queue (older work) ahead of commands received since the last +dispatch. Test: +`tests/test_fast_callbacks.py::test_threadsafe_producer_order_survives_run_restarts`. + +On v0.1.64 (Docker epoll, `--cpus 2`): cross-thread ping-pong 11.4 → 7.8 µs +CPU per op (uvloop 5.9), task ping-pong 10.4 → 6.9 (uvloop 6.9). + +### Park instead of spinning by default + +`wake_spin_window` defaults to 0; `RSLOOP_WAKE_SPIN_US` stays as an opt-in, +documented in `docs/configuration.md`. A spin only pays off when the waking +threads never hold the GIL. + +On v0.1.64 (Docker epoll, `--cpus 2`, medians of 3): Granian-shaped +requests 54.6 → 41.1 µs CPU (uvloop 50.0), cross-thread ping-pong 14.1 → +6.5 µs (uvloop 7.6), single-connection TCP ping-pong 15.2 → 10.5 µs (uvloop +8.0), and the pool repro 598 → 2,593 req/s at 61.3 → 0.2 connects per 1k +(uvloop 2,166, 0.1). + +### One callback generation per turn; poll I/O without the GIL + +- `run_forever`: a turn runs the callbacks queued when it started; newer + ones wait for the next turn, after a non-blocking reactor poll. That poll + runs under `py.detach`, as uvloop polls without the GIL (with the GIL held, + back-to-back turns kept it for the whole switch interval while + `call_soon_threadsafe` producers waited, and the pool repro fell to 851 + req/s at 61 connects per 1k). +- A queued `stop()` keeps the turn going until its Stop item runs, so stop + still takes effect in the turn that requested it. +- **Scheduling order.** The loop thread queues its own `call_soon` handles in + a lock-free local queue and other threads queue into the run's pending + queue. Every off-thread item records how many items the loop thread had + queued for itself when it was queued (`LoopCore::local_enqueues`, written + only by the loop thread), and each generation is merged from both queues + on that mark (`merge_ready_generation`): an off-thread item goes behind + exactly the local items queued before it. Marks are stored once per change, + so a burst of off-thread items shares one; work queued before a run + carries the mark read before the run could queue anything. v0.1.64 already + failed the tightest case (50 alternating `call_soon` / thread-posted + `call_soon_threadsafe` calls from one callback). +- **Tests.** `tests/test_local_sockets.py::test_tcp_data_is_delivered_between_call_soon_generations` + (a self-rescheduling `call_soon` chain delayed a readable protocol by 513 + callbacks before, 1 after); `tests/test_fast_callbacks.py`: + `test_earlier_call_soon_runs_before_later_threadsafe_call`, + `test_call_soon_and_threadsafe_calls_interleave_in_scheduling_order` and + `test_threadsafe_call_runs_before_later_call_soon`, each against asyncio + and rsloop; Rust `generation_order_tests` for the merge. +- **Cost.** A self-rescheduling `call_soon` chain is now one turn per step, + as in asyncio and uvloop. See [section 1](#1-results-against-v0165-and-uvloop) + for what that does to the chained-callback benchmark. + +On v0.1.64 with the two changes above, session read 1 at c64 went from +1,618 to 1,936 rps and from 20 to 1.3–3.0 new sessions per 1k (uvloop +1,746). + +### Resolve host names natively, without the GIL + +- `src/bindings/loop_api/resolver.rs`: on Unix, an exact loop resolves + natively when it has no custom default executor, `socket.getaddrinfo` is + still the standard library's function, and host and port are plain (an + ASCII name the IDNA codec passes unchanged, or `bytes`; an `int` port in + range, `str` or `bytes`). `getaddrinfo(3)` runs on the blocking pool + without the GIL. The loop thread raises the `socket.getaddrinfo` audit + event through CPython's C API (`PySys_Audit`, as `_socket` does, so a + replaced `sys.audit` skips no hook) and builds the same rows (enum members, glibc's `inet_ntop` text) or + the same `gaierror` / `OSError` (`EAI_SYSTEM` as `PyErr_SetFromErrno` + raises it, errno 0 included). Every other shape keeps the executor path. +- A replaced or wrapped `socket.getaddrinfo` still receives every lookup. + The standard function is recognised by what a `functools.wraps` wrapper + cannot copy: its globals are the `socket` module's namespace, its code is + named `getaddrinfo`, and it has no `__wrapped__`. The `_socket.getaddrinfo` + it delegates to must be `_socket`'s builtin, so a replaced delegate also + keeps the executor path. Both are checked on every lookup. +- `src/blocking.rs`: at most `min(32, CPUs + 4)` lookups run at once, the + size of asyncio's default executor (CPUs from the affinity mask, as + `os.process_cpu_count()` counts them); further lookups wait in order and + `run_in_executor` jobs never wait for a lookup slot. Without the cap a + burst of 1,000 lookups grew the pool to hundreds of threads. +- `src/engine/commands.rs`: a `LoopWork` ready item carries the loop-thread + half of such work, so the worker never takes the GIL. +- **Tests.** `tests/test_native_addrinfo.py`: 25 argument shapes compared + with `socket.getaddrinfo` (rows, enum types and errors, including the + shapes left to CPython), 64 concurrent lookups, cancelled lookups, a lookup + created before `run_until_complete`, the audit event (raised on the loop + thread; a refusing hook fails the lookup, also with `sys.audit` replaced), + a `functools.wraps` wrapper installed before the first lookup, a replaced + `_socket.getaddrinfo` before and after native lookups, and the thread + count under a burst. + Rust: the bounded slots, glibc's IPv6 text, the IDNA fast path, the cap + formula and the `EAI_SYSTEM` mapping. + +## 4. What v0.1.65 already did, and what was dropped + +- **Blocking workers.** The fork replaced v0.1.64's thread per + `blocking::run` call with its own cached pool. v0.1.65 moved + `blocking::run` onto `smol::unblock`, which reuses threads too, so the + fork's pool was dropped; only the bounded lookup slots remain, on top of + `smol::unblock`. +- **Fork safety, investigated and left alone.** `smol`'s blocking pool keeps + its idle count in a static, so a job submitted in a child forked within + 500 ms of the parent's last blocking job waits for a thread that does not + exist (a Rust test on `blocking::run` reproduces it). In Python, though, + `run_in_executor` and `rsloop.run` already hang in a child forked after + the parent ran any rsloop loop, on v0.1.64, v0.1.65 and this branch alike, + so a fork-aware pool alone fixes nothing visible; it is not part of this + branch. +- **Wake overhead.** v0.1.65's root-waker reuse, `block_on_until` and the + turn counter are kept as they are; the changes above sit on top of them. + +## 5. Correctness + +All on this branch's head (release builds), on the host (io_uring) unless +noted: + +| check | result | +|---|---| +| `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings` | clean (also on every commit of the series) | +| `scripts/run_rust_tests.py`, default and `--all-features` | 383 and 474 passed | +| Python core suite (`scripts/run_python_tests.py tests/test_*.py`, CPython 3.14.7) | 689 passed, 12 skipped; `tests/test_run.py::TestRun::test_run_fallback_handles_sigint` passes when run in a foreground process (it fails under a runner that ignores SIGINT) | +| free-threaded CPython 3.14.7t: `tests/test_native_addrinfo.py`, `tests/test_free_threading.py` | 37 passed | +| ruff, ruff format, pyright on the changed tests | clean; pyright reports one error in `tests/test_local_sockets.py:298`, which is upstream's own test and reports the same on master | +| #115 greenlet lost wakeup / SIGSEGV repro, every depth | 0 stuck, 0 SIGSEGV, 0 nonzero exits (55 host runs, 110 Docker epoll runs) | +| #116 protocol leak repro; stdlib `StreamReaderProtocol` churn; asyncpg connect/close churn | 0 protocols alive, +0–1 B per connection (host and Docker) | +| #118 cancelled `TimerHandle` retention (`repro.py`, `asyncpg_repro.py`) | 0/20000 alive in every case; 0 statement caches, states or protocols alive | +| wake-race stress tests on v0.1.65 without the fences | `block_on` stalls in 2 of 3 runs (rounds 20,765 and 79,563); with the fences both pass | + +## 6. Remaining differences + +- **Chained callbacks.** A self-rescheduling `call_soon` chain costs one turn + per step now, as on asyncio and uvloop + ([section 1](#1-results-against-v0165-and-uvloop)). +- **Spin under saturation.** On v0.1.64 the 50 µs spin let rsloop beat + uvloop by 4–7% on the session read 2, count read and list read c64 rows (a saturated container, frequent + wakes) while it cost 22–47% at c16. `RSLOOP_WAKE_SPIN_US=50` restores it + for workloads whose wakes come from GIL-free threads. +- **Single serial TCP connection.** On a strict one-connection ping-pong the + branch costs about 0.7–1.4 µs more CPU per round trip than uvloop on + v0.1.64's base; with 8 connections it is level. The difference sits in the + loop-thread reader task and turn bookkeeping. +- **Latency under load.** With the loop busy, an I/O event reaches the task + it wakes one turn later than on uvloop, as on asyncio (a loaded asyncpg + connect took 126 ms vs uvloop's 86 ms on v0.1.64's base). Running the + reactor's events before taking each generation brought it to 94 ms but + changed nothing in the full-service runs and changes asyncio's callback order, so + it was not adopted. +- **Health probe at c64** sat 0.5–1.4% below uvloop on v0.1.64's base, + at asyncio's level, after name resolution brought its CPU per request to + uvloop's. Each request holds one pooled connection while it waits out a + 0.25 s timeout for a second one, so the rate depends on which waiter wins + released connections; timer punctuality, timer order and I/O turn depth + were each measured and ruled out. It is not among the rows in section 1. + +## Reproducing + +The tools in `perf/bench/` are the ones these measurements used. They +expect a benchmark workspace outside this repository, `$RSPERF_ROOT` (see +the `perf/bench/harness.py` docstring for its variables): the application's +images and compose file, a compose override that renames every container, +network and volume to the `rsperf` project and removes host ports, `sites/` +with unpacked wheels (one directory per build) and the outputs under +`run/bench/`. + +| tool | what | +|---|---| +| `harness.py` | the service harness: `harness.py api [rounds]`. Arms `uvloop`, `asyncio`, `rsloop` (the image's pin), `fork:` (a build under `sites/`), `@` (extra environment, e.g. `@nospin`). The workload (service, endpoints, test-account login) comes from `RSPERF__*`; quiet-gated (`QUIET_*`); `BENCH_ENDPOINTS`, `BENCH_CONCURRENCY`, `BENCH_ROUND_BASE` select rows and rotation | +| `quiet.py` | the quiet-machine probe: CPU used outside the benchmark's containers over a window | +| `compare.py` | the tables: `compare.py ... --arm