From c7515e9262472e3178a5b70da2912a113b9fd6ee Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Sat, 26 Sep 2026 13:14:52 +0800 Subject: [PATCH] fix(control): pace the response poller by what a poll costs The response poller cleared its idle counter on any routed response and went straight back to the fast path. A poll of that loop is not cheap: it locks the dispatcher and, per core, drains responses and flushes the WFQ. Because the counter was cleared by work, a response rate above roughly one per second held it below the loop's first wait for the life of the process, and the loop's share of a worker was then whatever a poll costs - up to a pinned core. Bound the share of a worker instead of the length of the fast run. The pacer times each poll and derives its wait from that cost, so a cycle of polls plus the wait cannot exceed the duty budget however expensive a poll turns out to be on a given build; a cheap poll still falls back to the 1 ms wait. `idle` keeps counting polls that routed nothing and widens the wait to 10 ms at the same 1024 threshold, so a node that goes quiet settles on the timeline it had before, and a response that lands just behind a drained batch still earns one poll without a wait. A fixed wait cannot do this: at the ~4.3 ms per poll measured on a debug node a 1 ms wait leaves the loop near 90% of a worker, which is why the wait is derived rather than chosen. The duty bound is asserted across three orders of magnitude of poll cost, and that test fails on the previous pacing, which reported a 10 us poll holding 100% of a worker. --- nodedb/src/bootstrap/background_loops.rs | 20 +- nodedb/src/bootstrap/mod.rs | 1 + nodedb/src/bootstrap/poll_pacer.rs | 250 +++++++++++++++++++++++ 3 files changed, 258 insertions(+), 13 deletions(-) create mode 100644 nodedb/src/bootstrap/poll_pacer.rs diff --git a/nodedb/src/bootstrap/background_loops.rs b/nodedb/src/bootstrap/background_loops.rs index 17d95db11..2a3187cb0 100644 --- a/nodedb/src/bootstrap/background_loops.rs +++ b/nodedb/src/bootstrap/background_loops.rs @@ -7,6 +7,7 @@ use std::time::Duration; use tracing::info; +use super::poll_pacer::{PollPace, PollPacer}; use crate::ServerConfig; use crate::control::state::SharedState; use crate::event::bus::EventConsumerRx; @@ -396,7 +397,7 @@ pub fn spawn_response_poller( "response_poller", crate::control::shutdown::ShutdownPhase::DrainingDataPlane, move |_shutdown| async move { - let mut idle_iters: u32 = 0; + let mut pacer = PollPacer::new(); loop { if shared_poller.data_plane_drain.is_complete() { // One last pass so a response the drain's own final poll @@ -404,19 +405,12 @@ pub fn spawn_response_poller( shared_poller.poll_and_route_responses(); break; } + let started = std::time::Instant::now(); let routed = shared_poller.poll_and_route_responses(); - if routed > 0 { - idle_iters = 0; - tokio::task::yield_now().await; - continue; - } - idle_iters = idle_iters.saturating_add(1); - if idle_iters <= 256 { - tokio::task::yield_now().await; - } else if idle_iters <= 1024 { - tokio::time::sleep(std::time::Duration::from_millis(1)).await; - } else { - tokio::time::sleep(std::time::Duration::from_millis(10)).await; + let cost = started.elapsed(); + match pacer.step(routed, cost) { + PollPace::Fast => tokio::task::yield_now().await, + PollPace::Tick(wait) => tokio::time::sleep(wait).await, } } drain_guard.report_drained(); diff --git a/nodedb/src/bootstrap/mod.rs b/nodedb/src/bootstrap/mod.rs index 8faa98793..e740d41c7 100644 --- a/nodedb/src/bootstrap/mod.rs +++ b/nodedb/src/bootstrap/mod.rs @@ -12,6 +12,7 @@ pub mod diagnostics; pub mod index_registry_seed; pub mod listeners; pub mod panic_hook; +pub mod poll_pacer; pub mod quota_replay; pub mod schema_rehydrate; pub mod signal; diff --git a/nodedb/src/bootstrap/poll_pacer.rs b/nodedb/src/bootstrap/poll_pacer.rs new file mode 100644 index 000000000..f0207ec0d --- /dev/null +++ b/nodedb/src/bootstrap/poll_pacer.rs @@ -0,0 +1,250 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Pacing for the response poller. + +use std::time::Duration; + +/// How long the response poller waits before its next poll. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum PollPace { + /// Poll again without waiting. + Fast, + /// Wait one tick. A response that lands meanwhile waits at most this long. + Tick(Duration), +} + +/// Pacing for the response poller. +/// +/// The loop has two jobs that pull against each other: route a response as soon +/// as it lands, and avoid spending a worker asking a question the cores have not +/// answered yet. Polling at yield speed serves the first and loses the second, +/// so the fast path is bounded here rather than granted on the strength of the +/// last poll. +/// +/// Granting it on the strength of the last poll is what made the loop's own wait +/// unreachable. Any routed response cleared the idle counter, and a poll of this +/// loop is expensive - the dispatcher mutex plus a `flush_wfq` per core - so the +/// 256 empty polls needed to reach the first wait take about a second on a live +/// node. One routed response inside that window clears the counter, so any +/// response rate above roughly one per second holds `idle_iters` below the first +/// wait for the life of the process; the loop then runs at fast-path speed, +/// which is a pinned worker. +/// +/// The two counters are therefore deliberately asymmetric: `idle` counts polls +/// that routed nothing and widens the wait, while `fast` counts the fast polls +/// taken since the last wait and is cleared by the wait itself, not by work. +/// Work narrows the wait but cannot buy another fast poll, which bounds the loop +/// for every response rate rather than for the ones we predicted. +pub(super) struct PollPacer { + /// Fast polls taken since the last wait. + fast: u32, + /// Consecutive polls that routed nothing. + idle: u32, +} + +/// Share of a worker the poller may spend polling, in percent. +/// +/// The wait is derived from what a poll actually cost, so this holds whatever +/// that turns out to be: a poll locks the dispatcher and flushes the WFQ once +/// per core, which is milliseconds on a debug build and much less on a release +/// one. A fixed wait cannot bound CPU for both. +const MAX_DUTY_PERCENT: u32 = 10; + +/// Fast polls allowed back-to-back before the pacer has to wait. +/// +/// One: the poll after a drain is what catches a response that landed just +/// behind the batch that was just routed. Every further fast poll is CPU spent +/// re-reading a queue that has already said it is empty. +const FAST_POLLS: u32 = 1; + +/// Polls that routed nothing before the wait widens to [`IDLE_TICK`]. +/// +/// 1024 is the threshold the loop already used, kept so a node that goes quiet +/// settles on the same timeline it did before this change. +const QUIET_POLLS: u32 = 1024; + +/// Wait while responses are still arriving. One millisecond is the latency the +/// loop already accepted once its idle counter passed 256. +const TICK: Duration = Duration::from_millis(1); + +/// Wait once they stop: 100 polls/second, off a core. +const IDLE_TICK: Duration = Duration::from_millis(10); + +impl PollPacer { + pub(super) const fn new() -> Self { + Self { fast: 0, idle: 0 } + } + + /// Wait that holds this loop to [`MAX_DUTY_PERCENT`] of a worker given what + /// a poll just cost. + /// + /// Spending `cost` in every `cost + wait` bounds the share at + /// `cost / (cost + wait)`, so the wait scales with the cost. A cheap poll + /// falls back to [`TICK`], and an expensive one is paced out far enough that + /// it cannot pin a worker however costly the poll turns out to be. + fn wait_for(cost: Duration) -> Duration { + // A cycle is `FAST_POLLS + 1` polls followed by this wait, so the loop's + // share of a worker is `per_cycle / (per_cycle + wait)`. + let per_cycle = cost * (FAST_POLLS + 1); + let scaled = per_cycle * (100 / MAX_DUTY_PERCENT - 1); + if scaled > TICK { scaled } else { TICK } + } + + /// Record one poll outcome and its cost, and say how long to wait before the + /// next poll. + pub(super) fn step(&mut self, routed: usize, cost: Duration) -> PollPace { + if routed > 0 { + self.idle = 0; + if self.fast < FAST_POLLS { + self.fast = self.fast.saturating_add(1); + return PollPace::Fast; + } + } else { + self.idle = self.idle.saturating_add(1); + } + self.fast = 0; + let wait = Self::wait_for(cost); + if self.idle > QUIET_POLLS { + PollPace::Tick(if wait > IDLE_TICK { wait } else { IDLE_TICK }) + } else { + PollPace::Tick(wait) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Nominal cost of one poll in the simulations below. + const POLL_COST: Duration = Duration::from_micros(200); + + /// Longest run of `Fast` paces the pacer hands out for `outcomes`. + /// + /// The run, not the poll count, is the quantity that matters: a fast poll + /// hands the runtime a task that is ready again immediately, so an + /// unbounded run is the pin. + fn longest_fast_run(outcomes: impl Iterator) -> u32 { + let mut pacer = PollPacer::new(); + let (mut run, mut longest) = (0u32, 0u32); + for routed in outcomes { + match pacer.step(routed, POLL_COST) { + PollPace::Fast => { + run += 1; + longest = longest.max(run); + } + PollPace::Tick(_) => run = 0, + } + } + longest + } + + /// Share of a worker this loop spends polling when a poll costs `cost` and + /// every poll routes work - the worst case, and the one that pinned a + /// worker before: the counter was cleared by work, so the loop yielded + /// again and again and its share of the worker was whatever a poll cost. + fn duty_percent(cost: Duration) -> u128 { + let mut pacer = PollPacer::new(); + let (mut polling, mut wall) = (Duration::ZERO, Duration::ZERO); + for _ in 0..1_000 { + match pacer.step(1, cost) { + PollPace::Fast => { + polling += cost; + wall += cost; + } + PollPace::Tick(wait) => { + polling += cost; + wall += cost + wait; + } + } + } + polling.as_micros() * 100 / wall.as_micros() + } + + /// The bound this pacer exists for: for any cost a poll turns out to have, a + /// poll on every iteration must stay inside the duty budget. + /// + /// The wait is derived from the measured cost rather than fixed, so the + /// bound does not depend on how expensive a poll is on a given build - which + /// is the mistake a fixed 1 ms wait makes, and why this is asserted across + /// three orders of magnitude. + #[test] + fn a_poll_on_every_iteration_stays_within_the_duty_budget() { + for micros in [10u64, 100, 1_000, 4_300, 10_000] { + let duty = duty_percent(Duration::from_micros(micros)); + assert!( + duty <= MAX_DUTY_PERCENT as u128, + "a {micros} us poll held {duty}% of a worker" + ); + } + } + + /// The old shape at the same rate: work cleared the counter and the loop + /// took the fast path again, so there was no wait to bound anything. + #[test] + fn work_on_every_poll_cannot_hold_the_fast_path() { + let longest = longest_fast_run(std::iter::repeat_n(1, 10_000)); + assert!( + longest <= FAST_POLLS, + "fast path ran {longest} polls without a wait" + ); + } + + /// The same defect at the rate a live node actually reaches: one response + /// per few polls keeps clearing the counter. + #[test] + fn a_trickle_of_responses_cannot_hold_the_fast_path() { + let longest = longest_fast_run((0..10_000).map(|i| usize::from(i % 3 == 0))); + assert!( + longest <= FAST_POLLS, + "trickle held the fast path for {longest} polls" + ); + } + + /// An idle node still has to reach the 10 ms wait, or a pinned worker is + /// traded for a permanently hot one. + #[test] + fn an_idle_poller_still_widens_to_the_idle_tick() { + let mut pacer = PollPacer::new(); + let widened = (0..10_000).any(|_| pacer.step(0, POLL_COST) == PollPace::Tick(IDLE_TICK)); + assert!(widened, "idle poller never reached the idle tick"); + } + + /// Work must still narrow the wait, or a single response after a quiet + /// spell would leave the loop on the wide idle wait while a burst follows. + #[test] + fn work_narrows_the_wait_again() { + let mut pacer = PollPacer::new(); + let mut idle_wait = Duration::ZERO; + for _ in 0..10_000 { + if let PollPace::Tick(wait) = pacer.step(0, POLL_COST) { + idle_wait = wait; + } + } + assert!( + idle_wait >= IDLE_TICK, + "an idle poller must reach the idle wait" + ); + assert_eq!( + pacer.step(1, POLL_COST), + PollPace::Fast, + "the response just routed earns one fast poll" + ); + let after_work = match pacer.step(1, POLL_COST) { + PollPace::Tick(wait) => wait, + PollPace::Fast => panic!("work must not hold the fast path open"), + }; + assert!( + after_work < idle_wait, + "work narrowed the wait to {after_work:?}, which is not below {idle_wait:?}" + ); + } + + /// The latency the fast path exists for: a response that landed just behind + /// the batch just routed is taken without a wait. + #[test] + fn the_poll_after_a_batch_is_still_fast() { + let mut pacer = PollPacer::new(); + assert_eq!(pacer.step(1, POLL_COST), PollPace::Fast); + } +}