diff --git a/README.md b/README.md index 4bd6610..599e07d 100644 --- a/README.md +++ b/README.md @@ -409,9 +409,10 @@ it is unhealthy. Below that it shows: - **Relays**: each relay's state, and the protocol its session negotiated. - **Uplink**: the round trip to the first connected relay, the bandwidth the - connection estimates it can carry, the media every track writes, and the - share lost and resent. Loss and headroom are graded only while someone - watches, since the relay pulls nothing otherwise. + connection estimates it can carry, the media the program's tracks write, and + the share lost and resent. Loss and headroom are graded only while someone + watches, since the relay pulls nothing otherwise. Source tracks are left out + of the media, since they encode whether or not anyone pulls them. - **Watching**: the sessions subscribed to the broadcast, from the relays' `.stats` broadcasts. A relay shares those only with a connection allowed to read them, so most show "unknown". In a cluster, a relay forwarding the @@ -420,8 +421,10 @@ it is unhealthy. Below that it shows: the newer instance, or the first relay's when they are one instance. - **Tracks**: each track's bitrate and frame rate, and, from viewers that report, the lowest bitrate any received, its worst grade, and the latency - from the program clock to the slowest viewer. Latency compares the two - hosts' wall clocks, so keep both synced. + from the program clock to the slowest viewer. The program's tracks come + first, then each source's own tracks, grouped by source; a still image or + color carries no media to grade. Latency compares the two hosts' wall + clocks, so keep both synced. - **Viewers**: each viewer that reports, with its round trip, latency, and playout delay. The studio reads 64 at most, a sample the grades rest on: it counts the rest, and a reader one frees goes to the next viewer to arrive. diff --git a/quest/m0/README.md b/quest/m0/README.md index 7c7296f..461ee6b 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -87,7 +87,6 @@ repository. - [Broadcast sources play in sync](/quest/m0/broadcast-sync.md) - subscribed broadcasts and guests play audio and video on the publisher's timestamps, with `moq play`'s delay - [Wayland display and window sources](/quest/m0/wayland-sources.md) - portal-picked sources that reopen without the picker - [Live encode control](/quest/m0/encode-control.md) - simulcast rungs that follow the bandwidth grant, live bitrate and resolution, keyframes on request -- [Health grades source tracks](/quest/m0/health-sources.md) - the Health panel covers each published source track, not only the program - [Audio sources report stalls](/quest/m0/audio-stall.md) - a dead microphone or system audio reads stalled like a camera - [Instant replay and DVR](/quest/m0/replay.md) - replays and clips from held groups, and viewer scrubbing through the archive - [HTML/CSS overlays](/quest/m0/overlays.md) - script-free HTML and CSS graphics through Blitz natively and the DOM in the browser diff --git a/quest/m0/health-sources.md b/quest/m0/health-sources.md deleted file mode 100644 index 95354ea..0000000 --- a/quest/m0/health-sources.md +++ /dev/null @@ -1,22 +0,0 @@ -# [S] Health grades source tracks - -## Goal - -The Health panel grades each published source track the way it grades the -program's: sent rate, viewers reporting, and the worst grade and latency they -report, so a source someone watches directly is covered too. - -## Plan - -Source tracks are listed in their own nested catalogs under the studio's -catalog (see `README.md`, Source tracks), and the health sampler reads only -the root renditions today. Viewers report per rendition through `.echo`, so -the per-track table and its causes extend to the source renditions rather than -adding a second path. Keep the table readable when there are many sources, for -example grouped by source. Raised on -[#3](https://github.com/moq-dev/studio/pull/3). - -## Related - -- [Source tracks](/quest/m0/compose/sources.md) - the audio source tracks this - grades once they land diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 3fbb6b8..3d54ad3 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -60,11 +60,13 @@ pub struct Publish { owners: Arc>, } -/// Which [`Claim`] owns each source name's entry in the studio catalog. +/// Which [`Claim`] owns each source name's entry in the studio catalog, and +/// each open [`Listing`]'s catalog by its claim. #[derive(Default)] struct Owners { next: u64, by_name: BTreeMap, + catalogs: BTreeMap, } impl Publish { @@ -90,6 +92,17 @@ impl Publish { } } + /// What each source track the studio catalog lists has written, keyed by + /// rendition ID, as [`Catalog::stats`](moq_mux::catalog::Producer::stats) + /// reports the program's. The broadcast names every track uniquely, so all + /// sources share one map. A replaced source's tracks drop out with its + /// catalog entry, even while it is still stopping. + pub fn stats(&self) -> BTreeMap { + let owners = self.owners.lock().unwrap(); + let listed = owners.by_name.values().filter_map(|id| owners.catalogs.get(id)); + listed.flat_map(|catalog| catalog.stats()).collect() + } + /// Claim `name`'s entry in the studio catalog for a source that is opening, /// taking it from any older source of that name. pub(crate) fn claim(&self, name: &str) -> Claim { @@ -155,6 +168,7 @@ impl<'a> Listing<'a> { .with_bandwidth(publish.bandwidth.clone()) .with_hls(None); let catalog = moq_mux::catalog::Producer::new(&mut scratch, config)?; + claim.owners.lock().unwrap().catalogs.insert(claim.id, catalog.clone()); Ok(Self { claim, kind: publish.kind.clone(), @@ -194,6 +208,7 @@ impl<'a> Listing<'a> { impl Drop for Listing<'_> { fn drop(&mut self) { + self.claim.owners.lock().unwrap().catalogs.remove(&self.claim.id); // Fails only once the studio catalog closed, which took the entry with it. let _ = self.studio.mutate(|root| { if self.claim.owned() { @@ -539,6 +554,20 @@ mod tests { } } => {} } + + // Every frame written counts for the Health panel, while the + // studio catalog lists the source. + let stats = publish.stats(); + assert_eq!( + stats.values().map(|rendition| rendition.frames).sum::(), + 5, + "{stats:?}" + ); + // A replacement takes the listing, and the old tracks with it. + let _newer = publish.claim("camera"); + assert!(publish.stats().is_empty()); + drop(listing); + assert!(publish.owners.lock().unwrap().catalogs.is_empty()); }) .await; } diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 2c0cb44..51d4922 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -693,6 +693,7 @@ async fn publish( relays: links, feedback, watching, + sources: sources.clone(), }; let output = Output { diff --git a/rs/moq-studio/src/health.rs b/rs/moq-studio/src/health.rs index 215aabf..5e19011 100644 --- a/rs/moq-studio/src/health.rs +++ b/rs/moq-studio/src/health.rs @@ -47,7 +47,8 @@ pub struct Report { pub relays: Vec, pub uplink: Uplink, pub viewers: Viewers, - /// Each track of the broadcast, in catalog order. + /// Each track of the broadcast in catalog order: the program's, then each + /// source's by source name. pub tracks: Vec, /// What is wrong, worst first, each naming its cause. pub problems: Vec, @@ -84,7 +85,9 @@ pub struct Uplink { pub rtt: Option, /// What the congestion controller estimates the connection can carry. pub capacity: Option, - /// The media every track wrote over the last interval. + /// The media the program's tracks wrote over the last interval. Source + /// tracks are left out: they encode whether or not anyone pulls them, and + /// the relay uploads only what someone does. pub media: Option, /// Transport bytes lost per media byte written over the last interval. pub loss: Option, @@ -135,6 +138,8 @@ pub struct Viewer { /// One track of the broadcast. #[derive(Debug)] pub struct Track { + /// The source whose own track it is, or `None` for the program's. + pub source: Option, /// Its catalog rendition ID. pub id: String, /// What it carries, such as `video H.264 1920x1080`. @@ -153,6 +158,16 @@ pub struct Track { pub latency: Option, } +impl Track { + /// How a problem names it. + fn name(&self) -> String { + match &self.source { + Some(source) => format!("source `{source}`'s {}", self.label), + None => self.label.clone(), + } + } +} + /// Something wrong with the stream, and why. #[derive(Debug, Clone, PartialEq)] pub struct Problem { @@ -196,6 +211,8 @@ pub struct Publisher { pub relays: Vec, pub feedback: Feedback, pub watching: Watching, + /// Where the sources publish their own tracks, for what those wrote. + pub sources: moq_scene::publish::Publish, } /// The connection to one relay. @@ -217,6 +234,7 @@ struct Previous { at: Instant, local: Local, report: Option>, + sources: Option>, /// Which relay's link the report's transport readings came from. serving: Option, } @@ -240,33 +258,39 @@ impl Sampler { let mut report = Report::default(); let mut graded = Graded::default(); - let mut current = None; + let (mut current, mut sources) = (None, None); let mut serving = None; if let Some(publisher) = &self.publisher { report.relays = publisher.relays.iter().map(relay).collect(); serving = report.relays.iter().position(|relay| relay.state == State::Connected); let link = serving.map(|index| &publisher.relays[index]); - let snapshot = snapshot(publisher, link.map(|link| &link.monitor)); - let ours = health::Report::new(publisher.observer.clone(), snapshot); - let mut sample = health::Sample::new(&ours, Duration::ZERO); + let ours = health::Report::new( + publisher.observer.clone(), + snapshot(publisher, link.map(|link| &link.monitor)), + ); // Another relay's link counted its transport from another start. - let previous_report = previous + let previous_ours = previous .as_ref() .filter(|previous| previous.serving == serving) .and_then(|previous| previous.report.as_ref()); - if let (Some(previous), Some(interval)) = (previous_report, interval) { - sample = sample.with_previous(previous, interval); - } - let graded_ours = health::publisher(&sample, &self.thresholds); + let graded_ours = self.grade(&ours, previous_ours.zip(interval)); report.uplink = uplink(&ours.snapshot, &graded_ours, local.watched); report.uplink.relay = link.map(|link| link.endpoint.clone()); graded.publisher = Some(graded_ours); + // Graded apart from the program's, so their media stays out of the + // uplink's; see [`Uplink::media`]. + let mut snapshot = hang::stats::Snapshot::default(); + snapshot.renditions = publisher.sources.stats(); + let theirs = health::Report::new(publisher.observer.clone(), snapshot); + let previous_theirs = previous.as_ref().and_then(|previous| previous.sources.as_ref()); + graded.sources = Some(self.grade(&theirs, previous_theirs.zip(interval))); graded.viewers = publisher.feedback.grade(now, &self.thresholds); report.viewers.unread = publisher.feedback.unread(); report.viewers.watching = publisher.watching.get(); - report.tracks = tracks(publisher, &graded); + report.tracks = tracks(&publisher.catalog.snapshot(), &publisher.clock, &graded); report.viewers.reports = viewers(publisher, &graded.viewers); current = Some(ours); + sources = Some(theirs); } let drops = previous.as_ref().map(|previous| Drops { @@ -288,10 +312,24 @@ impl Sampler { at: now, local, report: current, + sources, serving, }); report } + + /// Grade what the studio wrote, against `previous` and the interval since. + fn grade( + &self, + current: &health::Report, + previous: Option<(&health::Report, Duration)>, + ) -> Health { + let mut sample = health::Sample::new(current, Duration::ZERO); + if let Some((previous, interval)) = previous { + sample = sample.with_previous(previous, interval); + } + health::publisher(&sample, &self.thresholds) + } } fn relay(link: &Link) -> Relay { @@ -361,28 +399,21 @@ fn uplink(snapshot: &hang::stats::Snapshot, graded: &Health, watched: bool) -> U } } -/// Each catalog track, with what it sent and what its viewers received. -fn tracks(publisher: &Publisher, graded: &Graded) -> Vec { - let catalog = publisher.catalog.snapshot(); - let sent = graded.publisher.as_ref(); - let video = catalog.video.renditions.iter().map(|(id, config)| { - let size = match (config.coded_width, config.coded_height) { - (Some(width), Some(height)) => format!(" {width}x{height}"), - _ => String::new(), - }; - (id, format!("video {}{size}", config.codec)) +/// Each media track `catalog` lists, the program's and then each source's, +/// with what it sent and what its viewers received. A source's stills carry +/// no media to grade, so they are left out. +fn tracks(catalog: &hang::Catalog, clock: &moq_mux::Clock, graded: &Graded) -> Vec { + let program = renditions(catalog).map(|(id, label)| (None, id, label, graded.publisher.as_ref())); + let sources = catalog.ext.studio.sources.iter().flat_map(|(name, catalog)| { + renditions(catalog).map(move |(id, label)| (Some(name), id, label, graded.sources.as_ref())) }); - let audio = catalog - .audio - .renditions - .iter() - .map(|(id, config)| (id, format!("audio {} {} ch", config.codec, config.channel_count))); - video - .chain(audio) - .map(|(id, label)| { + program + .chain(sources) + .map(|(source, id, label, sent)| { let ours = sent.and_then(|health| health.renditions.get(id)); let interval = sent.and_then(|health| health.interval); let mut track = Track { + source: source.cloned(), id: id.clone(), label, sent: ours.and_then(|rendition| rendition.bitrate).map(Rate::from_bps), @@ -405,7 +436,7 @@ fn tracks(publisher: &Publisher, graded: &Graded) -> Vec { if let Some(bitrate) = rendition.bitrate.map(Rate::from_bps) { track.received = Some(track.received.map_or(bitrate, |received| received.min(bitrate))); } - let latency = viewer.latency(id, &publisher.clock); + let latency = viewer.latency(id, clock); track.latency = track.latency.max(latency); } track @@ -413,6 +444,23 @@ fn tracks(publisher: &Publisher, graded: &Graded) -> Vec { .collect() } +/// Each video and audio rendition `catalog` lists, by ID, with what it carries. +fn renditions(catalog: &hang::Catalog) -> impl Iterator { + let video = catalog.video.renditions.iter().map(|(id, config)| { + let size = match (config.coded_width, config.coded_height) { + (Some(width), Some(height)) => format!(" {width}x{height}"), + _ => String::new(), + }; + (id, format!("video {}{size}", config.codec)) + }); + let audio = catalog + .audio + .renditions + .iter() + .map(|(id, config)| (id, format!("audio {} {} ch", config.codec, config.channel_count))); + video.chain(audio) +} + fn viewers(publisher: &Publisher, graded: &[GradedViewer]) -> Vec { graded .iter() @@ -445,7 +493,10 @@ fn viewers(publisher: &Publisher, graded: &[GradedViewer]) -> Vec { /// The grades of one interval. #[derive(Default)] struct Graded { + /// The program's tracks and the uplink. publisher: Option, + /// The sources' own tracks. + sources: Option, viewers: Vec, } @@ -654,7 +705,7 @@ fn viewer_problems(facts: &Facts<'_>, add: &mut impl FnMut(Verdict, String)) { let track = |id: Option<&str>| { id.and_then(|id| facts.labels.iter().find(|track| track.id == id)) - .map_or_else(|| "the stream".to_string(), |track| track.label.clone()) + .map_or_else(|| "the stream".to_string(), Track::name) }; for (input, verdict, count, id) in worst { let of = format!("{count} of {reporting} viewers"); @@ -958,6 +1009,7 @@ mod tests { fn track(id: &str, label: &str) -> Track { Track { + source: None, id: id.into(), label: label.into(), sent: None, @@ -1091,6 +1143,80 @@ mod tests { ); } + /// What the studio wrote over one interval, graded: `bytes` and 30 frames + /// for each rendition. + fn wrote(renditions: &[(&str, u64)]) -> Health { + let observer = Observer::new(moq_net::Path::new("show").unwrap(), None); + let report = |scale: u64| { + let mut snapshot = hang::stats::Snapshot::<()>::default(); + for (id, bytes) in renditions { + let mut rendition = hang::stats::Rendition::default(); + rendition.bytes = bytes * scale; + rendition.frames = 30 * scale; + snapshot.renditions.insert(id.to_string(), rendition); + } + health::Report::new(observer.clone(), snapshot) + }; + let (previous, current) = (report(0), report(1)); + let sample = health::Sample::new(¤t, Duration::ZERO).with_previous(&previous, INTERVAL); + health::publisher(&sample, &Thresholds::default()) + } + + /// Each source's media tracks are graded like the program's, after them and + /// by source name, from what the source wrote and what viewers report on + /// them; a still has no media to grade. A viewer stalling on one names + /// its source. + #[test] + fn source_tracks_are_graded_by_source() { + let video = |width: u32, height: u32| { + let codec: hang::catalog::VideoCodec = "avc3.42c01f".parse().unwrap(); + let mut config = hang::catalog::VideoConfig::new(codec); + config.coded_width = Some(width); + config.coded_height = Some(height); + config + }; + let mut catalog = hang::Catalog::::default(); + catalog.video.renditions.insert("0.avc3".into(), video(1920, 1080)); + let mut camera = hang::Catalog::default(); + camera.video.renditions.insert("2.avc3".into(), video(1280, 720)); + let mut logo = hang::Catalog::default(); + let png = hang::catalog::BinaryConfig::new(hang::catalog::Mode::Snapshot); + logo.binary.tracks.insert("1.png".into(), png); + let sources = &mut catalog.ext.studio.sources; + sources.insert("logo".into(), logo); + sources.insert("camera".into(), camera); + + // 375 kB in a second is 3 Mb/s, and 125 kB is 1 Mb/s. + let graded = Graded { + publisher: Some(wrote(&[("0.avc3", 375_000)])), + sources: Some(wrote(&[("2.avc3", 125_000)])), + viewers: vec![stalling("alice", "2.avc3", Duration::from_millis(100))], + }; + let tracks = tracks(&catalog, &moq_mux::Clock::new(), &graded); + let rows: Vec<_> = tracks + .iter() + .map(|track| (track.source.as_deref(), track.id.as_str(), track.sent.map(rate))) + .collect(); + assert_eq!( + rows, + [ + (None, "0.avc3", Some("3.0 Mb/s".into())), + (Some("camera"), "2.avc3", Some("1.0 Mb/s".into())), + ] + ); + let camera = &tracks[1]; + assert_eq!(camera.framerate, Some(30.0)); + assert_eq!((camera.viewers, camera.verdict), (1, Verdict::Unhealthy)); + assert_eq!((tracks[0].viewers, tracks[0].verdict), (0, Verdict::Unknown)); + + let local = Local::default(); + let uplink = Uplink::default(); + assert_eq!( + causes(problems(&facts(&local, &uplink, &graded.viewers, &tracks))), + ["1 of 1 viewers are stalling on source `camera`'s video avc3.42c01f 1280x720."] + ); + } + /// Problems list worst first: an unreachable relay and a frozen source /// above a slow round trip. #[test] @@ -1235,6 +1361,12 @@ mod tests { }], feedback: feedback.clone(), watching: Watching::default(), + sources: moq_scene::publish::Publish::new( + broadcast.clone(), + catalog.clone(), + moq_net::bandwidth::Allocator::unlimited(), + moq_video::encode::Kind::Auto, + ), })); let alice = origin.create_broadcast("show/viewers/alice.echo").unwrap(); diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 0e47ef6..8a24348 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1613,6 +1613,7 @@ impl Panels { egui::Grid::new("tracks").striped(true).show(ui, |ui| { for heading in [ + "Source", "Track", "Sent", "Frames", @@ -1624,7 +1625,18 @@ impl Panels { ui.strong(heading); } ui.end_row(); + // Each source named once, beside its first track. The program's + // tracks come first and name none, so no source name can pass + // for them. + let mut group = None; for track in &report.tracks { + let source = track.source.as_deref(); + ui.label(if source == group { + "" + } else { + source.unwrap_or_default() + }); + group = source; ui.label(&track.label).on_hover_text(&track.id); ui.label(or_unknown(track.sent.map(health::rate))); ui.label(or_unknown(track.framerate.map(|framerate| format!("{framerate:.0}/s"))));