diff --git a/Cargo.lock b/Cargo.lock index 6cd030f7..8e12ff24 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2389,6 +2389,7 @@ dependencies = [ name = "nesprotocol" version = "0.1.0" dependencies = [ + "libc", "serde", "serde_json", ] diff --git a/apps/nescapture/src/encode.rs b/apps/nescapture/src/encode.rs index 6a54ffbd..566062a7 100644 --- a/apps/nescapture/src/encode.rs +++ b/apps/nescapture/src/encode.rs @@ -1080,7 +1080,6 @@ impl PipelineHandle { encode_ms: encode_avg_ms.clone(), dropped: dropped_frames.clone(), idr_requested: idr_requested.clone(), - epoch: Instant::now(), reconfig_tx: reconfig_tx.clone(), }; let ipc_prog = ipc_progress.clone(); @@ -2709,14 +2708,14 @@ fn bgra_to_yuv420(pixels: &[u8], width: u32, height: u32, vk_format: u32) -> Vec /// has: it only has to tell one frame from the next. fn pyro_messages( frame: &nespyro::EncodedFrame, - ts_ms: u32, + captured_us: u64, width: u16, height: u16, ) -> Result>, nesprotocol::pyrowave::ChunkError> { nesprotocol::pyrowave::chunk_frame( frame.index as u32, nesprotocol::pyrowave::FrameMeta { - ts_ms, + captured_us, width, height, }, @@ -2741,8 +2740,6 @@ struct IpcConfig { dropped: Arc, /// Shared with the encoder thread, which honours it on the next frame. idr_requested: Arc, - /// Zero point for wire timestamps. - epoch: Instant, /// For the rate probe to command its own steps. /// /// The probe has to live where the encoded sizes are, which is here, and @@ -2799,9 +2796,6 @@ fn ipc_send_thread( } }; - // Fixed before the first frame arrives, so every timestamp shares an epoch - // even though frames are stamped from their own present. - let start_time = cfg.epoch; let mut frame_count: u64 = 0; // So a PyroWave encoder short of scratch space is said once per episode. let mut scratch_overflowing = false; @@ -2920,9 +2914,11 @@ fn ipc_send_thread( // however long the frame spent queued for the encoder into the // timestamp, so the receiver could not tell capture time from // backlog and had nothing honest to pace on. - let timestamp_ms = present_time - .saturating_duration_since(start_time) - .as_millis() as u32; + // + // On the clock every media stamp on this machine shares, so the hub + // can measure capture-to-send exactly and the client can measure + // the path from it. + let captured_us = nesprotocol::clock::instant_us(present_time); // Set FLAG_RECONFIG on the first frame after an encoder reconfig. // Clear it after setting so only the first frame is marked. Taken // for a PyroWave frame too, which carries no flags, so that the @@ -2948,7 +2944,7 @@ fn ipc_send_thread( STREAM_VIDEO, protocol_codec, flags, - timestamp_ms, + captured_us, width, height, &pkt.data, @@ -2974,7 +2970,7 @@ fn ipc_send_thread( log::info!("PyroWave encoder has scratch space again"); } scratch_overflowing = overflowed; - match pyro_messages(frame, timestamp_ms, width, height) { + match pyro_messages(frame, captured_us, width, height) { // Every PyroWave frame stands alone, so none is a key // frame in the sense the counters below mean. Ok(messages) => (messages, false, frame.data.len()), @@ -4808,7 +4804,7 @@ mod pyrowave_tests { assert_eq!(got.frame, 41); assert_eq!(got.critical, 4); assert_eq!( - (got.meta.ts_ms, got.meta.width, got.meta.height), + (got.meta.captured_us, got.meta.width, got.meta.height), (1234, 2560, 1440) ); assert_eq!(got.packets.len(), frame.packets.len()); diff --git a/apps/neshub/examples/pyro_bench.rs b/apps/neshub/examples/pyro_bench.rs index cd5a5f9f..b12ebc87 100644 --- a/apps/neshub/examples/pyro_bench.rs +++ b/apps/neshub/examples/pyro_bench.rs @@ -102,11 +102,11 @@ async fn send(conn: Connection, o: &Opts) -> Result<()> { while Instant::now() < end { tick.tick().await; let t0 = Instant::now(); - f.meta.ts_ms = seq as u32; + f.meta.captured_us = u64::from(seq); // Read per frame, as the hub does: with several paths up, the limit // is the smallest of their MTUs and moves as paths come and go. let now_max = conn.max_datagram_size().context("datagrams went away")?; - let d = pack_datagrams(seq, &f, now_max)?; + let d = pack_datagrams(seq, &f, now_max, nesprotocol::clock::monotonic_us())?; let backlog = DGRAM_BUFFER_BYTES.saturating_sub(conn.datagram_send_buffer_space()); if should_skip(backlog, d.buf.len(), DGRAM_BUFFER_BYTES).is_some() { skipped += 1; diff --git a/apps/neshub/src/control.rs b/apps/neshub/src/control.rs index ad9bd58b..ab5fab3c 100644 --- a/apps/neshub/src/control.rs +++ b/apps/neshub/src/control.rs @@ -477,6 +477,7 @@ mod tests { incomplete, never_arrived: 0, rtt_ms: 180, + ..Default::default() } } diff --git a/apps/neshub/src/dgram.rs b/apps/neshub/src/dgram.rs index 3743210f..ac94d0a6 100644 --- a/apps/neshub/src/dgram.rs +++ b/apps/neshub/src/dgram.rs @@ -16,8 +16,8 @@ use crate::pyro::PyroSender; use nesprotocol::MSG_DATA; use nesprotocol::datagram::{ - DGRAM_BUFFER_BYTES, DGRAM_HDR_LEN, MAX_FRAGMENTS_PER_FRAME, MIN_DGRAM_PAYLOAD, fragment_count, - write_datagram_header, + DGRAM_BUFFER_BYTES, DGRAM_HDR_LEN, DatagramHeader, MAX_FRAGMENTS_PER_FRAME, MIN_DGRAM_PAYLOAD, + fragment_count, write_datagram_header, }; use nesprotocol::pyrowave::PyroFrame; use nesprotocol::reliable::video_wants_reliable; @@ -136,10 +136,22 @@ impl DatagramSender { // this buffer, so a 300-fragment keyframe costs one allocation, not 300. let mut buf = BytesMut::with_capacity(total * DGRAM_HDR_LEN + body.len()); let mut header = [0u8; DGRAM_HDR_LEN]; + // One reading for the whole frame: every fragment is handed over in the + // single call below, so they all leave this end at the same moment. + let sent_us = nesprotocol::clock::monotonic_us(); for index in 0..total { let start = index * payload_size; let end = (start + payload_size).min(body.len()); - write_datagram_header(&mut header, self.kind, seq, index as u16, total as u16); + write_datagram_header( + &mut header, + &DatagramHeader { + kind: self.kind, + seq, + index: index as u16, + total: total as u16, + sent_us, + }, + ); buf.extend_from_slice(&header); buf.extend_from_slice(&body[start..end]); } diff --git a/apps/neshub/src/e2e_tests.rs b/apps/neshub/src/e2e_tests.rs index d474c0a4..c9220844 100644 --- a/apps/neshub/src/e2e_tests.rs +++ b/apps/neshub/src/e2e_tests.rs @@ -42,7 +42,7 @@ pub(crate) fn synthetic_frame(id: u32, bytes: usize, packet: usize) -> PyroFrame PyroFrame { frame: id, meta: FrameMeta { - ts_ms: id * 16, + captured_us: u64::from(id) * 16_000, width: 1920, height: 1080, }, @@ -236,7 +236,7 @@ async fn frames_cross_the_hub_byte_for_byte() { for (i, (g, s)) in got.iter().zip(&sent).enumerate() { assert!(g.is_whole(), "frame {i}: {}/{}", g.received, g.total); assert_eq!(g.seq, i as u16); - assert_eq!(g.ts_ms, s.meta.ts_ms); + assert_eq!(g.captured_us, s.meta.captured_us); assert_eq!(packets(g), originals(s), "frame {i} changed in transit"); } } diff --git a/apps/neshub/src/ipc_listener.rs b/apps/neshub/src/ipc_listener.rs index c9b83731..1e9aae9f 100644 --- a/apps/neshub/src/ipc_listener.rs +++ b/apps/neshub/src/ipc_listener.rs @@ -86,13 +86,16 @@ pub async fn run_video_listener( session_manager .reassert_codec(decoded.codec, &controller, &cmd_tx) .await; - let mut frame_data = Vec::with_capacity(1 + decoded.data.len()); - frame_data.push(decoded.codec); - frame_data.push(decoded.flags); - frame_data.extend_from_slice(&decoded.timestamp_ms.to_le_bytes()); - frame_data.extend_from_slice(&decoded.width.to_le_bytes()); - frame_data.extend_from_slice(&decoded.height.to_le_bytes()); - frame_data.extend_from_slice(decoded.data); + let frame_data = nesprotocol::encode_video_payload( + &nesprotocol::VideoHead { + codec: decoded.codec, + flags: decoded.flags, + captured_us: decoded.stamp, + width: decoded.width, + height: decoded.height, + }, + decoded.data, + ); session_manager.broadcast_video(frame_data).await; } else { @@ -150,7 +153,9 @@ pub async fn run_audio_listener(socket_path: PathBuf, session_manager: Arc bool { } }; - if let Err(e) = send.write_all(&[STREAM_KEYFRAME, STREAM_VERSION]).await { + let header = keyframe_stream_header(nesprotocol::clock::monotonic_us()); + if let Err(e) = send.write_all(&header).await { debug!("keyframe header write failed: {e}"); send.reset(0u32.into()).ok(); return false; diff --git a/apps/neshub/src/pyro.rs b/apps/neshub/src/pyro.rs index 524c3fc7..00ca9c0f 100644 --- a/apps/neshub/src/pyro.rs +++ b/apps/neshub/src/pyro.rs @@ -44,13 +44,14 @@ impl PyroSender { debug!("{label}: peer does not support QUIC datagrams"); return; }; - let datagrams = match pack_datagrams(self.seq, frame, max) { - Ok(d) => d, - Err(e) => { - debug!("{label}: dropping PyroWave frame {}: {e}", frame.frame); - return; - } - }; + let datagrams = + match pack_datagrams(self.seq, frame, max, nesprotocol::clock::monotonic_us()) { + Ok(d) => d, + Err(e) => { + debug!("{label}: dropping PyroWave frame {}: {e}", frame.frame); + return; + } + }; let backlog = DGRAM_BUFFER_BYTES.saturating_sub(self.conn.datagram_send_buffer_space()); if let Some(skip) = should_skip(backlog, datagrams.buf.len(), DGRAM_BUFFER_BYTES) { diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 9e5742bf..7beba930 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -899,12 +899,11 @@ pub struct SessionManager { last_keyframes: AtomicU64, /// How much this box's own pipeline delay varied, frame to frame. /// - /// **Variation, not absolute delay**, and it cannot be otherwise: the - /// encoder's `timestamp_ms` counts from its own start rather than from any - /// epoch, so the difference to wall clock holds an unknown constant even - /// though both run on this machine. Variation is also the comparable - /// quantity -- the client measures the same thing about the total, using the - /// same code, so the difference between the two is what the network added. + /// Variation above the pipeline's own best, measured with the same code the + /// client measures the whole path with, so the two are comparable and the + /// difference between them is what the network added. The encoder stamps + /// capture on the clock this hub reads at hand-off, so there is no offset + /// to remove here and no skew either. pipeline: std::sync::Mutex, audio_bytes: AtomicU64, last_audio_bytes: AtomicU64, @@ -1044,25 +1043,19 @@ impl SessionManager { /// Note how long this box took over a frame, relative to its own best. /// - /// `ts_ms` is stamped at capture by an encoder on this same machine, so the - /// clock is shared even though its origin is arbitrary -- which is exactly - /// what [`DelayTracker`] is built for. + /// The frame's capture time is stamped by an encoder on this same machine, + /// on the clock read here at hand-off (`nesprotocol::clock`). /// /// [`DelayTracker`]: nesprotocol::delay::DelayTracker fn note_pipeline_delay(&self, payload: &[u8]) { - let Some(ts_bytes) = payload.get(2..6) else { - return; - }; - self.note_pipeline_ts(u32::from_le_bytes(ts_bytes.try_into().unwrap())); + if let Some((head, _)) = nesprotocol::decode_video_payload(payload) { + self.note_pipeline_ts(head.captured_us); + } } - fn note_pipeline_ts(&self, ts_ms: u32) { - let now_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or_default(); + fn note_pipeline_ts(&self, captured_us: u64) { if let Ok(mut pipeline) = self.pipeline.lock() { - pipeline.observe(ts_ms, now_ms, std::time::Instant::now()); + pipeline.observe(captured_us, nesprotocol::clock::monotonic_us()); } } @@ -1073,12 +1066,16 @@ impl SessionManager { /// same as a second with no variation, but the wire has no room to say so /// and the client can tell from the frame counters beside it. pub fn pipeline_delays(&self) -> (u16, u16, u16) { + // The stats stream carries whole milliseconds; the measurement is finer. + fn ms(us: u32) -> u16 { + (us / 1000).min(u32::from(u16::MAX)) as u16 + } let Ok(mut pipeline) = self.pipeline.lock() else { return (0, 0, 0); }; pipeline .take() - .map(|s| (s.p50_ms, s.p95_ms, s.max_ms)) + .map(|s| (ms(s.p50_us), ms(s.p95_us), ms(s.max_us))) .unwrap_or((0, 0, 0)) } @@ -1108,7 +1105,7 @@ impl SessionManager { /// counting them as keyframes would report sixty keyframes a second, which /// describes nothing. pub async fn broadcast_pyrowave(&self, frame: PyroFrame) { - self.note_pipeline_ts(frame.meta.ts_ms); + self.note_pipeline_ts(frame.meta.captured_us); self.video_delta_bytes .fetch_add(frame.data.len() as u64, Ordering::Relaxed); let item = MediaItem::PyroWave(Arc::new(frame)); diff --git a/apps/neswire/src/encoder.rs b/apps/neswire/src/encoder.rs index a7eedcc1..f0ada001 100644 --- a/apps/neswire/src/encoder.rs +++ b/apps/neswire/src/encoder.rs @@ -2,7 +2,6 @@ use anyhow::{Result, bail}; use crossbeam_channel::Receiver; use opus_head_sys::*; use std::os::unix::net::UnixDatagram; -use std::time::Instant; use nesprotocol::{CODEC_OPUS, STREAM_AUDIO, encode_ipc_frame}; @@ -133,18 +132,19 @@ fn send_frame( encoder: &MsEncoder, socket: &UnixDatagram, config: &EncoderConfig, - frame: &[f32], - start: &Instant, + packet: &crate::sink::Packet, opus_buf: &mut Vec, ) -> Result<()> { - let encoded_len = encoder.encode_float(frame, config.frame_size, opus_buf)?; + let encoded_len = encoder.encode_float(&packet.samples, config.frame_size, opus_buf)?; let encoded_data = &opus_buf[..encoded_len]; - let timestamp_ms = start.elapsed().as_millis() as u32; + // The packet's place on the graph clock, not when it was encoded: the + // receiver needs the audio timeline, and encode time only says when this + // thread got to it. let ipc_frame = encode_ipc_frame( STREAM_AUDIO, CODEC_OPUS, 0, - timestamp_ms, + packet.position, 0, 0, encoded_data, @@ -153,7 +153,7 @@ fn send_frame( Ok(()) } -pub fn run(config: EncoderConfig, rx: Receiver>) -> Result<()> { +pub fn run(config: EncoderConfig, rx: Receiver) -> Result<()> { tracing::info!( "encoder started — IPC {}, {}ch, {} samples/frame", config.ipc_path, @@ -165,7 +165,6 @@ pub fn run(config: EncoderConfig, rx: Receiver>) -> Result<()> { let encoder = MsEncoder::create_surround(&config)?; let mut opus_buf = vec![0u8; 4000]; - let start = Instant::now(); // Connect to hub with retry loop { @@ -185,8 +184,8 @@ pub fn run(config: EncoderConfig, rx: Receiver>) -> Result<()> { } // Send loop - while let Ok(frame) = rx.recv() { - send_frame(&encoder, &socket, &config, &frame, &start, &mut opus_buf)?; + while let Ok(packet) = rx.recv() { + send_frame(&encoder, &socket, &config, &packet, &mut opus_buf)?; } tracing::info!("encoder shut down"); @@ -399,10 +398,13 @@ mod tests { let config = config(path.to_str().expect("utf-8 path")); let encoder = MsEncoder::create_surround(&config).expect("encoder"); - let start = Instant::now(); let mut opus_buf = vec![0u8; 4000]; + let packet = crate::sink::Packet { + position: 123_456, + samples: tone(0), + }; - send_frame(&encoder, &sender, &config, &tone(0), &start, &mut opus_buf).expect("send"); + send_frame(&encoder, &sender, &config, &packet, &mut opus_buf).expect("send"); let mut buf = vec![0u8; 65536]; let n = listener.recv(&mut buf).expect("recv"); @@ -410,6 +412,10 @@ mod tests { assert_eq!(frame.stream_type, STREAM_AUDIO); assert_eq!(frame.codec, CODEC_OPUS); + assert_eq!( + frame.stamp, 123_456, + "the graph position did not reach the hub" + ); assert!(!frame.data.is_empty(), "frame carried no payload"); // And the payload is Opus the far end can actually decode. diff --git a/apps/neswire/src/main.rs b/apps/neswire/src/main.rs index 67671758..86699804 100644 --- a/apps/neswire/src/main.rs +++ b/apps/neswire/src/main.rs @@ -36,7 +36,7 @@ fn main() -> anyhow::Result<()> { let sample_rate = 48_000u32; let frame_size = (sample_rate * args.packet_duration_ms) / 1000; - let (tx, rx) = crossbeam_channel::bounded::>(256); + let (tx, rx) = crossbeam_channel::bounded::(256); let encoder_config = encoder::EncoderConfig { channels: args.channels, diff --git a/apps/neswire/src/sink.rs b/apps/neswire/src/sink.rs index bbce068a..e9cd1b9c 100644 --- a/apps/neswire/src/sink.rs +++ b/apps/neswire/src/sink.rs @@ -16,7 +16,25 @@ fn missed_frames(expected: Option, now: Option) -> u64 { } } -pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender>) -> Result<()> { +/// One packet's worth of interleaved samples, and where it starts. +pub struct Packet { + /// The packet's first sample on the graph clock, in frames at the graph + /// rate. Consecutive packets are exactly a packet apart, missed cycles + /// included, because a missed cycle is filled (see `process`). + pub position: u64, + pub samples: Vec, +} + +/// Where the first of `held` frames sits, given where the last one ends. +/// +/// The accumulator is continuous up to `end`: a missed cycle is filled before +/// the next one is appended. So the frames in it are the `held` frames before +/// `end`, and the oldest is the next packet's start. +fn first_position(end: u64, held: u64) -> u64 { + end.saturating_sub(held) +} + +pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender) -> Result<()> { pw::init(); tracing::info!("pipewire init ok"); @@ -54,7 +72,7 @@ pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender let samples_per_frame = (frame_size * channels) as usize; struct State { - tx: Sender>, + tx: Sender, accum: Vec, samples_per_frame: usize, channels: usize, @@ -62,6 +80,9 @@ pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender /// The graph position the next cycle's data starts at, if the stream /// has run a cycle yet. See `process` for what it is for. next_ticks: Option, + /// Where the accumulator ends when the graph gives no position: a count + /// of the frames received, so packets still carry a continuous timeline. + counted_end: u64, } let state = State { @@ -71,6 +92,7 @@ pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender channels: channels as usize, sample_rate, next_ticks: None, + counted_end: 0, }; let _listener = stream @@ -131,8 +153,10 @@ pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender tracing::debug!(missed, "graph cycle missed; filled with silence"); } state.next_ticks = ticks.map(|t| t + frames); + state.counted_end += missed as u64 + frames; state.accum.extend_from_slice(samples); + let end = state.next_ticks.unwrap_or(state.counted_end); tracing::trace!( "pw delivered {} samples, accum now {}, frame size {}", @@ -143,10 +167,13 @@ pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender // Drain complete frames while state.accum.len() >= state.samples_per_frame { - let frame: Vec = - state.accum.drain(..state.samples_per_frame).collect(); + let held = (state.accum.len() / state.channels) as u64; + let packet = Packet { + position: first_position(end, held), + samples: state.accum.drain(..state.samples_per_frame).collect(), + }; - if state.tx.try_send(frame).is_err() { + if state.tx.try_send(packet).is_err() { tracing::warn!("encoder falling behind, dropping frame"); } } @@ -237,7 +264,17 @@ pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender #[cfg(test)] mod tests { - use super::missed_frames; + use super::{first_position, missed_frames}; + + /// Two packets drained from one accumulator are one packet apart, and the + /// first starts where the oldest held frame does. + #[test] + fn packets_are_placed_on_the_graph_clock() { + // 600 frames held, ending at graph position 10_600. + assert_eq!(first_position(10_600, 600), 10_000); + // After one 240-frame packet is drained, 360 remain before the end. + assert_eq!(first_position(10_600, 360), 10_240); + } #[test] fn a_gap_is_exactly_the_frames_the_clock_moved_past() { diff --git a/crates/nesprotocol/Cargo.toml b/crates/nesprotocol/Cargo.toml index 4d036fa6..084a28b5 100644 --- a/crates/nesprotocol/Cargo.toml +++ b/crates/nesprotocol/Cargo.toml @@ -15,3 +15,7 @@ lifecycle = ["dep:serde", "dep:serde_json"] [dependencies] serde = { workspace = true, optional = true } serde_json = { workspace = true, optional = true } + +# CLOCK_MONOTONIC, the one clock every media timestamp on the sender is on. +[target.'cfg(unix)'.dependencies] +libc.workspace = true diff --git a/crates/nesprotocol/src/clock.rs b/crates/nesprotocol/src/clock.rs new file mode 100644 index 00000000..038e025b --- /dev/null +++ b/crates/nesprotocol/src/clock.rs @@ -0,0 +1,81 @@ +//! The one clock media time is measured on. +//! +//! # Why one clock, and why this one +//! +//! Every media unit carries when it was captured and when it was handed to the +//! transport, and those stamps are taken by different processes: nescapture +//! stamps a video frame, neswire an audio packet's graph position, neshub the +//! send. They can only be subtracted if they are on one clock, so the sender's +//! clock is the kernel's `CLOCK_MONOTONIC`, which every process on a machine +//! shares and nothing can step. Wall clock would be shared too, and NTP moves +//! it. +//! +//! Microseconds in a `u64`: fine enough to see a millisecond of jitter in, and +//! it does not wrap in any session that will ever run, so nothing downstream +//! needs wrapping arithmetic. +//! +//! # The receiver +//! +//! The two machines' clocks are unrelated, so a receiver never compares its own +//! reading with a sender's stamp except through `delay`, which only uses the +//! difference's variation. Off unix (the client on Windows) this is therefore +//! the process's own monotonic clock, anchored at first use: consistent within +//! the process, which is all a receiver needs. + +/// Now, in microseconds, on the clock media timestamps are on. +#[cfg(unix)] +pub fn monotonic_us() -> u64 { + let mut ts = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + // SAFETY: `ts` is a valid out-pointer, and `CLOCK_MONOTONIC` exists on + // every unix this builds for, so the call cannot fail. + unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) }; + ts.tv_sec as u64 * 1_000_000 + ts.tv_nsec as u64 / 1_000 +} + +/// Now, in microseconds, on this process's own monotonic clock. +#[cfg(not(unix))] +pub fn monotonic_us() -> u64 { + static ANCHOR: std::sync::OnceLock = std::sync::OnceLock::new(); + ANCHOR + .get_or_init(std::time::Instant::now) + .elapsed() + .as_micros() as u64 +} + +/// When `instant` was, on the same clock as [`monotonic_us`]. +/// +/// For a moment something else stamped as an `Instant`, such as a compositor's +/// present time, taken as how long ago it was. An instant in the future reads +/// as now. +pub fn instant_us(instant: std::time::Instant) -> u64 { + monotonic_us().saturating_sub(instant.elapsed().as_micros() as u64) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_clock_moves_forward_in_microseconds() { + let a = monotonic_us(); + std::thread::sleep(std::time::Duration::from_millis(2)); + let b = monotonic_us(); + assert!(b - a >= 2_000, "{a} -> {b}"); + assert!(b - a < 1_000_000, "{a} -> {b}"); + } + + #[test] + fn an_instant_lands_on_the_same_clock() { + let then = std::time::Instant::now(); + std::thread::sleep(std::time::Duration::from_millis(3)); + let now = monotonic_us(); + let stamped = instant_us(then); + assert!( + (2_000..1_000_000).contains(&(now - stamped)), + "{stamped} vs {now}" + ); + } +} diff --git a/crates/nesprotocol/src/datagram.rs b/crates/nesprotocol/src/datagram.rs index 03409aee..8bb64ef1 100644 --- a/crates/nesprotocol/src/datagram.rs +++ b/crates/nesprotocol/src/datagram.rs @@ -18,9 +18,14 @@ //! # Wire format //! //! ```text -//! [1B kind] [2B frame seq LE] [2B fragment index LE] [2B fragment count LE] [fragment] +//! [1B kind] [2B frame seq LE] [2B fragment index LE] [2B fragment count LE] +//! [8B sent_us LE] [fragment] //! ``` //! +//! `sent_us` is when the hub handed the datagram to QUIC, on the sender's +//! [`clock`](crate::clock). Every fragment carries it, so the receiver can +//! measure the path from any one of them (see [`delay`](crate::delay)). +//! //! The fragments carry a frame *body*: exactly the bytes [`encode_frame`] produces //! minus its four-byte length prefix, which only a stream needs since a datagram //! already knows its own length. Reassembled, the body is `[1B type][2B seq][payload]` @@ -33,8 +38,8 @@ //! //! [`encode_frame`]: crate::encode_frame -/// kind(1) + frame seq(2) + fragment index(2) + fragment count(2). -pub const DGRAM_HDR_LEN: usize = 7; +/// kind(1) + frame seq(2) + fragment index(2) + fragment count(2) + sent_us(8). +pub const DGRAM_HDR_LEN: usize = 15; /// Datagram kinds. These share the numbering of the stream types they replace, /// so a frame's identity does not change with the transport under it. @@ -79,17 +84,21 @@ pub struct DatagramHeader { pub index: u16, /// How many fragments the whole frame was split into. Always at least 1. pub total: u16, + /// When the sender handed this datagram to QUIC, in microseconds on its + /// [`clock`](crate::clock). + pub sent_us: u64, } /// Write a datagram header into the first [`DGRAM_HDR_LEN`] bytes of `buf`. /// /// # Panics /// If `buf` is shorter than [`DGRAM_HDR_LEN`]. -pub fn write_datagram_header(buf: &mut [u8], kind: u8, seq: u16, index: u16, total: u16) { - buf[0] = kind; - buf[1..3].copy_from_slice(&seq.to_le_bytes()); - buf[3..5].copy_from_slice(&index.to_le_bytes()); - buf[5..7].copy_from_slice(&total.to_le_bytes()); +pub fn write_datagram_header(buf: &mut [u8], header: &DatagramHeader) { + buf[0] = header.kind; + buf[1..3].copy_from_slice(&header.seq.to_le_bytes()); + buf[3..5].copy_from_slice(&header.index.to_le_bytes()); + buf[5..7].copy_from_slice(&header.total.to_le_bytes()); + buf[7..15].copy_from_slice(&header.sent_us.to_le_bytes()); } /// Split a received datagram into its header and fragment payload. @@ -112,6 +121,7 @@ pub fn decode_datagram(buf: &[u8]) -> Option<(DatagramHeader, &[u8])> { seq: u16::from_le_bytes([buf[1], buf[2]]), index, total, + sent_us: u64::from_le_bytes(buf[7..15].try_into().unwrap()), }; Some((header, &buf[DGRAM_HDR_LEN..])) } @@ -144,19 +154,18 @@ mod tests { #[test] fn header_roundtrips() { let mut buf = [0u8; DGRAM_HDR_LEN + 3]; - write_datagram_header(&mut buf, DGRAM_VIDEO, 0xBEEF, 2, 7); + let sent = DatagramHeader { + kind: DGRAM_VIDEO, + seq: 0xBEEF, + index: 2, + total: 7, + sent_us: 0x0123_4567_89AB_CDEF, + }; + write_datagram_header(&mut buf, &sent); buf[DGRAM_HDR_LEN..].copy_from_slice(&[1, 2, 3]); let (header, payload) = decode_datagram(&buf).expect("valid datagram"); - assert_eq!( - header, - DatagramHeader { - kind: DGRAM_VIDEO, - seq: 0xBEEF, - index: 2, - total: 7 - } - ); + assert_eq!(header, sent); assert_eq!(payload, &[1, 2, 3]); } @@ -166,15 +175,42 @@ mod tests { // total == 0 describes a frame with no fragments. let mut buf = [0u8; DGRAM_HDR_LEN]; - write_datagram_header(&mut buf, DGRAM_VIDEO, 1, 0, 0); + write_datagram_header( + &mut buf, + &DatagramHeader { + kind: DGRAM_VIDEO, + seq: 1, + index: 0, + total: 0, + sent_us: 0, + }, + ); assert!(decode_datagram(&buf).is_none()); // index outside the fragment count. - write_datagram_header(&mut buf, DGRAM_VIDEO, 1, 4, 4); + write_datagram_header( + &mut buf, + &DatagramHeader { + kind: DGRAM_VIDEO, + seq: 1, + index: 4, + total: 4, + sent_us: 0, + }, + ); assert!(decode_datagram(&buf).is_none()); // A count that would size an implausible allocation. - write_datagram_header(&mut buf, DGRAM_VIDEO, 1, 0, u16::MAX); + write_datagram_header( + &mut buf, + &DatagramHeader { + kind: DGRAM_VIDEO, + seq: 1, + index: 0, + total: u16::MAX, + sent_us: 0, + }, + ); assert!(decode_datagram(&buf).is_none()); } diff --git a/crates/nesprotocol/src/delay.rs b/crates/nesprotocol/src/delay.rs index e7176c91..c396e6d8 100644 --- a/crates/nesprotocol/src/delay.rs +++ b/crates/nesprotocol/src/delay.rs @@ -1,144 +1,180 @@ -//! How much a delay varies, measured the same way at both ends. +//! How late a media unit arrived, measured the same way at both ends. //! //! # What this measures //! -//! Every frame carries the timestamp the sender stamped on it at capture. The -//! two clocks are unrelated, so the difference between that and the moment the -//! frame arrives here is meaningless on its own -- it contains an unknown, -//! roughly constant offset. Its *variation* is not meaningless, and variation is -//! the only thing a playout buffer exists to absorb: a path that delivers every -//! frame exactly 300 ms late needs no buffer at all, while one that alternates -//! between 10 ms and 60 ms needs 50 ms whatever its average. +//! Every media unit carries the time the sender stamped on it, on the sender's +//! [`clock`](crate::clock). The receiver's clock is unrelated, so the one-way +//! delay `d = observed - stamp` contains an unknown constant -- the two clocks' +//! offset -- plus propagation, plus queueing. Only the queueing varies, and it +//! is the only part a playout or a rate decision has any use for: a path that +//! delivers everything exactly 300 ms late needs no buffer, while one that +//! alternates between 10 and 60 ms needs 50 whatever its average. //! -//! So this tracks the smallest difference seen recently -- the best the path has -//! managed, which stands in for the unknown offset -- and reports how far above -//! it each frame lands. +//! So the lowest `d` the path has produced stands in for offset plus +//! propagation, and a unit's queueing delay is how far above that it landed. //! -//! # What it deliberately does not do +//! # Clock skew, measured instead of expired //! -//! It does not distinguish the sender's contribution from the network's. The -//! sender reports its own pipeline delay separately, and the two are compared -//! rather than subtracted: a frame that was late because the encoder stalled -//! should not raise a buffer, because buffering is latency spent hiding a fault -//! that should be fixed instead. +//! Two crystals never run at quite the same rate. At 100 ppm the receiver's +//! clock gains 0.1 ms a second on the sender's, so `d` drifts linearly, and a +//! lowest-ever `d` taken an hour ago would be 360 ms off. This used to be +//! handled by letting the best case expire after 30 s, which was a guess about +//! how fast paths change, and threw away a good baseline on a timer. +//! +//! Instead the drift is measured: the lowest `d` of each second lies on a line +//! whose slope is the skew, found by least squares over every second the path +//! has run. The baseline is that line at its lowest, and it never expires -- +//! nothing invalidates it but a change of path, which the caller knows about +//! and says so with [`DelayTracker::reset_baseline`]. //! //! # Why this is shared //! -//! Both ends measure a delay and the two are compared: the hub reports how much -//! its own pipeline varied before a frame left, the client reports how much the -//! total varied by the time it arrived. A comparison between two different -//! measures would be meaningless, so there is one measure, defined once. -//! -//! It also has to be *variation* on both sides rather than absolute delay, -//! because neither side can know the absolute. The hub's `timestamp_ms` counts -//! from the encoder's own start, not from any epoch, so even on one machine the -//! difference to wall clock contains an unknown constant -- and across two -//! machines there is no shared clock at all. -//! -//! Nothing here adapts anything yet. It measures, so that the decision about -//! what to adapt is made against a distribution rather than an intuition. +//! The hub measures its own pipeline with it (capture to hand-off, one clock, +//! no skew) and the client measures the whole path. Two different measures +//! could not be compared, so there is one, defined once. -use std::collections::VecDeque; -use std::time::{Duration, Instant}; - -/// How long a best-case observation stays authoritative. -/// -/// The minimum has to expire. Paths change -- a relay is dropped for a direct -/// connection, a phone moves between cells -- and a minimum from the old path -/// makes every frame on the new one look permanently late, which would pin a -/// buffer at a size nothing needs. Long enough that an ordinary quiet spell does -/// not reset the baseline, short enough that a genuine change is noticed. -const BASELINE_WINDOW: Duration = Duration::from_secs(30); - -/// Ceiling on retained observations, so a stalled consumer cannot grow this. -const MAX_SAMPLES: usize = 4096; - -/// One second's worth of lateness, summarised. +/// One second's queueing delay, summarised. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct DelaySummary { - /// Frames measured. + /// Units measured. pub frames: u32, - /// Median lateness above the best the path has managed, in milliseconds. - pub p50_ms: u16, + /// Median queueing delay above the path's best, in microseconds. + pub p50_us: u32, /// The 95th percentile of the same. - pub p95_ms: u16, - /// The worst single frame. - pub max_ms: u16, + pub p95_us: u32, + /// The worst single unit. + pub max_us: u32, } -/// Tracks arrival lateness for one media stream. -#[derive(Debug)] +/// `[4B frames][4B p50_us][4B p95_us][4B max_us]`, little-endian. +pub const DELAY_SUMMARY_LEN: usize = 16; + +impl DelaySummary { + pub fn encode(&self, buf: &mut Vec) { + buf.extend_from_slice(&self.frames.to_le_bytes()); + buf.extend_from_slice(&self.p50_us.to_le_bytes()); + buf.extend_from_slice(&self.p95_us.to_le_bytes()); + buf.extend_from_slice(&self.max_us.to_le_bytes()); + } + + /// `None` when `buf` is shorter than [`DELAY_SUMMARY_LEN`]. + pub fn decode(buf: &[u8]) -> Option { + let at = |o: usize| Some(u32::from_le_bytes(buf.get(o..o + 4)?.try_into().ok()?)); + Some(Self { + frames: at(0)?, + p50_us: at(4)?, + p95_us: at(8)?, + max_us: at(12)?, + }) + } +} + +/// Ceiling on units held for one summary, so a consumer that stops taking +/// cannot grow this without bound. Far past a second of any stream we send. +const MAX_SAMPLES: usize = 4096; + +/// Tracks queueing delay for one media stream on one path. +#[derive(Debug, Default)] pub struct DelayTracker { - /// Best-case delay observations, each with when it was taken, oldest first. - /// - /// A deque rather than a single value because a minimum that can only fall - /// never recovers from a path that improved, and one that is simply reset on - /// a timer throws away a good baseline for no reason. Holding the recent - /// candidates lets the oldest expire while a better one is still standing. - baseline: VecDeque<(Instant, i32)>, - /// Lateness of each frame this second, in milliseconds. - samples: Vec, - /// The first delay seen, as an anchor for every later one. - /// - /// Delays are compared *relative to this*, in wrapping arithmetic. Both - /// clocks are unrelated and the sender's stamp is a `u32` of milliseconds - /// that wraps every 49 days, so an absolute subtraction is a number with no - /// meaning and a wrap in the middle of a session makes it jump by 2^32 -- - /// which, taken as lateness, reads as every frame being weeks late until the - /// baseline expires. Anchoring and then interpreting the difference as - /// signed makes the wrap a non-event, because the quantity that matters is - /// only ever a few hundred milliseconds wide. - anchor: Option, + /// The first observation, which every later one is measured relative to, + /// so the fit works on small numbers: `(observed_us, d_us)`. + origin: Option<(u64, i64)>, + /// The second being collected, and the lowest `d` seen in it so far. + current: Option<(u64, f64)>, + /// Every completed second's lowest `d`: `(seconds since origin, d)`. + minima: Vec<(f64, f64)>, + /// Running sums for the least-squares line through `minima`. + fit: Fit, + /// The skew, in microseconds of `d` per second of the receiver's clock. + slope: f64, + /// The lowest `d - slope * x` the path has produced: where the baseline + /// line sits. + floor: Option, + /// Queueing delay of each unit since the last [`take`](Self::take). + samples: Vec, } -impl Default for DelayTracker { - fn default() -> Self { - Self::new() +#[derive(Debug, Default, Clone, Copy)] +struct Fit { + n: f64, + sx: f64, + sy: f64, + sxx: f64, + sxy: f64, +} + +impl Fit { + fn add(&mut self, x: f64, y: f64) { + self.n += 1.0; + self.sx += x; + self.sy += y; + self.sxx += x * x; + self.sxy += x * y; + } + + /// The least-squares slope, once there are two distinct seconds to draw a + /// line through. + fn slope(&self) -> Option { + let denominator = self.n * self.sxx - self.sx * self.sx; + (self.n >= 2.0 && denominator > 0.0) + .then(|| (self.n * self.sxy - self.sx * self.sy) / denominator) } } impl DelayTracker { pub fn new() -> Self { - Self { - baseline: VecDeque::new(), - samples: Vec::new(), - anchor: None, - } + Self::default() } - /// Record one frame's arrival. + /// Record one unit, and return its queueing delay in microseconds. /// - /// `stamp_ms` is the timestamp the frame carries; `observed_ms` is the local - /// clock at the point being measured -- arrival, for a receiver; the moment - /// the frame is handed to the transport, for a sender. See [`anchor`](Self::anchor) for why the difference is taken - /// in wrapping arithmetic rather than widened. - pub fn observe(&mut self, stamp_ms: u32, observed_ms: u64, now: Instant) { - let raw = (observed_ms as u32).wrapping_sub(stamp_ms); - let anchor = *self.anchor.get_or_insert(raw); - let delay = raw.wrapping_sub(anchor) as i32; - self.expire(now); + /// `stamp_us` is the time the unit carries, on the sender's clock; + /// `observed_us` is this end's clock at the point being measured -- arrival + /// for a receiver, hand-off to the transport for a sender. + pub fn observe(&mut self, stamp_us: u64, observed_us: u64) -> u32 { + // Both are microsecond counts far below 2^63, so the difference of the + // two as signed is exact. + let d = observed_us as i64 - stamp_us as i64; + let (origin_us, origin_d) = *self.origin.get_or_insert((observed_us, d)); + let x_us = observed_us.saturating_sub(origin_us); + let x = x_us as f64 / 1e6; + let y = (d - origin_d) as f64; - // Anything at or below the running minimum becomes the new baseline, and - // supersedes the candidates it beats -- they can only be worse, so - // keeping them would let a stale, higher value resurface on expiry. - while self.baseline.back().is_some_and(|(_, d)| *d >= delay) { - self.baseline.pop_back(); + let second = x_us / 1_000_000; + match self.current { + Some((at, _)) if at != second => { + self.close_second(); + self.current = Some((second, y)); + } + Some((_, ref mut low)) => *low = low.min(y), + None => self.current = Some((second, y)), } - self.baseline.push_back((now, delay)); - let best = self.baseline.front().map(|(_, d)| *d).unwrap_or(delay); - let lateness = i64::from(delay) - .saturating_sub(i64::from(best)) - .clamp(0, i64::from(u16::MAX)) as u16; + let residual = y - self.slope * x; + let floor = match self.floor { + Some(floor) if floor <= residual => floor, + _ => residual, + }; + self.floor = Some(floor); + + let q = (residual - floor).clamp(0.0, f64::from(u32::MAX)) as u32; if self.samples.len() < MAX_SAMPLES { - self.samples.push(lateness); + self.samples.push(q); } + q } - /// Summarise and clear the frames seen since the last call. + /// The skew this path has been measured at, in parts per million: how much + /// faster this end's clock runs than the sender's. Zero until there are two + /// seconds to measure it from. + pub fn skew_ppm(&self) -> f64 { + self.slope + } + + /// Summarise and clear the units seen since the last call. /// /// Returns `None` for a second in which nothing arrived. That is not the - /// same as a second with no lateness, and reporting zero would say the path + /// same as a second with no queueing, and reporting zero would say the path /// is behaving perfectly at the moment it has stopped delivering anything. pub fn take(&mut self) -> Option { if self.samples.is_empty() { @@ -148,29 +184,44 @@ impl DelayTracker { let at = |q: f64| self.samples[((self.samples.len() - 1) as f64 * q) as usize]; let summary = DelaySummary { frames: self.samples.len() as u32, - p50_ms: at(0.5), - p95_ms: at(0.95), - max_ms: self.samples[self.samples.len() - 1], + p50_us: at(0.5), + p95_us: at(0.95), + max_us: self.samples[self.samples.len() - 1], }; self.samples.clear(); Some(summary) } - /// Forget the baseline entirely. + /// Forget the path entirely: baseline, skew and every second behind them. /// - /// For a path change, where the old best case says nothing about the new - /// path and keeping it would make every frame look late until it expired. + /// For a path change, where the old path's propagation delay says nothing + /// about the new one's, and keeping it would read the difference as + /// queueing for the rest of the session. pub fn reset_baseline(&mut self) { - self.baseline.clear(); + *self = Self { + samples: std::mem::take(&mut self.samples), + ..Self::default() + }; } - fn expire(&mut self, now: Instant) { - while self - .baseline - .front() - .is_some_and(|(seen, _)| now.duration_since(*seen) > BASELINE_WINDOW) - { - self.baseline.pop_front(); + /// Fold the second just finished into the fit, and move the baseline line + /// to the new slope. + fn close_second(&mut self) { + let Some((second, low)) = self.current.take() else { + return; + }; + // The minimum is placed at the middle of its second: it could have been + // anywhere in it, and the middle is where that error averages out. + let x = second as f64 + 0.5; + self.minima.push((x, low)); + self.fit.add(x, low); + if let Some(slope) = self.fit.slope() { + self.slope = slope; + self.floor = self + .minima + .iter() + .map(|(x, y)| y - slope * x) + .reduce(f64::min); } } } @@ -179,132 +230,110 @@ impl DelayTracker { mod tests { use super::*; - /// A frame sent at `ts_ms` arriving `transit` milliseconds later, where the - /// receiver's clock is offset from the sender's by an arbitrary amount. - const OFFSET: u64 = 1_700_000_000_000; + /// The receiver's clock is offset from the sender's by an arbitrary amount. + const OFFSET: u64 = 1_700_000_000_000_000; - fn at(j: &mut DelayTracker, ts_ms: u32, transit: u64, now: Instant) { - j.observe(ts_ms, OFFSET + u64::from(ts_ms) + transit, now); + /// A unit stamped at `sent_us`, `transit_us` in flight, arriving on a + /// receiver clock that runs `skew_ppm` fast. + fn arrive(j: &mut DelayTracker, sent_us: u64, transit_us: u64, skew_ppm: f64) -> u32 { + let true_arrival = sent_us + transit_us; + let observed = OFFSET + true_arrival + (true_arrival as f64 * skew_ppm / 1e6) as u64; + j.observe(sent_us, observed) } #[test] - fn a_constant_delay_is_not_jitter() { - // The point of the whole measure. A path that delivers every frame - // exactly 300 ms late needs no buffer at all; only variation does. + fn a_constant_delay_is_not_queueing() { let mut j = DelayTracker::new(); - let now = Instant::now(); - for i in 0..120u32 { - at( - &mut j, - i * 16, - 300, - now + Duration::from_millis(u64::from(i) * 16), - ); + for i in 0..120u64 { + arrive(&mut j, i * 16_000, 300_000, 0.0); } - let s = j.take().expect("frames arrived"); - assert_eq!(s.max_ms, 0, "a constant offset was read as lateness"); + let s = j.take().expect("units arrived"); + assert_eq!(s.max_us, 0, "a constant offset was read as queueing"); assert_eq!(s.frames, 120); } #[test] - fn lateness_is_measured_against_the_best_the_path_managed() { + fn queueing_is_measured_against_the_best_the_path_managed() { let mut j = DelayTracker::new(); - let now = Instant::now(); - at(&mut j, 0, 20, now); - at(&mut j, 16, 70, now + Duration::from_millis(16)); - at(&mut j, 32, 20, now + Duration::from_millis(32)); - let s = j.take().expect("frames arrived"); - assert_eq!( - s.max_ms, 50, - "the 70 ms frame is 50 ms above the 20 ms best" - ); + arrive(&mut j, 0, 20_000, 0.0); + let q = arrive(&mut j, 16_000, 70_000, 0.0); + arrive(&mut j, 32_000, 20_000, 0.0); + assert_eq!(q, 50_000, "the 70 ms unit is 50 ms above the 20 ms best"); + assert_eq!(j.take().unwrap().max_us, 50_000); } #[test] - fn a_silent_second_reports_nothing_rather_than_no_jitter() { - // Zero would say the path is behaving perfectly at the moment it has - // stopped delivering anything at all. + fn a_better_unit_becomes_the_baseline_at_once() { let mut j = DelayTracker::new(); - assert_eq!(j.take(), None); + arrive(&mut j, 0, 90_000, 0.0); + assert_eq!(arrive(&mut j, 16_000, 10_000, 0.0), 0); + assert_eq!(arrive(&mut j, 32_000, 50_000, 0.0), 40_000); } #[test] - fn the_baseline_expires_so_an_improved_path_is_noticed() { - // A minimum that can only fall never recovers: one lucky early frame - // would make every later frame look late for the rest of the session. - let mut j = DelayTracker::new(); - let now = Instant::now(); - at(&mut j, 0, 10, now); - j.take(); + fn a_silent_second_reports_nothing_rather_than_no_queueing() { + assert_eq!(DelayTracker::new().take(), None); + } - // Much later, the path settles at a steady 200 ms. - let later = now + BASELINE_WINDOW + Duration::from_secs(1); - for i in 0..10u32 { - at( - &mut j, - 1000 + i * 16, - 200, - later + Duration::from_millis(u64::from(i) * 16), - ); + /// The reason the baseline no longer expires: at 100 ppm a fixed baseline + /// would read 0.1 ms a second of drift as queueing, which over ten minutes + /// is 60 ms on a path with none at all. + #[test] + fn clock_skew_is_measured_and_not_read_as_queueing() { + let mut j = DelayTracker::new(); + // Ten minutes at 200 units a second, a steady 40 ms path, 100 ppm. + for i in 0..(600 * 200u64) { + let q = arrive(&mut j, i * 5_000, 40_000, 100.0); + if i > 200 * 5 { + assert!(q < 200, "unit {i} read {q} µs of queueing from skew alone"); + } } - let s = j.take().expect("frames arrived"); - assert_eq!( - s.max_ms, 0, - "a stale best case from a previous path made a steady path look late", + assert!( + (j.skew_ppm() - 100.0).abs() < 1.0, + "measured {} ppm", + j.skew_ppm() ); } #[test] - fn a_better_observation_supersedes_worse_ones_still_in_the_window() { - // Otherwise a higher candidate resurfaces when the better one expires, - // and the baseline walks upwards for no reason the path can account for. + fn queueing_on_a_skewed_path_is_still_exact() { let mut j = DelayTracker::new(); - let now = Instant::now(); - at(&mut j, 0, 90, now); - at(&mut j, 16, 10, now + Duration::from_millis(16)); - // The 90 ms candidate is gone, so this sits 40 ms above the 10 ms best. - at(&mut j, 32, 50, now + Duration::from_millis(32)); - let s = j.take().expect("frames arrived"); - assert_eq!(s.max_ms, 40); + for i in 0..(60 * 200u64) { + arrive(&mut j, i * 5_000, 40_000, -150.0); + } + let stalled = arrive(&mut j, 60 * 200 * 5_000, 40_000 + 25_000, -150.0); + assert!((24_500..25_500).contains(&stalled), "read {stalled} µs"); } #[test] fn a_reset_forgets_the_old_path_entirely() { let mut j = DelayTracker::new(); - let now = Instant::now(); - at(&mut j, 0, 10, now); + arrive(&mut j, 0, 10_000, 0.0); j.take(); j.reset_baseline(); - at(&mut j, 16, 400, now + Duration::from_millis(16)); - let s = j.take().expect("frames arrived"); - assert_eq!(s.max_ms, 0, "the new path was judged against the old one"); + assert_eq!( + arrive(&mut j, 16_000, 400_000, 0.0), + 0, + "the new path was judged against the old one" + ); } #[test] fn the_tail_is_kept_apart_from_the_middle() { - // The number that sizes a buffer is the tail. A run of prompt frames - // with one bad one must not average into "slightly late". let mut j = DelayTracker::new(); - let now = Instant::now(); - for i in 0..99u32 { - at( - &mut j, - i * 16, - 10, - now + Duration::from_millis(u64::from(i) * 16), - ); + for i in 0..99u64 { + arrive(&mut j, i * 16_000, 10_000, 0.0); } - at(&mut j, 99 * 16, 260, now + Duration::from_millis(99 * 16)); - let s = j.take().expect("frames arrived"); - assert_eq!(s.p50_ms, 0); - assert_eq!(s.max_ms, 250); + arrive(&mut j, 99 * 16_000, 260_000, 0.0); + let s = j.take().expect("units arrived"); + assert_eq!(s.p50_us, 0); + assert_eq!(s.max_us, 250_000); } #[test] fn taking_clears_so_each_answer_describes_one_second() { let mut j = DelayTracker::new(); - let now = Instant::now(); - at(&mut j, 0, 10, now); + arrive(&mut j, 0, 10_000, 0.0); assert!(j.take().is_some()); assert_eq!( j.take(), @@ -314,14 +343,17 @@ mod tests { } #[test] - fn a_wrapped_sender_timestamp_does_not_poison_the_baseline() { - // ts_ms wraps every 49 days. A negative delay taken as the best case - // would make every subsequent frame appear weeks late. - let mut j = DelayTracker::new(); - let now = Instant::now(); - j.observe(u32::MAX - 10, OFFSET, now); - j.observe(5, OFFSET + 16, now + Duration::from_millis(16)); - let s = j.take().expect("frames arrived"); - assert!(s.max_ms < u16::MAX, "a wrap produced a nonsense lateness"); + fn a_summary_survives_the_wire() { + let s = DelaySummary { + frames: 200, + p50_us: 1_200, + p95_us: 18_000, + max_us: 41_200, + }; + let mut buf = Vec::new(); + s.encode(&mut buf); + assert_eq!(buf.len(), DELAY_SUMMARY_LEN); + assert_eq!(DelaySummary::decode(&buf), Some(s)); + assert_eq!(DelaySummary::decode(&buf[..15]), None); } } diff --git a/crates/nesprotocol/src/lib.rs b/crates/nesprotocol/src/lib.rs index 44241656..95023b53 100644 --- a/crates/nesprotocol/src/lib.rs +++ b/crates/nesprotocol/src/lib.rs @@ -3,6 +3,7 @@ // can drift from each other silently. pub mod attach; +pub mod clock; pub mod datagram; pub mod delay; pub mod gamepad; @@ -67,7 +68,8 @@ impl Carrier { pub const IPC_MAGIC: [u8; 4] = [b'N', b'S', b'T', b'R']; -pub const IPC_HEADER_LEN: usize = 20; +/// `[4B magic][1B type][1B codec][1B flags][1B reserved][8B stamp][2B w][2B h][4B len]`. +pub const IPC_HEADER_LEN: usize = 24; // ── QUIC stream protocol ──────────────────────────────────────── @@ -83,7 +85,11 @@ pub const IPC_HEADER_LEN: usize = 20; /// stream type it does not know, so it would receive no keyframes at all and /// never decode anything — the two cannot interoperate either, and the mismatch /// is worth reporting rather than presenting as a frozen picture. -pub const STREAM_VERSION: u8 = 3; +/// +/// v4 put time on every media unit (see [`clock`]): a send time on every +/// datagram and keyframe stream, a capture time in microseconds on video where +/// v3 had milliseconds, and the graph position on audio, which had nothing. +pub const STREAM_VERSION: u8 = 4; /// Uniform frame header overhead: [4B u32 LE frame_len] [1B type] [2B u16 LE seq] pub const FRAME_HDR_LEN: usize = 7; @@ -208,25 +214,32 @@ pub const FLAG_KEYFRAME: u8 = 0x01; pub const FLAG_RECONFIG: u8 = 0x02; // stream reconfiguration (codec/bit-depth change) /// Encode an IPC frame into a byte buffer. -/// Format: [4B magic] [1B type] [1B codec] [1B flags] [1B reserved] [4B ts_ms LE] [4B w_h LE] [4B len LE] [N data] +/// +/// Format: `[4B magic][1B type][1B codec][1B flags][1B reserved][8B stamp LE] +/// [2B width LE][2B height LE][4B len LE][N data]`. +/// +/// `stamp` is the media unit's time, and what it counts depends on the stream: +/// for video, when the frame was presented, in microseconds on +/// [`clock::monotonic_us`]; for audio, the packet's first sample on the audio +/// graph's clock, in 48 kHz frames. Both are the producer's to know and the +/// hub's to carry; neither is ever converted into the other. pub fn encode_ipc_frame( stream_type: u8, codec: u8, flags: u8, - timestamp_ms: u32, + stamp: u64, width: u16, height: u16, data: &[u8], ) -> Vec { - let header_len = IPC_HEADER_LEN; - let mut buf = Vec::with_capacity(header_len + data.len()); + let mut buf = Vec::with_capacity(IPC_HEADER_LEN + data.len()); buf.extend_from_slice(&IPC_MAGIC); buf.push(stream_type); buf.push(codec); buf.push(flags); buf.push(0); // reserved - buf.extend_from_slice(×tamp_ms.to_le_bytes()); + buf.extend_from_slice(&stamp.to_le_bytes()); buf.extend_from_slice(&width.to_le_bytes()); buf.extend_from_slice(&height.to_le_bytes()); buf.extend_from_slice(&(data.len() as u32).to_le_bytes()); @@ -241,7 +254,8 @@ pub struct DecodedIpcFrame<'a> { pub stream_type: u8, pub codec: u8, pub flags: u8, - pub timestamp_ms: u32, + /// See [`encode_ipc_frame`] for what it counts on each stream. + pub stamp: u64, pub width: u16, pub height: u16, pub data: &'a [u8], @@ -258,10 +272,10 @@ pub fn decode_ipc_frame(buf: &[u8]) -> Option> { let stream_type = buf[4]; let codec = buf[5]; let flags = buf[6]; - let timestamp_ms = u32::from_le_bytes([buf[8], buf[9], buf[10], buf[11]]); - let width = u16::from_le_bytes([buf[12], buf[13]]); - let height = u16::from_le_bytes([buf[14], buf[15]]); - let data_len = u32::from_le_bytes([buf[16], buf[17], buf[18], buf[19]]) as usize; + let stamp = u64::from_le_bytes(buf[8..16].try_into().unwrap()); + let width = u16::from_le_bytes([buf[16], buf[17]]); + let height = u16::from_le_bytes([buf[18], buf[19]]); + let data_len = u32::from_le_bytes(buf[20..24].try_into().unwrap()) as usize; if buf.len() < IPC_HEADER_LEN + data_len { return None; @@ -271,13 +285,85 @@ pub fn decode_ipc_frame(buf: &[u8]) -> Option> { stream_type, codec, flags, - timestamp_ms, + stamp, width, height, data: &buf[IPC_HEADER_LEN..IPC_HEADER_LEN + data_len], }) } +// ── Media payloads ────────────────────────────────────────────── +// +// What a media frame body carries after its `[type][seq]`, defined once for the +// hub that builds them and every receiver that reads them. + +/// `[1B codec][1B flags][8B captured_us LE][2B width LE][2B height LE]`. +pub const VIDEO_PAYLOAD_HDR_LEN: usize = 14; + +/// The head of a video frame payload. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct VideoHead { + pub codec: u8, + pub flags: u8, + /// When the frame was presented, in microseconds on the sender's + /// [`clock::monotonic_us`]. + pub captured_us: u64, + pub width: u16, + pub height: u16, +} + +/// Build a video frame payload: the head, then the coded frame. +pub fn encode_video_payload(head: &VideoHead, data: &[u8]) -> Vec { + let mut buf = Vec::with_capacity(VIDEO_PAYLOAD_HDR_LEN + data.len()); + buf.push(head.codec); + buf.push(head.flags); + buf.extend_from_slice(&head.captured_us.to_le_bytes()); + buf.extend_from_slice(&head.width.to_le_bytes()); + buf.extend_from_slice(&head.height.to_le_bytes()); + buf.extend_from_slice(data); + buf +} + +/// Split a video frame payload into its head and the coded frame. +pub fn decode_video_payload(payload: &[u8]) -> Option<(VideoHead, &[u8])> { + if payload.len() < VIDEO_PAYLOAD_HDR_LEN { + return None; + } + let head = VideoHead { + codec: payload[0], + flags: payload[1], + captured_us: u64::from_le_bytes(payload[2..10].try_into().unwrap()), + width: u16::from_le_bytes([payload[10], payload[11]]), + height: u16::from_le_bytes([payload[12], payload[13]]), + }; + Some((head, &payload[VIDEO_PAYLOAD_HDR_LEN..])) +} + +/// `[8B position LE]`, then the Opus packet. +pub const AUDIO_PAYLOAD_HDR_LEN: usize = 8; + +/// Build an audio packet payload. +/// +/// `position` is the packet's first sample on the sender's audio graph clock, +/// in 48 kHz frames. It is the audio timeline: two packets whose positions are +/// further apart than the first one's length have samples missing between +/// them, by exactly that count, whether those packets were lost or are late. +pub fn encode_audio_payload(position: u64, opus: &[u8]) -> Vec { + let mut buf = Vec::with_capacity(AUDIO_PAYLOAD_HDR_LEN + opus.len()); + buf.extend_from_slice(&position.to_le_bytes()); + buf.extend_from_slice(opus); + buf +} + +/// Split an audio packet payload into its position and the Opus packet. +pub fn decode_audio_payload(payload: &[u8]) -> Option<(u64, &[u8])> { + if payload.len() < AUDIO_PAYLOAD_HDR_LEN { + return None; + } + let position = u64::from_le_bytes(payload[..8].try_into().unwrap()); + Some((position, &payload[AUDIO_PAYLOAD_HDR_LEN..])) +} + /// Codec ID to human-readable name. pub fn codec_name(codec: u8) -> &'static str { match codec { @@ -611,6 +697,15 @@ pub struct ReceiverReport { pub never_arrived: u32, /// The receiver's own round-trip estimate, in milliseconds. pub rtt_ms: u32, + /// How long video spent queued on the path this second, from each + /// datagram's send time (see [`delay`]). + /// + /// The leading signal. Loss and RTT say a path was overrun some time ago; + /// one-way queueing delay rises the moment a queue starts to build, which + /// is tens of milliseconds before either of them moves. + pub video_queue: delay::DelaySummary, + /// The same for audio, on its own connection. + pub audio_queue: delay::DelaySummary, } impl ReceiverReport { @@ -629,9 +724,9 @@ impl ReceiverReport { } } -/// `[8B goodput_bps][4B released][4B incomplete][4B never_arrived][4B rtt_ms]`, -/// all little-endian. -pub const RECEIVER_REPORT_LEN: usize = 24; +/// `[8B goodput_bps][4B released][4B incomplete][4B never_arrived][4B rtt_ms] +/// [16B video_queue][16B audio_queue]`, all little-endian. +pub const RECEIVER_REPORT_LEN: usize = 24 + 2 * delay::DELAY_SUMMARY_LEN; pub fn encode_receiver_report(buf: &mut Vec, report: &ReceiverReport) { buf.reserve(RECEIVER_REPORT_LEN); @@ -640,12 +735,11 @@ pub fn encode_receiver_report(buf: &mut Vec, report: &ReceiverReport) { 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()); + report.video_queue.encode(buf); + report.audio_queue.encode(buf); } /// 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; @@ -657,6 +751,8 @@ pub fn decode_receiver_report(payload: &[u8]) -> Option { incomplete: u32_at(12), never_arrived: u32_at(16), rtt_ms: u32_at(20), + video_queue: delay::DelaySummary::decode(&payload[24..])?, + audio_queue: delay::DelaySummary::decode(&payload[24 + delay::DELAY_SUMMARY_LEN..])?, }) } @@ -720,6 +816,18 @@ mod media_control_tests { incomplete: 12, never_arrived: 1, rtt_ms: 182, + video_queue: delay::DelaySummary { + frames: 59, + p50_us: 2_100, + p95_us: 9_800, + max_us: 28_800, + }, + audio_queue: delay::DelaySummary { + frames: 200, + p50_us: 1_900, + p95_us: 15_300, + max_us: 41_200, + }, } } @@ -740,16 +848,6 @@ mod media_control_tests { } } - #[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 @@ -1154,3 +1252,62 @@ mod media_control_tests { assert_eq!(codec_name(CODEC_PYROWAVE), "pyrowave"); } } + +#[cfg(test)] +mod media_payload_tests { + use super::*; + + #[test] + fn an_ipc_frame_carries_its_stamp_whole() { + // A u64 of microseconds, so a stamp past 2^32 must come back intact: + // the width is the reason it changed. + let stamp = 0x0000_0123_4567_89AB; + let ipc = encode_ipc_frame( + STREAM_VIDEO, + CODEC_AV1, + FLAG_KEYFRAME, + stamp, + 1920, + 1080, + &[7; 5], + ); + assert_eq!(ipc.len(), IPC_HEADER_LEN + 5); + let d = decode_ipc_frame(&ipc).expect("valid"); + assert_eq!( + (d.stamp, d.width, d.height, d.data), + (stamp, 1920, 1080, &[7u8; 5][..]) + ); + assert!( + decode_ipc_frame(&ipc[..IPC_HEADER_LEN + 4]).is_none(), + "a short frame was read" + ); + } + + #[test] + fn a_video_payload_round_trips_and_keeps_its_flags_where_the_hub_looks() { + let head = VideoHead { + codec: CODEC_H265, + flags: FLAG_KEYFRAME, + captured_us: 9_876_543_210, + width: 2560, + height: 1440, + }; + let payload = encode_video_payload(&head, &[1, 2, 3]); + assert_eq!( + decode_video_payload(&payload), + Some((head, &[1u8, 2, 3][..])) + ); + assert!(reliable::video_is_keyframe(&payload)); + assert!(decode_video_payload(&payload[..VIDEO_PAYLOAD_HDR_LEN - 1]).is_none()); + } + + #[test] + fn an_audio_payload_carries_its_graph_position() { + let payload = encode_audio_payload(48_000 * 3600, &[0xFC, 0xFF]); + assert_eq!( + decode_audio_payload(&payload), + Some((48_000 * 3600, &[0xFCu8, 0xFF][..])) + ); + assert!(decode_audio_payload(&payload[..7]).is_none()); + } +} diff --git a/crates/nesprotocol/src/pyrowave.rs b/crates/nesprotocol/src/pyrowave.rs index 36bc97a5..241ffbb5 100644 --- a/crates/nesprotocol/src/pyrowave.rs +++ b/crates/nesprotocol/src/pyrowave.rs @@ -71,8 +71,9 @@ pub const ABANDON_MS: u64 = 100; /// Datagram kind, beside `DGRAM_VIDEO` and `DGRAM_AUDIO`. pub const DGRAM_PYROWAVE: u8 = 2; -/// `[kind][seq u16][index u16][total u16][critical u16][ts_ms u32][flags]`. -pub const PYRO_DGRAM_HDR_LEN: usize = 14; +/// `[kind][seq u16][index u16][total u16][critical u16][captured_us u64] +/// [sent_us u64][flags]`. +pub const PYRO_DGRAM_HDR_LEN: usize = 26; /// Bound on datagrams per frame, about 19 MB at a typical MTU. Past it a /// frame did not come from a sender of ours, and the bound keeps a corrupt @@ -114,7 +115,8 @@ pub const MAX_IN_FLIGHT: usize = 8; /// The per-frame fields of the IPC header. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub struct FrameMeta { - pub ts_ms: u32, + /// When the frame was presented, on the sender's [`clock`](crate::clock). + pub captured_us: u64, pub width: u16, pub height: u16, } @@ -213,7 +215,7 @@ pub fn chunk_frame( STREAM_PYROWAVE, CODEC_PYROWAVE, 0, - meta.ts_ms, + meta.captured_us, meta.width, meta.height, &body, @@ -337,7 +339,7 @@ impl FrameAssembler { } let meta = FrameMeta { - ts_ms: ipc.timestamp_ms, + captured_us: ipc.stamp, width: ipc.width, height: ipc.height, }; @@ -430,7 +432,10 @@ pub struct PyroDatagramHeader { /// broken on frames that arrived complete. The sender knows which /// datagrams those bands went in. pub critical: u16, - pub ts_ms: u32, + /// When the frame was presented, on the sender's [`clock`](crate::clock). + pub captured_us: u64, + /// When the hub handed this datagram to QUIC, on the same clock. + pub sent_us: u64, pub flags: u8, } @@ -441,8 +446,9 @@ pub fn write_pyro_header(buf: &mut [u8], h: &PyroDatagramHeader) { buf[3..5].copy_from_slice(&h.index.to_le_bytes()); buf[5..7].copy_from_slice(&h.total.to_le_bytes()); buf[7..9].copy_from_slice(&h.critical.to_le_bytes()); - buf[9..13].copy_from_slice(&h.ts_ms.to_le_bytes()); - buf[13] = h.flags; + buf[9..17].copy_from_slice(&h.captured_us.to_le_bytes()); + buf[17..25].copy_from_slice(&h.sent_us.to_le_bytes()); + buf[25] = h.flags; } /// Split a PyroWave datagram into header and payload. `None` for anything that @@ -457,8 +463,9 @@ pub fn decode_pyro_datagram(buf: &[u8]) -> Option<(PyroDatagramHeader, &[u8])> { index: u16_at(3), total: u16_at(5), critical: u16_at(7), - ts_ms: u32::from_le_bytes([buf[9], buf[10], buf[11], buf[12]]), - flags: buf[13], + captured_us: u64::from_le_bytes(buf[9..17].try_into().unwrap()), + sent_us: u64::from_le_bytes(buf[17..25].try_into().unwrap()), + flags: buf[25], }; if h.total == 0 || usize::from(h.total) > PYRO_MAX_DATAGRAMS || h.index >= h.total { return None; @@ -520,6 +527,7 @@ pub fn pack_datagrams( seq: u16, frame: &PyroFrame, max_datagram: usize, + sent_us: u64, ) -> Result { let payload = max_datagram.saturating_sub(PYRO_DGRAM_HDR_LEN); if payload < MIN_DGRAM_PAYLOAD { @@ -582,7 +590,8 @@ pub fn pack_datagrams( index: index as u16, total, critical: critical_entries as u16, - ts_ms: frame.meta.ts_ms, + captured_us: frame.meta.captured_us, + sent_us, flags: flags | duplicate, }, ); @@ -652,7 +661,7 @@ impl> AsRef<[u8]> for CollectedPacket

{ #[derive(Debug, Clone, PartialEq, Eq)] pub struct CollectedFrame

{ pub seq: u16, - pub ts_ms: u32, + pub captured_us: u64, /// Datagrams that arrived, out of `total`. pub received: u16, pub total: u16, @@ -698,7 +707,7 @@ pub struct CollectorStats { struct Flight

{ seq: u16, - ts_ms: u32, + captured_us: u64, total: u16, critical: u16, slots: Vec>, @@ -780,7 +789,7 @@ impl> PyroCollector

{ pos, Flight { seq: h.seq, - ts_ms: h.ts_ms, + captured_us: h.captured_us, total: h.total, critical: h.critical, slots: (0..h.total).map(|_| None).collect(), @@ -912,7 +921,7 @@ impl> PyroCollector

{ CollectedFrame { seq: flight.seq, - ts_ms: flight.ts_ms, + captured_us: flight.captured_us, received: flight.received, total: flight.total, critical_complete, @@ -939,7 +948,7 @@ mod tests { PyroFrame { frame: 7, meta: FrameMeta { - ts_ms: 1234, + captured_us: 1234, width: 1920, height: 1080, }, @@ -1122,7 +1131,7 @@ mod tests { #[test] fn a_packet_that_fits_takes_one_datagram_whole() { let frame = frame_of(&[8, 500, 1200 - PYRO_DGRAM_HDR_LEN, 300], 0); - let d = pack_datagrams(3, &frame, 1200).unwrap(); + let d = pack_datagrams(3, &frame, 1200, 77).unwrap(); let got = unpack(&d); assert_eq!(d.total, 4); assert_eq!(got.len(), 4); @@ -1130,7 +1139,8 @@ mod tests { assert_eq!(h.index as usize, i); assert_eq!(h.flags, 0); assert_eq!(h.seq, 3); - assert_eq!(h.ts_ms, 1234); + assert_eq!(h.captured_us, 1234); + assert_eq!(h.sent_us, 77, "the send time did not reach the header"); assert_eq!( p.as_slice(), frame.packet(i), @@ -1142,7 +1152,7 @@ mod tests { #[test] fn critical_datagrams_go_twice_under_the_same_index() { let frame = frame_of(&[8, 900, 900, 900, 900], 2); - let got = unpack(&pack_datagrams(0, &frame, 1200).unwrap()); + let got = unpack(&pack_datagrams(0, &frame, 1200, 0).unwrap()); assert_eq!(got.len(), 5 + 2); let tail: Vec<_> = got[5..].iter().map(|(h, _)| (h.index, h.flags)).collect(); assert_eq!(tail, [(0, FLAG_DUPLICATE), (1, FLAG_DUPLICATE)]); @@ -1153,7 +1163,7 @@ mod tests { #[test] fn an_oversized_block_goes_in_marked_pieces() { let frame = frame_of(&[8, 2488, 100], 2); - let got = unpack(&pack_datagrams(0, &frame, 1200).unwrap()); + let got = unpack(&pack_datagrams(0, &frame, 1200, 0).unwrap()); let flags: Vec = got.iter().map(|(h, _)| h.flags).collect(); let part = FLAG_PART; assert_eq!( @@ -1179,12 +1189,12 @@ mod tests { fn packing_refuses_a_path_too_small_or_a_frame_too_large() { let frame = frame_of(&[8], 0); assert!(matches!( - pack_datagrams(0, &frame, PYRO_DGRAM_HDR_LEN + MIN_DGRAM_PAYLOAD - 1), + pack_datagrams(0, &frame, PYRO_DGRAM_HDR_LEN + MIN_DGRAM_PAYLOAD - 1, 0), Err(PackError::PathTooSmall { .. }) )); let frame = frame_of(&vec![100; PYRO_MAX_DATAGRAMS + 1], 0); assert!(matches!( - pack_datagrams(0, &frame, 1200), + pack_datagrams(0, &frame, 1200, 0), Err(PackError::TooMany { .. }) )); } @@ -1197,7 +1207,8 @@ mod tests { index: 0, total: 2, critical: 1, - ts_ms: 0, + captured_us: 0, + sent_us: 0, flags: 0, }; write_pyro_header(&mut buf, &ok); @@ -1232,7 +1243,7 @@ mod tests { type Wire = Vec<(PyroDatagramHeader, Vec)>; fn wire(seq: u16, frame: &PyroFrame) -> Wire { - unpack(&pack_datagrams(seq, frame, 1200).unwrap()) + unpack(&pack_datagrams(seq, frame, 1200, 0).unwrap()) } fn packets_of(f: &CollectedFrame>) -> Vec> { diff --git a/crates/nesprotocol/src/reliable.rs b/crates/nesprotocol/src/reliable.rs index 00768f3e..e64568b2 100644 --- a/crates/nesprotocol/src/reliable.rs +++ b/crates/nesprotocol/src/reliable.rs @@ -81,11 +81,27 @@ pub const MAX_KEYFRAME_STREAMS_IN_FLIGHT: usize = 2; /// Offset of the flags byte in a video frame payload. /// -/// The payload the hub broadcasts is `[1B codec] [1B flags] [4B ts_ms LE] -/// [2B width LE] [2B height LE] [data]` — see `encode_ipc_frame`, whose header -/// fields these mirror. +/// The payload the hub broadcasts is [`crate::encode_video_payload`]'s. pub const VIDEO_PAYLOAD_FLAGS_OFFSET: usize = 1; +/// `[1B STREAM_KEYFRAME][1B STREAM_VERSION][8B sent_us LE]`, then the frame body. +/// +/// `sent_us` is when the hub began writing the stream, on its +/// [`clock`](crate::clock). A keyframe has arrived when its last byte lands, +/// which on a reliable stream includes any retransmission, so its queueing +/// delay reads higher than a datagram's on the same path. That is what it +/// really cost, and why keyframes are measured apart from datagrams. +pub const KEYFRAME_STREAM_HDR_LEN: usize = 10; + +/// The header a keyframe stream opens with. +pub fn keyframe_stream_header(sent_us: u64) -> [u8; KEYFRAME_STREAM_HDR_LEN] { + let mut header = [0u8; KEYFRAME_STREAM_HDR_LEN]; + header[0] = crate::STREAM_KEYFRAME; + header[1] = crate::STREAM_VERSION; + header[2..].copy_from_slice(&sent_us.to_le_bytes()); + header +} + /// Whether a video frame payload warrants a reliable stream over datagrams. /// /// True for a keyframe, and for the reconfiguration frame that follows a codec @@ -120,7 +136,10 @@ pub fn video_is_keyframe(payload: &[u8]) -> bool { #[cfg(test)] mod tests { use super::*; - use crate::{CODEC_H264, STREAM_VIDEO, decode_ipc_frame, encode_ipc_frame}; + use crate::{ + CODEC_H264, STREAM_VIDEO, VideoHead, decode_ipc_frame, encode_ipc_frame, + encode_video_payload, + }; /// Build the payload the hub broadcasts out of an IPC frame, the way /// `ipc_listener` does, so the offset is checked against its real producer @@ -136,12 +155,16 @@ mod tests { &[9, 9, 9], ); let decoded = decode_ipc_frame(&ipc).expect("valid ipc frame"); - let mut payload = vec![decoded.codec, decoded.flags]; - payload.extend_from_slice(&decoded.timestamp_ms.to_le_bytes()); - payload.extend_from_slice(&decoded.width.to_le_bytes()); - payload.extend_from_slice(&decoded.height.to_le_bytes()); - payload.extend_from_slice(decoded.data); - payload + encode_video_payload( + &VideoHead { + codec: decoded.codec, + flags: decoded.flags, + captured_us: decoded.stamp, + width: decoded.width, + height: decoded.height, + }, + decoded.data, + ) } #[test] diff --git a/crates/nespyro/tests/transport.rs b/crates/nespyro/tests/transport.rs index 753bf48c..4a3fe751 100644 --- a/crates/nespyro/tests/transport.rs +++ b/crates/nespyro/tests/transport.rs @@ -50,7 +50,7 @@ fn readiness<'a>(config: &EncodeConfig, packets: impl Iterator) /// the network drops. fn across_the_wire(frame: &EncodedFrame, lose: impl Fn(usize) -> bool) -> CollectedFrame> { let meta = FrameMeta { - ts_ms: 16, + captured_us: 16_000, width: 1920, height: 1080, }; @@ -71,7 +71,7 @@ fn across_the_wire(frame: &EncodedFrame, lose: impl Fn(usize) -> bool) -> Collec } let whole = whole.expect("the hub never completed the frame"); - let datagrams = pack_datagrams(0, &whole, DATAGRAM).unwrap(); + let datagrams = pack_datagrams(0, &whole, DATAGRAM, 0).unwrap(); let mut client = PyroCollector::>::new(); let mut out = Vec::new(); let mut at = 0;