diff --git a/crates/punktfunk-core/src/abr.rs b/crates/punktfunk-core/src/abr.rs new file mode 100644 index 00000000..857fd39c --- /dev/null +++ b/crates/punktfunk-core/src/abr.rs @@ -0,0 +1,300 @@ +//! Adaptive bitrate: the client-side AIMD controller behind the "Automatic" bitrate setting. +//! +//! Runs inside [`crate::client`]'s data-plane pump on the same 750 ms cadence as the adaptive-FEC +//! [`crate::quic::LossReport`], deciding when to ask the host for a different encoder bitrate via +//! [`crate::quic::SetBitrate`]. Division of labour with adaptive FEC: **FEC answers fast, random +//! loss** (Wi-Fi bursts, RF noise — recoverable redundancy is the right tool); **bitrate answers +//! persistent congestion** (the link simply can't carry the rate — more FEC only adds load). The +//! controller therefore reacts to *sustained* signals only: +//! +//! - **unrecoverable frames** — loss exceeded the FEC budget (the stream visibly froze/recovered); +//! - **heavy loss** — a window whose shard loss is beyond what FEC should be left to absorb alone; +//! - **one-way-delay rise** — capture→received latency (host-clock skew corrected) climbing above +//! its rolling baseline: standing queue growth, the *pre-loss* signature of a saturated link +//! (bufferbloat) — this is the early-warning signal loss-based control lacks; +//! - **a jump-to-live flush** — the pump discarded its backlog, the strongest "we were behind" +//! evidence there is. +//! +//! AIMD shape: two consecutive bad windows ⇒ multiplicative decrease (×0.7, floored); ~10 s of +//! clean windows ⇒ additive-ish increase (+~6 %, ceilinged at the session's starting rate — the +//! controller recovers *back to* what was negotiated, never beyond it). Changes are rate-limited +//! (each one costs the IDR the host's rebuilt encoder opens with) and the whole controller +//! disables itself against a host that never answers [`crate::quic::BitrateChanged`] (an older +//! build that ignores unknown control messages). + +use std::collections::VecDeque; +use std::time::{Duration, Instant}; + +/// Never ask for less than this — below it the stream is unusable anyway and the floor keeps a +/// mis-measured window from cratering the session. +const FLOOR_KBPS: u32 = 5_000; +/// Consecutive bad windows before a decrease — one window can be a scheduler blip or a single +/// Wi-Fi scan; two in a row (1.5 s) is a condition. +const BAD_WINDOWS_TO_DECREASE: u32 = 2; +/// Consecutive clean windows before probing back up (~10 s at the 750 ms cadence): recovery is +/// deliberately much slower than backoff, classic AIMD. +const CLEAN_WINDOWS_TO_INCREASE: u32 = 13; +/// Minimum gap between requested changes — every accepted change costs an encoder rebuild + IDR +/// on the host, and back-to-back steps would outrun the ack/effect round trip. +const CHANGE_COOLDOWN: Duration = Duration::from_secs(3); +/// Window shard loss beyond which the window counts bad even without an unrecoverable frame: +/// 2 % sustained is congestion territory, not the random tail FEC exists for. +const HEAVY_LOSS_PPM: u32 = 20_000; +/// How far the window's mean one-way delay may sit above the rolling baseline before it counts +/// as queue growth. 25 ms is far beyond jitter at any streamable frame rate. +const OWD_RISE_US: i64 = 25_000; +/// Rolling window (in 750 ms report windows, ~30 s) whose minimum mean is the OWD baseline. +/// Long enough to remember the uncongested floor, short enough to follow genuine path changes. +const BASELINE_WINDOWS: usize = 40; +/// Requests sent without a single [`crate::quic::BitrateChanged`] ack before concluding the host +/// predates bitrate renegotiation and going quiet for the rest of the session. +const MAX_UNACKED: u32 = 3; + +/// One decision per report window; `Some(kbps)` = send a [`crate::quic::SetBitrate`]. +pub(crate) struct BitrateController { + /// `false` = permanently off (explicit user bitrate, an old host, or ack silence). + enabled: bool, + /// The rate we believe the host encodes at (updated by acks; requests are not assumed). + current_kbps: u32, + /// The session's starting (negotiated) rate — the recovery ceiling. + ceiling_kbps: u32, + floor_kbps: u32, + /// Recent window mean OWDs (µs); the rolling min is the uncongested baseline. + owd_means: VecDeque, + bad_windows: u32, + clean_windows: u32, + last_change: Option, + /// Requests since the last ack — reaching [`MAX_UNACKED`] disables the controller. + unacked: u32, +} + +impl BitrateController { + /// `start_kbps` is the Welcome-resolved session bitrate when the user chose Automatic, or `0` + /// to build a permanently-disabled controller (explicit bitrate / an old host that didn't + /// echo one — no known ceiling to work against). + pub(crate) fn new(start_kbps: u32) -> Self { + BitrateController { + enabled: start_kbps > 0, + current_kbps: start_kbps, + ceiling_kbps: start_kbps, + floor_kbps: FLOOR_KBPS.min(start_kbps.max(1)), + owd_means: VecDeque::with_capacity(BASELINE_WINDOWS), + bad_windows: 0, + clean_windows: 0, + last_change: None, + unacked: 0, + } + } + + /// The host's [`crate::quic::BitrateChanged`] ack: its clamp is authoritative for what the + /// encoder now targets, and any ack proves the host renegotiates (resets the silence counter). + pub(crate) fn on_ack(&mut self, kbps: u32) { + if kbps > 0 { + self.current_kbps = kbps; + } + self.unacked = 0; + } + + /// Feed one report window; returns the rate to request now, if any. `dropped` = frames that + /// went FEC-unrecoverable in the window, `loss_ppm` the window's [`crate::quic::LossReport`] + /// figure, `owd_mean_us` the window's mean skew-corrected capture→received latency (`None` + /// without a clock handshake), `flushed` = the pump's jump-to-live fired in the window. + pub(crate) fn on_window( + &mut self, + now: Instant, + dropped: u64, + loss_ppm: u32, + owd_mean_us: Option, + flushed: bool, + ) -> Option { + if !self.enabled { + return None; + } + if self.unacked >= MAX_UNACKED { + // The host never answered: an older build. Go quiet instead of spamming a message it + // logs as unknown every few seconds. + self.enabled = false; + tracing::info!("adaptive bitrate off — host never acked a SetBitrate (older host)"); + return None; + } + // OWD: compare against the rolling-min baseline of PRIOR windows (so a rising window + // doesn't drag its own baseline up), then record it. + let owd_bad = match owd_mean_us { + Some(mean) => { + let bad = self + .owd_means + .iter() + .min() + .is_some_and(|&base| mean > base + OWD_RISE_US); + if self.owd_means.len() == BASELINE_WINDOWS { + self.owd_means.pop_front(); + } + self.owd_means.push_back(mean); + bad + } + None => false, + }; + let bad = dropped > 0 || loss_ppm >= HEAVY_LOSS_PPM || owd_bad || flushed; + if bad { + self.bad_windows += 1; + self.clean_windows = 0; + } else { + self.clean_windows += 1; + self.bad_windows = 0; + } + let cooled = self + .last_change + .is_none_or(|t| now.duration_since(t) >= CHANGE_COOLDOWN); + if !cooled { + return None; + } + if self.bad_windows >= BAD_WINDOWS_TO_DECREASE && self.current_kbps > self.floor_kbps { + let next = ((self.current_kbps as u64 * 7 / 10) as u32).max(self.floor_kbps); + self.bad_windows = 0; + return self.request(next, now); + } + if self.clean_windows >= CLEAN_WINDOWS_TO_INCREASE && self.current_kbps < self.ceiling_kbps + { + let next = (self.current_kbps + self.current_kbps / 16 + 1).min(self.ceiling_kbps); + self.clean_windows = 0; + return self.request(next, now); + } + None + } + + fn request(&mut self, kbps: u32, now: Instant) -> Option { + self.last_change = Some(now); + self.unacked += 1; + // `current_kbps` is NOT updated here — the host's ack is authoritative. A lost/ignored + // request just recomputes from the same base next time (and counts toward MAX_UNACKED). + Some(kbps) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A window cadence matching the pump's 750 ms tick, safely past the change cooldown when + /// stepped 5× between decisions. + const TICK: Duration = Duration::from_millis(750); + + fn ticks(start: Instant, n: u32) -> Instant { + start + TICK * n + } + + /// Drive `n` clean windows, asserting no decision fires before the clean threshold. + fn run_clean(c: &mut BitrateController, start: Instant, from: u32, n: u32) -> Option { + let mut out = None; + for i in from..from + n { + out = c.on_window(ticks(start, i), 0, 0, Some(10_000), false); + if out.is_some() { + return out; + } + } + out + } + + #[test] + fn disabled_when_not_automatic_or_old_host() { + // start 0 = explicit user bitrate or a host that didn't echo one → permanently off. + let mut c = BitrateController::new(0); + let now = Instant::now(); + assert_eq!(c.on_window(now, 5, 900_000, Some(500_000), true), None); + } + + #[test] + fn two_bad_windows_step_down_multiplicatively() { + let mut c = BitrateController::new(20_000); + let start = Instant::now(); + // One bad window is a blip — no reaction. + assert_eq!(c.on_window(ticks(start, 0), 1, 0, None, false), None); + // The second consecutive bad window backs off ×0.7. + assert_eq!(c.on_window(ticks(start, 1), 1, 0, None, false), Some(14_000)); + c.on_ack(14_000); + // Still bad after the cooldown → another ×0.7 step from the ACKED rate. + assert_eq!(c.on_window(ticks(start, 6), 1, 0, None, false), None); // bad #1 again + assert_eq!(c.on_window(ticks(start, 7), 1, 0, None, false), Some(9_800)); + } + + #[test] + fn cooldown_blocks_back_to_back_steps() { + let mut c = BitrateController::new(20_000); + let start = Instant::now(); + assert_eq!(c.on_window(ticks(start, 0), 1, 0, None, false), None); + assert_eq!(c.on_window(ticks(start, 1), 1, 0, None, false), Some(14_000)); + c.on_ack(14_000); + // Two more bad windows land INSIDE the 3 s cooldown (ticks 2,3 = 1.5/2.25 s) → held. + assert_eq!(c.on_window(ticks(start, 2), 1, 0, None, false), None); + assert_eq!(c.on_window(ticks(start, 3), 1, 0, None, false), None); + } + + #[test] + fn floor_is_never_crossed() { + let mut c = BitrateController::new(6_000); + let start = Instant::now(); + assert_eq!(c.on_window(ticks(start, 0), 1, 0, None, false), None); + // ×0.7 of 6000 = 4200 < floor → clamped to 5000. + assert_eq!(c.on_window(ticks(start, 1), 1, 0, None, false), Some(5_000)); + c.on_ack(5_000); + // At the floor, further bad windows request nothing. + assert_eq!(c.on_window(ticks(start, 6), 1, 0, None, false), None); + assert_eq!(c.on_window(ticks(start, 7), 1, 0, None, false), None); + } + + #[test] + fn sustained_clean_recovers_toward_ceiling_only() { + let mut c = BitrateController::new(20_000); + let start = Instant::now(); + assert_eq!(c.on_window(ticks(start, 0), 1, 0, None, false), None); + assert_eq!(c.on_window(ticks(start, 1), 1, 0, None, false), Some(14_000)); + c.on_ack(14_000); + // 13 clean windows → one additive step up (14000 + 14000/16 + 1 = 14876). + let up = run_clean(&mut c, start, 2, 13); + assert_eq!(up, Some(14_876)); + c.on_ack(14_876); + // Fully recovered → clean windows at the ceiling stay quiet (never probe past start). + c.on_ack(20_000); + assert_eq!(run_clean(&mut c, start, 40, 20), None); + } + + #[test] + fn owd_rise_alone_is_a_congestion_signal() { + let mut c = BitrateController::new(20_000); + let start = Instant::now(); + // Establish a ~10 ms baseline over a few clean windows. + for i in 0..4 { + assert_eq!(c.on_window(ticks(start, i), 0, 0, Some(10_000), false), None); + } + // Delay climbs 40 ms above baseline with ZERO loss — bufferbloat. Two windows → back off. + assert_eq!(c.on_window(ticks(start, 4), 0, 0, Some(50_000), false), None); + assert_eq!( + c.on_window(ticks(start, 5), 0, 0, Some(52_000), false), + Some(14_000) + ); + } + + #[test] + fn ack_silence_disables_the_controller() { + let mut c = BitrateController::new(20_000); + let start = Instant::now(); + let mut sent = 0; + let mut i = 0; + // Keep every window bad and never ack: exactly MAX_UNACKED requests, then silence. + while i < 60 { + if c.on_window(ticks(start, i), 1, 0, None, false).is_some() { + sent += 1; + } + i += 1; + } + assert_eq!(sent, MAX_UNACKED); + } + + #[test] + fn flush_counts_as_a_bad_window() { + let mut c = BitrateController::new(20_000); + let start = Instant::now(); + assert_eq!(c.on_window(ticks(start, 0), 0, 0, None, true), None); + assert_eq!(c.on_window(ticks(start, 1), 0, 0, None, true), Some(14_000)); + } +} diff --git a/crates/punktfunk-core/src/client.rs b/crates/punktfunk-core/src/client.rs index c75c5501..b022e1c6 100644 --- a/crates/punktfunk-core/src/client.rs +++ b/crates/punktfunk-core/src/client.rs @@ -15,9 +15,11 @@ use crate::config::{CompositorPref, GamepadPref, Mode, Role}; use crate::error::{PunktfunkError, Result}; use crate::input::InputEvent; use crate::packet::FLAG_PROBE; +use crate::abr::BitrateController; use crate::quic::{ - endpoint, io, window_loss_ppm, ColorInfo, HdrMeta, Hello, HidOutput, LossReport, ProbeRequest, - ProbeResult, Reconfigure, Reconfigured, RequestKeyframe, RichInput, Start, Welcome, + endpoint, io, window_loss_ppm, BitrateChanged, ColorInfo, HdrMeta, Hello, HidOutput, + LossReport, ProbeRequest, ProbeResult, Reconfigure, Reconfigured, RequestKeyframe, RichInput, + SetBitrate, Start, Welcome, }; use crate::session::{Frame, Session}; use crate::transport::UdpTransport; @@ -27,6 +29,19 @@ use std::sync::mpsc::{Receiver, RecvTimeoutError, SyncSender}; use std::sync::{Arc, Condvar, Mutex}; use std::time::{Duration, Instant}; +/// Join `host` and `port` for `SocketAddr` parsing, bracketing a bare IPv6 literal +/// (`fd00::1` → `[fd00::1]:4770`) — without the brackets the joined string can never parse and +/// the error blames the caller's input. The control/data sockets are still IPv4-bound today, so +/// a v6 dial fails at connect (with an honest IO error); this is the parse-side groundwork for +/// IPv6 support. V4 literals, hostnames, and already-bracketed input pass through unchanged. +fn join_host_port(host: &str, port: u16) -> String { + if host.contains(':') && !host.starts_with('[') { + format!("[{host}]:{port}") + } else { + format!("{host}:{port}") + } +} + /// A control-stream request the embedder makes on the open handshake stream: a mode switch or a /// speed test. One outbound channel carries both so the worker's `select!` has a single writer /// (two `&mut ctrl_send` borrows across select branches don't compile). @@ -35,6 +50,9 @@ enum CtrlRequest { Probe(ProbeRequest), Keyframe, Loss(LossReport), + /// Adaptive bitrate: ask the host to re-target its encoder (kbps). Sent by the pump's + /// [`BitrateController`] when the user's bitrate setting is Automatic. + SetBitrate(u32), } /// What the worker reports to [`NativeClient::connect`] once the handshake lands: the negotiated @@ -642,7 +660,7 @@ impl NativeClient { .map_err(PunktfunkError::Io)?; let pin = pin.to_string(); let name = name.to_string(); - let remote: std::net::SocketAddr = format!("{host}:{port}") + let remote: std::net::SocketAddr = join_host_port(host, port) .parse() .map_err(|_| PunktfunkError::InvalidArg("host:port"))?; @@ -1096,7 +1114,7 @@ async fn worker_main(args: WorkerArgs) { hot_tids, } = args; let setup = async { - let remote: std::net::SocketAddr = format!("{host}:{port}") + let remote: std::net::SocketAddr = join_host_port(&host, port) .parse() .map_err(|_| PunktfunkError::InvalidArg("host:port"))?; let (ep, observed) = endpoint::client_pinned_with_identity( @@ -1289,6 +1307,10 @@ async fn worker_main(args: WorkerArgs) { } }); + // Adaptive bitrate ack slot: the control task parks the latest BitrateChanged here; the + // pump's controller drains it on its report tick (`take()` — an ack is consumed once). + let bitrate_ack: Arc>> = Arc::new(Mutex::new(None)); + // Control task: the handshake stream stays open for mid-stream renegotiation + speed tests. // Outbound requests (mode switch, probe) and inbound replies (Reconfigured, ProbeResult) are // multiplexed with `select!`; a single outbound channel (`ctrl_rx`) keeps one writer so the @@ -1296,6 +1318,7 @@ async fn worker_main(args: WorkerArgs) { { let mode_slot = mode_slot.clone(); let probe = probe.clone(); + let bitrate_ack = bitrate_ack.clone(); tokio::spawn(async move { loop { tokio::select! { @@ -1306,6 +1329,7 @@ async fn worker_main(args: WorkerArgs) { CtrlRequest::Probe(p) => p.encode(), CtrlRequest::Keyframe => RequestKeyframe.encode(), CtrlRequest::Loss(r) => r.encode(), + CtrlRequest::SetBitrate(k) => SetBitrate { bitrate_kbps: k }.encode(), }; if io::write_msg(&mut ctrl_send, &bytes).await.is_err() { break; @@ -1342,6 +1366,15 @@ async fn worker_main(args: WorkerArgs) { delivered_packets = p.delivered_packets, "speed-test probe result" ); + } else if let Ok(ack) = BitrateChanged::decode(&msg) { + // Adaptive bitrate: the host's clamp is authoritative — park it for + // the pump's controller (which also reads any ack as "this host + // renegotiates", arming further steps). + tracing::info!( + kbps = ack.bitrate_kbps, + "host re-targeted encoder bitrate" + ); + *bitrate_ack.lock().unwrap() = Some(ack.bitrate_kbps); } else { tracing::warn!("unknown control message — ignoring"); } @@ -1417,6 +1450,18 @@ async fn worker_main(args: WorkerArgs) { const ADAPT_REPORT_INTERVAL: Duration = Duration::from_millis(750); let mut last_report = Instant::now(); let (mut last_recovered, mut last_received, mut last_dropped) = (0u64, 0u64, 0u64); + // Adaptive bitrate (see `crate::abr`): armed only when the embedder asked for Automatic + // (`bitrate_kbps == 0`) and the host echoed the rate it actually configured (an old host + // echoes 0 → controller stays permanently off). Fed once per report window with the same + // deltas the LossReport uses, plus the window's mean skew-corrected one-way delay and + // whether a jump-to-live flush fired. + let mut abr = BitrateController::new(if bitrate_kbps == 0 { + resolved_bitrate_kbps + } else { + 0 + }); + let (mut owd_sum_ns, mut owd_frames) = (0i128, 0u32); + let mut flush_in_window = false; // Jump-to-live state (see the guard in the loop below): the clock-based over-bound run // (`stale_frames`, armed only when the skew handshake succeeded so the clocks are comparable), // the clock-free non-draining-queue run (`standing_frames`), and the last-jump instant for the @@ -1443,12 +1488,33 @@ async fn worker_main(args: WorkerArgs) { p.active && !p.done }; if !probe_active && last_report.elapsed() >= ADAPT_REPORT_INTERVAL { + let window_dropped = st.frames_dropped.wrapping_sub(last_dropped); let loss_ppm = window_loss_ppm( st.fec_recovered_shards.wrapping_sub(last_recovered), st.packets_received.wrapping_sub(last_received), - st.frames_dropped.wrapping_sub(last_dropped), + window_dropped, ); let _ = ctrl_tx.send(CtrlRequest::Loss(LossReport { loss_ppm })); + // Adaptive bitrate: drain any host ack first (its clamp is authoritative), then + // feed the controller this window's congestion signals; a decision becomes a + // SetBitrate on the control stream. + if let Some(acked) = bitrate_ack.lock().unwrap().take() { + abr.on_ack(acked); + } + let owd_mean_us = + (owd_frames > 0).then(|| (owd_sum_ns / owd_frames as i128 / 1000) as i64); + (owd_sum_ns, owd_frames) = (0, 0); + if let Some(kbps) = abr.on_window( + Instant::now(), + window_dropped, + loss_ppm, + owd_mean_us, + flush_in_window, + ) { + tracing::info!(kbps, "adaptive bitrate: requesting encoder re-target"); + let _ = ctrl_tx.send(CtrlRequest::SetBitrate(kbps)); + } + flush_in_window = false; last_report = Instant::now(); last_recovered = st.fec_recovered_shards; last_received = st.packets_received; @@ -1486,6 +1552,13 @@ async fn worker_main(args: WorkerArgs) { } else { 0 }; + // Feed the adaptive-bitrate controller's OWD window (mean capture→received + // delay): rising delay under zero loss is queue growth — the pre-loss + // congestion signal. Only meaningful with a clock handshake. + if clock_offset_ns != 0 && lat_ns > 0 { + owd_sum_ns += lat_ns; + owd_frames += 1; + } if clock_offset_ns != 0 && lat_ns > FLUSH_LATENCY.as_nanos() as i128 { stale_frames += 1; } else { @@ -1505,6 +1578,7 @@ async fn worker_main(args: WorkerArgs) { stale_frames = 0; standing_frames = 0; last_flush = Some(Instant::now()); + flush_in_window = true; // strongest "link can't hold the rate" signal let flushed = session.flush_backlog().unwrap_or(0); let dropped = frames.clear(); let _ = ctrl_tx.send(CtrlRequest::Keyframe); @@ -1543,6 +1617,23 @@ async fn worker_main(args: WorkerArgs) { conn.close(close_code.into(), b"client closed"); } +#[cfg(test)] +mod host_port_tests { + use super::join_host_port; + + #[test] + fn brackets_bare_ipv6_only() { + assert_eq!(join_host_port("192.168.1.9", 4770), "192.168.1.9:4770"); + assert_eq!(join_host_port("myhost", 4770), "myhost:4770"); + assert_eq!(join_host_port("fd00::1", 4770), "[fd00::1]:4770"); + assert_eq!(join_host_port("[fd00::1]", 4770), "[fd00::1]:4770"); + // The bracketed form is what SocketAddr's parser actually accepts. + assert!(join_host_port("fd00::1", 4770) + .parse::() + .is_ok()); + } +} + #[cfg(test)] mod frame_channel_tests { use super::{FrameChannel, FramePop, FRAME_QUEUE_HARD_CAP}; diff --git a/crates/punktfunk-core/src/lib.rs b/crates/punktfunk-core/src/lib.rs index 60c201ee..f4c77936 100644 --- a/crates/punktfunk-core/src/lib.rs +++ b/crates/punktfunk-core/src/lib.rs @@ -25,6 +25,8 @@ #![forbid(unsafe_op_in_unsafe_fn)] pub mod abi; +#[cfg(feature = "quic")] +mod abr; pub mod audio; #[cfg(feature = "quic")] pub mod client; diff --git a/crates/punktfunk-core/src/quic.rs b/crates/punktfunk-core/src/quic.rs index ff739047..f2508e54 100644 --- a/crates/punktfunk-core/src/quic.rs +++ b/crates/punktfunk-core/src/quic.rs @@ -369,6 +369,32 @@ pub struct LossReport { pub loss_ppm: u32, } +/// `client → host`, any time after [`Start`]: reconfigure the encoder to a new target bitrate +/// without reconnecting — the mid-stream lever of adaptive bitrate. The host clamps the request +/// exactly like [`Hello::bitrate_kbps`] (its `[MIN, MAX]` band; `0` → host default), answers with +/// [`BitrateChanged`] carrying the value it actually configured, and rebuilds the encoder in +/// place at the same mode — the first new-rate frame is an IDR with in-band parameter sets, which +/// every client decoder already follows (same discipline as a [`Reconfigure`] mode switch). +/// +/// Sent by the client's automatic-bitrate controller (active when the user's bitrate setting is +/// "Automatic", i.e. `Hello::bitrate_kbps == 0`) when the link can't sustain the current rate — +/// or can sustain more again. A host that predates this ignores it (unknown control message) and +/// never answers; the client's controller detects the silence and disables itself. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct SetBitrate { + /// Requested encoder bitrate in kilobits per second (`0` = host default, like Hello's field). + pub bitrate_kbps: u32, +} + +/// `host → client`: answer to [`SetBitrate`] — the bitrate the host actually configured (the +/// request clamped to its supported band). The encoder switches on the next frame (an IDR); the +/// stream never pauses. Also the controller's liveness signal: no answer ⇒ an old host that +/// doesn't renegotiate bitrate. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct BitrateChanged { + pub bitrate_kbps: u32, +} + /// `client → host`, any time after [`Start`]: run a bandwidth speed test. The host bursts /// filler access units (flagged [`crate::packet::FLAG_PROBE`]) over the data plane at /// `target_kbps` of application goodput for `duration_ms`, *pausing video for the duration*, then @@ -455,6 +481,10 @@ pub const MSG_RECONFIGURED: u8 = 0x02; pub const MSG_REQUEST_KEYFRAME: u8 = 0x03; /// Type byte of [`LossReport`]. pub const MSG_LOSS_REPORT: u8 = 0x04; +/// Type byte of [`SetBitrate`]. +pub const MSG_SET_BITRATE: u8 = 0x05; +/// Type byte of [`BitrateChanged`]. +pub const MSG_BITRATE_CHANGED: u8 = 0x06; /// Type byte of [`ProbeRequest`]. pub const MSG_PROBE_REQUEST: u8 = 0x20; /// Type byte of [`ProbeResult`]. @@ -1128,6 +1158,46 @@ impl LossReport { } } +impl SetBitrate { + pub fn encode(&self) -> Vec { + // magic[0..4] type[4] bitrate_kbps[5..9] + let mut b = Vec::with_capacity(9); + b.extend_from_slice(CTL_MAGIC); + b.push(MSG_SET_BITRATE); + b.extend_from_slice(&self.bitrate_kbps.to_le_bytes()); + b + } + + pub fn decode(b: &[u8]) -> Result { + if b.len() != 9 || &b[0..4] != CTL_MAGIC || b[4] != MSG_SET_BITRATE { + return Err(PunktfunkError::InvalidArg("bad SetBitrate")); + } + Ok(SetBitrate { + bitrate_kbps: u32::from_le_bytes(b[5..9].try_into().unwrap()), + }) + } +} + +impl BitrateChanged { + pub fn encode(&self) -> Vec { + // magic[0..4] type[4] bitrate_kbps[5..9] + let mut b = Vec::with_capacity(9); + b.extend_from_slice(CTL_MAGIC); + b.push(MSG_BITRATE_CHANGED); + b.extend_from_slice(&self.bitrate_kbps.to_le_bytes()); + b + } + + pub fn decode(b: &[u8]) -> Result { + if b.len() != 9 || &b[0..4] != CTL_MAGIC || b[4] != MSG_BITRATE_CHANGED { + return Err(PunktfunkError::InvalidArg("bad BitrateChanged")); + } + Ok(BitrateChanged { + bitrate_kbps: u32::from_le_bytes(b[5..9].try_into().unwrap()), + }) + } +} + /// Compute a [`LossReport`] `loss_ppm` from one window's session-stat deltas: shards FEC recovered /// (the loss it absorbed), shards received, and frames that went unrecoverable. Loss ≈ recovered / /// (received + recovered) — the fraction of shards that arrived missing. A frame drop means loss @@ -2684,6 +2754,23 @@ mod tests { assert!(window_loss_ppm(u64::MAX, 1, 9) <= 1_000_000); } + #[test] + fn bitrate_messages_roundtrip() { + let req = SetBitrate { + bitrate_kbps: 14_000, + }; + assert_eq!(SetBitrate::decode(&req.encode()).unwrap(), req); + let ack = BitrateChanged { + bitrate_kbps: 14_000, + }; + assert_eq!(BitrateChanged::decode(&ack.encode()).unwrap(), ack); + // Same payload shape as LossReport — the type byte alone must keep them disjoint. + assert!(LossReport::decode(&req.encode()).is_err()); + assert!(SetBitrate::decode(&ack.encode()).is_err()); + assert!(BitrateChanged::decode(&req.encode()).is_err()); + assert!(SetBitrate::decode(&LossReport { loss_ppm: 7 }.encode()).is_err()); + } + #[test] fn probe_messages_roundtrip() { let req = ProbeRequest { diff --git a/crates/punktfunk-host/src/punktfunk1.rs b/crates/punktfunk-host/src/punktfunk1.rs index 2cd6ec6a..0e747e36 100644 --- a/crates/punktfunk-host/src/punktfunk1.rs +++ b/crates/punktfunk-host/src/punktfunk1.rs @@ -32,14 +32,14 @@ use punktfunk_core::config::{ use punktfunk_core::input::{InputEvent, InputKind}; use punktfunk_core::packet::{FLAG_PIC, FLAG_PROBE, FLAG_SOF}; use punktfunk_core::quic::{ - endpoint, io, ClockEcho, ClockProbe, ColorInfo, Hello, LossReport, PairChallenge, PairProof, - PairRequest, PairResult, ProbeRequest, ProbeResult, Reconfigure, Reconfigured, RequestKeyframe, - Start, Welcome, + endpoint, io, BitrateChanged, ClockEcho, ClockProbe, ColorInfo, Hello, LossReport, + PairChallenge, PairProof, PairRequest, PairResult, ProbeRequest, ProbeResult, Reconfigure, + Reconfigured, RequestKeyframe, SetBitrate, Start, Welcome, }; use punktfunk_core::transport::UdpTransport; use punktfunk_core::Session; use rand::RngCore; -use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering}; use std::sync::Arc; #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -1054,6 +1054,7 @@ async fn serve_session( // (inbound requests, outbound probe results) are multiplexed with `select!`. let (reconfig_tx, reconfig_rx) = std::sync::mpsc::channel::(); let (keyframe_tx, keyframe_rx) = std::sync::mpsc::channel::<()>(); + let (bitrate_tx, bitrate_rx) = std::sync::mpsc::channel::(); let (probe_tx, probe_rx) = std::sync::mpsc::channel::(); let (probe_result_tx, mut probe_result_rx) = tokio::sync::mpsc::unbounded_channel::(); @@ -1124,6 +1125,27 @@ async fn serve_session( ); } } + } else if let Ok(req) = SetBitrate::decode(&msg) { + // Mid-stream bitrate renegotiation (adaptive bitrate): clamp exactly like + // the Hello request, ack the resolved value, then hand it to the data-plane + // thread, which rebuilds the encoder in place at the same mode — the fresh + // encoder's first frame is an IDR with in-band parameter sets, so the + // client's decoder follows without a reconnect. + let resolved = resolve_bitrate_kbps(req.bitrate_kbps); + tracing::info!( + requested_kbps = req.bitrate_kbps, + resolved_kbps = resolved, + "mid-stream bitrate change requested" + ); + let ack = BitrateChanged { + bitrate_kbps: resolved, + }; + if io::write_msg(&mut ctrl_send, &ack.encode()).await.is_err() { + break; + } + if bitrate_tx.send(resolved).is_err() { + break; // data plane gone + } } else if let Ok(req) = ProbeRequest::decode(&msg) { tracing::info!( target_kbps = req.target_kbps, @@ -1445,6 +1467,7 @@ async fn serve_session( quit: quit_stream, reconfig: reconfig_rx, keyframe: keyframe_rx, + bitrate_rx, compositor, bitrate_kbps, bit_depth, @@ -2738,7 +2761,9 @@ struct SendStats { fps: u32, codec: &'static str, client: String, - bitrate_kbps: u32, + /// Live encoder bitrate (kbps) — the capture thread updates it on a mid-stream adaptive + /// bitrate change, so the web-console sample reports what the encoder is ACTUALLY targeting. + bitrate_kbps: Arc, } #[allow(clippy::too_many_arguments)] @@ -2921,7 +2946,7 @@ fn send_loop( fps: (new_frames as f64 / secs) as f32, repeat_fps: (repeat_frames as f64 / secs) as f32, mbps: tx_mbps as f32, - bitrate_kbps: stats.bitrate_kbps, + bitrate_kbps: stats.bitrate_kbps.load(Ordering::Relaxed), frames_dropped: s.frames_dropped.saturating_sub(last_frames_dropped) as u32, packets_dropped: s.packets_dropped.saturating_sub(last_packets_dropped) as u32, send_dropped: s.packets_send_dropped.saturating_sub(last_send_dropped) as u32, @@ -3078,6 +3103,9 @@ struct SessionContext { reconfig: std::sync::mpsc::Receiver, /// Client decode-recovery keyframe requests. keyframe: std::sync::mpsc::Receiver<()>, + /// Accepted mid-stream bitrate changes (adaptive bitrate, already clamped) — the encoder + /// alone is rebuilt in place at the new rate; capture + virtual output are untouched. + bitrate_rx: std::sync::mpsc::Receiver, /// The resolved compositor backend (moot on Windows — `vdisplay::open` ignores it there). compositor: crate::vdisplay::Compositor, /// Negotiated encoder bitrate (kbps). @@ -3137,8 +3165,9 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { quit, reconfig, keyframe, + bitrate_rx, compositor, - bitrate_kbps, + mut bitrate_kbps, bit_depth, // The resolved chroma is already captured in `plan` (above); ignore the duplicate here. chroma: _, @@ -3236,6 +3265,9 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { // The bounded channel applies backpressure (the encode thread blocks if the send falls behind, // so frames slow down rather than a dropped frame freezing the infinite-GOP stream). let (frame_tx, frame_rx) = std::sync::mpsc::sync_channel::(3); + // Live encoder bitrate, shared with the send thread's stats sample: a mid-stream adaptive + // bitrate change (bitrate_rx below) updates it so the console shows the actual target. + let live_bitrate = Arc::new(AtomicU32::new(bitrate_kbps)); // The send thread emits the web-console stats sample (it owns `session.stats()`); clone the // recorder so the capture loop keeps its own handle for the per-frame `is_armed()` gate. let send_stats = SendStats { @@ -3245,7 +3277,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { fps: mode.refresh_hz, codec: plan.codec.label(), client: client_label, - bitrate_kbps, + bitrate_kbps: live_bitrate.clone(), }; let send_thread = std::thread::Builder::new() .name("punktfunk-send".into()) @@ -3447,6 +3479,51 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { } } } + // Adaptive bitrate: drain to the NEWEST requested rate (the client's controller may step + // several times while we stream) and rebuild the ENCODER ONLY in place — the mode didn't + // change, so capture and the virtual output are untouched and the switch costs exactly the + // IDR the fresh encoder opens with (the same resync discipline as a mode switch, minus the + // pipeline churn). Rates arrive pre-clamped by the control task (`resolve_bitrate_kbps`). + let mut want_kbps = None; + while let Ok(k) = bitrate_rx.try_recv() { + want_kbps = Some(k); + } + if let Some(new_kbps) = want_kbps.filter(|&k| k != bitrate_kbps) { + // `interval` was built as 1/effective_hz, so the round-trip recovers the integer rate. + let hz = (1.0 / interval.as_secs_f64()).round() as u32; + match crate::encode::open_video( + plan.codec, + frame.format, + frame.width, + frame.height, + hz, + new_kbps as u64 * 1000, + frame.is_cuda(), + bit_depth, + plan.chroma, + ) { + Ok(new_enc) => { + tracing::info!( + from_kbps = bitrate_kbps, + to_kbps = new_kbps, + "encoder rebuilt at new bitrate (adaptive bitrate)" + ); + enc = new_enc; + bitrate_kbps = new_kbps; + live_bitrate.store(new_kbps, Ordering::Relaxed); + // The owed AUs died with the old encoder — same bookkeeping as a mode-switch + // rebuild; the fresh encoder opens on an IDR, so anchor the IDR cooldown too. + inflight.clear(); + last_au_at = std::time::Instant::now(); + encoder_resets = 0; + last_forced_idr = Some(std::time::Instant::now()); + } + Err(e) => { + tracing::error!(error = %format!("{e:#}"), to_kbps = new_kbps, + "bitrate-change encoder rebuild failed — keeping the current rate"); + } + } + } // Client recovery: it asked for a fresh IDR (its decoder wedged on the cold opening // GOP). Coalesce the backlog — several requests fire before the IDR lands — and force // the next encoded frame to be a keyframe. (A reconfig rebuild above already opens with diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index fffeaee4..c757de4f 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -356,6 +356,16 @@ #define MSG_LOSS_REPORT 4 #endif +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Type byte of [`SetBitrate`]. +#define MSG_SET_BITRATE 5 +#endif + +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Type byte of [`BitrateChanged`]. +#define MSG_BITRATE_CHANGED 6 +#endif + #if defined(PUNKTFUNK_FEATURE_QUIC) // Type byte of [`ProbeRequest`]. #define MSG_PROBE_REQUEST 32