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 7b0ea34aa..642f3414d 100644 --- a/nodedb/src/bootstrap/mod.rs +++ b/nodedb/src/bootstrap/mod.rs @@ -13,6 +13,7 @@ pub mod index_registry_seed; pub mod listeners; pub mod panic_hook; pub mod permission_tree_load; +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); + } +}