diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index 28b8afde..5c56b35e 100644 --- a/apps/neshub/src/main.rs +++ b/apps/neshub/src/main.rs @@ -73,6 +73,16 @@ struct Args { #[arg(long, env = "NESTRI_AUDIO_BITRATE", default_value_t = 64)] audio_bitrate_per_channel: u32, + /// Ceiling on the video bitrate, in kbps. + /// + /// Set by `nesinit` from the boot descriptor's video limits, which come + /// from the tier the box was sized for. Absent means nobody said -- which is + /// not a licence to send whatever the encoder defaults to, since that is + /// precisely how every session came to offer 10 Mbps regardless of what the + /// path could carry. Unset is reported, and a conservative ceiling is used. + #[arg(long, env = "NESTRI_MAX_BITRATE")] + max_bitrate_kbps: Option, + /// Socket nescope sends screenshots on. neshub listens; nescope dials out. #[arg( long, @@ -82,6 +92,14 @@ struct Args { screenshot_ipc: PathBuf, } +/// What to assume when nobody said. +/// +/// Deliberately modest. A ceiling that was never set should not behave like an +/// unlimited one: the whole failure this exists to fix was a session offering +/// 10 Mbps into a path carrying under three, because no number had ever been +/// chosen and the encoder's own default stood in for one. +const DEFAULT_MAX_BITRATE_KBPS: u32 = 4_000; + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -117,6 +135,15 @@ async fn main() -> Result<()> { } } + match args.max_bitrate_kbps { + Some(kbps) => tracing::info!("video ceiling: {kbps} kbps, from the boot descriptor"), + None => tracing::warn!( + "no video ceiling on the boot descriptor; using {DEFAULT_MAX_BITRATE_KBPS} kbps. \ + A box sized by a tier is told its ceiling -- if this is one, the descriptor did \ + not carry it." + ), + } + let endpoint = builder.bind().await?; let endpoint_addr = endpoint.addr(); let ep_id = endpoint_addr.id; diff --git a/apps/nesinit/src/services.rs b/apps/nesinit/src/services.rs index 3d7ee018..9dcd685e 100644 --- a/apps/nesinit/src/services.rs +++ b/apps/nesinit/src/services.rs @@ -41,6 +41,7 @@ use tokio::sync::mpsc::{Receiver, Sender}; use crate::reap::{Waiters, Watched}; use crate::workload::Failure; +use nesprotocol::lifecycle::VideoLimits; /// A service that died, and how. #[derive(Debug, Clone, PartialEq, Eq)] @@ -60,7 +61,13 @@ pub trait Services { /// Called once, after the shares are mounted and before anything may be /// launched. An empty stack is legitimate: a box with no services still /// boots, and a caller can still launch something that needs none. - fn bring_up(&mut self) -> Result, Failure>; + /// + /// `video` comes from the descriptor and reaches the services that read it. + /// It has to arrive here rather than later because a service configured + /// after it is already running has a window in which it is not configured, + /// and for a bitrate ceiling that window is a session streaming at whatever + /// default it started with. + fn bring_up(&mut self, video: VideoLimits) -> Result, Failure>; /// Deaths, as they happen. /// @@ -308,6 +315,13 @@ pub struct Stack { running: Vec<(&'static str, Watched)>, deaths: Receiver, reported: Sender, + /// What the host said this box may spend on video, from the descriptor. + /// + /// Held here because `spawn` is where it reaches a service, and `spawn` + /// takes a `&'static Service` whose `env` is a fixed table -- a value that + /// arrives at runtime has no route through it otherwise. The same problem + /// `RUST_LOG` has, solved the same way. + video: VideoLimits, } impl Stack { @@ -328,6 +342,7 @@ impl Stack { running: Vec::new(), deaths, reported, + video: VideoLimits::default(), } } @@ -390,6 +405,15 @@ impl Stack { if let Ok(filter) = std::env::var("RUST_LOG") { command.env("RUST_LOG", filter); } + // The descriptor's video limits, for the services that read them. Same + // shape of problem as `RUST_LOG` above -- `env_clear` drops everything + // and the service table is a fixed list of literals, so a value that + // only exists at runtime has no other route in. `neshub` reads this + // through the clap `env =` attribute it already uses for every other + // setting. + if let Some(kbps) = self.video.bitrate_kbps { + command.env("NESTRI_MAX_BITRATE", kbps.to_string()); + } // The service's own entry last, so a service that states one of these // for itself wins over the defaults above. command.envs(service.env.iter().copied()); @@ -451,7 +475,8 @@ impl Stack { } impl Services for Stack { - fn bring_up(&mut self) -> Result, Failure> { + fn bring_up(&mut self, video: VideoLimits) -> Result, Failure> { + self.video = video; let mut up = Vec::new(); // Lifted out so the loop does not hold a borrow of `self` across the // start it is asking for. @@ -604,6 +629,9 @@ pub mod double { /// only thing under test. pub struct Double { pub brought_up: usize, + /// What the last `bring_up` was told, so a test can assert the limits + /// reached the stack rather than assuming they did. + pub video: VideoLimits, pub failure: Option, pub names: Vec, deaths: Receiver, @@ -625,6 +653,7 @@ pub mod double { names: vec!["dbus-system".into(), "neshub".into()], deaths, report, + video: VideoLimits::default(), } } @@ -637,8 +666,9 @@ pub mod double { } impl Services for Double { - fn bring_up(&mut self) -> Result, Failure> { + fn bring_up(&mut self, video: VideoLimits) -> Result, Failure> { self.brought_up += 1; + self.video = video; match &self.failure { Some(failure) => Err(failure.clone()), None => Ok(self.names.clone()), diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index 9cdda4f9..c71262c7 100644 --- a/apps/nesinit/src/session.rs +++ b/apps/nesinit/src/session.rs @@ -311,7 +311,7 @@ where // A box whose own services will not come up cannot be launched // into, so this is refused rather than reported and carried on // from — unlike a launch, which is the caller's to correct. - match services.bring_up() { + match services.bring_up(descriptor.video) { Ok(up) => { tracing::info!(services = up.len(), "the box is ready to be launched into"); send(&mut writer, &GuestToHost::Initialized { services: up }).await? @@ -517,6 +517,7 @@ mod tests { ro: false, }], drives: Vec::new(), + video: Default::default(), } } @@ -715,6 +716,53 @@ mod tests { ); } + #[tokio::test] + async fn the_descriptors_video_limits_reach_the_services() { + // The ceiling is useless if it stops at the descriptor. `neshub` is the + // only thing that can enforce it and it is a service, so the number has + // to survive the whole way from the boot document to the spawn. + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let session = spawn(guest, Given::new(Double::exits_when_stopped(Exit::code(0)))); + + let mut given = descriptor(); + given.video.bitrate_kbps = Some(8_000); + + caller.expect_ready().await; + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(given), + }) + .await; + caller.expect_booted().await; + caller.say(&HostToGuest::Shutdown).await; + + let (_, _, services) = session.await.unwrap(); + assert_eq!(services.video.bitrate_kbps, Some(8_000)); + } + + #[tokio::test] + async fn a_box_told_nothing_about_video_says_so_rather_than_inventing_a_limit() { + // "Unsaid" must not arrive as a number. A stack that cannot tell the two + // apart cannot log that it was never told, and a ceiling nobody set is + // exactly how every session came to offer 10 Mbps. + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let session = spawn(guest, Given::new(Double::exits_when_stopped(Exit::code(0)))); + + caller.expect_ready().await; + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .await; + caller.expect_booted().await; + caller.say(&HostToGuest::Shutdown).await; + + let (_, _, services) = session.await.unwrap(); + assert_eq!(services.video.bitrate_kbps, None); + } + #[tokio::test] async fn a_launch_runs_what_it_names_and_is_reported_by_its_id() { let (guest, host) = tokio::io::duplex(4096); diff --git a/apps/nesinit/tests/services_stop.rs b/apps/nesinit/tests/services_stop.rs index 9859a186..f4cb1ea2 100644 --- a/apps/nesinit/tests/services_stop.rs +++ b/apps/nesinit/tests/services_stop.rs @@ -50,7 +50,7 @@ fn alive(pid: i32) -> bool { async fn a_stack_that_goes_away_takes_its_services_with_it() { let waiters = Waiters::new(); let mut stack = Stack::from_table(waiters, SLEEPERS); - let up = stack.bring_up().expect("two sleeps did not start"); + let up = stack.bring_up(Default::default()).expect("two sleeps did not start"); assert_eq!(up.len(), 2); let pids = stack.pids(); diff --git a/crates/nesprotocol/src/lib.rs b/crates/nesprotocol/src/lib.rs index f4bfd1ec..d78e29ec 100644 --- a/crates/nesprotocol/src/lib.rs +++ b/crates/nesprotocol/src/lib.rs @@ -40,6 +40,17 @@ pub const FRAME_HDR_LEN: usize = 7; pub const MSG_DATA: u8 = 0; // generic data frame (video / audio) pub const MSG_IDR_REQUEST: u8 = 0x10; // request a keyframe (desktop → hub → hudless) pub const MSG_ENCODE_SETTINGS: u8 = 0x12; // change encoder settings (desktop → hub → hudless) +/// What the receiver actually got, once a second (desktop → hub). +/// +/// The hub cannot see this. Its own view of the path -- RTT, congestion window, +/// whether a datagram send returned an error -- was measured saying the path was +/// healthy while the client was receiving almost nothing, and one reason is +/// structural: `send_datagram` evicts the oldest queued datagrams and returns +/// `Ok`, so the send side has no backpressure signal at all. Only the far end +/// knows what arrived. +pub const MSG_RECEIVER_REPORT: u8 = 0x13; +/// Who decides the bitrate, and the ceiling to decide within (desktop → hub). +pub const MSG_CONTROL_MODE: u8 = 0x14; pub const MSG_INPUT_BATCH: u8 = 0xFE; // batched input events (desktop → hub) /// Build a frame body: `[u8 type] [u16 LE seq] [payload]`. @@ -224,3 +235,242 @@ pub fn decode_encode_settings(payload: &[u8]) -> Option<(u8, u8, u32, Option }; Some((codec_id, rc, value, depth)) } + +// ── Receiver report ───────────────────────────────────────────── + +/// What one second looked like from the receiving end. +/// +/// Counts are per-second deltas, not totals: a controller wants to know what is +/// happening now, and a total makes every reading depend on how long the session +/// has been running. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct ReceiverReport { + /// Video bits per second actually reassembled and released to the decoder. + /// + /// Not what was sent, and not what arrived -- what *completed*. A frame + /// missing one fragment contributes nothing here, which is right: it + /// contributed nothing to the picture either. When the sender is saturating + /// the path this is the measured capacity of it. + pub goodput_bps: u64, + /// Frames released to the decoder. + pub released: u32, + /// Frames that started arriving and never completed. + pub incomplete: u32, + /// Frames no fragment of which ever arrived. + pub never_arrived: u32, + /// The receiver's own round-trip estimate, in milliseconds. + pub rtt_ms: u32, +} + +impl ReceiverReport { + /// The fraction of frames that did not make it, in `0.0..=1.0`. + /// + /// `None` when no frames were accounted for at all, which is not the same + /// as no loss -- a second in which nothing was sent and a second in which + /// nothing arrived look identical here, and only the caller knows which it + /// is expecting. + pub fn loss(&self) -> Option { + let total = self.released + self.incomplete + self.never_arrived; + if total == 0 { + return None; + } + Some((self.incomplete + self.never_arrived) as f32 / total as f32) + } +} + +/// `[8B goodput_bps][4B released][4B incomplete][4B never_arrived][4B rtt_ms]`, +/// all little-endian. +pub const RECEIVER_REPORT_LEN: usize = 24; + +pub fn encode_receiver_report(buf: &mut Vec, report: &ReceiverReport) { + buf.reserve(RECEIVER_REPORT_LEN); + buf.extend_from_slice(&report.goodput_bps.to_le_bytes()); + buf.extend_from_slice(&report.released.to_le_bytes()); + buf.extend_from_slice(&report.incomplete.to_le_bytes()); + buf.extend_from_slice(&report.never_arrived.to_le_bytes()); + buf.extend_from_slice(&report.rtt_ms.to_le_bytes()); +} + +/// Decode a receiver report. `None` when the payload is short. +/// +/// A payload *longer* than expected is accepted and its tail ignored, so a newer +/// client that appends a field still reports usefully to an older hub. +pub fn decode_receiver_report(payload: &[u8]) -> Option { + if payload.len() < RECEIVER_REPORT_LEN { + return None; + } + let u32_at = |o: usize| u32::from_le_bytes(payload[o..o + 4].try_into().unwrap()); + Some(ReceiverReport { + goodput_bps: u64::from_le_bytes(payload[0..8].try_into().unwrap()), + released: u32_at(8), + incomplete: u32_at(12), + never_arrived: u32_at(16), + rtt_ms: u32_at(20), + }) +} + +// ── Control mode ──────────────────────────────────────────────── + +/// Who is choosing the bitrate. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum ControlMode { + /// The hub's controller decides, within the ceiling. + #[default] + Auto, + /// A person decided, and the controller stands down until told otherwise. + /// + /// Kept because it is how this class of bug gets diagnosed at all: the + /// original "the bitrate is already lowered" report was wrong, and the only + /// way anyone established that was by setting one by hand and watching the + /// picture come back. + Manual, +} + +pub const CONTROL_MODE_AUTO: u8 = 0; +pub const CONTROL_MODE_MANUAL: u8 = 1; + +/// `[1B mode][4B ceiling_kbps LE]`. A ceiling of 0 means "no opinion, keep +/// whatever the hub was given". +pub const CONTROL_MODE_LEN: usize = 5; + +pub fn encode_control_mode(buf: &mut Vec, mode: ControlMode, ceiling_kbps: u32) { + buf.reserve(CONTROL_MODE_LEN); + buf.push(match mode { + ControlMode::Auto => CONTROL_MODE_AUTO, + ControlMode::Manual => CONTROL_MODE_MANUAL, + }); + buf.extend_from_slice(&ceiling_kbps.to_le_bytes()); +} + +/// Returns `(mode, ceiling_kbps)`; the ceiling is `None` when it was left at 0. +pub fn decode_control_mode(payload: &[u8]) -> Option<(ControlMode, Option)> { + if payload.len() < CONTROL_MODE_LEN { + return None; + } + let mode = match payload[0] { + CONTROL_MODE_AUTO => ControlMode::Auto, + CONTROL_MODE_MANUAL => ControlMode::Manual, + // An unknown mode is not a reason to stop controlling the bitrate, and + // guessing "manual" would silently disable the controller. + _ => return None, + }; + let ceiling = u32::from_le_bytes(payload[1..5].try_into().unwrap()); + Some((mode, (ceiling != 0).then_some(ceiling))) +} + +#[cfg(test)] +mod media_control_tests { + use super::*; + + fn report() -> ReceiverReport { + ReceiverReport { + goodput_bps: 2_850_000, + released: 47, + incomplete: 12, + never_arrived: 1, + rtt_ms: 182, + } + } + + #[test] + fn a_report_survives_the_wire() { + let mut buf = Vec::new(); + encode_receiver_report(&mut buf, &report()); + assert_eq!(buf.len(), RECEIVER_REPORT_LEN); + assert_eq!(decode_receiver_report(&buf), Some(report())); + } + + #[test] + fn a_short_report_is_refused_rather_than_guessed() { + let mut buf = Vec::new(); + encode_receiver_report(&mut buf, &report()); + for n in 0..RECEIVER_REPORT_LEN { + assert_eq!(decode_receiver_report(&buf[..n]), None, "{n} bytes"); + } + } + + #[test] + fn a_longer_report_is_read_and_its_tail_ignored() { + // So a newer client that appends a field still reports usefully to a + // hub that predates it. + let mut buf = Vec::new(); + encode_receiver_report(&mut buf, &report()); + buf.extend_from_slice(&[0xAA; 8]); + assert_eq!(decode_receiver_report(&buf), Some(report())); + } + + #[test] + fn loss_counts_every_frame_that_did_not_arrive_whole() { + // An incomplete frame is a lost frame. It cost bandwidth and produced no + // picture, which is worse than never having been sent. + let r = ReceiverReport { + released: 90, + incomplete: 8, + never_arrived: 2, + ..Default::default() + }; + assert_eq!(r.loss(), Some(0.1)); + } + + #[test] + fn a_silent_second_has_no_loss_figure() { + // Nothing sent and nothing arrived look identical from here. Reporting + // 0% would tell a controller the path is healthy; reporting 100% would + // tell it to collapse the bitrate. Neither is known, so neither is said. + assert_eq!(ReceiverReport::default().loss(), None); + } + + #[test] + fn total_loss_is_reported_as_total() { + let r = ReceiverReport { + released: 0, + incomplete: 46, + never_arrived: 14, + ..Default::default() + }; + assert_eq!(r.loss(), Some(1.0)); + } + + #[test] + fn a_control_mode_survives_the_wire() { + for (mode, ceiling) in [ + (ControlMode::Auto, 8_000u32), + (ControlMode::Manual, 1_000), + (ControlMode::Auto, 0), + ] { + let mut buf = Vec::new(); + encode_control_mode(&mut buf, mode, ceiling); + assert_eq!(buf.len(), CONTROL_MODE_LEN); + assert_eq!( + decode_control_mode(&buf), + Some((mode, (ceiling != 0).then_some(ceiling))), + ); + } + } + + #[test] + fn an_unknown_mode_is_refused_rather_than_defaulted() { + // Defaulting to manual would silently switch the controller off, which + // is the failure this whole change exists to remove. + let mut buf = vec![0x7F]; + buf.extend_from_slice(&8_000u32.to_le_bytes()); + assert_eq!(decode_control_mode(&buf), None); + } + + #[test] + fn the_new_message_types_do_not_collide() { + let all = [ + MSG_DATA, + MSG_IDR_REQUEST, + MSG_ENCODE_SETTINGS, + MSG_RECEIVER_REPORT, + MSG_CONTROL_MODE, + MSG_INPUT_BATCH, + ]; + for (i, a) in all.iter().enumerate() { + for b in &all[i + 1..] { + assert_ne!(a, b, "two message types share a value"); + } + } + } +} diff --git a/crates/nesprotocol/src/lifecycle.rs b/crates/nesprotocol/src/lifecycle.rs index 72a732a4..c2fd119a 100644 --- a/crates/nesprotocol/src/lifecycle.rs +++ b/crates/nesprotocol/src/lifecycle.rs @@ -179,6 +179,39 @@ pub struct BootDescriptor { pub mounts: Vec, #[serde(default)] pub drives: Vec, + /// What the box may spend on video. + /// + /// Here rather than on a launch because its consumer is `neshub`, which is + /// a service and comes up with the box. Geometry went the other way for the + /// same reason: its consumer is the compositor, which is started per launch. + /// A number travels to where the thing that reads it is started. + #[serde(default)] + pub video: VideoLimits, +} + +/// Ceilings on what a box's video may cost. +/// +/// A struct rather than a bare number so the next video-shaped limit joins it +/// instead of arriving loose alongside it. +/// +/// Note what `deny_unknown_fields` on [`BootDescriptor`] means for this: a host +/// that sends `video` to a guest too old to know the field is *refused*, not +/// quietly accepted. That is the intended direction of failure -- the +/// alternative is a box that comes up, streams, and ignores the ceiling it was +/// given, which is exactly the shape of failure this stack produces too easily. +/// The guest image is rebuilt before a host starts sending it. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct VideoLimits { + /// Ceiling on the video bitrate, in kbps. + /// + /// `None` means nobody said, which is not the same as zero and is not the + /// same as unlimited. A reader that was told nothing should say so and pick + /// a conservative default of its own; a reader that treats "unsaid" as + /// "unlimited" reproduces the bug this exists to fix, where every session + /// offered 10 Mbps because nothing had ever set a number. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub bitrate_kbps: Option, } /// How a workload ended. @@ -391,6 +424,7 @@ mod tests { ro: true, }], drives: Vec::new(), + video: VideoLimits::default(), } } @@ -633,3 +667,64 @@ mod tests { assert!(!on_exit.terminal); } } + +#[cfg(test)] +mod video_limits_tests { + use super::*; + + #[test] + fn a_descriptor_without_video_still_reads() { + // An older host says nothing about video. That has to keep working, and + // it has to be distinguishable from a host that said "no limit". + let d: BootDescriptor = serde_json::from_str(r#"{"mounts":[],"drives":[]}"#).unwrap(); + assert_eq!(d.video.bitrate_kbps, None); + } + + #[test] + fn an_unsaid_ceiling_is_not_serialised() { + // So a host that has nothing to say produces the same bytes it always + // did, and an older guest keeps accepting it. + let d = BootDescriptor { + mounts: Vec::new(), + drives: Vec::new(), + video: VideoLimits::default(), + }; + let json = serde_json::to_string(&d).unwrap(); + assert!(!json.contains("bitrate"), "{json}"); + } + + #[test] + fn a_ceiling_survives_the_round_trip() { + let d = BootDescriptor { + mounts: Vec::new(), + drives: Vec::new(), + video: VideoLimits { + bitrate_kbps: Some(8_000), + }, + }; + let back: BootDescriptor = + serde_json::from_str(&serde_json::to_string(&d).unwrap()).unwrap(); + assert_eq!(back, d); + } + + #[test] + fn zero_is_not_the_same_as_unsaid() { + // A reader that conflates them cannot tell "the host wants no video" from + // "the host never mentioned it", and the second must not be read as a + // licence to send whatever it likes. + let said: BootDescriptor = serde_json::from_str(r#"{"video":{"bitrate_kbps":0}}"#).unwrap(); + let unsaid: BootDescriptor = serde_json::from_str("{}").unwrap(); + assert_eq!(said.video.bitrate_kbps, Some(0)); + assert_eq!(unsaid.video.bitrate_kbps, None); + } + + #[test] + fn an_unknown_video_field_is_refused() { + // Same reasoning as the descriptor's own `deny_unknown_fields`: a limit + // this build does not understand is one it would otherwise ignore while + // reporting success. + let r: Result = + serde_json::from_str(r#"{"video":{"bitrate_kbps":8000,"fps_cap":30}}"#); + assert!(r.is_err(), "an unknown video limit must not be ignored"); + } +}