diff --git a/apps/neshub/src/congestion.rs b/apps/neshub/src/congestion.rs new file mode 100644 index 00000000..14a5eaa4 --- /dev/null +++ b/apps/neshub/src/congestion.rs @@ -0,0 +1,67 @@ +//! The transport's congestion control, handed to the media's own. +//! +//! The video rate is decided per frame from what the path measurably did +//! (`rate.rs`), and that decision needs each frame to leave as one train: +//! its fragments back to back, so the bottleneck spaces them by its own rate +//! and the client can measure it. A transport controller in between would +//! window and pace them by its own estimate, smearing the very spacing being +//! measured, and second-guess a rate already chosen from better evidence. +//! +//! So this controller holds nothing back. Its window is larger than noq paces +//! at all (it stops pacing above `u32::MAX`), and it ignores loss, which the +//! media's own controller sees sooner, as queueing. +//! +//! Endpoint-wide, because transport settings are per endpoint, not per +//! connection: audio, input and control get it too. They are small flows the +//! video rate leaves room for; nothing they send can build a queue. + +use std::any::Any; +use std::sync::Arc; + +use iroh::endpoint::{Controller, ControllerFactory}; + +/// Past this noq does not pace: see `noq_proto::connection::pacing`. +const UNPACED_WINDOW: u64 = u32::MAX as u64 * 2; + +#[derive(Debug, Clone, Copy, Default)] +pub struct MediaRate; + +impl Controller for MediaRate { + fn on_congestion_event( + &mut self, + _now: std::time::Instant, + _sent: std::time::Instant, + _is_persistent_congestion: bool, + _is_ecn: bool, + _lost_bytes: u64, + _largest_lost_pn: u64, + ) { + } + + fn on_mtu_update(&mut self, _new_mtu: u16) {} + + fn window(&self) -> u64 { + UNPACED_WINDOW + } + + fn clone_box(&self) -> Box { + Box::new(*self) + } + + fn initial_window(&self) -> u64 { + UNPACED_WINDOW + } + + fn into_any(self: Box) -> Box { + self + } +} + +#[derive(Debug, Default)] +pub struct MediaRateFactory; + +impl ControllerFactory for MediaRateFactory { + fn build(self: Arc, _now: std::time::Instant, _current_mtu: u16) -> Box { + Box::new(MediaRate) + } +} diff --git a/apps/neshub/src/control.rs b/apps/neshub/src/control.rs index ab5fab3c..29436589 100644 --- a/apps/neshub/src/control.rs +++ b/apps/neshub/src/control.rs @@ -5,87 +5,46 @@ //! Nothing used to decide. The encoder took a bitrate from an environment //! variable nobody set, so every session offered the same 10 Mbps whatever the //! path between the two ends could carry. On a path sustaining under 3 Mbps that -//! does not degrade the picture, it removes it: most fragments are dropped, -//! every frame made of several fragments arrives incomplete, and the keyframe -- -//! the largest frame there is -- never completes either, so nothing ever -//! recovers. +//! does not degrade the picture, it removes it. //! -//! # Why the receiver decides, and not this end +//! # From the path, frame by frame //! -//! The obvious inputs are here: the hub holds the QUIC connection and can read -//! its round-trip time, its congestion window and its remaining datagram buffer. -//! They were measured, over an entire multi-session run, saying the path was -//! healthy while the client was receiving almost nothing -- 182 ms round trip, -//! *zero* QUIC packet loss, and arrivals falling from 1776 to 239 datagrams a -//! second. +//! Each client's path is modelled from its per-frame feedback (`rate.rs`): the +//! capacity its frames measure, whether the bottleneck is still busy with one +//! frame when the next arrives, and the share it delivers when it is. Down at +//! once when the queue grows, to that measured share; up only to a rate the path +//! has proven, held on the path's own recurrence. See nescore's +//! `docs/superpowers/specs/2026-10-16-video-rate-from-the-path-design.md`. //! -//! Part of that is structural rather than bad luck. `send_datagram` evicts the -//! oldest queued datagrams and returns `Ok`, so a sender that is overrunning the -//! path is told nothing at all; the "frames dropped" counter it feeds cannot be -//! non-zero however badly things are going. So the primary evidence is the -//! receiver's own report of what arrived, and this end's view is the fallback -//! for when no report has come in. +//! This replaced a controller built from thresholds: loss above 10% to back +//! off, below 2% to climb, back off to 0.85 of goodput, climb a twentieth of the +//! ceiling a second, a 10% deadband, queue limits of 250 and 80 ms. It could +//! only find the path's limit by hitting it, so it sawtoothed, and it heard about +//! the queue it built once a second, after the queue was tens of milliseconds +//! deep. //! -//! # Shape +//! # One encoder, the slowest client //! -//! Slow and dull on purpose. This is not a congestion controller in the QUIC -//! sense and does not try to be: it finds the right order of magnitude for a -//! video bitrate and stays there, at one decision a second. The transport -//! underneath is still doing real congestion control on its own packets. +//! One encoder serves every client, so it can answer only one question: the +//! rate the slowest path allows. Averaging would leave that client never +//! recovering while the numbers look acceptable. -use nesprotocol::{ControlMode, ReceiverReport}; +use std::collections::HashMap; -/// Loss above this means the path is being overrun and the target comes down. -const LOSS_DECREASE: f32 = 0.10; -/// Loss below this means there is room to climb. -const LOSS_INCREASE: f32 = 0.02; +use iroh::EndpointId; +use nesprotocol::ControlMode; +use nesprotocol::datagram::FrameFeedback; -/// What to keep of the measured goodput when backing off. -/// -/// Backing off to *below* what actually got through, rather than to a fraction -/// of what we were asking for, is what makes recovery quick. At 10 Mbps into a -/// 3 Mbps path the next target is about 2.5 Mbps rather than 8.5, so one step -/// does what twenty multiplicative decreases would. -const BACKOFF: f32 = 0.85; - -/// How much of the ceiling to add per calm second. Twenty steps from floor to -/// ceiling: slow enough that a brief quiet spell does not undo a back-off. -const CLIMB_FRACTION: u32 = 20; - -/// Don't actuate for less than this fraction of the current target. -/// -/// Retuning is cheap now -- it no longer rebuilds the encoder or forces a -/// keyframe -- but a line in the log every second saying the bitrate moved by -/// 1% is noise that hides the changes that matter. -const DEADBAND: f32 = 0.10; +use crate::rate::{self, PathModel}; /// Never go below this, whatever the ceiling is. /// -/// Under this a 1080p stream is not worth sending and the right answer is fewer -/// pixels rather than fewer bits, which is a decision about the tier and not one -/// this can make. It also has to leave room for audio, which shares the -/// connection and is not counted against the video ceiling. +/// A policy bound like the ceiling, not a guess about the network: under this a +/// 1080p stream is not worth sending, and the right answer is fewer pixels +/// rather than fewer bits, which is a decision about the tier and not one this +/// can make. const ABSOLUTE_FLOOR_KBPS: u32 = 800; -/// Seconds without a receiver report before this end's own view is used instead. -const SILENT_TICKS_BEFORE_FALLBACK: u32 = 3; - -/// Backlog, in milliseconds of video, that means the path is being overrun. -/// -/// Not a tuned number: it is a latency budget. At sixty frames a second a frame -/// is under 17 ms, so a quarter second of backlog is fifteen frames already -/// handed over and not yet gone -- a player is looking at a picture from before -/// their last four keypresses. There is no bitrate worth that, so past this the -/// target comes down whatever the loss says. -const QUEUE_DECREASE_MS: u32 = 250; - -/// Backlog, in milliseconds, under which the path is considered clear. -/// -/// Climbing needs a stronger warrant than holding does, because climbing is -/// what digs the queue. Five frames or so of backlog is the most that can be -/// outstanding and still be called live. -const QUEUE_CLIMB_MS: u32 = 80; - /// The band a target must stay inside. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct Limits { @@ -112,11 +71,8 @@ impl Limits { } } -/// What this end can see about the path, for when the far end is not talking. -/// -/// Every field is optional because every one of them can be unavailable: a -/// connection with no established path has no round-trip time, and a congestion -/// window is only meaningful once something has been sent. +/// What this end can see about a client's connection: for the stats, not for +/// the decision. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct PathView { /// Congestion window in bytes, as the transport currently believes it. @@ -124,70 +80,54 @@ pub struct PathView { /// Round-trip time in milliseconds. pub rtt_ms: Option, /// Datagram bytes handed to the transport that have not yet left. - /// - /// The backlog is the one thing this end can see that says the path is - /// being overrun *while it is still only late*. Loss says so too, but only - /// afterwards, and on a path that queues rather than drops, "afterwards" - /// can be several seconds. pub backlog_bytes: Option, } impl PathView { /// How long the backlog takes to drain at `drain_kbps`, in milliseconds. - /// - /// Bits divided by kilobits-per-second is milliseconds. The rate to divide - /// by is the one the queue actually drains at -- what is getting through -- - /// and not what we are asking for, which is the number that is too high - /// whenever this matters. pub fn backlog_ms(&self, drain_kbps: u32) -> Option { let backlog = self.backlog_bytes?; let drain = u64::from(drain_kbps.max(1)); Some((backlog.saturating_mul(8) / drain).min(u64::from(u32::MAX)) as u32) } - - /// Delivery rate the window and round trip imply, in kbps. - /// - /// A window is a quantity of bytes in flight for one round trip, so the two - /// together are a rate. Treated as an estimate of last resort: it describes - /// the transport's own opinion of the path, and that opinion is exactly what - /// was observed to be wrong when this mattered most. - pub fn estimate_kbps(&self) -> Option { - let (cwnd, rtt) = (self.cwnd_bytes?, self.rtt_ms?.max(1)); - let bits_per_second = cwnd.saturating_mul(8) * 1000 / u64::from(rtt); - Some((bits_per_second / 1000).min(u64::from(u32::MAX)) as u32) - } } /// Why the target is what it is, for the overlay and the log. #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(u8)] pub enum Reason { - /// Loss was high; backed off to under what was actually getting through. - Congested, - /// Loss was in the middle band; left alone. - Holding, - /// Loss was low; climbing towards the ceiling. - Climbing, - /// No receiver report recently; using this end's view of the path. - Fallback, - /// No report and no usable view either; decaying towards the floor. - Blind, + /// Where the session started; no path has said otherwise yet. + Start, + /// A path's queue was growing; sending below its share to drain it. + Draining, + /// Settled at the share the slowest path delivered while queueing. + Measured, + /// Raised to what the slowest path has proven it carries. + Stepped, + /// No frame came back from the slowest path; held at its last share. + Silent, /// Somebody set the bitrate by hand. Manual, /// The encoder is not under a bitrate at all, so there is nothing to decide. ConstantQuality, /// Nobody is connected, so there is no path to have an opinion about. NoClients, - /// This second's numbers describe frames the hub chose not to send, so they - /// say nothing about the path. - SelfInflicted, - /// Nothing was lost, but the send queue is standing deep enough that the - /// picture is arriving late. Backed off on the queue rather than on loss. - Backlogged, +} + +impl From for Reason { + fn from(r: rate::Reason) -> Self { + match r { + rate::Reason::Start => Self::Start, + rate::Reason::Draining => Self::Draining, + rate::Reason::Measured => Self::Measured, + rate::Reason::Stepped => Self::Stepped, + rate::Reason::Silent => Self::Silent, + } + } } /// The controller's whole state. -#[derive(Debug, Clone, Copy)] +#[derive(Debug)] pub struct Controller { /// The ceiling the box was given, which a client may lower but never raise. /// @@ -196,13 +136,6 @@ pub struct Controller { /// it again, because the only number left to compare against is the one it /// just lowered. box_ceiling_kbps: u32, - /// What the controller currently believes the bitrate should be. - /// - /// Kept apart from `sent_kbps` deliberately. Folding the two together looks - /// simpler and stalls the climb: the step up is a fraction of the *ceiling*, - /// so once the target passes half of it each step is under the deadband, and - /// a target that only moved when it actuated could never accumulate past - /// that point. Belief moves every second; only saying so is rationed. target_kbps: u32, /// What the encoder was last told, or `None` before anything was sent. sent_kbps: Option, @@ -210,19 +143,18 @@ pub struct Controller { mode: ControlMode, /// Set when the encoder is in constant-QP; there is no bitrate to control. constant_quality: bool, - silent_ticks: u32, reason: Reason, - /// Last measured send-queue depth, kept for reporting rather than control. + /// Last measured send-queue depth, for reporting. backlog_ms: u32, + /// Each streaming client's path. + paths: HashMap, } impl Controller { /// Start at the ceiling. /// /// Optimistic on purpose: the common case is a path that can carry the tier - /// that was sold, and a session that opened at its floor and climbed would - /// take twenty seconds to look like what was paid for. The first bad report - /// undoes it in one step. + /// that was sold, and a path that cannot says so within frames. pub fn new(limits: Limits) -> Self { Self { box_ceiling_kbps: limits.ceiling_kbps, @@ -231,9 +163,9 @@ impl Controller { limits, mode: ControlMode::Auto, constant_quality: false, - silent_ticks: 0, - reason: Reason::Holding, + reason: Reason::NoClients, backlog_ms: 0, + paths: HashMap::new(), } } @@ -250,21 +182,31 @@ impl Controller { self.reason } - /// How far behind the send queue was at the last decision, in milliseconds. + /// How far behind the send queue was at the last tick, in milliseconds. pub fn backlog_ms(&self) -> u32 { self.backlog_ms } + /// The slowest path's measured capacity, in kbps; 0 before one is measured. + pub fn capacity_kbps(&self) -> u32 { + self.slowest() + .and_then(PathModel::capacity_kbps) + .unwrap_or(0) + } + + /// The slowest path's last delivered share, in kbps; 0 before it queued. + pub fn delivery_kbps(&self) -> u32 { + self.slowest() + .and_then(PathModel::delivery_kbps) + .unwrap_or(0) + } + /// The ceiling the box was given, whatever a client has since asked for. pub fn box_ceiling_kbps(&self) -> u32 { self.box_ceiling_kbps } /// Adopt a ceiling a client asked for, never above the one the box was given. - /// - /// A client may lower its own ceiling -- to test a path, or because it knows - /// something about its link that this end does not -- but it may not raise - /// the one the tier bought. pub fn set_ceiling(&mut self, ceiling_kbps: u32) { self.limits = Limits::new(ceiling_kbps.min(self.box_ceiling_kbps).max(1)); self.target_kbps = self.limits.clamp(self.target_kbps); @@ -293,39 +235,45 @@ impl Controller { self.constant_quality = constant_quality; } - /// One second's decision. - /// - /// `report` is the worst report across the clients attached, or `None` when - /// none of them said anything. Returns the new target when it is worth - /// sending, and `None` when nothing should be sent -- which is most seconds. - /// One second's decision. - /// - /// `self_inflicted` says this second contained frames the hub deliberately - /// withheld -- a client resynchronising, whose deltas were undecodable and - /// were dropped rather than sent. Those seconds cannot be read as evidence - /// about the path: fewer frames were sent, so fewer arrived, so the measured - /// goodput is low and the loss is high, and a controller anchoring a backoff - /// on that would cut the bitrate in response to its own decision. Every - /// stutter would then also cost bandwidth, which is the opposite of what a - /// resynchronisation needs. - pub fn tick( + /// One frame's feedback from one client. Returns a new target when the + /// encoder should hear one. + pub fn on_feedback( &mut self, - clients: usize, - report: Option, - path: PathView, - self_inflicted: bool, + client: EndpointId, + frame: &FrameFeedback, + now_us: u64, ) -> Option { - if clients == 0 { + let start = self.target_kbps; + self.paths + .entry(client) + .or_insert_with(|| PathModel::new(start, now_us)) + .observe(frame, now_us); + self.decide() + } + + /// Once a second, with the clients that are streaming and this end's view + /// of their connections. Returns a new target when the encoder should hear + /// one. + pub fn tick(&mut self, streaming: &[EndpointId], path: PathView) -> Option { + self.paths.retain(|id, _| streaming.contains(id)); + for model in self.paths.values_mut() { + model.tick(); + } + self.backlog_ms = path.backlog_ms(self.target_kbps).unwrap_or(0); + if streaming.is_empty() { // Nothing is connected, so nothing is being carried and there is no - // path to form an opinion about. Deciding here means deciding on the - // absence of evidence: the controller used to read it as silence and - // decay to the floor, so a box waiting for its first client spent - // that time winding itself down and then jumped back up the moment - // somebody arrived. + // path to form an opinion about. self.reason = Reason::NoClients; - self.silent_ticks = 0; return None; } + self.decide() + } + + fn slowest(&self) -> Option<&PathModel> { + self.paths.values().min_by_key(|m| m.allowed_kbps()) + } + + fn decide(&mut self) -> Option { if self.constant_quality { self.reason = Reason::ConstantQuality; return None; @@ -334,124 +282,44 @@ impl Controller { self.reason = Reason::Manual; return None; } - if self_inflicted { - // Held, not decayed. The path may be fine; this second simply - // cannot say, and silence about the path is not evidence against it. - self.reason = Reason::SelfInflicted; - self.silent_ticks = 0; + let (allowed, reason) = match self.slowest() { + Some(m) => (m.allowed_kbps(), m.reason().into()), + None if self.sent_kbps.is_none() => (self.target_kbps, Reason::Start), + None => return None, + }; + self.reason = reason; + self.target_kbps = self.limits.clamp(allowed); + // Every change is said: they are events -- a queue found, a share + // measured, a step proven -- not a level wandering by a percent. The + // first decision is always said, because the encoder started at + // whatever its environment gave it. + if self.sent_kbps == Some(self.target_kbps) { return None; } - - // What the queue drains at is what is getting through. With no report - // to say, the target is the best guess available -- and a target that - // is too high only makes the backlog look shorter than it is, so this - // errs towards patience rather than towards cutting the rate. - let drain_kbps = match report.as_ref().map(|r| (r.goodput_bps / 1000) as u32) { - Some(measured) if measured > 0 => measured, - _ => self.target_kbps, - }; - self.backlog_ms = path.backlog_ms(drain_kbps).unwrap_or(0); - - let next = match report.as_ref().and_then(|r| r.loss().map(|l| (r, l))) { - Some((report, loss)) => { - self.silent_ticks = 0; - self.decide_from_report(report, loss, path) - } - None => { - self.silent_ticks = self.silent_ticks.saturating_add(1); - if self.silent_ticks < SILENT_TICKS_BEFORE_FALLBACK { - self.reason = Reason::Holding; - return None; - } - self.decide_from_path(path) - } - }; - - self.target_kbps = self.limits.clamp(next); - // Every second moves the belief; only a change worth acting on is said. - // The first decision is always said -- the encoder started at whatever - // its environment gave it, and this end has no way to know it matches. - match self.sent_kbps { - Some(sent) if !worth_sending(sent, self.target_kbps, self.limits) => None, - _ => { - self.sent_kbps = Some(self.target_kbps); - Some(self.target_kbps) - } - } - } - - fn decide_from_report(&mut self, report: &ReceiverReport, loss: f32, path: PathView) -> u32 { - let measured = (report.goodput_bps / 1000) as u32; - // `backlog_ms` is zero both when the queue is empty and when the - // transport cannot say, which are the same thing for this purpose: a - // queue nobody can see is not evidence of one. - let known = path.backlog_bytes.is_some(); - let queued = known && self.backlog_ms > QUEUE_DECREASE_MS; - - if loss > LOSS_DECREASE || queued { - self.reason = if queued { - Reason::Backlogged - } else { - Reason::Congested - }; - // Anchored on what actually arrived, not on what we were asking for. - // `goodput` of zero means nothing completed at all, in which case - // there is no measurement to anchor to and the target is simply - // halved -- the alternative, backing off to zero, would take the - // stream below the floor on a single bad second. - let anchor = if measured == 0 { - self.target_kbps / 2 - } else { - measured.min(self.target_kbps) - }; - return (anchor as f32 * BACKOFF) as u32; - } - // Climbing is what digs the queue, so it needs the queue to be empty as - // well as the loss to be low. Without this the controller climbs all - // the way to the ceiling against a path it is already overrunning, - // because a path that queues instead of dropping reports no loss at all - // until the buffer finally overflows -- and by then the picture is - // seconds behind. - if loss < LOSS_INCREASE && (!known || self.backlog_ms < QUEUE_CLIMB_MS) { - self.reason = Reason::Climbing; - return self - .target_kbps - .saturating_add(self.limits.ceiling_kbps / CLIMB_FRACTION); - } - self.reason = Reason::Holding; - self.target_kbps - } - - fn decide_from_path(&mut self, path: PathView) -> u32 { - match path.estimate_kbps() { - Some(estimate) => { - self.reason = Reason::Fallback; - estimate - } - None => { - // Nothing said and nothing visible. Holding a high target on no - // evidence is how the original failure sustained itself, so this - // decays rather than holds. - self.reason = Reason::Blind; - (self.target_kbps as f32 * BACKOFF) as u32 - } - } + self.sent_kbps = Some(self.target_kbps); + Some(self.target_kbps) } } -/// Whether a change is big enough to be worth actuating. -/// -/// Reaching either end of the band always is: a target pinned to the floor or -/// the ceiling is a fact worth stating even when the step to it was small. -pub fn worth_sending(current: u32, next: u32, limits: Limits) -> bool { - if current == next { - return false; +/// Tell the encoder a target the controller has just decided. +pub fn actuate( + cmd_tx: &tokio::sync::mpsc::UnboundedSender>, + controller: &Controller, + kbps: u32, +) { + let mut cmd = vec![nesprotocol::MSG_ENCODE_SETTINGS]; + nesprotocol::encode_bitrate_only(&mut cmd, kbps); + if cmd_tx.send(cmd).is_err() { + tracing::warn!("encoder command channel closed"); + return; } - if next == limits.floor_kbps() || next == limits.ceiling_kbps { - return true; - } - let change = current.abs_diff(next) as f32; - change >= current as f32 * DEADBAND + // Debug, not info: each is an event, but a path that keeps changing can + // make several a second, which at info buries the lines that matter. + tracing::debug!( + "video target {kbps}/{} kbps: {:?}", + controller.limits().ceiling_kbps, + controller.reason(), + ); } #[cfg(test)] @@ -460,456 +328,130 @@ mod tests { const CEILING: u32 = 10_000; - fn limits() -> Limits { - Limits::new(CEILING) - } - fn controller() -> Controller { - Controller::new(limits()) + Controller::new(Limits::new(CEILING)) } - /// A second in which `released` of `total` frames made it, carrying - /// `goodput_kbps` of completed video. - fn report(released: u32, incomplete: u32, goodput_kbps: u64) -> ReceiverReport { - ReceiverReport { - goodput_bps: goodput_kbps * 1000, - released, - incomplete, - never_arrived: 0, - rtt_ms: 180, - ..Default::default() + fn client(n: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[n; 32]).public() + } + + /// A frame of `bytes` sent at `t_us`, arriving over a link of `kbps` + /// that `busy` says was still carrying the previous frame. + fn frame(seq: u16, t_us: u64, bytes: u32, kbps: u32, queue_us: u32) -> FrameFeedback { + let spread = (u64::from(bytes - 1_200) * 8_000 / u64::from(kbps)) as u32; + FrameFeedback { + seq, + bytes, + fragments: bytes.div_ceil(1_200) as u16, + first_bytes: 1_200, + sent_us: t_us, + first_arrival_us: t_us + 40_000 + u64::from(queue_us), + spread_us: spread, + first_q_us: queue_us, } } - fn healthy() -> ReceiverReport { - report(60, 0, 9_800) - } - - /// A path with `backlog` bytes handed over and not yet gone. - fn backlogged(backlog: u64) -> PathView { - PathView { - cwnd_bytes: Some(13_000), - rtt_ms: Some(180), - backlog_bytes: Some(backlog), - } - } - - /// The second reported failure, as measured over a 1000-mile link. - /// - /// Every frame arrived and every frame completed -- zero loss, zero QUIC - /// congestion events, a flat 180 ms round trip for the whole session -- and - /// the picture was still several seconds behind, because the offer was - /// several times what the path carried and the difference was sitting in - /// the hub's own send buffer. A controller reading loss alone sees a - /// perfect path here and climbs to the ceiling against it. - #[test] - fn a_path_that_queues_instead_of_dropping_must_not_read_as_healthy() { - let mut c = controller(); - // Two megabits get through; a megabyte is already queued behind them. - let arriving = report(60, 0, 2_000); - let path = backlogged(1024 * 1024); - - for _ in 0..30 { - c.tick(1, Some(arriving), path, false); - } - - assert_ne!( - c.reason(), - Reason::Climbing, - "climbing against a path with four seconds of backlog" - ); - assert!( - c.target_kbps() < 2_000, - "target {} kbps is at or above what is getting through, so the \ - queue can only grow", - c.target_kbps() - ); - } - - /// The sawtooth: floor, climb to ceiling, collapse, repeat every 20 s. - /// - /// Loss alone cannot break this cycle, because on a queueing path loss only - /// appears once the buffer finally overflows -- long after the latency has - /// made the session unplayable, and by then the queue is deep enough that - /// backing off to the floor is the only way out. - #[test] - fn the_target_settles_instead_of_sawtoothing() { - let mut c = controller(); - // A steady 2 Mbps path. The backlog is what the last second of - // over-sending left behind, drained at what actually gets through. - let mut backlog: i64 = 0; - let mut seen = Vec::new(); - - for _ in 0..60 { - let target = c.target_kbps(); - // Whatever was asked for above 2 Mbps piles up; the rest drains. - backlog = (backlog + (i64::from(target) - 2_000) * 1000 / 8).clamp(0, 4 * 1024 * 1024); - c.tick( - 1, - Some(report(60, 0, 2_000)), - backlogged(backlog as u64), - false, - ); - seen.push(c.target_kbps()); - } - - let settled = &seen[30..]; - let (lo, hi) = ( - *settled.iter().min().unwrap(), - *settled.iter().max().unwrap(), - ); - assert!( - hi - lo <= 1_000, - "target still swinging between {lo} and {hi} kbps after 30 seconds" - ); - assert!( - hi <= 2_400, - "settled at {hi} kbps against a path carrying 2000" - ); - } - - /// The reported depth is what the next session will be judged on, so it - /// has to be right: bytes over the rate they leave at, in milliseconds. - #[test] - fn the_queue_depth_is_reported_as_measured() { - let mut c = controller(); - // 250 kB queued behind a 2 Mbps drain is exactly one second. - c.tick(1, Some(report(60, 0, 2_000)), backlogged(250_000), false); - assert_eq!(c.backlog_ms(), 1_000); - - // A transport that cannot say must report no queue, not a wrong one. - c.tick(1, Some(report(60, 0, 2_000)), PathView::default(), false); - assert_eq!(c.backlog_ms(), 0); - } - - /// The backlog must not become a reason never to climb again. - #[test] - fn a_clear_queue_still_climbs() { - let mut c = controller(); - c.set_ceiling(4_000); - let clear = backlogged(0); - for _ in 0..5 { - c.tick(1, Some(report(60, 0, 3_900)), clear, false); - } - assert_eq!(c.reason(), Reason::Climbing); - } - - /// The reported failure, as measured: a 10 Mbps offer into a path carrying - /// under three, where *no* frame completed for minutes. - #[test] - fn the_reported_failure_is_corrected_in_seconds() { - let mut c = controller(); - assert_eq!(c.target_kbps(), CEILING); - - let mut ticks = 0; - while c.target_kbps() > 2_800 && ticks < 10 { - // 46 frames a second starting to arrive, none of them completing. - c.tick(1, Some(report(0, 46, 0)), PathView::default(), false); - ticks += 1; - } - assert!( - c.target_kbps() <= 2_800, - "still at {} kbps after {ticks} seconds", - c.target_kbps(), - ); - assert!( - ticks <= 5, - "took {ticks} seconds to stop overrunning the path" - ); - } - - /// With a goodput measurement to anchor on it should take one step, not - /// several -- that is the whole reason for backing off to what arrived - /// rather than to a fraction of what was asked for. - #[test] - fn a_measured_goodput_is_corrected_in_one_step() { - let mut c = controller(); - let sent = c.tick(1, Some(report(12, 48, 2_850)), PathView::default(), false); - assert_eq!(sent, Some(c.target_kbps())); - assert!( - c.target_kbps() < 2_850, - "backed off to {} kbps, which is above what actually arrived", - c.target_kbps(), - ); - } - - /// The negative that matters most: a path that is fine must be left alone. - #[test] - fn a_healthy_path_is_not_wandered_away_from() { - let mut c = controller(); - // The first decision is always stated. The encoder started at whatever - // its environment gave it and this end cannot know that matches, so the - // target is asserted once rather than assumed. - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - Some(CEILING) - ); - for _ in 0..60 { - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - None, - "a healthy path produced a bitrate change", - ); - } - assert_eq!(c.target_kbps(), CEILING); - } - #[test] fn the_first_decision_is_always_stated() { - // Otherwise a session whose path happens to match its ceiling never - // tells the encoder anything, and the encoder keeps whatever its - // environment gave it -- which is the failure this replaces. let mut c = controller(); - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - Some(CEILING), - "the opening target was never stated", - ); - // And not repeated, now that the encoder has been told. - assert_eq!(c.tick(1, Some(healthy()), PathView::default(), false), None); + assert_eq!(c.tick(&[client(1)], PathView::default()), Some(CEILING)); + assert_eq!(c.tick(&[client(1)], PathView::default()), None); } #[test] - fn a_small_change_is_not_worth_saying() { - // Below the deadband nothing is sent, so the log and the encoder are not - // touched sixty times a minute for changes nobody could see. - let l = limits(); - assert!(!worth_sending(5_000, 5_200, l)); - assert!(worth_sending(5_000, 5_600, l)); - } - - #[test] - fn reaching_either_end_of_the_band_is_always_worth_saying() { - let l = limits(); - assert!(worth_sending(l.floor_kbps() + 1, l.floor_kbps(), l)); - assert!(worth_sending(CEILING - 1, CEILING, l)); - } - - #[test] - fn the_ceiling_is_never_exceeded() { + fn the_slowest_client_decides() { let mut c = controller(); - for _ in 0..200 { - c.tick(1, Some(healthy()), PathView::default(), false); - assert!(c.target_kbps() <= CEILING, "{} kbps", c.target_kbps()); - } - } - - #[test] - fn the_floor_is_never_gone_below() { - let mut c = controller(); - for _ in 0..200 { - c.tick(1, Some(report(0, 60, 0)), PathView::default(), false); - assert!( - c.target_kbps() >= c.limits().floor_kbps(), - "{} kbps is below the floor", - c.target_kbps(), - ); - } - } - - #[test] - fn it_climbs_back_after_a_bad_patch() { - let mut c = controller(); - for _ in 0..5 { - c.tick(1, Some(report(0, 46, 0)), PathView::default(), false); - } - let bottom = c.target_kbps(); - for _ in 0..40 { - c.tick(1, Some(healthy()), PathView::default(), false); + let (fast, slow) = (client(1), client(2)); + let mut t = 0; + let mut q = 0; + for seq in 0..120u16 { + // The fast client's frames drain at once; the slow one's queue + // grows by 6 ms a frame, a train arriving straight behind the last. + c.on_feedback(fast, &frame(seq, t, 20_000, 40_000, 0), t); + let mut f = frame(seq, t, 20_000, 6_000, q); + if seq > 0 { + f.first_arrival_us = t + 40_000 + u64::from(q); + } + c.on_feedback(slow, &f, t); + q += 6_000; + t += 16_667; } assert!( - c.target_kbps() > bottom, - "never recovered from {bottom} kbps" + c.target_kbps() < CEILING, + "the slow client did not bring the rate down" ); - assert_eq!(c.target_kbps(), CEILING, "did not climb all the way back"); } #[test] - fn a_second_the_hub_starved_is_not_read_as_a_bad_path() { - // While a client resynchronises its deltas are withheld, so fewer frames - // are sent, fewer arrive, goodput reads low and loss reads high. Backing - // off on that would cut the bitrate in response to the hub's own - // decision -- and make every stutter cost bandwidth as well. + fn the_ceiling_is_never_exceeded_and_the_floor_never_gone_below() { let mut c = controller(); - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - Some(CEILING) - ); - - let starved = report(0, 46, 0); - for _ in 0..10 { - assert_eq!(c.tick(1, Some(starved), PathView::default(), true), None); - } - assert_eq!(c.target_kbps(), CEILING, "the hub cut its own bitrate"); - assert_eq!(c.reason(), Reason::SelfInflicted); - } - - #[test] - fn a_genuinely_bad_second_still_acts_once_the_hub_stops_starving_it() { - // The flag holds, it does not blind. The moment a second is the path's - // own, the same evidence is acted on. - let mut c = controller(); - c.tick(1, Some(healthy()), PathView::default(), false); - c.tick(1, Some(report(0, 46, 0)), PathView::default(), true); - assert_eq!(c.target_kbps(), CEILING); - - c.tick(1, Some(report(12, 48, 2_850)), PathView::default(), false); - assert!(c.target_kbps() < 2_850, "a real bad second was ignored too"); - } - - #[test] - fn a_starved_second_does_not_count_towards_silence() { - // It is not a missing report -- one arrived, it just cannot be read. - // Counting it as silence would slide towards the path-based fallback and - // then to decaying blind, which is a different wrong answer. - let mut c = controller(); - c.tick(1, Some(healthy()), PathView::default(), false); - for _ in 0..10 { - c.tick(1, Some(report(0, 60, 0)), PathView::default(), true); - } - assert_eq!(c.reason(), Reason::SelfInflicted); - assert_eq!(c.target_kbps(), CEILING); + c.tick(&[client(1)], PathView::default()); + assert!(c.target_kbps() <= CEILING); + c.set_ceiling(500); + assert!(c.target_kbps() <= 500); + assert!(c.target_kbps() >= c.limits().floor_kbps()); } #[test] fn a_hand_set_bitrate_stands_the_controller_down() { - // How this class of bug gets diagnosed. A controller that overrode a - // person's setting a second later would take the tool away. let mut c = controller(); - c.note_manual_target(1_000); - for _ in 0..30 { - assert_eq!( - c.tick(1, Some(report(0, 60, 0)), PathView::default(), false), - None - ); - } - assert_eq!(c.target_kbps(), 1_000); + c.note_manual_target(3_000); + assert_eq!(c.tick(&[client(1)], PathView::default()), None); assert_eq!(c.reason(), Reason::Manual); + assert_eq!(c.target_kbps(), 3_000); } #[test] fn constant_quality_has_no_bitrate_to_decide() { let mut c = controller(); c.set_constant_quality(true); - assert_eq!( - c.tick(1, Some(report(0, 60, 0)), PathView::default(), false), - None - ); + assert_eq!(c.tick(&[client(1)], PathView::default()), None); assert_eq!(c.reason(), Reason::ConstantQuality); } - #[test] - fn silence_is_tolerated_briefly_and_then_acted_on() { - let mut c = controller(); - // A missed report or two says nothing; the path was fine a second ago. - for _ in 0..(SILENT_TICKS_BEFORE_FALLBACK - 1) { - assert_eq!(c.tick(1, None, PathView::default(), false), None); - assert_eq!(c.target_kbps(), CEILING); - } - // Past that, with nothing visible from this end either, it decays rather - // than holding a high target on no evidence at all -- holding is how the - // original failure sustained itself. - c.tick(1, None, PathView::default(), false); - assert_eq!(c.reason(), Reason::Blind); - assert!(c.target_kbps() < CEILING); - } - - #[test] - fn this_ends_view_is_used_only_when_the_far_end_is_silent() { - let mut c = controller(); - let path = PathView { - cwnd_bytes: Some(120_000), - rtt_ms: Some(200), - backlog_bytes: None, - }; - // A report present means the path view is ignored, however tempting. - c.tick(1, Some(healthy()), path, false); - assert_eq!(c.reason(), Reason::Climbing); - - for _ in 0..SILENT_TICKS_BEFORE_FALLBACK { - c.tick(1, None, path, false); - } - assert_eq!(c.reason(), Reason::Fallback); - // 120 KB in flight per 200 ms is 4.8 Mbps. - assert_eq!(c.target_kbps(), 4_800); - } - #[test] fn nothing_is_decided_while_nobody_is_connected() { - // A box waiting for its first client used to read the silence as a dead - // path and wind itself down to the floor, then jump back up the moment - // somebody arrived. There is no path to have an opinion about yet. let mut c = controller(); - for _ in 0..30 { - assert_eq!(c.tick(0, None, PathView::default(), false), None); - } - assert_eq!(c.target_kbps(), CEILING); + assert_eq!(c.tick(&[], PathView::default()), None); assert_eq!(c.reason(), Reason::NoClients); } #[test] - fn a_lowered_ceiling_can_be_raised_again() { - // The trap in folding the box's ceiling together with the one in force: - // after lowering, the only number left to compare against is the lowered - // one, so the client can never get back up. + fn a_client_that_leaves_takes_its_path_with_it() { let mut c = controller(); - c.set_ceiling(2_000); - assert_eq!(c.limits().ceiling_kbps, 2_000); - c.set_ceiling(5_000); - assert_eq!(c.limits().ceiling_kbps, 5_000); - assert_eq!(c.box_ceiling_kbps(), CEILING); + c.on_feedback(client(1), &frame(0, 0, 20_000, 6_000, 0), 0); + c.tick(&[], PathView::default()); + assert!(c.paths.is_empty()); } #[test] - fn a_ceiling_change_is_restated_to_the_encoder() { - // The encoder is the thing that has to act on it, and a target that - // happens to land on the same number is still a different instruction - // when the band around it moved. + fn a_lowered_ceiling_can_be_raised_again() { let mut c = controller(); - assert!( - c.tick(1, Some(healthy()), PathView::default(), false) - .is_some() - ); - c.set_ceiling(3_000); - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - Some(3_000) - ); + c.set_ceiling(4_000); + assert_eq!(c.limits().ceiling_kbps, 4_000); + c.set_ceiling(9_000); + assert_eq!(c.limits().ceiling_kbps, 9_000); } #[test] fn a_client_may_lower_its_ceiling_but_not_raise_it() { let mut c = controller(); - c.set_ceiling(4_000); - assert_eq!(c.limits().ceiling_kbps, 4_000); - assert!(c.target_kbps() <= 4_000, "the target outlived its ceiling"); + c.set_ceiling(CEILING * 2); + assert_eq!(c.limits().ceiling_kbps, CEILING); + } - c.set_ceiling(50_000); - assert_eq!( - c.limits().ceiling_kbps, - CEILING, - "a client raised the ceiling the tier bought", - ); + #[test] + fn a_ceiling_change_is_restated_to_the_encoder() { + let mut c = controller(); + c.tick(&[client(1)], PathView::default()); + c.set_ceiling(CEILING); + assert_eq!(c.tick(&[client(1)], PathView::default()), Some(CEILING)); } #[test] fn a_ceiling_under_the_floor_does_not_invert_the_band() { - // A misconfigured tier must not produce a floor above its own ceiling, - // which would put every target above the limit it exists to enforce. - let l = Limits::new(500); - assert!(l.floor_kbps() <= l.ceiling_kbps); - assert_eq!(l.clamp(10_000), 500); - } - - #[test] - fn a_silent_second_is_not_read_as_a_healthy_one() { - // No frames at all accounted for: nothing sent, or nothing arrived. - // Reading it as no loss would climb straight into a path that may be - // carrying nothing. - let mut c = controller(); - let empty = ReceiverReport::default(); - assert_eq!(empty.loss(), None); - c.tick(1, Some(empty), PathView::default(), false); - assert_eq!(c.target_kbps(), CEILING, "climbed on an empty report"); + let limits = Limits::new(500); + assert!(limits.floor_kbps() <= limits.ceiling_kbps); } } diff --git a/apps/neshub/src/dgram.rs b/apps/neshub/src/dgram.rs index ac94d0a6..be6f2015 100644 --- a/apps/neshub/src/dgram.rs +++ b/apps/neshub/src/dgram.rs @@ -38,11 +38,13 @@ pub enum MediaItem { /// Transport settings for an endpoint carrying media datagrams. /// -/// See [`DGRAM_BUFFER_BYTES`] for why the defaults are not enough. +/// See [`DGRAM_BUFFER_BYTES`] for why the defaults are not enough, and +/// `congestion.rs` for why the transport does not decide the rate. pub fn media_transport_config() -> QuicTransportConfig { QuicTransportConfig::builder() .datagram_send_buffer_size(DGRAM_BUFFER_BYTES) .datagram_receive_buffer_size(Some(DGRAM_BUFFER_BYTES)) + .congestion_controller_factory(std::sync::Arc::new(crate::congestion::MediaRateFactory)) .build() } diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index ab5b9c0a..b4716bf1 100644 --- a/apps/neshub/src/main.rs +++ b/apps/neshub/src/main.rs @@ -1,3 +1,4 @@ +mod congestion; mod control; mod dgram; #[cfg(test)] @@ -6,6 +7,7 @@ mod ipc_listener; mod keyframe; mod negotiation; mod pyro; +mod rate; mod screenshot; mod session; mod ticket; @@ -362,29 +364,15 @@ async fn main() -> Result<()> { loop { interval.tick().await; - // One decision a second, on the same tick as the stats, because - // a report describes the second that just passed and there is - // nothing to gain from deciding more often than they arrive. + // The rate is decided per frame, as feedback arrives (see + // `session.rs`). Once a second is for what feedback cannot say: + // a path that went silent, and clients that left. { - let clients = mgr.client_count().await; - let (report, path, self_inflicted) = mgr.worst_report().await; + let streaming = mgr.streaming_clients().await; + let (_, path, _) = mgr.worst_report().await; let mut controller = controller.lock().await; - if let Some(kbps) = controller.tick(clients, report, path, self_inflicted) { - let mut cmd = vec![nesprotocol::MSG_ENCODE_SETTINGS]; - nesprotocol::encode_bitrate_only(&mut cmd, kbps); - if cmd_tx.send(cmd).is_err() { - tracing::warn!("encoder command channel closed"); - } else { - // Trace, not info: under a path that keeps moving - // this fires every second, and a per-second line at - // info buries everything worth reading. - tracing::trace!( - "video ceiling {}/{} kbps: {:?}", - kbps, - controller.limits().ceiling_kbps, - controller.reason(), - ); - } + if let Some(kbps) = controller.tick(&streaming, path) { + crate::control::actuate(&cmd_tx, &controller, kbps); } } @@ -435,6 +423,8 @@ async fn main() -> Result<()> { pipeline_p95_ms, pipeline_max_ms, backlog_ms: controller.backlog_ms().min(u32::from(u16::MAX)) as u16, + capacity_kbps: controller.capacity_kbps(), + delivery_kbps: controller.delivery_kbps(), }, ); } diff --git a/apps/neshub/src/rate.rs b/apps/neshub/src/rate.rs new file mode 100644 index 00000000..d135a7a1 --- /dev/null +++ b/apps/neshub/src/rate.rs @@ -0,0 +1,453 @@ +//! What rate one client's path can carry, from how its frames arrived. +//! +//! See nescore's `docs/superpowers/specs/2026-10-16-video-rate-from-the-path-design.md`. +//! +//! # Every frame is a measurement +//! +//! A frame's fragments leave this end back to back (`congestion.rs` sees that +//! nothing holds them), so they reach the client spaced by the slowest link +//! they crossed. Two things follow, both read from the client's per-frame +//! feedback rather than guessed: +//! +//! - **Capacity.** A frame's bytes after its first fragment, over the time its +//! fragments took to arrive, is the bottleneck's rate. Every multi-fragment +//! frame measures it without sending above it for longer than one frame. +//! - **Whether we are over it.** Below capacity a frame's train has drained +//! before the next frame reaches the bottleneck, so consecutive trains arrive +//! with a gap between them. When the next train starts arriving no later than +//! the previous one's next fragment would have, the bottleneck never went idle +//! between them: it is busy with our traffic. Busy is not yet too much -- a +//! keyframe keeps it busy for a few frames and then it drains -- so the test +//! is whether the queue the first fragments find is *growing* across the busy +//! period, and whether more is being sent into it than comes out. +//! +//! # Down at once, up on evidence +//! +//! When the queue grows, what the path delivered over the busy period is its +//! measured share, and the queue already built is drained exactly: sending as +//! far below that share as the rate went above it, for as long, until the +//! trains have a gap between them again. Then the rate sits at the share. +//! +//! Going up uses the capacity the frames measured, and holds each step on the +//! same evidence the audio playout uses: a rate may be used again once the path +//! has gone quiet for longer than the longest it has ever gone between two +//! overloads at or below it, playout's start counting as the first. + +use nesprotocol::datagram::FrameFeedback; + +/// Why a client's rate is what it is. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum Reason { + /// Where the session started; nothing has said otherwise yet. + Start, + /// The queue was growing; sending below the path's share to drain it. + Draining, + /// Settled at the share the path delivered while it was queueing. + Measured, + /// Raised to what the path has proven it carries. + Stepped, + /// No frame has come back; held at the last rate the path proved. + Silent, +} + +/// One time the path was overloaded. +#[derive(Debug, Clone, Copy)] +struct Overload { + at_us: u64, + /// What was being sent into it. + sent_kbps: u32, +} + +/// A run of frames during which the bottleneck never went idle. +#[derive(Debug, Clone, Copy)] +struct Busy { + start_sent_us: u64, + start_arrival_us: u64, + /// Bytes of the frames in the run before the latest one. + bytes: u64, + /// The queue the second frame of the run found: the first frame's own + /// burst included, so a run that only drains never reads as growing. + first_q_us: u32, + /// Whether this run has already been counted as an overload. + counted: bool, +} + +#[derive(Debug)] +pub struct PathModel { + allowed_kbps: u32, + reason: Reason, + started_us: u64, + prev: Option, + busy: Option, + draining: bool, + /// The share the path last delivered while queueing. + delivery_kbps: Option, + /// The highest capacity a frame has measured since the last overload. + capacity_kbps: Option, + overloads: Vec, + /// Whether any feedback arrived since the last [`tick`](Self::tick). + heard: bool, +} + +/// Bits per millisecond, which is kbps, from bytes over microseconds. +fn kbps(bytes: u64, us: u64) -> u32 { + (bytes.saturating_mul(8_000) / us.max(1)).min(u64::from(u32::MAX)) as u32 +} + +impl PathModel { + pub fn new(start_kbps: u32, now_us: u64) -> Self { + Self { + allowed_kbps: start_kbps, + reason: Reason::Start, + started_us: now_us, + prev: None, + busy: None, + draining: false, + delivery_kbps: None, + capacity_kbps: None, + overloads: Vec::new(), + heard: false, + } + } + + pub fn allowed_kbps(&self) -> u32 { + self.allowed_kbps + } + pub fn reason(&self) -> Reason { + self.reason + } + pub fn capacity_kbps(&self) -> Option { + self.capacity_kbps + } + pub fn delivery_kbps(&self) -> Option { + self.delivery_kbps + } + + /// Take one frame's feedback. + pub fn observe(&mut self, frame: &FrameFeedback, now_us: u64) { + self.heard = true; + if frame.fragments >= 2 && frame.spread_us > 0 { + let measured = kbps( + u64::from(frame.bytes.saturating_sub(u32::from(frame.first_bytes))), + u64::from(frame.spread_us), + ); + self.capacity_kbps = Some(self.capacity_kbps.map_or(measured, |c| c.max(measured))); + } + + if let Some(prev) = self.prev.replace(*frame) { + if busy_between(&prev, frame) { + self.continue_busy(&prev, frame, now_us); + } else { + self.busy = None; + if self.draining { + self.draining = false; + if let Some(share) = self.delivery_kbps { + self.allowed_kbps = share; + self.reason = Reason::Measured; + } + } else { + self.step_up(now_us); + } + } + } + } + + fn continue_busy(&mut self, prev: &FrameFeedback, frame: &FrameFeedback, now_us: u64) { + let busy = self.busy.get_or_insert(Busy { + start_sent_us: prev.sent_us, + start_arrival_us: prev.first_arrival_us, + bytes: 0, + first_q_us: frame.first_q_us, + counted: false, + }); + busy.bytes += u64::from(prev.bytes); + let delivered = kbps( + busy.bytes, + frame.first_arrival_us.saturating_sub(busy.start_arrival_us), + ); + let sent = kbps(busy.bytes, frame.sent_us.saturating_sub(busy.start_sent_us)); + let growing = frame.first_q_us > busy.first_q_us; + if !(growing && sent > delivered) { + return; + } + // Drained at the rate it was built: as far below the share as the + // sender went above it, for as long as the run has lasted. + self.allowed_kbps = (2 * u64::from(delivered)) + .saturating_sub(u64::from(sent)) + .max(1) as u32; + self.delivery_kbps = Some(delivered); + self.draining = true; + self.reason = Reason::Draining; + if !busy.counted { + busy.counted = true; + self.overloads.push(Overload { + at_us: now_us, + sent_kbps: sent, + }); + // What the frames measured above this share was not true of the + // path as it is now; it is measured again from here. + self.capacity_kbps = Some(delivered); + } + } + + /// Raise the rate to the highest the path has proven it carries. + fn step_up(&mut self, now_us: u64) { + let Some(capacity) = self.capacity_kbps else { + return; + }; + if capacity <= self.allowed_kbps { + return; + } + // The capacity frames measured, or the highest rate below it that the + // path has gone quiet at for long enough to be trusted. + let mut candidates: Vec = self + .overloads + .iter() + .map(|o| o.sent_kbps) + .filter(|&r| r > self.allowed_kbps && r < capacity) + .chain(std::iter::once(capacity)) + .collect(); + candidates.sort_unstable_by(|a, b| b.cmp(a)); + let Some(&to) = candidates.iter().find(|&&r| self.proven(r, now_us)) else { + return; + }; + if to > self.allowed_kbps { + self.allowed_kbps = to; + self.reason = Reason::Stepped; + } + } + + /// Whether `rate` has been quiet long enough: since the last overload at or + /// below it, longer than the longest gap between two such overloads. + fn proven(&self, rate: u32, now_us: u64) -> bool { + let mut last = self.started_us; + let mut longest = 0; + let mut any = false; + for o in self.overloads.iter().filter(|o| o.sent_kbps <= rate) { + longest = longest.max(o.at_us.saturating_sub(last)); + last = o.at_us; + any = true; + } + !any || now_us.saturating_sub(last) > longest + } + + /// Once a second. A second in which no frame came back is itself a + /// measurement: whatever was sent did not arrive, so the rate goes to the + /// last share the path proved, as for a queue. + pub fn tick(&mut self) { + if !std::mem::take(&mut self.heard) && self.prev.is_some() { + if let Some(share) = self.delivery_kbps { + self.allowed_kbps = self.allowed_kbps.min(share); + } + self.reason = Reason::Silent; + } + } +} + +/// Whether the bottleneck went from `prev`'s train straight into `frame`'s. +/// +/// On a link that never idled, `frame`'s first fragment lands exactly its own +/// serialisation after `prev`'s last, at the rate a train measured. So the +/// test is whether it landed no later than that. With no train long enough to +/// show a rate there is nothing to compare with, and two single-fragment frames +/// cannot tell a busy link from an idle one. +fn busy_between(prev: &FrameFeedback, frame: &FrameFeedback) -> bool { + let per_byte = [prev, frame] + .into_iter() + .find(|f| f.fragments >= 2 && f.bytes > u32::from(f.first_bytes)) + .map(|f| { + ( + u64::from(f.spread_us), + u64::from(f.bytes - u32::from(f.first_bytes)), + ) + }); + let Some((spread, bytes)) = per_byte else { + return false; + }; + let first = (u64::from(frame.first_bytes) * spread).div_ceil(bytes); + let prev_end = prev.first_arrival_us + u64::from(prev.spread_us); + frame.first_arrival_us <= prev_end + first +} + +#[cfg(test)] +mod tests { + use super::*; + + const FRAGMENT: u64 = 1_200; + const INTERVAL_US: u64 = 16_667; + const PROPAGATION_US: u64 = 40_000; + + /// A path with a FIFO bottleneck, fed one frame at a time. + struct Bottleneck { + /// What the bottleneck carries for us, in kbps, by time. + capacity: Box u32>, + /// Extra bytes other traffic puts into it, by frame. + cross: Box u64>, + /// When the bottleneck is next free. + free_at: u64, + seq: u16, + } + + impl Bottleneck { + fn new(capacity: impl Fn(u64) -> u32 + 'static) -> Self { + Self { + capacity: Box::new(capacity), + cross: Box::new(|_| 0), + free_at: 0, + seq: 0, + } + } + + /// Send a frame of `bytes` at `sent_us`; return what the client says. + fn send(&mut self, n: u64, sent_us: u64, bytes: u64) -> FrameFeedback { + let c = u64::from((self.capacity)(sent_us)).max(1); + let at = sent_us + PROPAGATION_US; + // Cross traffic lands just ahead of the frame. + self.free_at = self.free_at.max(at) + (self.cross)(n) * 8_000 / c; + let start = self.free_at.max(at); + let first_q = start - at; + let fragments = bytes.div_ceil(FRAGMENT).max(1); + let serialise = |b: u64| b * 8_000 / c; + let first_done = start + serialise(FRAGMENT.min(bytes)); + let last_done = start + serialise(bytes); + self.free_at = last_done; + self.seq = self.seq.wrapping_add(1); + FrameFeedback { + seq: self.seq, + bytes: bytes as u32, + fragments: fragments as u16, + first_bytes: FRAGMENT.min(bytes) as u16, + sent_us, + first_arrival_us: first_done, + spread_us: (last_done - first_done) as u32, + first_q_us: first_q as u32, + } + } + } + + /// Run `frames` frames at whatever the model allows, the encoder holding + /// its target as an average, as a CBR encoder does. `keyframe_every` + /// frames, one is 8x a delta, and the deltas around it are smaller by what + /// it took. + fn run( + path: &mut Bottleneck, + model: &mut PathModel, + frames: u64, + keyframe_every: Option, + ) -> Vec<(u32, u32)> { + let mut out = Vec::new(); + for n in 0..frames { + let sent = n * INTERVAL_US; + let budget = u64::from(model.allowed_kbps()) * INTERVAL_US / 8_000; + let size = match keyframe_every { + Some(gop) => { + let delta = budget * gop / (gop + 7); + if n % gop == 0 { delta * 8 } else { delta } + } + None => budget, + }; + let fb = path.send(n, sent, size.max(1)); + model.observe(&fb, fb.first_arrival_us); + out.push((model.allowed_kbps(), fb.first_q_us)); + } + out + } + + #[test] + fn a_rate_above_the_path_comes_down_to_what_it_delivers() { + let mut path = Bottleneck::new(|_| 6_000); + let mut model = PathModel::new(10_000, 0); + let trace = run(&mut path, &mut model, 600, None); + let (settled, q) = *trace.last().unwrap(); + assert!( + (5_400..=6_000).contains(&settled), + "settled at {settled} kbps on a 6000 kbps path" + ); + assert!(q < 2_000, "the queue never drained: {q} us"); + let first_low = trace.iter().position(|(r, _)| *r <= 6_000).unwrap(); + assert!(first_low < 20, "took {first_low} frames to come down"); + } + + #[test] + fn a_rate_below_the_path_steps_up_to_what_frames_measured() { + let mut path = Bottleneck::new(|_| 12_000); + let mut model = PathModel::new(3_000, 0); + let trace = run(&mut path, &mut model, 120, None); + let (settled, q) = *trace.last().unwrap(); + assert!( + (11_000..=12_050).contains(&settled), + "settled at {settled} kbps on a 12000 kbps path" + ); + assert!(q < 2_000, "stepping up built a queue: {q} us"); + } + + #[test] + fn a_keyframe_is_not_mistaken_for_overload() { + let mut path = Bottleneck::new(|_| 12_000); + let mut model = PathModel::new(6_000, 0); + run(&mut path, &mut model, 600, Some(120)); + assert!( + model.overloads.is_empty(), + "{} overloads", + model.overloads.len() + ); + } + + #[test] + fn a_capacity_drop_is_followed_down_and_back_up_without_a_sawtooth() { + let mut path = Bottleneck::new(|t| { + if (10_000_000..20_000_000).contains(&t) { + 4_000 + } else { + 9_000 + } + }); + let mut model = PathModel::new(9_000, 0); + let trace = run(&mut path, &mut model, 60 * 40, None); + let during = trace[60 * 15].0; + assert!( + (3_600..=4_000).contains(&during), + "{during} kbps during the drop" + ); + let after = trace.last().unwrap().0; + assert!( + (8_000..=9_000).contains(&after), + "{after} kbps after the drop" + ); + assert!( + model.overloads.len() <= 4, + "{} overloads: a sawtooth", + model.overloads.len() + ); + } + + #[test] + fn recurring_cross_traffic_is_learned_and_not_stepped_into_forever() { + let mut path = Bottleneck::new(|_| 10_000); + // Another flow sends 60 KB every 5 s. + path.cross = Box::new(|n| if n % 300 == 150 { 60_000 } else { 0 }); + let mut model = PathModel::new(9_000, 0); + run(&mut path, &mut model, 60 * 120, None); + let second_minute = model + .overloads + .iter() + .filter(|o| o.at_us > 60_000_000) + .count(); + assert!( + second_minute <= 2, + "{second_minute} overloads in the second minute" + ); + } + + #[test] + fn silence_holds_the_rate_at_the_last_proven_share() { + let mut path = Bottleneck::new(|_| 6_000); + let mut model = PathModel::new(10_000, 0); + run(&mut path, &mut model, 300, None); + model.tick(); + model.tick(); + assert_eq!(model.reason(), Reason::Silent); + assert!(model.allowed_kbps() <= model.delivery_kbps().unwrap()); + } +} diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 8e377516..2ae70787 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -188,6 +188,14 @@ impl ClientSession { // it is sent, as for any client waiting on one. self.awaiting_keyframe.store(true, Ordering::Relaxed); let _ = self.idr_cmd_tx.send(vec![MSG_IDR_REQUEST]); + // The client says how each frame arrived, back on this same + // connection; the rate is decided on each one as it comes. + self.tasks.push(tokio::spawn(run_feedback_reader( + conn.clone(), + self.client, + self.controller.clone(), + self.idr_cmd_tx.clone(), + ))); let relay_ms = self.relay_ms.clone(); let awaiting = self.awaiting_keyframe.clone(); let withheld = self.withheld.clone(); @@ -890,6 +898,37 @@ async fn run_stats_sender(conn: Connection, mut rx: tokio::sync::mpsc::Unbounded debug!("stats sender exiting"); } +/// Read a client's per-frame feedback and decide the rate on each frame. +/// +/// Acted on here, as it arrives, rather than collected for a once-a-second +/// tick: a queue that starts building is answered a frame interval later, not +/// up to a second later, which at 60 frames a second is the difference +/// between one frame's queue and sixty. +async fn run_feedback_reader( + conn: Connection, + client: iroh::EndpointId, + controller: Arc>, + cmd_tx: tokio::sync::mpsc::UnboundedSender>, +) { + loop { + let datagram = match conn.read_datagram().await { + Ok(d) => d, + Err(e) => { + debug!("feedback reader ended: {e}"); + return; + } + }; + let Some(frame) = nesprotocol::datagram::decode_feedback(&datagram) else { + continue; + }; + let now_us = nesprotocol::clock::monotonic_us(); + let mut controller = controller.lock().await; + if let Some(kbps) = controller.on_feedback(client, &frame, now_us) { + crate::control::actuate(&cmd_tx, &controller, kbps); + } + } +} + /// Whether a broadcast video payload is a keyframe. /// /// The payload is `[1B codec][1B flags][4B ts][2B w][2B h][data]`, the same @@ -1266,6 +1305,18 @@ impl SessionManager { .count() } + /// Which clients have somewhere to send a picture: the paths the rate + /// controller has to answer to. + pub async fn streaming_clients(&self) -> Vec { + self.sessions + .lock() + .await + .iter() + .filter(|(_, s)| s.is_streaming()) + .map(|(id, _)| *id) + .collect() + } + /// Opus actually received from neswire since the last call, in kbps. /// /// Measured, not configured. The reported figure used to be diff --git a/crates/nesprotocol/src/datagram.rs b/crates/nesprotocol/src/datagram.rs index 8bb64ef1..bac7523f 100644 --- a/crates/nesprotocol/src/datagram.rs +++ b/crates/nesprotocol/src/datagram.rs @@ -49,6 +49,75 @@ pub const DGRAM_AUDIO: u8 = 1; /// [`crate::pyrowave`]. pub const DGRAM_PYROWAVE: u8 = crate::pyrowave::DGRAM_PYROWAVE; +/// One completed video frame as the client saw it arrive, client to hub on the +/// video connection. +/// +/// Sent once per frame, so the hub hears about every frame a round trip after +/// it left rather than once a second. What the hub's rate decision rests on: +/// whether the bottleneck was still busy with the last frame when this one +/// reached it, and how fast it delivered this one's fragments. See nescore's +/// `docs/superpowers/specs/2026-10-16-video-rate-from-the-path-design.md`. +pub const DGRAM_FEEDBACK: u8 = 3; + +/// `[1B kind][2B seq][4B bytes][2B fragments][2B first_bytes][8B sent_us] +/// [8B first_arrival_us][4B spread_us][4B first_q_us]`, little-endian. +pub const FEEDBACK_LEN: usize = 35; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct FrameFeedback { + /// The frame's sequence number. + pub seq: u16, + /// Datagram payload bytes that arrived for it, header excluded. + pub bytes: u32, + /// Fragments that arrived. + pub fragments: u16, + /// Payload bytes of the first fragment to arrive, which the arrival spread + /// does not cover: the spread starts when it has landed. + pub first_bytes: u16, + /// When the hub sent the frame, from its datagrams' `sent_us`. + pub sent_us: u64, + /// When the first fragment arrived, on the client's own clock. Only ever + /// compared with another frame's, never with the hub's clock. + pub first_arrival_us: u64, + /// The last fragment's arrival minus the first's. + pub spread_us: u32, + /// Queueing delay of the first fragment on the path (`delay`). + pub first_q_us: u32, +} + +pub fn encode_feedback(f: &FrameFeedback) -> [u8; FEEDBACK_LEN] { + let mut b = [0u8; FEEDBACK_LEN]; + b[0] = DGRAM_FEEDBACK; + b[1..3].copy_from_slice(&f.seq.to_le_bytes()); + b[3..7].copy_from_slice(&f.bytes.to_le_bytes()); + b[7..9].copy_from_slice(&f.fragments.to_le_bytes()); + b[9..11].copy_from_slice(&f.first_bytes.to_le_bytes()); + b[11..19].copy_from_slice(&f.sent_us.to_le_bytes()); + b[19..27].copy_from_slice(&f.first_arrival_us.to_le_bytes()); + b[27..31].copy_from_slice(&f.spread_us.to_le_bytes()); + b[31..35].copy_from_slice(&f.first_q_us.to_le_bytes()); + b +} + +/// `None` for anything that is not a whole feedback datagram. +pub fn decode_feedback(b: &[u8]) -> Option { + if b.len() < FEEDBACK_LEN || b[0] != DGRAM_FEEDBACK { + return None; + } + let u16_at = |o: usize| u16::from_le_bytes([b[o], b[o + 1]]); + let u32_at = |o: usize| u32::from_le_bytes(b[o..o + 4].try_into().unwrap()); + let u64_at = |o: usize| u64::from_le_bytes(b[o..o + 8].try_into().unwrap()); + Some(FrameFeedback { + seq: u16_at(1), + bytes: u32_at(3), + fragments: u16_at(7), + first_bytes: u16_at(9), + sent_us: u64_at(11), + first_arrival_us: u64_at(19), + spread_us: u32_at(27), + first_q_us: u32_at(31), + }) +} /// Bounds reassembly work. A 4K keyframe runs to a few hundred fragments at a /// typical MTU; anything past this did not come from a sender of ours. pub const MAX_FRAGMENTS_PER_FRAME: usize = 4096; @@ -214,6 +283,30 @@ mod tests { assert!(decode_datagram(&buf).is_none()); } + #[test] + fn feedback_round_trips() { + let f = FrameFeedback { + seq: 0xABCD, + bytes: 17_000, + fragments: 15, + first_bytes: 1_180, + sent_us: 0x0102_0304_0506_0708, + first_arrival_us: 0x1112_1314_1516_1718, + spread_us: 9_800, + first_q_us: 450, + }; + let wire = encode_feedback(&f); + assert_eq!(decode_feedback(&wire), Some(f)); + assert_eq!(decode_feedback(&wire[..FEEDBACK_LEN - 1]), None); + let mut other = wire; + other[0] = DGRAM_VIDEO; + assert_eq!( + decode_feedback(&other), + None, + "a video datagram read as feedback" + ); + } + #[test] fn empty_body_still_takes_one_fragment() { assert_eq!(fragment_count(0, 1000), 1); diff --git a/crates/nesprotocol/src/stats.rs b/crates/nesprotocol/src/stats.rs index 5879319e..46dab1e9 100644 --- a/crates/nesprotocol/src/stats.rs +++ b/crates/nesprotocol/src/stats.rs @@ -120,12 +120,18 @@ pub struct VideoBreakdown { /// eight seconds old: loss cannot show a queue, only its overflow. Anything /// but near-zero here means latency is being spent on backlog. pub backlog_ms: u16, + /// The bottleneck capacity the controller has measured from frame + /// dispersion, in kbps; 0 before any frame has measured it. + pub capacity_kbps: u32, + /// The rate the path last delivered while it was queueing, in kbps: the + /// share it proved it could carry. 0 before it has queued at all. + pub delivery_kbps: u32, } /// `[4B key_bps][4B delta_bps][1B keyframes][4B target][4B ceiling][1B reason] /// [1B manual][4B box_ceiling][2B pipeline_p50][2B pipeline_p95][2B pipeline_max] -/// [2B backlog_ms]` -pub const VIDEO_BREAKDOWN_LEN: usize = 31; +/// [2B backlog_ms][4B capacity_kbps][4B delivery_kbps]` +pub const VIDEO_BREAKDOWN_LEN: usize = 39; pub fn encode_video_breakdown(buf: &mut Vec, b: &VideoBreakdown) { buf.reserve(VIDEO_BREAKDOWN_LEN); @@ -141,6 +147,8 @@ pub fn encode_video_breakdown(buf: &mut Vec, b: &VideoBreakdown) { buf.extend_from_slice(&b.pipeline_p95_ms.to_le_bytes()); buf.extend_from_slice(&b.pipeline_max_ms.to_le_bytes()); buf.extend_from_slice(&b.backlog_ms.to_le_bytes()); + buf.extend_from_slice(&b.capacity_kbps.to_le_bytes()); + buf.extend_from_slice(&b.delivery_kbps.to_le_bytes()); } /// Decoded stats from any source. @@ -211,6 +219,8 @@ pub fn decode_stats(msg_type: u8, data: &[u8], stats: &mut PipelineStats) { pipeline_p95_ms: u16::from_le_bytes([d[25], d[26]]), pipeline_max_ms: u16::from_le_bytes([d[27], d[28]]), backlog_ms: u16::from_le_bytes([d[29], d[30]]), + capacity_kbps: u32_at(31), + delivery_kbps: u32_at(35), }); } } @@ -236,6 +246,8 @@ mod breakdown_tests { pipeline_p95_ms: 24, pipeline_max_ms: 61, backlog_ms: 40, + capacity_kbps: 23_500, + delivery_kbps: 7_800, } } diff --git a/docs/media-transport.md b/docs/media-transport.md index 058caab8..e741ff89 100644 --- a/docs/media-transport.md +++ b/docs/media-transport.md @@ -124,6 +124,18 @@ rate, while the pacer underneath is using the measured window and RTT. If the application is producing faster than the pacer drains, the answer is to produce less — which is the bitrate controller's job — not to add a second pacer. +**Since 2026-10-16 the hub turns this pacer off** (`neshub/src/congestion.rs`). +Its controller reports a window above `u32::MAX`, past which noq does not pace, +and ignores loss. The video rate is decided per frame from client feedback +(`neshub/src/rate.rs`), and that needs each frame to leave as a back-to-back +train, so the bottleneck spaces the fragments by its own rate and the client can +measure it. The transport pacing by its window estimate would smear exactly that +spacing. The rate is still produced to fit the path; it is the media's +controller that decides it, on the path's queueing delay, which leads loss and +RTT by tens of milliseconds. Transport settings are per endpoint, so the hub's +audio, input and control connections are unpaced too: they are small flows the +video rate leaves room for. + ## Loss is a lagging indicator; the queue is the leading one A path that is overrun does not necessarily drop anything. It can simply be