From eb5297ff405a0afe52a86e6842ad188174141edd Mon Sep 17 00:00:00 2001 From: David Chen Date: Tue, 28 Jul 2026 16:02:15 -0700 Subject: [PATCH 1/9] Add latency metrics CSV logging --- .changeset/local-video-latency-report.md | 5 + examples/local_video/README.md | 35 ++ .../scripts/generate_frame_report.py | 483 ++++++++++++++++++ examples/local_video/src/frame_log.rs | 124 +++++ examples/local_video/src/publisher.rs | 220 +++++++- examples/local_video/src/subscriber.rs | 104 +++- examples/local_video/src/subscriber_timing.rs | 237 ++++++++- 7 files changed, 1185 insertions(+), 23 deletions(-) create mode 100644 .changeset/local-video-latency-report.md create mode 100755 examples/local_video/scripts/generate_frame_report.py create mode 100644 examples/local_video/src/frame_log.rs diff --git a/.changeset/local-video-latency-report.md b/.changeset/local-video-latency-report.md new file mode 100644 index 000000000..ee8e33fa3 --- /dev/null +++ b/.changeset/local-video-latency-report.md @@ -0,0 +1,5 @@ +--- +"local_video": patch +--- + +Add frame-range CSV timing and delivery-quality logging to the local video publisher and subscriber, plus a PDF report generator for either or both logs. diff --git a/examples/local_video/README.md b/examples/local_video/README.md index 966b18145..731b621aa 100644 --- a/examples/local_video/README.md +++ b/examples/local_video/README.md @@ -100,6 +100,15 @@ Publisher usage: --room-name demo \ --identity cam-1 \ --display-video + + # log publisher metrics for frames 301 through 1200 (inclusive) + cargo run --release -p local_video -F desktop --bin publisher -- \ + --camera-index 0 \ + --room-name demo \ + --identity cam-1 \ + --log-csv publisher.csv \ + --log-start-frame-id 301 \ + --log-end-frame-id 1200 ``` List devices usage: @@ -136,6 +145,9 @@ Publisher flags (in addition to the common connection flags above): - `--attach-frame-id`: Attach a monotonically increasing frame ID to each published frame via the packet trailer. The subscriber displays this in the timestamp overlay when `--display-timestamp` is used. - `--display-video`: Open a window that displays the video frames being published. - `--display-timing`: Burn publisher timing metrics into the local preview window. Requires `--display-video`. +- `--log-csv `: Write one CSV row per packetized frame with capture, encoder, packetization, frame-gap, and inter-frame timing metrics. This automatically enables timestamp and frame-ID metadata. +- `--log-start-frame-id `: Start CSV logging at this frame ID (inclusive). Requires `--log-csv`. +- `--log-end-frame-id `: Stop CSV logging after this frame ID (inclusive). Requires `--log-csv`. - `--e2ee-key `: Enable end-to-end encryption with the given shared key. The subscriber must use the same key to decrypt. Subscriber usage: @@ -174,14 +186,37 @@ Subscriber usage: --room-name demo \ --identity viewer-1 \ --e2ee-key my-secret-key + + # log rendered-frame metrics for the same inclusive frame-ID window + cargo run --release -p local_video -F desktop --bin subscriber -- \ + --room-name demo \ + --identity viewer-1 \ + --log-csv subscriber.csv \ + --log-start-frame-id 301 \ + --log-end-frame-id 1200 ``` Subscriber flags (in addition to the common connection flags above): - `--participant `: Only subscribe to video tracks from the specified participant. - `--low-latency`: Force zero video playout delay so received frames render as soon as possible. This can increase visible stutter when packets arrive late or out of order. - `--display-timestamp`: Show detailed frame ID, publisher timestamp, subscriber timing stages, and end-to-end latency in the separate diagnostics window. Timestamp fields require the publisher to use `--attach-timestamp`; frame ID requires `--attach-frame-id`. +- `--log-csv `: Write one CSV row per rendered frame with receive, decode, sink, paint, end-to-end latency, frame-gap, inter-frame timing, and WebRTC loss/freeze metrics. The publisher must use `--log-csv` or both `--attach-timestamp` and `--attach-frame-id`. +- `--log-start-frame-id `: Start CSV logging at this frame ID (inclusive). Requires `--log-csv`. +- `--log-end-frame-id `: Stop CSV logging after this frame ID (inclusive). Requires `--log-csv`. - `--e2ee-key `: Enable end-to-end decryption with the given shared key. Must match the key used by the publisher. +Generate a PDF report from the publisher log, subscriber log, or both: +``` + python3 -m pip install reportlab + + python3 examples/local_video/scripts/generate_frame_report.py \ + --publisher publisher.csv \ + --subscriber subscriber.csv \ + --output frame-report.pdf +``` + +Omit either `--publisher` or `--subscriber` to create a single-sided report. The report plots latency over the logged duration and marks frame-ID gaps and freezes. With paired logs, frame loss is the set of packetized publisher frame IDs that were not rendered by the subscriber. Subscriber freezes use WebRTC's reported freeze counters; publisher-only reports infer a freeze from an inter-frame gap greater than three times the median interval. + Notes: - If the active video track is unsubscribed or unpublished, the app clears its state and will automatically attach to the next matching video track when it appears. - For E2EE to work, both publisher and subscriber must specify the same `--e2ee-key` value. If the keys don't match, the subscriber will not be able to decode the video. diff --git a/examples/local_video/scripts/generate_frame_report.py b/examples/local_video/scripts/generate_frame_report.py new file mode 100755 index 000000000..6fec5836e --- /dev/null +++ b/examples/local_video/scripts/generate_frame_report.py @@ -0,0 +1,483 @@ +#!/usr/bin/env python3 +"""Generate a concise PDF report from local_video per-frame CSV logs.""" + +from __future__ import annotations + +import argparse +import csv +import math +import statistics +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable, Sequence + +try: + from reportlab.lib.colors import HexColor, white + from reportlab.lib.pagesizes import landscape, letter + from reportlab.pdfgen import canvas +except ImportError as error: + raise SystemExit( + "reportlab is required; install it with: python3 -m pip install reportlab" + ) from error + + +NAVY = HexColor("#102A43") +BLUE = HexColor("#147D92") +CYAN = HexColor("#2CB1BC") +INK = HexColor("#243B53") +MUTED = HexColor("#627D98") +GRID = HexColor("#D9E2EC") +PANEL = HexColor("#F0F4F8") +RED = HexColor("#D64545") +ORANGE = HexColor("#E88D14") + + +@dataclass(frozen=True) +class LogData: + kind: str + path: Path + rows: list[dict[str, str]] + latency_column: str + interval_column: str + + @property + def label(self) -> str: + return "Publisher" if self.kind == "publisher" else "Subscriber" + + +@dataclass(frozen=True) +class Event: + elapsed_ms: float + count: int + duration_ms: float = 0.0 + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Generate a PDF from publisher and/or subscriber --log-csv output." + ) + parser.add_argument("--publisher", type=Path, help="Publisher CSV log") + parser.add_argument("--subscriber", type=Path, help="Subscriber CSV log") + parser.add_argument("-o", "--output", type=Path, help="Output PDF path") + parser.add_argument("--title", default="Local Video Frame Metrics") + args = parser.parse_args() + if args.publisher is None and args.subscriber is None: + parser.error("at least one of --publisher or --subscriber is required") + if args.output is None: + source = args.subscriber or args.publisher + assert source is not None + args.output = source.with_suffix(".pdf") + return args + + +def number(value: str | None) -> float | None: + if value is None or not value.strip(): + return None + try: + parsed = float(value) + except ValueError: + return None + return parsed if math.isfinite(parsed) else None + + +def values(rows: Iterable[dict[str, str]], column: str) -> list[float]: + return [parsed for row in rows if (parsed := number(row.get(column))) is not None] + + +def read_log(path: Path, kind: str) -> LogData: + latency_column = "capture_to_packetize_ms" if kind == "publisher" else "e2e_latency_ms" + interval_column = "packetize_interval_ms" if kind == "publisher" else "render_interval_ms" + with path.open(newline="", encoding="utf-8") as source: + reader = csv.DictReader(source) + required = {"elapsed_ms", "frame_id", latency_column} + missing = required.difference(reader.fieldnames or ()) + if missing: + raise ValueError(f"{path} is not a {kind} frame log; missing {', '.join(sorted(missing))}") + rows = [row for row in reader if number(row.get(latency_column)) is not None] + if not rows: + raise ValueError(f"{path} contains no completed {kind} frame samples") + return LogData(kind, path, rows, latency_column, interval_column) + + +def percentile(samples: Sequence[float], percent: float) -> float: + ordered = sorted(samples) + if len(ordered) == 1: + return ordered[0] + position = (len(ordered) - 1) * percent / 100.0 + lower = math.floor(position) + upper = math.ceil(position) + if lower == upper: + return ordered[lower] + return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower) + + +def series(log: LogData) -> list[tuple[float, float]]: + result = [] + for row in log.rows: + elapsed = number(row.get("elapsed_ms")) + latency = number(row.get(log.latency_column)) + if elapsed is not None and latency is not None: + result.append((elapsed, latency)) + return result + + +def gap_events(log: LogData) -> list[Event]: + events = [] + for row in log.rows: + elapsed = number(row.get("elapsed_ms")) + gap = number(row.get("frame_id_gap")) + if elapsed is not None and gap is not None and gap > 0: + events.append(Event(elapsed, round(gap))) + return events + + +def inferred_freeze_events(log: LogData) -> list[Event]: + intervals = [value for value in values(log.rows, log.interval_column) if value > 0] + if not intervals: + return [] + expected = statistics.median(intervals) + threshold = expected * 3.0 + events = [] + for row in log.rows: + elapsed = number(row.get("elapsed_ms")) + interval = number(row.get(log.interval_column)) + if elapsed is not None and interval is not None and interval > threshold: + events.append(Event(elapsed, 1, interval - expected)) + return events + + +def subscriber_freeze_events(log: LogData) -> list[Event]: + if log.kind != "subscriber": + return inferred_freeze_events(log) + counts = values(log.rows, "freeze_count") + if not counts: + return inferred_freeze_events(log) + events = [] + previous_count = 0 + previous_duration = 0.0 + for row in log.rows: + elapsed = number(row.get("elapsed_ms")) + count = number(row.get("freeze_count")) + duration = number(row.get("total_freeze_duration_ms")) + if elapsed is None or count is None: + continue + rounded_count = round(count) + if rounded_count > previous_count: + duration_delta = max(0.0, (duration or previous_duration) - previous_duration) + events.append(Event(elapsed, rounded_count - previous_count, duration_delta)) + previous_count = max(previous_count, rounded_count) + if duration is not None: + previous_duration = max(previous_duration, duration) + return events + + +def last_value(log: LogData, column: str) -> float | None: + return next( + (parsed for row in reversed(log.rows) if (parsed := number(row.get(column))) is not None), + None, + ) + + +def paired_loss_events(publisher: LogData, subscriber: LogData) -> list[Event]: + publisher_ids = {round(value) for value in values(publisher.rows, "frame_id")} + subscriber_ids = {round(value) for value in values(subscriber.rows, "frame_id")} + if not publisher_ids or not subscriber_ids: + return [] + low = max(min(publisher_ids), min(subscriber_ids)) + high = min(max(publisher_ids), max(subscriber_ids)) + missing_ids = { + frame_id for frame_id in publisher_ids if low <= frame_id <= high + } - subscriber_ids + events = [] + for row in publisher.rows: + frame_id = number(row.get("frame_id")) + elapsed = number(row.get("elapsed_ms")) + if frame_id is not None and elapsed is not None and round(frame_id) in missing_ids: + events.append(Event(elapsed, 1)) + return events + + +def format_count(value: float | None) -> str: + return "NA" if value is None else f"{round(value):,}" + + +def draw_header(pdf: canvas.Canvas, title: str, subtitle: str) -> None: + width, height = landscape(letter) + pdf.setFillColor(white) + pdf.rect(0, 0, width, height, fill=1, stroke=0) + pdf.setFillColor(NAVY) + pdf.rect(0, height - 72, width, 72, fill=1, stroke=0) + pdf.setFillColor(white) + pdf.setFont("Helvetica-Bold", 21) + pdf.drawString(38, height - 33, title) + pdf.setFillColor(HexColor("#D9F2F4")) + pdf.setFont("Helvetica", 8.5) + pdf.drawString(39, height - 51, subtitle) + + +def draw_card(pdf: canvas.Canvas, x: float, y: float, width: float, label: str, value: str) -> None: + pdf.setFillColor(PANEL) + pdf.roundRect(x, y, width, 48, 5, fill=1, stroke=0) + pdf.setFillColor(MUTED) + pdf.setFont("Helvetica-Bold", 6.8) + pdf.drawString(x + 9, y + 32, label.upper()) + pdf.setFillColor(INK) + pdf.setFont("Helvetica-Bold", 15) + pdf.drawString(x + 9, y + 11, value) + + +def draw_time_series( + pdf: canvas.Canvas, + logs: Sequence[LogData], + loss_events: Sequence[Event], + freeze_events: Sequence[Event], + x: float, + y: float, + width: float, + height: float, +) -> None: + all_series = [(log, series(log)) for log in logs] + latency_values = [latency for _, samples in all_series for _, latency in samples] + duration = max(elapsed for _, samples in all_series for elapsed, _ in samples) + y_max = max(1.0, percentile(latency_values, 99) * 1.2) + + pdf.setFillColor(INK) + pdf.setFont("Helvetica-Bold", 11) + pdf.drawString(x, y + height + 17, "Latency over time") + pdf.setFont("Helvetica", 7.5) + pdf.setFillColor(MUTED) + pdf.drawRightString(x + width, y + height + 17, "milliseconds") + + for tick in range(5): + tick_y = y + height * tick / 4 + pdf.setStrokeColor(GRID) + pdf.line(x, tick_y, x + width, tick_y) + pdf.setFillColor(MUTED) + pdf.setFont("Helvetica", 7) + pdf.drawRightString(x - 7, tick_y - 2, f"{y_max * tick / 4:.0f}") + + colors = {"publisher": CYAN, "subscriber": BLUE} + for log, samples in all_series: + stride = max(1, math.ceil(len(samples) / 1800)) + path = pdf.beginPath() + for index, (elapsed, latency) in enumerate(samples[::stride]): + point_x = x if duration <= 0 else x + width * elapsed / duration + point_y = y + height * min(latency, y_max) / y_max + (path.moveTo if index == 0 else path.lineTo)(point_x, point_y) + pdf.setStrokeColor(colors[log.kind]) + pdf.setLineWidth(1.05) + pdf.drawPath(path, stroke=1, fill=0) + + for event, color, offset in [ + *((event, RED, -1.0) for event in loss_events), + *((event, ORANGE, 1.0) for event in freeze_events), + ]: + event_x = ( + x + if duration <= 0 + else x + width * min(event.elapsed_ms, duration) / duration + offset + ) + event_x = max(x, min(x + width, event_x)) + pdf.setStrokeColor(color) + pdf.setLineWidth(0.55) + pdf.setDash(2, 2) + pdf.line(event_x, y, event_x, y + height) + pdf.setDash() + + legend_x = x + 8 + for label, color in [ + *((log.label, colors[log.kind]) for log in logs), + ("Frame loss", RED), + ("Freeze", ORANGE), + ]: + pdf.setStrokeColor(color) + pdf.setLineWidth(2) + pdf.line(legend_x, y + height - 12, legend_x + 14, y + height - 12) + pdf.setFillColor(MUTED) + pdf.setFont("Helvetica", 7) + pdf.drawString(legend_x + 18, y + height - 15, label) + legend_x += 73 + + pdf.setFillColor(MUTED) + pdf.setFont("Helvetica", 7) + for tick in range(5): + tick_x = x + width * tick / 4 + pdf.drawCentredString(tick_x, y - 13, f"{duration * tick / 4000:.1f}s") + pdf.setStrokeColor(INK) + pdf.rect(x, y, width, height, fill=0, stroke=1) + + +def latency_rows(logs: Sequence[LogData]) -> list[tuple[str, list[float]]]: + metrics = [] + for log in logs: + if log.kind == "publisher": + columns = ( + ("Publisher capture to buffer", "capture_to_buffer_ms"), + ("Publisher encode", "encode_ms"), + ("Publisher capture to packetize", "capture_to_packetize_ms"), + ) + else: + columns = ( + ("Subscriber exposure to receive", "exposure_to_receive_ms"), + ("Subscriber receive to decode", "receive_to_decode_ms"), + ("Subscriber receive to paint", "receive_to_paint_ms"), + ("Subscriber end to end", "e2e_latency_ms"), + ) + metrics.extend((label, values(log.rows, column)) for label, column in columns) + return [(label, samples) for label, samples in metrics if samples] + + +def draw_latency_table( + pdf: canvas.Canvas, logs: Sequence[LogData], x: float, y: float, width: float +) -> None: + rows = latency_rows(logs) + pdf.setFillColor(INK) + pdf.setFont("Helvetica-Bold", 10.5) + pdf.drawString(x, y + 18, "Latency summary") + headers = (("Stage", 0), ("Mean", width - 135), ("P50", width - 88), ("P95", width - 41)) + pdf.setFillColor(NAVY) + pdf.rect(x, y - 4, width, 21, fill=1, stroke=0) + pdf.setFillColor(white) + pdf.setFont("Helvetica-Bold", 7) + for label, offset in headers: + pdf.drawString(x + offset + 7, y + 4, label) + row_y = y - 20 + for index, (label, samples) in enumerate(rows): + pdf.setFillColor(PANEL if index % 2 == 0 else white) + pdf.rect(x, row_y, width, 15, fill=1, stroke=0) + pdf.setFillColor(INK) + pdf.setFont("Helvetica", 7.3) + pdf.drawString(x + 7, row_y + 4.5, label) + for offset, value in zip( + (width - 128, width - 81, width - 34), + (statistics.fmean(samples), percentile(samples, 50), percentile(samples, 95)), + ): + pdf.drawRightString(x + offset, row_y + 4.5, f"{value:.1f}") + row_y -= 15 + + +def draw_delivery_table( + pdf: canvas.Canvas, + publisher: LogData | None, + subscriber: LogData | None, + losses: int, + freezes: Sequence[Event], + x: float, + y: float, + width: float, +) -> None: + if subscriber is not None: + packet_loss = last_value(subscriber, "packets_lost") + dropped = last_value(subscriber, "frames_dropped") + freeze_duration = last_value(subscriber, "total_freeze_duration_ms") + else: + packet_loss = dropped = freeze_duration = None + freeze_count = sum(event.count for event in freezes) + if publisher is not None and subscriber is not None: + loss_label = "Publisher IDs not rendered" + elif subscriber is not None: + loss_label = "Rendered frame-ID gaps" + else: + loss_label = "Packetized frame-ID gaps" + rows = ( + (loss_label, f"{losses:,}"), + ("RTP packets lost", format_count(packet_loss)), + ("WebRTC frames dropped", format_count(dropped)), + ("Freezes", f"{freeze_count:,}"), + ("Freeze duration", "NA" if freeze_duration is None else f"{freeze_duration:.0f} ms"), + ) + pdf.setFillColor(INK) + pdf.setFont("Helvetica-Bold", 10.5) + pdf.drawString(x, y + 18, "Delivery quality") + row_y = y - 4 + for index, (label, value) in enumerate(rows): + pdf.setFillColor(PANEL if index % 2 == 0 else white) + pdf.rect(x, row_y - 20, width, 20, fill=1, stroke=0) + pdf.setFillColor(MUTED) + pdf.setFont("Helvetica", 7.5) + pdf.drawString(x + 7, row_y - 13, label) + pdf.setFillColor(INK) + pdf.setFont("Helvetica-Bold", 8) + pdf.drawRightString(x + width - 7, row_y - 13, value) + row_y -= 20 + + +def generate_report( + publisher: LogData | None, + subscriber: LogData | None, + output: Path, + title: str, +) -> None: + logs = [log for log in (publisher, subscriber) if log is not None] + assert logs + primary = subscriber or publisher + assert primary is not None + primary_latencies = values(primary.rows, primary.latency_column) + duration_ms = max(values(primary.rows, "elapsed_ms"), default=0.0) + + event_log = subscriber or publisher + assert event_log is not None + loss_events = gap_events(event_log) + freeze_events = subscriber_freeze_events(event_log) + if publisher is not None and subscriber is not None: + loss_events = paired_loss_events(publisher, subscriber) + losses = sum(event.count for event in loss_events) + + sources = " + ".join(f"{log.label}: {log.path.name}" for log in logs) + subtitle = f"{sources} | inclusive logged frame range" + output.parent.mkdir(parents=True, exist_ok=True) + pdf = canvas.Canvas(str(output), pagesize=landscape(letter)) + pdf.setTitle(title) + pdf.setAuthor("LiveKit local_video") + draw_header(pdf, title, subtitle) + + cards = ( + ("Rendered frames" if subscriber else "Packetized frames", f"{len(primary.rows):,}"), + ("Duration", f"{duration_ms / 1000:.1f} s"), + ("Mean latency", f"{statistics.fmean(primary_latencies):.1f} ms"), + ("P50 latency", f"{percentile(primary_latencies, 50):.1f} ms"), + ("P95 latency", f"{percentile(primary_latencies, 95):.1f} ms"), + ("Frame losses", f"{losses:,}"), + ) + card_width = 112 + for index, (label, value) in enumerate(cards): + draw_card(pdf, 38 + index * (card_width + 11), 461, card_width, label, value) + + draw_time_series(pdf, logs, loss_events, freeze_events, 50, 206, 692, 205) + draw_latency_table(pdf, logs, 38, 145, 470) + draw_delivery_table(pdf, publisher, subscriber, losses, freeze_events, 530, 145, 224) + + pdf.setStrokeColor(GRID) + pdf.line(38, 28, 754, 28) + pdf.setFillColor(MUTED) + pdf.setFont("Helvetica", 6.8) + freeze_note = ( + "Freeze markers use subscriber WebRTC freeze counters." + if subscriber and values(subscriber.rows, "freeze_count") + else "Freeze markers are inter-frame gaps over 3x the median interval." + ) + pdf.drawString( + 38, + 17, + "Frame losses are frame-ID gaps; with paired logs they are publisher IDs not rendered by the subscriber. " + + freeze_note, + ) + pdf.save() + + +def main() -> int: + args = parse_args() + try: + publisher = read_log(args.publisher, "publisher") if args.publisher else None + subscriber = read_log(args.subscriber, "subscriber") if args.subscriber else None + generate_report(publisher, subscriber, args.output, args.title) + except (OSError, ValueError) as error: + print(f"error: {error}", file=sys.stderr) + return 1 + print(f"Wrote {args.output}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/local_video/src/frame_log.rs b/examples/local_video/src/frame_log.rs new file mode 100644 index 000000000..dbff5a934 --- /dev/null +++ b/examples/local_video/src/frame_log.rs @@ -0,0 +1,124 @@ +use anyhow::{bail, Result}; +use std::{ + fmt, + fs::File, + io::{BufWriter, Write}, + path::Path, +}; + +/// Inclusive frame-ID bounds for per-frame CSV logging. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) struct FrameLogRange { + start: Option, + end: Option, +} + +impl FrameLogRange { + /// Validates optional inclusive frame-ID bounds. + pub(crate) fn new(start: Option, end: Option) -> Result { + if let (Some(start), Some(end)) = (start, end) { + if start > end { + bail!("--log-start-frame-id ({start}) must not exceed --log-end-frame-id ({end})"); + } + } + Ok(Self { start, end }) + } + + /// Returns whether a frame ID falls within the configured inclusive bounds. + pub(crate) fn contains(self, frame_id: u32) -> bool { + self.start.is_none_or(|start| frame_id >= start) + && self.end.is_none_or(|end| frame_id <= end) + } + + /// Returns the frame ID immediately before an explicit start bound, when representable. + pub(crate) fn previous_to_start(self) -> Option { + self.start.and_then(|start| start.checked_sub(1)) + } + + /// Returns whether this frame ID is the configured inclusive end bound. + pub(crate) fn reaches_end(self, frame_id: u32) -> bool { + self.end == Some(frame_id) + } +} + +/// Creates a buffered CSV file, including missing parent directories, and writes its header. +pub(crate) fn create_csv(path: &Path, header: &str) -> std::io::Result> { + if let Some(parent) = path.parent().filter(|parent| !parent.as_os_str().is_empty()) { + std::fs::create_dir_all(parent)?; + } + let mut writer = BufWriter::new(File::create(path)?); + writeln!(writer, "{header}")?; + writer.flush()?; + Ok(writer) +} + +/// Displays an optional CSV cell without adding quoting or placeholder text. +pub(crate) struct CsvOption(pub(crate) Option); + +impl fmt::Display for CsvOption { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + if let Some(value) = &self.0 { + value.fmt(formatter) + } else { + Ok(()) + } + } +} + +/// Displays a timestamp delta in milliseconds when both endpoints are available and ordered. +pub(crate) struct CsvLatency(Option); + +impl CsvLatency { + /// Builds a latency cell from optional microsecond timestamps. + pub(crate) fn between(start_timestamp_us: Option, end_timestamp_us: Option) -> Self { + Self(match (start_timestamp_us, end_timestamp_us) { + (Some(start), Some(end)) => end.checked_sub(start), + _ => None, + }) + } +} + +impl fmt::Display for CsvLatency { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + if let Some(latency_us) = self.0 { + write!(formatter, "{:.3}", latency_us as f64 / 1_000.0) + } else { + Ok(()) + } + } +} + +/// Displays an optional floating-point CSV cell with millisecond precision. +pub(crate) struct CsvFloat(pub(crate) Option); + +impl fmt::Display for CsvFloat { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + if let Some(value) = self.0 { + write!(formatter, "{value:.3}") + } else { + Ok(()) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn frame_log_range_is_inclusive() { + let range = FrameLogRange::new(Some(10), Some(20)).expect("range should be valid"); + assert!(!range.contains(9)); + assert!(range.contains(10)); + assert!(range.contains(20)); + assert!(!range.contains(21)); + assert_eq!(range.previous_to_start(), Some(9)); + assert!(range.reaches_end(20)); + assert!(!range.reaches_end(19)); + } + + #[test] + fn frame_log_range_rejects_reversed_bounds() { + assert!(FrameLogRange::new(Some(20), Some(10)).is_err()); + } +} diff --git a/examples/local_video/src/publisher.rs b/examples/local_video/src/publisher.rs index 28932c5b9..2e7a29050 100644 --- a/examples/local_video/src/publisher.rs +++ b/examples/local_video/src/publisher.rs @@ -1,4 +1,4 @@ -use anyhow::Result; +use anyhow::{Context, Result}; use clap::{Parser, ValueEnum}; use livekit::e2ee::{key_provider::*, E2eeOptions, EncryptionType}; use livekit::options::{ @@ -12,7 +12,7 @@ use livekit::webrtc::video_source::{RtcVideoSource, VideoResolution}; use livekit_api::access_token; use livekit_api::services::room::{CreateRoomOptions, RoomClient}; use livekit_api::services::{ServiceError, TwirpError, TwirpErrorCode}; -use log::{debug, info}; +use log::{debug, info, warn}; use nokhwa::pixel_format::RgbFormat; use nokhwa::utils::{ ApiBackend, CameraFormat, CameraIndex, FrameFormat, RequestedFormat, RequestedFormatType, @@ -22,6 +22,8 @@ use nokhwa::Camera; use parking_lot::Mutex; use std::collections::{HashMap, VecDeque}; use std::env; +use std::io::{BufWriter, Write}; +use std::path::{Path, PathBuf}; use std::sync::{ atomic::{AtomicBool, Ordering}, Arc, @@ -32,6 +34,7 @@ use yuv_sys; #[cfg(all(target_os = "linux", target_arch = "aarch64"))] mod argus; mod codec_display; +mod frame_log; mod test_pattern; mod timestamp_burn; mod user_data; @@ -42,6 +45,8 @@ use test_pattern::TestPattern; use timestamp_burn::TimestampOverlay; use video_display::{align_up, PublisherTimingSample, SharedYuv}; +use frame_log::{create_csv, CsvFloat, CsvLatency, CsvOption, FrameLogRange}; + #[derive(Copy, Clone, Debug, ValueEnum)] enum PublisherCodec { H264, @@ -266,6 +271,18 @@ struct Args { #[arg(long, default_value_t = false, requires = "display_video")] display_timing: bool, + /// Write one row of publisher timing metrics per packetized frame to this CSV file + #[arg(long, value_name = "PATH")] + log_csv: Option, + + /// Start CSV logging at this frame ID (inclusive) + #[arg(long, requires = "log_csv")] + log_start_frame_id: Option, + + /// Stop CSV logging after this frame ID (inclusive) + #[arg(long, requires = "log_csv")] + log_end_frame_id: Option, + /// Shared encryption key for E2EE (enables AES-GCM end-to-end encryption when set) #[arg(long)] e2ee_key: Option, @@ -580,11 +597,108 @@ fn format_timing_line(timings: &PublisherTimingSummary) -> String { const MAX_PUBLISH_TIMING_SAMPLES: usize = 300; +const PUBLISHER_CSV_HEADER: &str = "sample,elapsed_ms,frame_id,capture_timestamp_us,frame_buffer_timestamp_us,encoder_upload_timestamp_us,encoder_output_timestamp_us,webrtc_packetize_timestamp_us,capture_to_buffer_ms,buffer_to_encoder_ms,encode_ms,encoder_to_packetize_ms,capture_to_packetize_ms,frame_id_gap,packetize_interval_ms"; + +struct PublisherCsvLogger { + writer: BufWriter, + range: FrameLogRange, + first_packetize_timestamp_us: Option, + previous_packetize_timestamp_us: Option, + previous_frame_id: Option, + sample_count: u64, + last_flush: Instant, +} + +impl PublisherCsvLogger { + fn new(path: &Path, range: FrameLogRange) -> std::io::Result { + Ok(Self { + writer: create_csv(path, PUBLISHER_CSV_HEADER)?, + range, + first_packetize_timestamp_us: None, + previous_packetize_timestamp_us: None, + previous_frame_id: range.previous_to_start(), + sample_count: 0, + last_flush: Instant::now(), + }) + } + + fn record(&mut self, sample: PublisherTimingSample) -> std::io::Result<()> { + let Some(frame_id) = sample.frame_id else { + return Ok(()); + }; + if !self.range.contains(frame_id) { + return Ok(()); + } + let Some(frame_buffer_timestamp_us) = sample.got_frame_buffer_timestamp_us else { + return Ok(()); + }; + let Some(encoder_upload_timestamp_us) = sample.encoder_upload_timestamp_us else { + return Ok(()); + }; + let Some(encoder_output_timestamp_us) = sample.encoder_output_timestamp_us else { + return Ok(()); + }; + let Some(packetize_timestamp_us) = sample.webrtc_packetize_timestamp_us else { + return Ok(()); + }; + + let first_packetize_timestamp_us = + *self.first_packetize_timestamp_us.get_or_insert(packetize_timestamp_us); + let frame_id_gap = self + .previous_frame_id + .and_then(|previous| frame_id.checked_sub(previous)) + .and_then(|delta| delta.checked_sub(1)); + let packetize_interval_ms = self + .previous_packetize_timestamp_us + .and_then(|previous| packetize_timestamp_us.checked_sub(previous)) + .map(|interval_us| interval_us as f64 / 1_000.0); + self.sample_count += 1; + + writeln!( + self.writer, + "{},{:.3},{},{},{},{},{},{},{},{},{},{},{},{},{}", + self.sample_count, + packetize_timestamp_us.saturating_sub(first_packetize_timestamp_us) as f64 / 1_000.0, + frame_id, + sample.sensor_exposure_timestamp_us, + frame_buffer_timestamp_us, + encoder_upload_timestamp_us, + encoder_output_timestamp_us, + packetize_timestamp_us, + CsvLatency::between( + Some(sample.sensor_exposure_timestamp_us), + Some(frame_buffer_timestamp_us), + ), + CsvLatency::between(Some(frame_buffer_timestamp_us), Some(encoder_upload_timestamp_us),), + CsvLatency::between( + Some(encoder_upload_timestamp_us), + Some(encoder_output_timestamp_us), + ), + CsvLatency::between(Some(encoder_output_timestamp_us), Some(packetize_timestamp_us),), + CsvLatency::between( + Some(sample.sensor_exposure_timestamp_us), + Some(packetize_timestamp_us), + ), + CsvOption(frame_id_gap), + CsvFloat(packetize_interval_ms), + )?; + + self.previous_frame_id = Some(frame_id); + self.previous_packetize_timestamp_us = Some(packetize_timestamp_us); + if self.range.reaches_end(frame_id) || self.last_flush.elapsed() >= Duration::from_secs(1) { + self.writer.flush()?; + self.last_flush = Instant::now(); + } + Ok(()) + } +} + #[derive(Default)] struct PublisherTimingState { samples: HashMap, order: VecDeque, latest_complete_sample: Option, + frame_log: Option, } impl PublisherTimingState { @@ -622,6 +736,12 @@ impl PublisherTimingState { if updated_sample.is_complete() { self.latest_complete_sample = Some(updated_sample); + if let Some(frame_log) = self.frame_log.as_mut() { + if let Err(error) = frame_log.record(updated_sample) { + warn!("Publisher CSV logging disabled after write failure: {error}"); + self.frame_log = None; + } + } Some(updated_sample) } else { None @@ -690,6 +810,54 @@ fn update_shared_timing_sample( mod tests { use super::*; + #[test] + fn publisher_frame_log_flags_parse_inclusive_bounds() { + let args = Args::try_parse_from([ + "publisher", + "--log-csv", + "publisher.csv", + "--log-start-frame-id", + "301", + "--log-end-frame-id", + "1200", + ]) + .expect("frame log flags should parse"); + assert_eq!(args.log_csv, Some(PathBuf::from("publisher.csv"))); + assert_eq!(args.log_start_frame_id, Some(301)); + assert_eq!(args.log_end_frame_id, Some(1200)); + } + + #[test] + fn publisher_frame_log_bounds_require_csv_path() { + assert!(Args::try_parse_from(["publisher", "--log-start-frame-id", "301"]).is_err()); + } + + #[test] + fn publisher_frame_log_writes_complete_samples_in_range() { + let path = std::env::temp_dir() + .join(format!("local-video-publisher-frame-log-{}.csv", std::process::id())); + let range = FrameLogRange::new(Some(301), Some(302)).expect("range should be valid"); + let mut logger = PublisherCsvLogger::new(&path, range).expect("log should be created"); + let sample = PublisherTimingSample { + frame_id: Some(301), + sensor_exposure_timestamp_us: 1_000, + got_frame_buffer_timestamp_us: Some(1_100), + encoder_upload_timestamp_us: Some(1_200), + encoder_output_timestamp_us: Some(1_300), + webrtc_packetize_timestamp_us: Some(1_400), + }; + logger.record(sample).expect("sample should be written"); + logger.writer.flush().expect("log should flush"); + drop(logger); + + let contents = std::fs::read_to_string(&path).expect("log should be readable"); + let lines: Vec<_> = contents.lines().collect(); + assert_eq!(lines.len(), 2); + assert_eq!(lines[0].split(',').count(), lines[1].split(',').count()); + assert!(lines[1].starts_with("1,0.000,301,")); + std::fs::remove_file(path).expect("temporary log should be removable"); + } + #[test] fn requested_playout_delay_is_absent_when_no_delay_flags_are_set() { assert_eq!(requested_playout_delay(None, None), None); @@ -874,6 +1042,11 @@ async fn main() -> Result<()> { } async fn run(args: Args, ctrl_c_received: Arc) -> Result<()> { + let log_range = FrameLogRange::new(args.log_start_frame_id, args.log_end_frame_id)?; + let logging_enabled = args.log_csv.is_some(); + let attach_timestamp = args.attach_timestamp || logging_enabled; + let attach_frame_id = args.attach_frame_id || logging_enabled; + if args.list_cameras { return list_cameras(); } @@ -1109,8 +1282,29 @@ async fn run(args: Args, ctrl_c_received: Arc) -> Result<()> { let track = LocalVideoTrack::create_video_track("camera", RtcVideoSource::Native(rtc_source.clone())); let display_shared = args.display_video.then(|| Arc::new(Mutex::new(SharedYuv::default()))); - let publish_timing_state = - args.display_timing.then(|| Arc::new(Mutex::new(PublisherTimingState::default()))); + let publisher_log = args + .log_csv + .as_deref() + .map(|path| PublisherCsvLogger::new(path, log_range)) + .transpose() + .with_context(|| { + format!( + "failed to create publisher frame log at {}", + args.log_csv.as_deref().expect("log path should be present").display() + ) + })?; + if let Some(path) = &args.log_csv { + info!( + "Writing publisher per-frame metrics to {} (frame-ID bounds are inclusive)", + path.display() + ); + } + let publish_timing_state = (args.display_timing || logging_enabled).then(|| { + Arc::new(Mutex::new(PublisherTimingState { + frame_log: publisher_log, + ..PublisherTimingState::default() + })) + }); if let Some(timing_state) = publish_timing_state.as_ref() { let timing_state = timing_state.clone(); @@ -1187,8 +1381,8 @@ async fn run(args: Args, ctrl_c_received: Arc) -> Result<()> { } let mut frame_metadata_features = FrameMetadataFeatures::default(); - frame_metadata_features.user_timestamp = args.attach_timestamp; - frame_metadata_features.frame_id = args.attach_frame_id; + frame_metadata_features.user_timestamp = attach_timestamp; + frame_metadata_features.frame_id = attach_frame_id; frame_metadata_features.user_data = args.attach_user_data; let publish_opts = |codec: VideoCodec| TrackPublishOptions { @@ -1222,12 +1416,11 @@ async fn run(args: Args, ctrl_c_received: Arc) -> Result<()> { info!("Published camera track"); requested_codec }; - let capture_config = CaptureConfig { fps: args.fps, - attach_timestamp: args.attach_timestamp, + attach_timestamp, burn_timestamp: args.burn_timestamp, - attach_frame_id: args.attach_frame_id, + attach_frame_id, display_timing: args.display_timing, }; @@ -1249,6 +1442,7 @@ async fn run(args: Args, ctrl_c_received: Arc) -> Result<()> { session, width, height, + publish_timing_state.clone(), user_data_channels.clone(), ) .await; @@ -1723,6 +1917,7 @@ async fn run_argus_capture_loop( session: argus::ArgusCaptureSession, width: u32, height: u32, + publish_timing_state: Option>>, user_data_channels: Option>>, ) -> Result<()> { let capture_handle = std::thread::Builder::new() @@ -1840,6 +2035,13 @@ async fn run_argus_capture_loop( } else { None }; + if let Some(timing_state) = publish_timing_state.as_ref() { + timing_state.lock().record_frame_buffer( + capture_wall_time_us, + fallback_wall_time_us, + fid, + ); + } let user_data = user_data_channels.as_ref().map(|targets| user_data::encode(&targets.lock())); let frame_metadata = if user_ts.is_some() || fid.is_some() || user_data.is_some() { diff --git a/examples/local_video/src/subscriber.rs b/examples/local_video/src/subscriber.rs index 68bcc4c52..6b09f777a 100644 --- a/examples/local_video/src/subscriber.rs +++ b/examples/local_video/src/subscriber.rs @@ -1,4 +1,4 @@ -use anyhow::Result; +use anyhow::{Context, Result}; use clap::Parser; use eframe::egui; use eframe::wgpu::{self, util::DeviceExt}; @@ -15,6 +15,7 @@ use parking_lot::Mutex; use std::{ collections::{HashMap, VecDeque}, env, + path::PathBuf, sync::OnceLock, sync::{ atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}, @@ -24,11 +25,13 @@ use std::{ }; mod codec_display; +mod frame_log; mod subscriber_timing; mod user_data; mod viewport_aspect; use codec_display::{codec_from_mime, codec_with_implementation}; +use frame_log::FrameLogRange; use subscriber_timing::SubscriberTimingHandle; use viewport_aspect::AspectConstrainedViewport; @@ -413,11 +416,35 @@ struct Args { #[arg(long)] display_timestamp: bool, + /// Write one row of subscriber timing and delivery metrics per rendered frame + #[arg(long, value_name = "PATH")] + log_csv: Option, + + /// Start CSV logging at this frame ID (inclusive) + #[arg(long, requires = "log_csv")] + log_start_frame_id: Option, + + /// Stop CSV logging after this frame ID (inclusive) + #[arg(long, requires = "log_csv")] + log_end_frame_id: Option, + /// Shared encryption key for E2EE (enables AES-GCM end-to-end encryption when set; must match publisher's key) #[arg(long)] e2ee_key: Option, } +fn record_received_frame_sample(frame: &BoxVideoFrame, subscriber_timing: &SubscriberTimingHandle) { + if let Some(metadata) = &frame.frame_metadata { + if let Some(capture_timestamp_us) = metadata.user_timestamp { + subscriber_timing.record_frame_received_by_sink( + capture_timestamp_us, + metadata.frame_id, + current_timestamp_us(), + ); + } + } +} + struct SharedYuv { room_name: String, self_identity: String, @@ -774,6 +801,21 @@ fn update_receive_bitrate_from_stats( } } +fn update_frame_log_quality( + stats: &[livekit::webrtc::stats::RtcStats], + subscriber_timing: &SubscriberTimingHandle, +) { + let Some(inbound) = find_video_inbound_stats(stats) else { + return; + }; + subscriber_timing.record_inbound_quality( + inbound.received.packets_lost, + inbound.inbound.frames_dropped, + inbound.inbound.freeze_count, + inbound.inbound.total_freeze_duration, + ); +} + struct TimestampAnchor { unix_timestamp_us: u64, instant: Instant, @@ -835,6 +877,28 @@ mod tests { assert!(low_latency_args.low_latency); } + #[test] + fn subscriber_frame_log_flags_parse_inclusive_bounds() { + let args = Args::try_parse_from([ + "subscriber", + "--log-csv", + "subscriber.csv", + "--log-start-frame-id", + "301", + "--log-end-frame-id", + "1200", + ]) + .expect("frame log flags should parse"); + assert_eq!(args.log_csv, Some(PathBuf::from("subscriber.csv"))); + assert_eq!(args.log_start_frame_id, Some(301)); + assert_eq!(args.log_end_frame_id, Some(1200)); + } + + #[test] + fn subscriber_frame_log_bounds_require_csv_path() { + assert!(Args::try_parse_from(["subscriber", "--log-end-frame-id", "1200"]).is_err()); + } + #[test] fn subscriber_diagnostics_show_status_without_timing() { let shared = Arc::new(Mutex::new(SharedYuv { @@ -924,6 +988,16 @@ async fn handle_track_subscribed( publication.dimension().1, publication.frame_metadata_features(), ); + if subscriber_timing.has_frame_log() { + let features = publication.frame_metadata_features(); + if !features.contains(&PacketTrailerFeature::PtfUserTimestamp) + || !features.contains(&PacketTrailerFeature::PtfFrameId) + { + log::warn!( + "Subscriber CSV logging requires publisher timestamp and frame-ID metadata; run the publisher with --log-csv or with both --attach-timestamp and --attach-frame-id" + ); + } + } { let mut s = shared.lock(); @@ -973,23 +1047,16 @@ async fn handle_track_subscribed( break; } let Some(mut frame) = sink.next().await else { break }; + record_received_frame_sample(&frame, &subscriber_timing_sink); let mut drained_frames = 0_u64; while let Some(Some(newer_frame)) = sink.next().now_or_never() { + record_received_frame_sample(&newer_frame, &subscriber_timing_sink); frame = newer_frame; drained_frames += 1; } if drained_frames > 0 { debug!("Dropped {drained_frames} stale decoded frames before render upload"); } - if let Some(metadata) = &frame.frame_metadata { - if let Some(capture_timestamp_us) = metadata.user_timestamp { - subscriber_timing_sink.record_frame_received_by_sink( - capture_timestamp_us, - metadata.frame_id, - current_timestamp_us(), - ); - } - } let channel_values = frame .frame_metadata .as_ref() @@ -1063,6 +1130,7 @@ async fn handle_track_subscribed( let my_sid_stats = sid.clone(); let simulcast_stats = simulcast.clone(); let shared_stats = shared.clone(); + let subscriber_timing_stats = subscriber_timing.clone(); tokio::spawn(async move { let mut logged_initial = false; let mut jitter_buffer_snapshot = None; @@ -1097,6 +1165,7 @@ async fn handle_track_subscribed( &mut receive_bitrate_snapshot, &shared_stats, ); + update_frame_log_quality(&stats, &subscriber_timing_stats); update_simulcast_quality_from_stats(&stats, &simulcast_stats); } Err(e) if !logged_initial => { @@ -1536,6 +1605,7 @@ async fn main() -> Result<()> { } async fn run(args: Args, ctrl_c_received: Arc) -> Result<()> { + let log_range = FrameLogRange::new(args.log_start_frame_id, args.log_end_frame_id)?; if args.low_latency { livekit::webrtc::enable_zero_playout_delay()?; info!("Low-latency mode enabled: WebRTC-ForcePlayoutDelay/min_ms:0,max_ms:0/"); @@ -1606,7 +1676,19 @@ async fn run(args: Args, ctrl_c_received: Arc) -> Result<()> { })); let frame_slot = Arc::new(LatestRenderFrameSlot::new()); let video_size = Arc::new(AtomicVideoSize::default()); - let subscriber_timing = SubscriberTimingHandle::new(); + let subscriber_timing = if let Some(path) = args.log_csv.as_deref() { + let timing = + SubscriberTimingHandle::with_frame_log(path, log_range).with_context(|| { + format!("failed to create subscriber frame log at {}", path.display()) + })?; + info!( + "Writing subscriber per-frame metrics to {} (frame-ID bounds are inclusive)", + path.display() + ); + timing + } else { + SubscriberTimingHandle::new() + }; let channel_history = Arc::new(Mutex::new(VecDeque::with_capacity(CHANNEL_HISTORY_LEN))); // Subscribe to room events: on first video track, start sink task diff --git a/examples/local_video/src/subscriber_timing.rs b/examples/local_video/src/subscriber_timing.rs index 7b9ed6bc1..446d4137c 100644 --- a/examples/local_video/src/subscriber_timing.rs +++ b/examples/local_video/src/subscriber_timing.rs @@ -1,13 +1,18 @@ use std::{ collections::{HashMap, VecDeque}, + fs::File, + io::{self, BufWriter, Write}, + path::Path, sync::Arc, time::{Duration, Instant}, }; use livekit::track::{SubscribeTimingEvent, SubscribeTimingStage}; -use log::info; +use log::{info, warn}; use parking_lot::Mutex; +use crate::frame_log::{create_csv, CsvFloat, CsvLatency, CsvOption, FrameLogRange}; + const MAX_SUBSCRIBER_TIMING_SAMPLES: usize = 300; const DISPLAY_UPDATE_INTERVAL: Duration = Duration::from_millis(100); const TIMING_LABEL_WIDTH: usize = 22; @@ -21,6 +26,7 @@ const TIMING_LINE_WIDTH: usize = #[derive(Clone, Default)] pub(crate) struct SubscriberTimingHandle { inner: Arc>, + frame_log: Option>>, } impl SubscriberTimingHandle { @@ -28,6 +34,14 @@ impl SubscriberTimingHandle { Self::default() } + /// Creates a timing handle that logs completed rendered frames to CSV. + pub(crate) fn with_frame_log(path: &Path, range: FrameLogRange) -> io::Result { + Ok(Self { + inner: Arc::default(), + frame_log: Some(Arc::new(Mutex::new(SubscriberCsvLogger::new(path, range)?))), + }) + } + pub(crate) fn record_subscribe_event(&self, event: SubscribeTimingEvent) { self.inner.lock().record_subscribe_event(event); } @@ -65,18 +79,46 @@ impl SubscriberTimingHandle { frame_prepare_timestamp_us: u64, frame_painted_timestamp_us: u64, ) { - self.inner.lock().record_frame_painted( + let sample = self.inner.lock().record_frame_painted( sensor_exposure_timestamp_us, frame_id, frame_prepare_timestamp_us, frame_painted_timestamp_us, ); + if let Some(frame_log) = &self.frame_log { + if let Err(error) = frame_log.lock().record(sample) { + warn!("Subscriber CSV logging disabled after write failure: {error}"); + } + } } pub(crate) fn display_overlay_lines(&self, now: Instant) -> Option> { self.inner.lock().display_overlay_lines(now) } + /// Returns whether per-frame CSV logging is enabled. + pub(crate) fn has_frame_log(&self) -> bool { + self.frame_log.is_some() + } + + /// Updates the cumulative WebRTC delivery-quality counters copied into future CSV rows. + pub(crate) fn record_inbound_quality( + &self, + packets_lost: i64, + frames_dropped: u32, + freeze_count: u32, + total_freeze_duration_secs: f64, + ) { + if let Some(frame_log) = &self.frame_log { + frame_log.lock().quality = Some(InboundQualitySnapshot { + packets_lost, + frames_dropped, + freeze_count, + total_freeze_duration_ms: total_freeze_duration_secs * 1_000.0, + }); + } + } + pub(crate) fn reset(&self) { self.inner.lock().reset(); } @@ -180,13 +222,14 @@ impl SubscriberTimingState { frame_id: Option, frame_prepare_timestamp_us: u64, frame_painted_timestamp_us: u64, - ) { + ) -> SubscriberTimingSample { let sample = self.get_or_insert_sample(sensor_exposure_timestamp_us, frame_id); sample.frame_prepare_timestamp_us.get_or_insert(frame_prepare_timestamp_us); sample.frame_painted_timestamp_us = Some(frame_painted_timestamp_us); let sample = *sample; self.latest_display_sample = Some(sample); self.render_latency_window.record(sample, Instant::now()); + sample } fn display_sample(&self) -> Option { @@ -293,6 +336,141 @@ impl SubscriberTimingState { } } +const SUBSCRIBER_CSV_HEADER: &str = "sample,elapsed_ms,frame_id,capture_timestamp_us,webrtc_receive_timestamp_us,decoder_upload_timestamp_us,decoder_output_timestamp_us,frame_sink_timestamp_us,frame_prepare_timestamp_us,frame_painted_timestamp_us,exposure_to_receive_ms,receive_to_decode_ms,decode_to_sink_ms,sink_to_prepare_ms,prepare_to_paint_ms,receive_to_paint_ms,e2e_latency_ms,frame_id_gap,render_interval_ms,packets_lost,frames_dropped,freeze_count,total_freeze_duration_ms"; + +#[derive(Clone, Copy)] +struct InboundQualitySnapshot { + packets_lost: i64, + frames_dropped: u32, + freeze_count: u32, + total_freeze_duration_ms: f64, +} + +struct SubscriberCsvLogger { + writer: BufWriter, + range: FrameLogRange, + first_painted_timestamp_us: Option, + previous_painted_timestamp_us: Option, + previous_frame_id: Option, + sample_count: u64, + quality: Option, + quality_baseline: Option, + last_flush: Instant, + failed: bool, +} + +impl SubscriberCsvLogger { + fn new(path: &Path, range: FrameLogRange) -> io::Result { + Ok(Self { + writer: create_csv(path, SUBSCRIBER_CSV_HEADER)?, + range, + first_painted_timestamp_us: None, + previous_painted_timestamp_us: None, + previous_frame_id: range.previous_to_start(), + sample_count: 0, + quality: None, + quality_baseline: None, + last_flush: Instant::now(), + failed: false, + }) + } + + fn record(&mut self, sample: SubscriberTimingSample) -> io::Result<()> { + if self.failed { + return Ok(()); + } + let Some(frame_id) = sample.frame_id else { + return Ok(()); + }; + if !self.range.contains(frame_id) { + return Ok(()); + } + let Some(frame_painted_timestamp_us) = sample.frame_painted_timestamp_us else { + return Ok(()); + }; + + let first_painted_timestamp_us = + *self.first_painted_timestamp_us.get_or_insert(frame_painted_timestamp_us); + let frame_id_gap = self + .previous_frame_id + .and_then(|previous| frame_id.checked_sub(previous)) + .and_then(|delta| delta.checked_sub(1)); + let render_interval_ms = self + .previous_painted_timestamp_us + .and_then(|previous| frame_painted_timestamp_us.checked_sub(previous)) + .map(|interval_us| interval_us as f64 / 1_000.0); + self.sample_count += 1; + let quality = self.quality.map(|current| { + let baseline = *self.quality_baseline.get_or_insert(current); + InboundQualitySnapshot { + packets_lost: current.packets_lost.saturating_sub(baseline.packets_lost), + frames_dropped: current.frames_dropped.saturating_sub(baseline.frames_dropped), + freeze_count: current.freeze_count.saturating_sub(baseline.freeze_count), + total_freeze_duration_ms: (current.total_freeze_duration_ms + - baseline.total_freeze_duration_ms) + .max(0.0), + } + }); + + let result = writeln!( + self.writer, + "{},{:.3},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{}", + self.sample_count, + frame_painted_timestamp_us.saturating_sub(first_painted_timestamp_us) as f64 / 1_000.0, + frame_id, + sample.sensor_exposure_timestamp_us, + CsvOption(sample.webrtc_receive_timestamp_us), + CsvOption(sample.decoder_upload_timestamp_us), + CsvOption(sample.decoder_output_timestamp_us), + CsvOption(sample.frame_sink_timestamp_us), + CsvOption(sample.frame_prepare_timestamp_us), + frame_painted_timestamp_us, + CsvLatency::between( + Some(sample.sensor_exposure_timestamp_us), + sample.webrtc_receive_timestamp_us, + ), + CsvLatency::between( + sample.webrtc_receive_timestamp_us, + sample.decoder_output_timestamp_us, + ), + CsvLatency::between(sample.decoder_output_timestamp_us, sample.frame_sink_timestamp_us), + CsvLatency::between(sample.frame_sink_timestamp_us, sample.frame_prepare_timestamp_us), + CsvLatency::between( + sample.frame_prepare_timestamp_us, + sample.frame_painted_timestamp_us, + ), + CsvLatency::between( + sample.webrtc_receive_timestamp_us, + sample.frame_painted_timestamp_us, + ), + CsvLatency::between( + Some(sample.sensor_exposure_timestamp_us), + sample.frame_painted_timestamp_us, + ), + CsvOption(frame_id_gap), + CsvFloat(render_interval_ms), + CsvOption(quality.map(|quality| quality.packets_lost)), + CsvOption(quality.map(|quality| quality.frames_dropped)), + CsvOption(quality.map(|quality| quality.freeze_count)), + CsvFloat(quality.map(|quality| quality.total_freeze_duration_ms)), + ); + + let should_flush = + self.range.reaches_end(frame_id) || self.last_flush.elapsed() >= Duration::from_secs(1); + let result = result.and_then(|()| if should_flush { self.writer.flush() } else { Ok(()) }); + if result.is_ok() { + self.previous_frame_id = Some(frame_id); + self.previous_painted_timestamp_us = Some(frame_painted_timestamp_us); + if should_flush { + self.last_flush = Instant::now(); + } + } else { + self.failed = true; + } + result + } +} + #[derive(Clone, Copy, Default)] struct LatencyStats { count: u64, @@ -573,6 +751,59 @@ fn assert_timing_lines_are_stable(lines: &[String]) { mod tests { use super::*; + #[test] + fn subscriber_frame_log_filters_range_and_rebases_quality() { + let path = std::env::temp_dir() + .join(format!("local-video-subscriber-frame-log-{}.csv", std::process::id())); + let range = FrameLogRange::new(Some(301), Some(303)).expect("range should be valid"); + let mut logger = SubscriberCsvLogger::new(&path, range).expect("log should be created"); + logger.quality = Some(InboundQualitySnapshot { + packets_lost: 5, + frames_dropped: 2, + freeze_count: 1, + total_freeze_duration_ms: 50.0, + }); + let sample = SubscriberTimingSample { + frame_id: Some(301), + sensor_exposure_timestamp_us: 1_000, + webrtc_receive_timestamp_us: Some(1_100), + decoder_upload_timestamp_us: Some(1_110), + decoder_output_timestamp_us: Some(1_200), + frame_sink_timestamp_us: Some(1_210), + frame_prepare_timestamp_us: Some(1_220), + frame_painted_timestamp_us: Some(1_300), + }; + logger.record(sample).expect("first sample should be written"); + logger.quality = Some(InboundQualitySnapshot { + packets_lost: 7, + frames_dropped: 3, + freeze_count: 2, + total_freeze_duration_ms: 150.0, + }); + logger + .record(SubscriberTimingSample { + frame_id: Some(303), + sensor_exposure_timestamp_us: 35_000, + webrtc_receive_timestamp_us: Some(35_100), + decoder_upload_timestamp_us: Some(35_110), + decoder_output_timestamp_us: Some(35_200), + frame_sink_timestamp_us: Some(35_210), + frame_prepare_timestamp_us: Some(35_220), + frame_painted_timestamp_us: Some(35_300), + }) + .expect("second sample should be written"); + logger.writer.flush().expect("log should flush"); + drop(logger); + + let contents = std::fs::read_to_string(&path).expect("log should be readable"); + let lines: Vec<_> = contents.lines().collect(); + assert_eq!(lines.len(), 3); + assert_eq!(lines[0].split(',').count(), lines[2].split(',').count()); + assert!(lines[1].ends_with(",0,,0,0,0,0.000")); + assert!(lines[2].ends_with(",1,34.000,2,1,1,100.000")); + std::fs::remove_file(path).expect("temporary log should be removable"); + } + fn timestamp_us(hour: u64, minute: u64, second: u64, millisecond: u64) -> u64 { (((hour * 3_600 + minute * 60 + second) * 1_000) + millisecond) * 1_000 } From 5bf977d0257472f8f7e1da2e5151c559ef9b0026 Mon Sep 17 00:00:00 2001 From: David Chen Date: Tue, 28 Jul 2026 16:04:31 -0700 Subject: [PATCH 2/9] remove changeset for example --- .changeset/local-video-latency-report.md | 5 ----- 1 file changed, 5 deletions(-) delete mode 100644 .changeset/local-video-latency-report.md diff --git a/.changeset/local-video-latency-report.md b/.changeset/local-video-latency-report.md deleted file mode 100644 index ee8e33fa3..000000000 --- a/.changeset/local-video-latency-report.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"local_video": patch ---- - -Add frame-range CSV timing and delivery-quality logging to the local video publisher and subscriber, plus a PDF report generator for either or both logs. From 57d60e2310d7bb6ee3edb0bef58b45e4d2135231 Mon Sep 17 00:00:00 2001 From: David Chen Date: Tue, 28 Jul 2026 23:20:21 -0700 Subject: [PATCH 3/9] Move latency CSV tests to module bottoms --- examples/local_video/src/publisher.rs | 96 ++++++++-------- examples/local_video/src/subscriber.rs | 44 ++++---- examples/local_video/src/subscriber_timing.rs | 106 +++++++++--------- 3 files changed, 123 insertions(+), 123 deletions(-) diff --git a/examples/local_video/src/publisher.rs b/examples/local_video/src/publisher.rs index 2e7a29050..ee5ae9047 100644 --- a/examples/local_video/src/publisher.rs +++ b/examples/local_video/src/publisher.rs @@ -810,54 +810,6 @@ fn update_shared_timing_sample( mod tests { use super::*; - #[test] - fn publisher_frame_log_flags_parse_inclusive_bounds() { - let args = Args::try_parse_from([ - "publisher", - "--log-csv", - "publisher.csv", - "--log-start-frame-id", - "301", - "--log-end-frame-id", - "1200", - ]) - .expect("frame log flags should parse"); - assert_eq!(args.log_csv, Some(PathBuf::from("publisher.csv"))); - assert_eq!(args.log_start_frame_id, Some(301)); - assert_eq!(args.log_end_frame_id, Some(1200)); - } - - #[test] - fn publisher_frame_log_bounds_require_csv_path() { - assert!(Args::try_parse_from(["publisher", "--log-start-frame-id", "301"]).is_err()); - } - - #[test] - fn publisher_frame_log_writes_complete_samples_in_range() { - let path = std::env::temp_dir() - .join(format!("local-video-publisher-frame-log-{}.csv", std::process::id())); - let range = FrameLogRange::new(Some(301), Some(302)).expect("range should be valid"); - let mut logger = PublisherCsvLogger::new(&path, range).expect("log should be created"); - let sample = PublisherTimingSample { - frame_id: Some(301), - sensor_exposure_timestamp_us: 1_000, - got_frame_buffer_timestamp_us: Some(1_100), - encoder_upload_timestamp_us: Some(1_200), - encoder_output_timestamp_us: Some(1_300), - webrtc_packetize_timestamp_us: Some(1_400), - }; - logger.record(sample).expect("sample should be written"); - logger.writer.flush().expect("log should flush"); - drop(logger); - - let contents = std::fs::read_to_string(&path).expect("log should be readable"); - let lines: Vec<_> = contents.lines().collect(); - assert_eq!(lines.len(), 2); - assert_eq!(lines[0].split(',').count(), lines[1].split(',').count()); - assert!(lines[1].starts_with("1,0.000,301,")); - std::fs::remove_file(path).expect("temporary log should be removable"); - } - #[test] fn requested_playout_delay_is_absent_when_no_delay_flags_are_set() { assert_eq!(requested_playout_delay(None, None), None); @@ -971,6 +923,54 @@ mod tests { assert_eq!(selected, 950); } + + #[test] + fn publisher_frame_log_flags_parse_inclusive_bounds() { + let args = Args::try_parse_from([ + "publisher", + "--log-csv", + "publisher.csv", + "--log-start-frame-id", + "301", + "--log-end-frame-id", + "1200", + ]) + .expect("frame log flags should parse"); + assert_eq!(args.log_csv, Some(PathBuf::from("publisher.csv"))); + assert_eq!(args.log_start_frame_id, Some(301)); + assert_eq!(args.log_end_frame_id, Some(1200)); + } + + #[test] + fn publisher_frame_log_bounds_require_csv_path() { + assert!(Args::try_parse_from(["publisher", "--log-start-frame-id", "301"]).is_err()); + } + + #[test] + fn publisher_frame_log_writes_complete_samples_in_range() { + let path = std::env::temp_dir() + .join(format!("local-video-publisher-frame-log-{}.csv", std::process::id())); + let range = FrameLogRange::new(Some(301), Some(302)).expect("range should be valid"); + let mut logger = PublisherCsvLogger::new(&path, range).expect("log should be created"); + let sample = PublisherTimingSample { + frame_id: Some(301), + sensor_exposure_timestamp_us: 1_000, + got_frame_buffer_timestamp_us: Some(1_100), + encoder_upload_timestamp_us: Some(1_200), + encoder_output_timestamp_us: Some(1_300), + webrtc_packetize_timestamp_us: Some(1_400), + }; + logger.record(sample).expect("sample should be written"); + logger.writer.flush().expect("log should flush"); + drop(logger); + + let contents = std::fs::read_to_string(&path).expect("log should be readable"); + let lines: Vec<_> = contents.lines().collect(); + assert_eq!(lines.len(), 2); + assert_eq!(lines[0].split(',').count(), lines[1].split(',').count()); + assert!(lines[1].starts_with("1,0.000,301,")); + std::fs::remove_file(path).expect("temporary log should be removable"); + } } fn list_cameras() -> Result<()> { diff --git a/examples/local_video/src/subscriber.rs b/examples/local_video/src/subscriber.rs index 6b09f777a..300c0961d 100644 --- a/examples/local_video/src/subscriber.rs +++ b/examples/local_video/src/subscriber.rs @@ -877,28 +877,6 @@ mod tests { assert!(low_latency_args.low_latency); } - #[test] - fn subscriber_frame_log_flags_parse_inclusive_bounds() { - let args = Args::try_parse_from([ - "subscriber", - "--log-csv", - "subscriber.csv", - "--log-start-frame-id", - "301", - "--log-end-frame-id", - "1200", - ]) - .expect("frame log flags should parse"); - assert_eq!(args.log_csv, Some(PathBuf::from("subscriber.csv"))); - assert_eq!(args.log_start_frame_id, Some(301)); - assert_eq!(args.log_end_frame_id, Some(1200)); - } - - #[test] - fn subscriber_frame_log_bounds_require_csv_path() { - assert!(Args::try_parse_from(["subscriber", "--log-end-frame-id", "1200"]).is_err()); - } - #[test] fn subscriber_diagnostics_show_status_without_timing() { let shared = Arc::new(Mutex::new(SharedYuv { @@ -928,6 +906,28 @@ mod tests { ] ); } + + #[test] + fn subscriber_frame_log_flags_parse_inclusive_bounds() { + let args = Args::try_parse_from([ + "subscriber", + "--log-csv", + "subscriber.csv", + "--log-start-frame-id", + "301", + "--log-end-frame-id", + "1200", + ]) + .expect("frame log flags should parse"); + assert_eq!(args.log_csv, Some(PathBuf::from("subscriber.csv"))); + assert_eq!(args.log_start_frame_id, Some(301)); + assert_eq!(args.log_end_frame_id, Some(1200)); + } + + #[test] + fn subscriber_frame_log_bounds_require_csv_path() { + assert!(Args::try_parse_from(["subscriber", "--log-end-frame-id", "1200"]).is_err()); + } } async fn handle_track_subscribed( diff --git a/examples/local_video/src/subscriber_timing.rs b/examples/local_video/src/subscriber_timing.rs index 446d4137c..106e03eaa 100644 --- a/examples/local_video/src/subscriber_timing.rs +++ b/examples/local_video/src/subscriber_timing.rs @@ -751,59 +751,6 @@ fn assert_timing_lines_are_stable(lines: &[String]) { mod tests { use super::*; - #[test] - fn subscriber_frame_log_filters_range_and_rebases_quality() { - let path = std::env::temp_dir() - .join(format!("local-video-subscriber-frame-log-{}.csv", std::process::id())); - let range = FrameLogRange::new(Some(301), Some(303)).expect("range should be valid"); - let mut logger = SubscriberCsvLogger::new(&path, range).expect("log should be created"); - logger.quality = Some(InboundQualitySnapshot { - packets_lost: 5, - frames_dropped: 2, - freeze_count: 1, - total_freeze_duration_ms: 50.0, - }); - let sample = SubscriberTimingSample { - frame_id: Some(301), - sensor_exposure_timestamp_us: 1_000, - webrtc_receive_timestamp_us: Some(1_100), - decoder_upload_timestamp_us: Some(1_110), - decoder_output_timestamp_us: Some(1_200), - frame_sink_timestamp_us: Some(1_210), - frame_prepare_timestamp_us: Some(1_220), - frame_painted_timestamp_us: Some(1_300), - }; - logger.record(sample).expect("first sample should be written"); - logger.quality = Some(InboundQualitySnapshot { - packets_lost: 7, - frames_dropped: 3, - freeze_count: 2, - total_freeze_duration_ms: 150.0, - }); - logger - .record(SubscriberTimingSample { - frame_id: Some(303), - sensor_exposure_timestamp_us: 35_000, - webrtc_receive_timestamp_us: Some(35_100), - decoder_upload_timestamp_us: Some(35_110), - decoder_output_timestamp_us: Some(35_200), - frame_sink_timestamp_us: Some(35_210), - frame_prepare_timestamp_us: Some(35_220), - frame_painted_timestamp_us: Some(35_300), - }) - .expect("second sample should be written"); - logger.writer.flush().expect("log should flush"); - drop(logger); - - let contents = std::fs::read_to_string(&path).expect("log should be readable"); - let lines: Vec<_> = contents.lines().collect(); - assert_eq!(lines.len(), 3); - assert_eq!(lines[0].split(',').count(), lines[2].split(',').count()); - assert!(lines[1].ends_with(",0,,0,0,0,0.000")); - assert!(lines[2].ends_with(",1,34.000,2,1,1,100.000")); - std::fs::remove_file(path).expect("temporary log should be removable"); - } - fn timestamp_us(hour: u64, minute: u64, second: u64, millisecond: u64) -> u64 { (((hour * 3_600 + minute * 60 + second) * 1_000) + millisecond) * 1_000 } @@ -988,4 +935,57 @@ mod tests { assert_eq!(lines[7], "Receive to Render: 50.0ms"); assert_eq!(lines[8], "e2e latency: 100.0ms"); } + + #[test] + fn subscriber_frame_log_filters_range_and_rebases_quality() { + let path = std::env::temp_dir() + .join(format!("local-video-subscriber-frame-log-{}.csv", std::process::id())); + let range = FrameLogRange::new(Some(301), Some(303)).expect("range should be valid"); + let mut logger = SubscriberCsvLogger::new(&path, range).expect("log should be created"); + logger.quality = Some(InboundQualitySnapshot { + packets_lost: 5, + frames_dropped: 2, + freeze_count: 1, + total_freeze_duration_ms: 50.0, + }); + let sample = SubscriberTimingSample { + frame_id: Some(301), + sensor_exposure_timestamp_us: 1_000, + webrtc_receive_timestamp_us: Some(1_100), + decoder_upload_timestamp_us: Some(1_110), + decoder_output_timestamp_us: Some(1_200), + frame_sink_timestamp_us: Some(1_210), + frame_prepare_timestamp_us: Some(1_220), + frame_painted_timestamp_us: Some(1_300), + }; + logger.record(sample).expect("first sample should be written"); + logger.quality = Some(InboundQualitySnapshot { + packets_lost: 7, + frames_dropped: 3, + freeze_count: 2, + total_freeze_duration_ms: 150.0, + }); + logger + .record(SubscriberTimingSample { + frame_id: Some(303), + sensor_exposure_timestamp_us: 35_000, + webrtc_receive_timestamp_us: Some(35_100), + decoder_upload_timestamp_us: Some(35_110), + decoder_output_timestamp_us: Some(35_200), + frame_sink_timestamp_us: Some(35_210), + frame_prepare_timestamp_us: Some(35_220), + frame_painted_timestamp_us: Some(35_300), + }) + .expect("second sample should be written"); + logger.writer.flush().expect("log should flush"); + drop(logger); + + let contents = std::fs::read_to_string(&path).expect("log should be readable"); + let lines: Vec<_> = contents.lines().collect(); + assert_eq!(lines.len(), 3); + assert_eq!(lines[0].split(',').count(), lines[2].split(',').count()); + assert!(lines[1].ends_with(",0,,0,0,0,0.000")); + assert!(lines[2].ends_with(",1,34.000,2,1,1,100.000")); + std::fs::remove_file(path).expect("temporary log should be removable"); + } } From 7e3eded0a29c851dc02726aa4d9fcbe03341aa8c Mon Sep 17 00:00:00 2001 From: David Chen Date: Thu, 6 Aug 2026 13:03:36 -0700 Subject: [PATCH 4/9] Move publisher diagnostics into a separate window --- examples/local_video/README.md | 7 +- examples/local_video/src/publisher.rs | 10 +- examples/local_video/src/video_display.rs | 199 ++++++++++++++-------- 3 files changed, 134 insertions(+), 82 deletions(-) diff --git a/examples/local_video/README.md b/examples/local_video/README.md index 731b621aa..36822ac85 100644 --- a/examples/local_video/README.md +++ b/examples/local_video/README.md @@ -94,7 +94,7 @@ Publisher usage: --identity cam-1 \ --e2ee-key my-secret-key - # publish and display the outgoing video locally + # publish and display the outgoing video locally with separate diagnostics cargo run -p local_video -F desktop --bin publisher -- \ --camera-index 0 \ --room-name demo \ @@ -143,8 +143,9 @@ Publisher flags (in addition to the common connection flags above): - `--attach-timestamp`: Attach the current wall-clock time (microseconds since UNIX epoch) as the user timestamp on each published frame. The subscriber can display this to measure end-to-end latency. - `--burn-timestamp`: Burn the attached timestamp into the video frame as a visible overlay. Has no effect unless `--attach-timestamp` is also set. - `--attach-frame-id`: Attach a monotonically increasing frame ID to each published frame via the packet trailer. The subscriber displays this in the timestamp overlay when `--display-timestamp` is used. -- `--display-video`: Open a window that displays the video frames being published. -- `--display-timing`: Burn publisher timing metrics into the local preview window. Requires `--display-video`. +- `--attach-user-data`: Attach six keyboard-controlled channel values to each frame. Focus the diagnostics window and use Q/A, W/S, E/D, R/F, T/G, and Y/H to adjust channels 1-6. Requires `--display-video`. +- `--display-video`: Open a video preview window and a separate publisher diagnostics window. The video window repaints only when a frame arrives; diagnostics update independently at 10 Hz. +- `--display-timing`: Show publisher timing metrics in the diagnostics window. Requires `--display-video`. - `--log-csv `: Write one CSV row per packetized frame with capture, encoder, packetization, frame-gap, and inter-frame timing metrics. This automatically enables timestamp and frame-ID metadata. - `--log-start-frame-id `: Start CSV logging at this frame ID (inclusive). Requires `--log-csv`. - `--log-end-frame-id `: Stop CSV logging after this frame ID (inclusive). Requires `--log-csv`. diff --git a/examples/local_video/src/publisher.rs b/examples/local_video/src/publisher.rs index ee5ae9047..3483f813f 100644 --- a/examples/local_video/src/publisher.rs +++ b/examples/local_video/src/publisher.rs @@ -258,16 +258,16 @@ struct Args { /// Attach keyboard-controlled 6-channel data (6x int16 fixed-point, 12 bytes) /// as the per-frame user_data trailer field. Control the channels from the - /// preview window: Q/A=CH1, W/S=CH2, E/D=CH3, R/F=CH4, T/G=CH5, Y/H=CH6. - /// Requires --display-video (the window provides keyboard focus). + /// diagnostics window: Q/A=CH1, W/S=CH2, E/D=CH3, R/F=CH4, T/G=CH5, Y/H=CH6. + /// Requires --display-video (the diagnostics window provides keyboard focus). #[arg(long, default_value_t = false, requires = "display_video")] attach_user_data: bool, - /// Open a window that displays the video frames being published + /// Open video preview and publisher diagnostics windows #[arg(long, default_value_t = false)] display_video: bool, - /// Burn publisher timing metrics into the local preview window + /// Show publisher timing metrics in the diagnostics window #[arg(long, default_value_t = false, requires = "display_video")] display_timing: bool, @@ -1424,7 +1424,7 @@ async fn run(args: Args, ctrl_c_received: Arc) -> Result<()> { display_timing: args.display_timing, }; - // Shared keyboard-controlled channel values, written by the preview window + // Shared keyboard-controlled channel values, written by the diagnostics window // and read by the capture loop to fill the user_data trailer. let user_data_channels = args.attach_user_data.then(|| Arc::new(Mutex::new([0.0f32; user_data::NUM_CHANNELS]))); diff --git a/examples/local_video/src/video_display.rs b/examples/local_video/src/video_display.rs index 4ce1b6ae2..484b09fbc 100644 --- a/examples/local_video/src/video_display.rs +++ b/examples/local_video/src/video_display.rs @@ -139,7 +139,7 @@ pub(crate) fn pack_i420_into_shared( s.repaint_ctx.clone() }; if let Some(ctx) = repaint_ctx { - ctx.request_repaint(); + ctx.request_repaint_of(egui::ViewportId::ROOT); } true } @@ -610,13 +610,25 @@ const CHANNEL_RATE_PER_S: f32 = 1.0; type ChannelValues = Arc>; +/// Diagnostics are intentionally slower than video rendering so UI work cannot pace video frames. +const DIAGNOSTICS_REPAINT_INTERVAL: Duration = Duration::from_millis(100); +const DIAGNOSTICS_CONTENT_PADDING: i8 = 10; +const DIAGNOSTICS_WINDOW_SIZE: [f32; 2] = [400.0, 320.0]; +const DIAGNOSTICS_WINDOW_MIN_SIZE: [f32; 2] = [320.0, 120.0]; + +fn diagnostics_viewport_id() -> egui::ViewportId { + egui::ViewportId::from_hash_of("publisher-diagnostics") +} + struct VideoApp { shared: Arc>, ctrl_c_received: Arc, viewport: AspectConstrainedViewport, - timing_overlay_state: PublisherTimingOverlayState, + timing_overlay_state: Arc>, /// Keyboard-controlled user_data channel values shared with the capture loop. channels: Option, + diagnostics_open: Arc, + diagnostics_started: Arc, } /// Apply held-key deltas to the shared channel values and return the current @@ -643,6 +655,113 @@ fn drive_channels( *values } +fn paint_channel_controls(ui: &mut egui::Ui, values: &[f32; crate::user_data::NUM_CHANNELS]) { + const KEY_LABELS: [&str; crate::user_data::NUM_CHANNELS] = + ["Q/A", "W/S", "E/D", "R/F", "T/G", "Y/H"]; + let mut lines = vec!["user_data channels".to_string()]; + for (idx, value) in values.iter().enumerate() { + lines.push(format!("CH{} [{}]: {:>+6.2}", idx + 1, KEY_LABELS[idx], value)); + } + + ui.add( + egui::Label::new( + egui::RichText::new(lines.join("\n")) + .monospace() + .size(12.0) + .color(egui::Color32::WHITE), + ) + .extend(), + ); +} + +fn paint_publisher_diagnostics( + ui: &mut egui::Ui, + shared: &Arc>, + timing_overlay_state: &Arc>, + channels: Option<&ChannelValues>, +) { + egui::Frame::NONE.inner_margin(egui::Margin::same(DIAGNOSTICS_CONTENT_PADDING)).show( + ui, + |ui| { + ui.visuals_mut().override_text_color = Some(egui::Color32::WHITE); + egui::ScrollArea::vertical().show(ui, |ui| { + let lines = publisher_overlay_lines( + shared, + &mut timing_overlay_state.lock(), + Instant::now(), + ) + .unwrap_or_else(|| vec!["Waiting for video...".to_string()]); + if lines.len() > 1 { + ui.set_min_width(PUBLISHER_TIMING_LINE_WIDTH as f32 * 8.0); + } + ui.add( + egui::Label::new( + egui::RichText::new(lines.join("\n")) + .monospace() + .size(12.0) + .color(egui::Color32::WHITE), + ) + .extend(), + ); + + if let Some(channels) = channels { + ui.separator(); + let values = drive_channels(ui.ctx(), channels); + paint_channel_controls(ui, &values); + } + }); + }, + ); +} + +impl VideoApp { + fn show_diagnostics_window(&self, ctx: &egui::Context) { + if !self.diagnostics_open.load(Ordering::Acquire) { + return; + } + + let shared = self.shared.clone(); + let timing_overlay_state = self.timing_overlay_state.clone(); + let channels = self.channels.clone(); + let diagnostics_open = self.diagnostics_open.clone(); + let diagnostics_started = self.diagnostics_started.clone(); + + ctx.show_viewport_deferred( + diagnostics_viewport_id(), + egui::ViewportBuilder::default() + .with_title("LiveKit Publisher Diagnostics") + .with_inner_size(DIAGNOSTICS_WINDOW_SIZE) + .with_min_inner_size(DIAGNOSTICS_WINDOW_MIN_SIZE), + move |ui, viewport_class| { + if !diagnostics_started.swap(true, Ordering::AcqRel) { + let viewport_class = match viewport_class { + egui::ViewportClass::Root => "root", + egui::ViewportClass::Deferred => "deferred", + egui::ViewportClass::Immediate => "immediate", + egui::ViewportClass::EmbeddedWindow => "embedded", + }; + log::info!( + "Publisher diagnostics window active: {}, refresh={}ms", + viewport_class, + DIAGNOSTICS_REPAINT_INTERVAL.as_millis() + ); + } + if ui.input(|input| input.viewport().close_requested()) { + diagnostics_open.store(false, Ordering::Release); + log::info!( + "Publisher diagnostics window closed; video rendering remains active" + ); + ui.ctx().request_repaint_of(egui::ViewportId::ROOT); + return; + } + + paint_publisher_diagnostics(ui, &shared, &timing_overlay_state, channels.as_ref()); + ui.ctx().request_repaint_after(DIAGNOSTICS_REPAINT_INTERVAL); + }, + ); + } +} + impl eframe::App for VideoApp { fn ui(&mut self, root_ui: &mut egui::Ui, _frame: &mut eframe::Frame) { let ctx = root_ui.ctx().clone(); @@ -658,11 +777,7 @@ impl eframe::App for VideoApp { self.viewport.set_video_size(&ctx, width, height); } - let channel_values = self.channels.as_ref().map(|targets| drive_channels(&ctx, targets)); - egui::CentralPanel::default().frame(egui::Frame::NONE).show(root_ui, |ui| { - ui.ctx().request_repaint(); - let size = viewport_aspect::fitted_video_size(ui.available_size(), self.viewport.aspect()); @@ -678,76 +793,10 @@ impl eframe::App for VideoApp { }, ); }); - - egui::Area::new("publisher_overlay".into()) - .anchor(egui::Align2::LEFT_TOP, egui::vec2(10.0, 10.0)) - .interactable(false) - .show(&ctx, |ui| { - let Some(lines) = publisher_overlay_lines( - &self.shared, - &mut self.timing_overlay_state, - Instant::now(), - ) else { - return; - }; - let has_timing = lines.len() > 1; - let text = lines.join("\n"); - egui::Frame::NONE - .fill(egui::Color32::from_black_alpha(160)) - .corner_radius(egui::CornerRadius::same(4)) - .inner_margin(egui::Margin::same(6)) - .show(ui, |ui| { - if has_timing { - ui.set_min_width(PUBLISHER_TIMING_LINE_WIDTH as f32 * 8.0); - } - ui.add( - egui::Label::new( - egui::RichText::new(text).monospace().color(egui::Color32::WHITE), - ) - .extend(), - ); - }); - }); - - if let Some(values) = channel_values { - paint_channel_controls(&ctx, &values); - } - - ctx.request_repaint_after(viewport_aspect::VIDEO_REPAINT_INTERVAL); + self.show_diagnostics_window(&ctx); } } -/// Bottom-left HUD listing the channel key bindings and current values. -fn paint_channel_controls(ctx: &egui::Context, values: &[f32; crate::user_data::NUM_CHANNELS]) { - const KEY_LABELS: [&str; crate::user_data::NUM_CHANNELS] = - ["Q/A", "W/S", "E/D", "R/F", "T/G", "Y/H"]; - let mut lines = vec!["user_data channels".to_string()]; - for (idx, value) in values.iter().enumerate() { - lines.push(format!("CH{} [{}]: {:>+6.2}", idx + 1, KEY_LABELS[idx], value)); - } - - egui::Area::new("channel_controls".into()) - .anchor(egui::Align2::LEFT_BOTTOM, egui::vec2(10.0, -10.0)) - .interactable(false) - .show(ctx, |ui| { - egui::Frame::NONE - .fill(egui::Color32::from_black_alpha(160)) - .corner_radius(egui::CornerRadius::same(4)) - .inner_margin(egui::Margin::same(6)) - .show(ui, |ui| { - ui.add( - egui::Label::new( - egui::RichText::new(lines.join("\n")) - .monospace() - .size(12.0) - .color(egui::Color32::WHITE), - ) - .extend(), - ); - }); - }); -} - pub(crate) fn run_display( title: &str, shared: Arc>, @@ -759,8 +808,10 @@ pub(crate) fn run_display( shared, ctrl_c_received: ctrl_c_received.clone(), viewport: AspectConstrainedViewport::new(initial_aspect), - timing_overlay_state: PublisherTimingOverlayState::default(), + timing_overlay_state: Arc::new(Mutex::new(PublisherTimingOverlayState::default())), channels, + diagnostics_open: Arc::new(AtomicBool::new(true)), + diagnostics_started: Arc::new(AtomicBool::new(false)), }; let native_options = viewport_aspect::native_options(initial_aspect); let result = eframe::run_native(title, native_options, Box::new(|_| Ok(Box::new(app)))); From 9b5c47763c364edb307aae016b9eff1dfdea11f3 Mon Sep 17 00:00:00 2001 From: David Chen Date: Thu, 6 Aug 2026 14:12:59 -0700 Subject: [PATCH 5/9] updating report layout --- .../scripts/generate_frame_report.py | 98 ++++++++++++------- 1 file changed, 65 insertions(+), 33 deletions(-) diff --git a/examples/local_video/scripts/generate_frame_report.py b/examples/local_video/scripts/generate_frame_report.py index 210324f91..df981da10 100755 --- a/examples/local_video/scripts/generate_frame_report.py +++ b/examples/local_video/scripts/generate_frame_report.py @@ -60,7 +60,7 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--publisher", type=Path, help="Publisher CSV log") parser.add_argument("--subscriber", type=Path, help="Subscriber CSV log") parser.add_argument("-o", "--output", type=Path, help="Output PDF path") - parser.add_argument("--title", default="Local Video Frame Metrics") + parser.add_argument("--title", default="Video Metrics") args = parser.parse_args() if args.publisher is None and args.subscriber is None: parser.error("at least one of --publisher or --subscriber is required") @@ -327,31 +327,63 @@ def draw_time_series( pdf.rect(x, y, width, height, fill=0, stroke=1) +def paired_transport_latencies(publisher: LogData, subscriber: LogData) -> list[float]: + packetize_by_frame_id = {} + for row in publisher.rows: + frame_id = number(row.get("frame_id")) + packetize_timestamp_us = number(row.get("webrtc_packetize_timestamp_us")) + if frame_id is not None and packetize_timestamp_us is not None: + packetize_by_frame_id[round(frame_id)] = packetize_timestamp_us + + latencies = [] + for row in subscriber.rows: + frame_id = number(row.get("frame_id")) + receive_timestamp_us = number(row.get("webrtc_receive_timestamp_us")) + if frame_id is None or receive_timestamp_us is None: + continue + packetize_timestamp_us = packetize_by_frame_id.get(round(frame_id)) + if packetize_timestamp_us is None: + continue + latency_us = receive_timestamp_us - packetize_timestamp_us + if latency_us >= 0: + latencies.append(latency_us / 1_000.0) + return latencies + + def latency_rows(logs: Sequence[LogData]) -> list[tuple[str, list[float]]]: metrics = [] - for log in logs: - if log.kind == "publisher": + publisher = next((log for log in logs if log.kind == "publisher"), None) + subscriber = next((log for log in logs if log.kind == "subscriber"), None) + + if publisher is not None: + columns = ( + ("[Publisher] exposure to buffer", "capture_to_buffer_ms"), + ("[Publisher] encode", "encode_ms"), + ("[Publisher] exposure to packetize", "capture_to_packetize_ms"), + ) + metrics.extend((label, values(publisher.rows, column)) for label, column in columns) + + if publisher is not None and subscriber is not None: + metrics.append( + ("[Transport] publish to receive", paired_transport_latencies(publisher, subscriber)) + ) + + if subscriber is not None: + if "e2e_to_gpu_complete_ms" in subscriber.rows[0]: columns = ( - ("Publisher capture to buffer", "capture_to_buffer_ms"), - ("Publisher encode", "encode_ms"), - ("Publisher capture to packetize", "capture_to_packetize_ms"), + ("[Subscriber] exposure to receive", "exposure_to_receive_ms"), + ("[Subscriber] receive to decode", "receive_to_decode_ms"), + ("[Subscriber] receive to GPU complete", "receive_to_gpu_complete_ms"), + ("End-to-end latency", "e2e_to_gpu_complete_ms"), ) else: - if "e2e_to_gpu_complete_ms" in log.rows[0]: - columns = ( - ("Subscriber exposure to receive", "exposure_to_receive_ms"), - ("Subscriber receive to decode", "receive_to_decode_ms"), - ("Subscriber receive to GPU complete", "receive_to_gpu_complete_ms"), - ("Subscriber end to GPU complete", "e2e_to_gpu_complete_ms"), - ) - else: - columns = ( - ("Subscriber exposure to receive", "exposure_to_receive_ms"), - ("Subscriber receive to decode", "receive_to_decode_ms"), - ("Subscriber receive to paint", "receive_to_paint_ms"), - ("Subscriber end to end", "e2e_latency_ms"), - ) - metrics.extend((label, values(log.rows, column)) for label, column in columns) + columns = ( + ("[Subscriber] exposure to receive", "exposure_to_receive_ms"), + ("[Subscriber] receive to decode", "receive_to_decode_ms"), + ("[Subscriber] receive to paint", "receive_to_paint_ms"), + ("End-to-end latency", "e2e_latency_ms"), + ) + metrics.extend((label, values(subscriber.rows, column)) for label, column in columns) return [(label, samples) for label, samples in metrics if samples] @@ -401,19 +433,19 @@ def draw_delivery_table( else: packet_loss = dropped = freeze_duration = None freeze_count = sum(event.count for event in freezes) - if publisher is not None and subscriber is not None: - loss_label = "Publisher IDs not rendered" - elif subscriber is not None: - loss_label = "Rendered frame-ID gaps" - else: - loss_label = "Packetized frame-ID gaps" - rows = ( - (loss_label, f"{losses:,}"), + rows = [ ("RTP packets lost", format_count(packet_loss)), ("WebRTC frames dropped", format_count(dropped)), ("Freezes", f"{freeze_count:,}"), ("Freeze duration", "NA" if freeze_duration is None else f"{freeze_duration:.0f} ms"), - ) + ] + if publisher is None or subscriber is None: + loss_label = ( + "Rendered frame-ID gaps" + if subscriber is not None + else "Packetized frame-ID gaps" + ) + rows.insert(0, (loss_label, f"{losses:,}")) pdf.setFillColor(INK) pdf.setFont("Helvetica-Bold", 10.5) pdf.drawString(x, y + 18, "Delivery quality") @@ -452,7 +484,7 @@ def generate_report( losses = sum(event.count for event in loss_events) sources = " + ".join(f"{log.label}: {log.path.name}" for log in logs) - subtitle = f"{sources} | inclusive logged frame range" + subtitle = sources output.parent.mkdir(parents=True, exist_ok=True) pdf = canvas.Canvas(str(output), pagesize=landscape(letter)) pdf.setTitle(title) @@ -472,8 +504,8 @@ def generate_report( draw_card(pdf, 38 + index * (card_width + 11), 461, card_width, label, value) draw_time_series(pdf, logs, loss_events, freeze_events, 50, 206, 692, 205) - draw_latency_table(pdf, logs, 38, 145, 470) - draw_delivery_table(pdf, publisher, subscriber, losses, freeze_events, 530, 145, 224) + draw_latency_table(pdf, logs, 38, 160, 470) + draw_delivery_table(pdf, publisher, subscriber, losses, freeze_events, 530, 160, 224) pdf.setStrokeColor(GRID) pdf.line(38, 28, 754, 28) From f5655ea2adb1a2142470b58443103fe8943943c4 Mon Sep 17 00:00:00 2001 From: David Chen Date: Mon, 10 Aug 2026 14:56:43 -0700 Subject: [PATCH 6/9] update report rendering --- .../scripts/generate_frame_report.py | 235 +++++++++++++----- 1 file changed, 173 insertions(+), 62 deletions(-) diff --git a/examples/local_video/scripts/generate_frame_report.py b/examples/local_video/scripts/generate_frame_report.py index df981da10..4475b3bf1 100755 --- a/examples/local_video/scripts/generate_frame_report.py +++ b/examples/local_video/scripts/generate_frame_report.py @@ -31,6 +31,19 @@ PANEL = HexColor("#F0F4F8") RED = HexColor("#D64545") ORANGE = HexColor("#E88D14") +PIPELINE_COLORS = ( + HexColor("#F6C344"), + HexColor("#F29E4C"), + HexColor("#E76F51"), + HexColor("#C8553D"), + HexColor("#8E5BD9"), + HexColor("#4C78A8"), + HexColor("#3A86FF"), + HexColor("#2CB1BC"), + HexColor("#1B998B"), + HexColor("#4ECDC4"), + HexColor("#6C63FF"), +) @dataclass(frozen=True) @@ -85,6 +98,17 @@ def values(rows: Iterable[dict[str, str]], column: str) -> list[float]: return [parsed for row in rows if (parsed := number(row.get(column))) is not None] +def summed_values( + rows: Iterable[dict[str, str]], columns: Sequence[str] +) -> list[float]: + samples = [] + for row in rows: + components = [number(row.get(column)) for column in columns] + if all(component is not None for component in components): + samples.append(sum(component for component in components if component is not None)) + return samples + + def first_available_column( fieldnames: Sequence[str], candidates: Sequence[str] ) -> str | None: @@ -191,13 +215,6 @@ def subscriber_freeze_events(log: LogData) -> list[Event]: return events -def last_value(log: LogData, column: str) -> float | None: - return next( - (parsed for row in reversed(log.rows) if (parsed := number(row.get(column))) is not None), - None, - ) - - def paired_loss_events(publisher: LogData, subscriber: LogData) -> list[Event]: publisher_ids = {round(value) for value in values(publisher.rows, "frame_id")} subscriber_ids = {round(value) for value in values(subscriber.rows, "frame_id")} @@ -215,12 +232,6 @@ def paired_loss_events(publisher: LogData, subscriber: LogData) -> list[Event]: if frame_id is not None and elapsed is not None and round(frame_id) in missing_ids: events.append(Event(elapsed, 1)) return events - - -def format_count(value: float | None) -> str: - return "NA" if value is None else f"{round(value):,}" - - def draw_header(pdf: canvas.Canvas, title: str, subtitle: str) -> None: width, height = landscape(letter) pdf.setFillColor(white) @@ -327,22 +338,33 @@ def draw_time_series( pdf.rect(x, y, width, height, fill=0, stroke=1) -def paired_transport_latencies(publisher: LogData, subscriber: LogData) -> list[float]: - packetize_by_frame_id = {} +def paired_frame_rows( + publisher: LogData, subscriber: LogData +) -> list[tuple[dict[str, str], dict[str, str]]]: + publisher_by_frame_id = {} for row in publisher.rows: frame_id = number(row.get("frame_id")) - packetize_timestamp_us = number(row.get("webrtc_packetize_timestamp_us")) - if frame_id is not None and packetize_timestamp_us is not None: - packetize_by_frame_id[round(frame_id)] = packetize_timestamp_us + if frame_id is not None: + publisher_by_frame_id[round(frame_id)] = row - latencies = [] + pairs = [] for row in subscriber.rows: frame_id = number(row.get("frame_id")) - receive_timestamp_us = number(row.get("webrtc_receive_timestamp_us")) - if frame_id is None or receive_timestamp_us is None: + if frame_id is None: + continue + publisher_row = publisher_by_frame_id.get(round(frame_id)) + if publisher_row is None: continue - packetize_timestamp_us = packetize_by_frame_id.get(round(frame_id)) - if packetize_timestamp_us is None: + pairs.append((publisher_row, row)) + return pairs + + +def paired_transport_latencies(publisher: LogData, subscriber: LogData) -> list[float]: + latencies = [] + for publisher_row, subscriber_row in paired_frame_rows(publisher, subscriber): + packetize_timestamp_us = number(publisher_row.get("webrtc_packetize_timestamp_us")) + receive_timestamp_us = number(subscriber_row.get("webrtc_receive_timestamp_us")) + if packetize_timestamp_us is None or receive_timestamp_us is None: continue latency_us = receive_timestamp_us - packetize_timestamp_us if latency_us >= 0: @@ -387,6 +409,85 @@ def latency_rows(logs: Sequence[LogData]) -> list[tuple[str, list[float]]]: return [(label, samples) for label, samples in metrics if samples] +def pipeline_stage_means(logs: Sequence[LogData]) -> list[tuple[str, float, object]]: + publisher = next((log for log in logs if log.kind == "publisher"), None) + subscriber = next((log for log in logs if log.kind == "subscriber"), None) + publisher_rows = publisher.rows if publisher is not None else [] + subscriber_rows = subscriber.rows if subscriber is not None else [] + + if publisher is not None and subscriber is not None: + pairs = paired_frame_rows(publisher, subscriber) + publisher_rows = [publisher_row for publisher_row, _ in pairs] + subscriber_rows = [subscriber_row for _, subscriber_row in pairs] + + stage_samples = [] + if publisher is not None: + stage_samples.extend( + ( + ("[P] exposure to buffer", values(publisher_rows, "capture_to_buffer_ms"), 0), + ( + "[P] frame encode", + summed_values( + publisher_rows, + ("buffer_to_encoder_ms", "encode_ms"), + ), + 1, + ), + ( + "[P] encoder to packetize", + values(publisher_rows, "encoder_to_packetize_ms"), + 3, + ), + ) + ) + if publisher is not None and subscriber is not None: + stage_samples.append( + ("[T] publish to receive", paired_transport_latencies(publisher, subscriber), 4) + ) + if subscriber is not None: + if "e2e_to_gpu_complete_ms" in subscriber.rows[0]: + stage_samples.extend( + ( + ("[S] receive to decode", values(subscriber_rows, "receive_to_decode_ms"), 5), + ( + "[S] decode to GPU render", + summed_values( + subscriber_rows, + ( + "decode_to_sink_ms", + "sink_to_select_ms", + "select_to_prepare_ms", + "prepare_to_draw_encoded_ms", + ), + ), + 6, + ), + ( + "[S] draw to GPU complete", + values(subscriber_rows, "draw_encoded_to_gpu_complete_ms"), + 10, + ), + ) + ) + else: + subscriber_stages = ( + ("[S] receive to decode", "receive_to_decode_ms", 5), + ("[S] decode to sink", "decode_to_sink_ms", 6), + ("[S] sink to prepare", "sink_to_prepare_ms", 8), + ("[S] prepare to paint", "prepare_to_paint_ms", 10), + ) + stage_samples.extend( + (label, values(subscriber_rows, column), color_index) + for label, column, color_index in subscriber_stages + ) + + return [ + (label, statistics.fmean(samples), PIPELINE_COLORS[color_index]) + for label, samples, color_index in stage_samples + if samples + ] + + def draw_latency_table( pdf: canvas.Canvas, logs: Sequence[LogData], x: float, y: float, width: float ) -> None: @@ -416,50 +517,61 @@ def draw_latency_table( row_y -= 15 -def draw_delivery_table( +def draw_pipeline_timeline( pdf: canvas.Canvas, - publisher: LogData | None, - subscriber: LogData | None, - losses: int, - freezes: Sequence[Event], + logs: Sequence[LogData], x: float, y: float, width: float, ) -> None: - if subscriber is not None: - packet_loss = last_value(subscriber, "packets_lost") - dropped = last_value(subscriber, "frames_dropped") - freeze_duration = last_value(subscriber, "total_freeze_duration_ms") - else: - packet_loss = dropped = freeze_duration = None - freeze_count = sum(event.count for event in freezes) - rows = [ - ("RTP packets lost", format_count(packet_loss)), - ("WebRTC frames dropped", format_count(dropped)), - ("Freezes", f"{freeze_count:,}"), - ("Freeze duration", "NA" if freeze_duration is None else f"{freeze_duration:.0f} ms"), - ] - if publisher is None or subscriber is None: - loss_label = ( - "Rendered frame-ID gaps" - if subscriber is not None - else "Packetized frame-ID gaps" - ) - rows.insert(0, (loss_label, f"{losses:,}")) + stages = pipeline_stage_means(logs) + total_ms = sum(mean_ms for _, mean_ms, _ in stages) pdf.setFillColor(INK) pdf.setFont("Helvetica-Bold", 10.5) - pdf.drawString(x, y + 18, "Delivery quality") - row_y = y - 4 - for index, (label, value) in enumerate(rows): - pdf.setFillColor(PANEL if index % 2 == 0 else white) - pdf.rect(x, row_y - 20, width, 20, fill=1, stroke=0) + pdf.drawString(x, y + 140, "Mean pipeline timeline") + if not stages or total_ms <= 0: pdf.setFillColor(MUTED) - pdf.setFont("Helvetica", 7.5) - pdf.drawString(x + 7, row_y - 13, label) + pdf.setFont("Helvetica", 8) + pdf.drawString(x, y + 116, "No complete pipeline-stage samples") + return + + pdf.setFillColor(MUTED) + pdf.setFont("Helvetica", 7.2) + timeline_summary = f"Segment width is proportional to mean duration | total {total_ms:.1f} ms" + pdf.drawString(x, y + 124, timeline_summary) + + bar_y = y + 96 + bar_height = 19 + cursor_x = x + for index, (_, mean_ms, color) in enumerate(stages): + segment_width = width * mean_ms / total_ms + pdf.setFillColor(color) + pdf.rect(cursor_x, bar_y, segment_width, bar_height, fill=1, stroke=0) + if segment_width >= 12: + pdf.setFillColor(white) + pdf.setFont("Helvetica-Bold", 6.2) + pdf.drawCentredString( + cursor_x + segment_width / 2, + bar_y + 6.2, + str(index + 1), + ) + cursor_x += segment_width + pdf.setStrokeColor(INK) + pdf.rect(x, bar_y, width, bar_height, fill=0, stroke=1) + + rows_per_column = 6 + column_width = width / 2 + legend_y = y + 78 + for index, (label, mean_ms, color) in enumerate(stages): + column = index // rows_per_column + row = index % rows_per_column + item_x = x + column * column_width + item_y = legend_y - row * 12 + pdf.setFillColor(color) + pdf.rect(item_x, item_y - 1, 7, 7, fill=1, stroke=0) pdf.setFillColor(INK) - pdf.setFont("Helvetica-Bold", 8) - pdf.drawRightString(x + width - 7, row_y - 13, value) - row_y -= 20 + pdf.setFont("Helvetica", 6.4) + pdf.drawString(item_x + 11, item_y, f"{index + 1}. {label} {mean_ms:.1f} ms") def generate_report( @@ -504,8 +616,8 @@ def generate_report( draw_card(pdf, 38 + index * (card_width + 11), 461, card_width, label, value) draw_time_series(pdf, logs, loss_events, freeze_events, 50, 206, 692, 205) - draw_latency_table(pdf, logs, 38, 160, 470) - draw_delivery_table(pdf, publisher, subscriber, losses, freeze_events, 530, 160, 224) + draw_latency_table(pdf, logs, 38, 160, 318) + draw_pipeline_timeline(pdf, logs, 380, 38, 374) pdf.setStrokeColor(GRID) pdf.line(38, 28, 754, 28) @@ -519,8 +631,7 @@ def generate_report( pdf.drawString( 38, 17, - "Frame losses are frame-ID gaps; with paired logs they are publisher IDs not rendered by the subscriber. " - + freeze_note, + "Frame-loss markers reflect frame-ID gaps. " + freeze_note, ) pdf.save() From 6ac8ab81161854bf666660f7406e509e0105ca07 Mon Sep 17 00:00:00 2001 From: David Chen Date: Tue, 11 Aug 2026 11:55:26 -0700 Subject: [PATCH 7/9] auto-stop process after logging --- examples/local_video/README.md | 4 +- .../scripts/generate_frame_report.py | 13 ++-- examples/local_video/src/publisher.rs | 64 ++++++++++++++----- examples/local_video/src/subscriber.rs | 20 +++++- examples/local_video/src/subscriber_timing.rs | 35 +++++----- 5 files changed, 92 insertions(+), 44 deletions(-) diff --git a/examples/local_video/README.md b/examples/local_video/README.md index 5f06d4be0..daab3ed28 100644 --- a/examples/local_video/README.md +++ b/examples/local_video/README.md @@ -148,7 +148,7 @@ Publisher flags (in addition to the common connection flags above): - `--display-timing`: Show publisher timing metrics in the diagnostics window. Requires `--display-video`. - `--log-csv `: Write one CSV row per packetized frame with capture, encoder, packetization, frame-gap, and inter-frame timing metrics. This automatically enables timestamp and frame-ID metadata. - `--log-start-frame-id `: Start CSV logging at this frame ID (inclusive). Requires `--log-csv`. -- `--log-end-frame-id `: Stop CSV logging after this frame ID (inclusive). Requires `--log-csv`. +- `--log-end-frame-id `: Flush the terminal packetized frame to CSV, then stop the publisher process. Requires `--log-csv`. - `--e2ee-key `: Enable end-to-end encryption with the given shared key. The subscriber must use the same key to decrypt. Subscriber usage: @@ -203,7 +203,7 @@ Subscriber flags (in addition to the common connection flags above): - `--display-timestamp`: Show detailed frame ID, publisher timestamp, subscriber timing stages, and end-to-end latency in the separate diagnostics window. Timestamp fields require the publisher to use `--attach-timestamp`; frame ID requires `--attach-frame-id`. - `--log-csv `: Write one CSV row per GPU-completed frame with receive, decode, sink, selection, CPU draw, GPU completion, end-to-end latency, frame-gap, inter-frame timing, and WebRTC loss/freeze metrics. The publisher must use `--log-csv` or both `--attach-timestamp` and `--attach-frame-id`. - `--log-start-frame-id `: Start CSV logging at this frame ID (inclusive). Requires `--log-csv`. -- `--log-end-frame-id `: Stop CSV logging after this frame ID (inclusive). Requires `--log-csv`. +- `--log-end-frame-id `: Flush the terminal GPU-completed frame to CSV, then stop the subscriber process. Requires `--log-csv`. - `--e2ee-key `: Enable end-to-end decryption with the given shared key. Must match the key used by the publisher. Generate a PDF report from the publisher log, subscriber log, or both: diff --git a/examples/local_video/scripts/generate_frame_report.py b/examples/local_video/scripts/generate_frame_report.py index 4475b3bf1..c77e03f69 100755 --- a/examples/local_video/scripts/generate_frame_report.py +++ b/examples/local_video/scripts/generate_frame_report.py @@ -426,18 +426,17 @@ def pipeline_stage_means(logs: Sequence[LogData]) -> list[tuple[str, float, obje ( ("[P] exposure to buffer", values(publisher_rows, "capture_to_buffer_ms"), 0), ( - "[P] frame encode", + "[P] encode and packetize", summed_values( publisher_rows, - ("buffer_to_encoder_ms", "encode_ms"), + ( + "buffer_to_encoder_ms", + "encode_ms", + "encoder_to_packetize_ms", + ), ), 1, ), - ( - "[P] encoder to packetize", - values(publisher_rows, "encoder_to_packetize_ms"), - 3, - ), ) ) if publisher is not None and subscriber is not None: diff --git a/examples/local_video/src/publisher.rs b/examples/local_video/src/publisher.rs index 3483f813f..094542fde 100644 --- a/examples/local_video/src/publisher.rs +++ b/examples/local_video/src/publisher.rs @@ -279,7 +279,7 @@ struct Args { #[arg(long, requires = "log_csv")] log_start_frame_id: Option, - /// Stop CSV logging after this frame ID (inclusive) + /// Stop the process after this packetized frame ID is written to CSV (inclusive) #[arg(long, requires = "log_csv")] log_end_frame_id: Option, @@ -622,24 +622,24 @@ impl PublisherCsvLogger { }) } - fn record(&mut self, sample: PublisherTimingSample) -> std::io::Result<()> { + fn record(&mut self, sample: PublisherTimingSample) -> std::io::Result { let Some(frame_id) = sample.frame_id else { - return Ok(()); + return Ok(false); }; if !self.range.contains(frame_id) { - return Ok(()); + return Ok(false); } let Some(frame_buffer_timestamp_us) = sample.got_frame_buffer_timestamp_us else { - return Ok(()); + return Ok(false); }; let Some(encoder_upload_timestamp_us) = sample.encoder_upload_timestamp_us else { - return Ok(()); + return Ok(false); }; let Some(encoder_output_timestamp_us) = sample.encoder_output_timestamp_us else { - return Ok(()); + return Ok(false); }; let Some(packetize_timestamp_us) = sample.webrtc_packetize_timestamp_us else { - return Ok(()); + return Ok(false); }; let first_packetize_timestamp_us = @@ -689,7 +689,7 @@ impl PublisherCsvLogger { self.writer.flush()?; self.last_flush = Instant::now(); } - Ok(()) + Ok(self.range.reaches_end(frame_id)) } } @@ -699,6 +699,7 @@ struct PublisherTimingState { order: VecDeque, latest_complete_sample: Option, frame_log: Option, + completed_log_frame_id: Option, } impl PublisherTimingState { @@ -737,9 +738,13 @@ impl PublisherTimingState { if updated_sample.is_complete() { self.latest_complete_sample = Some(updated_sample); if let Some(frame_log) = self.frame_log.as_mut() { - if let Err(error) = frame_log.record(updated_sample) { - warn!("Publisher CSV logging disabled after write failure: {error}"); - self.frame_log = None; + match frame_log.record(updated_sample) { + Ok(true) => self.completed_log_frame_id = updated_sample.frame_id, + Ok(false) => {} + Err(error) => { + warn!("Publisher CSV logging disabled after write failure: {error}"); + self.frame_log = None; + } } } Some(updated_sample) @@ -752,6 +757,10 @@ impl PublisherTimingState { self.latest_complete_sample } + fn take_completed_log_frame_id(&mut self) -> Option { + self.completed_log_frame_id.take() + } + fn get_or_insert_sample( &mut self, sensor_exposure_timestamp_us: u64, @@ -960,15 +969,25 @@ mod tests { encoder_output_timestamp_us: Some(1_300), webrtc_packetize_timestamp_us: Some(1_400), }; - logger.record(sample).expect("sample should be written"); - logger.writer.flush().expect("log should flush"); - drop(logger); + assert!(!logger.record(sample).expect("sample should be written")); + let end_sample = PublisherTimingSample { + frame_id: Some(302), + sensor_exposure_timestamp_us: 35_000, + got_frame_buffer_timestamp_us: Some(35_100), + encoder_upload_timestamp_us: Some(35_200), + encoder_output_timestamp_us: Some(35_300), + webrtc_packetize_timestamp_us: Some(35_400), + }; + assert!(logger.record(end_sample).expect("end sample should be written")); let contents = std::fs::read_to_string(&path).expect("log should be readable"); + drop(logger); let lines: Vec<_> = contents.lines().collect(); - assert_eq!(lines.len(), 2); + assert_eq!(lines.len(), 3); assert_eq!(lines[0].split(',').count(), lines[1].split(',').count()); + assert_eq!(lines[0].split(',').count(), lines[2].split(',').count()); assert!(lines[1].starts_with("1,0.000,301,")); + assert!(lines[2].starts_with("2,34.000,302,")); std::fs::remove_file(path).expect("temporary log should be removable"); } } @@ -1309,15 +1328,26 @@ async fn run(args: Args, ctrl_c_received: Arc) -> Result<()> { if let Some(timing_state) = publish_timing_state.as_ref() { let timing_state = timing_state.clone(); let display_shared_for_timing = display_shared.clone(); + let shutdown_on_log_end = ctrl_c_received.clone(); let mut events = track.publish_timing_events(); tokio::spawn(async move { use tokio_stream::StreamExt; while let Some(event) = events.next().await { - let sample = timing_state.lock().record_sdk_event(event); + let (sample, completed_frame_id) = { + let mut timing_state = timing_state.lock(); + let sample = timing_state.record_sdk_event(event); + let completed_frame_id = timing_state.take_completed_log_frame_id(); + (sample, completed_frame_id) + }; if let Some(sample) = sample { update_shared_timing_sample(display_shared_for_timing.as_ref(), sample); } + if let Some(frame_id) = completed_frame_id { + info!("Publisher completed --log-end-frame-id {frame_id}; shutting down..."); + shutdown_on_log_end.store(true, Ordering::Release); + break; + } } }); } diff --git a/examples/local_video/src/subscriber.rs b/examples/local_video/src/subscriber.rs index 1b7894273..161887076 100644 --- a/examples/local_video/src/subscriber.rs +++ b/examples/local_video/src/subscriber.rs @@ -425,7 +425,7 @@ struct Args { #[arg(long, requires = "log_csv")] log_start_frame_id: Option, - /// Stop CSV logging after this frame ID (inclusive) + /// Stop the process after this GPU-rendered frame ID is written to CSV (inclusive) #[arg(long, requires = "log_csv")] log_end_frame_id: Option, @@ -1715,6 +1715,8 @@ impl eframe::App for VideoApp { render_frame: Mutex::new(render_frame), video_size: self.video_size.clone(), subscriber_timing: self.subscriber_timing.clone(), + shutdown_on_log_end: self.ctrl_c_received.clone(), + repaint_ctx: ctx.clone(), }, ); ui.painter().add(cb); @@ -1935,6 +1937,8 @@ struct YuvPaintCallback { render_frame: Mutex>, video_size: Arc, subscriber_timing: SubscriberTimingHandle, + shutdown_on_log_end: Arc, + repaint_ctx: egui::Context, } struct YuvGpuState { @@ -2509,9 +2513,19 @@ impl CallbackTrait for YuvPaintCallback { ); let completion_probe = state.gpu_completion_poller.begin_probe(); let subscriber_timing = self.subscriber_timing.clone(); + let shutdown_on_log_end = self.shutdown_on_log_end.clone(); + let repaint_ctx = self.repaint_ctx.clone(); render_pass.on_submitted_work_done(move || { - subscriber_timing - .record_frame_gpu_complete(completion_token, current_timestamp_us()); + if let Some(frame_id) = subscriber_timing + .record_frame_gpu_complete(completion_token, current_timestamp_us()) + { + if !shutdown_on_log_end.swap(true, Ordering::AcqRel) { + info!( + "Subscriber completed --log-end-frame-id {frame_id}; shutting down..." + ); + } + repaint_ctx.request_repaint_of(egui::ViewportId::ROOT); + } drop(completion_probe); }); } diff --git a/examples/local_video/src/subscriber_timing.rs b/examples/local_video/src/subscriber_timing.rs index b9d5c182a..c633414ea 100644 --- a/examples/local_video/src/subscriber_timing.rs +++ b/examples/local_video/src/subscriber_timing.rs @@ -88,22 +88,27 @@ impl SubscriberTimingHandle { ) } - /// Records that the GPU submission containing the frame has completed. + /// Records GPU completion and returns the terminal logged frame ID, if reached. pub(crate) fn record_frame_gpu_complete( &self, token: FrameGpuCompletionToken, frame_gpu_complete_timestamp_us: u64, - ) { + ) -> Option { let sample = self.inner.lock().record_frame_gpu_complete(token, frame_gpu_complete_timestamp_us); let Some(sample) = sample else { - return; + return None; }; if let Some(frame_log) = &self.frame_log { - if let Err(error) = frame_log.lock().record(sample) { - warn!("Subscriber CSV logging disabled after write failure: {error}"); + match frame_log.lock().record(sample) { + Ok(true) => return sample.frame_id, + Ok(false) => {} + Err(error) => { + warn!("Subscriber CSV logging disabled after write failure: {error}"); + } } } + None } pub(crate) fn display_overlay_lines(&self, now: Instant) -> Option> { @@ -435,18 +440,18 @@ impl SubscriberCsvLogger { }) } - fn record(&mut self, sample: SubscriberTimingSample) -> io::Result<()> { + fn record(&mut self, sample: SubscriberTimingSample) -> io::Result { if self.failed { - return Ok(()); + return Ok(false); } let Some(frame_id) = sample.frame_id else { - return Ok(()); + return Ok(false); }; if !self.range.contains(frame_id) { - return Ok(()); + return Ok(false); } let Some(frame_gpu_complete_timestamp_us) = sample.frame_gpu_complete_timestamp_us else { - return Ok(()); + return Ok(false); }; let first_gpu_complete_timestamp_us = @@ -526,8 +531,8 @@ impl SubscriberCsvLogger { CsvFloat(quality.map(|quality| quality.total_freeze_duration_ms)), ); - let should_flush = - self.range.reaches_end(frame_id) || self.last_flush.elapsed() >= Duration::from_secs(1); + let reached_end = self.range.reaches_end(frame_id); + let should_flush = reached_end || self.last_flush.elapsed() >= Duration::from_secs(1); let result = result.and_then(|()| if should_flush { self.writer.flush() } else { Ok(()) }); if result.is_ok() { self.previous_frame_id = Some(frame_id); @@ -538,7 +543,7 @@ impl SubscriberCsvLogger { } else { self.failed = true; } - result + result.map(|()| reached_end) } } @@ -1088,9 +1093,9 @@ mod tests { let contents = std::fs::read_to_string(&path).expect("log should be readable"); assert_eq!(contents.lines().count(), 1); - timing.record_frame_gpu_complete(token, 1_400); - drop(timing); + assert_eq!(timing.record_frame_gpu_complete(token, 1_400), Some(301)); let contents = std::fs::read_to_string(&path).expect("log should be readable"); + drop(timing); let lines: Vec<_> = contents.lines().collect(); assert_eq!(lines.len(), 2); assert!(lines[1].starts_with("1,0.000,301,1000,")); From ca5fe1ded49885a735cf41f6ea44b72334e39cb3 Mon Sep 17 00:00:00 2001 From: David Chen Date: Tue, 11 Aug 2026 13:44:41 -0700 Subject: [PATCH 8/9] fix timing to separate receive / assembly from decode --- examples/local_video/README.md | 4 +- .../scripts/generate_frame_report.py | 23 +- examples/local_video/src/subscriber_timing.rs | 199 +++++++++--------- libwebrtc/src/native/packet_trailer.rs | 40 +++- libwebrtc/src/native/video_stream.rs | 29 ++- webrtc-sys/include/livekit/packet_trailer.h | 12 ++ webrtc-sys/include/livekit/video_frame.h | 2 + webrtc-sys/src/packet_trailer.cpp | 49 ++++- webrtc-sys/src/packet_trailer.rs | 9 + webrtc-sys/src/video_frame.cpp | 48 +++++ webrtc-sys/src/video_frame.rs | 4 + 11 files changed, 295 insertions(+), 124 deletions(-) diff --git a/examples/local_video/README.md b/examples/local_video/README.md index daab3ed28..b587fb4d3 100644 --- a/examples/local_video/README.md +++ b/examples/local_video/README.md @@ -200,8 +200,8 @@ Subscriber usage: Subscriber flags (in addition to the common connection flags above): - `--participant `: Only subscribe to video tracks from the specified participant. - `--low-latency`: Force zero video playout delay so received frames render as soon as possible. This can increase visible stutter when packets arrive late or out of order. -- `--display-timestamp`: Show detailed frame ID, publisher timestamp, subscriber timing stages, and end-to-end latency in the separate diagnostics window. Timestamp fields require the publisher to use `--attach-timestamp`; frame ID requires `--attach-frame-id`. -- `--log-csv `: Write one CSV row per GPU-completed frame with receive, decode, sink, selection, CPU draw, GPU completion, end-to-end latency, frame-gap, inter-frame timing, and WebRTC loss/freeze metrics. The publisher must use `--log-csv` or both `--attach-timestamp` and `--attach-frame-id`. +- `--display-timestamp`: Show detailed frame ID, publisher timestamp, first-packet receive/assembly (including jitter-buffer scheduling), actual decoder processing, render, and end-to-end timing in the separate diagnostics window. Timestamp fields require the publisher to use `--attach-timestamp`; frame ID requires `--attach-frame-id`. +- `--log-csv `: Write one CSV row per GPU-completed frame with receive/assembly through decode start, actual decode processing, render, detailed render-boundary timestamps, end-to-end latency, frame-gap, inter-frame timing, and WebRTC loss/freeze metrics. The publisher must use `--log-csv` or both `--attach-timestamp` and `--attach-frame-id`. - `--log-start-frame-id `: Start CSV logging at this frame ID (inclusive). Requires `--log-csv`. - `--log-end-frame-id `: Flush the terminal GPU-completed frame to CSV, then stop the subscriber process. Requires `--log-csv`. - `--e2ee-key `: Enable end-to-end decryption with the given shared key. Must match the key used by the publisher. diff --git a/examples/local_video/scripts/generate_frame_report.py b/examples/local_video/scripts/generate_frame_report.py index c77e03f69..d66e23048 100755 --- a/examples/local_video/scripts/generate_frame_report.py +++ b/examples/local_video/scripts/generate_frame_report.py @@ -40,7 +40,7 @@ HexColor("#4C78A8"), HexColor("#3A86FF"), HexColor("#2CB1BC"), - HexColor("#1B998B"), + HexColor("#44FF33"), HexColor("#4ECDC4"), HexColor("#6C63FF"), ) @@ -391,7 +391,14 @@ def latency_rows(logs: Sequence[LogData]) -> list[tuple[str, list[float]]]: ) if subscriber is not None: - if "e2e_to_gpu_complete_ms" in subscriber.rows[0]: + if "receive_and_assembly_ms" in subscriber.rows[0]: + columns = ( + ("[Subscriber] receive and assembly", "receive_and_assembly_ms"), + ("[Subscriber] decode", "decode_ms"), + ("[Subscriber] render", "render_ms"), + ("End-to-end latency", "e2e_to_gpu_complete_ms"), + ) + elif "e2e_to_gpu_complete_ms" in subscriber.rows[0]: columns = ( ("[Subscriber] exposure to receive", "exposure_to_receive_ms"), ("[Subscriber] receive to decode", "receive_to_decode_ms"), @@ -444,7 +451,17 @@ def pipeline_stage_means(logs: Sequence[LogData]) -> list[tuple[str, float, obje ("[T] publish to receive", paired_transport_latencies(publisher, subscriber), 4) ) if subscriber is not None: - if "e2e_to_gpu_complete_ms" in subscriber.rows[0]: + if "receive_and_assembly_ms" in subscriber.rows[0]: + subscriber_stages = ( + ("[S] receive and assembly", "receive_and_assembly_ms", 5), + ("[S] decode", "decode_ms", 6), + ("[S] render", "render_ms", 8), + ) + stage_samples.extend( + (label, values(subscriber_rows, column), color_index) + for label, column, color_index in subscriber_stages + ) + elif "e2e_to_gpu_complete_ms" in subscriber.rows[0]: stage_samples.extend( ( ("[S] receive to decode", values(subscriber_rows, "receive_to_decode_ms"), 5), diff --git a/examples/local_video/src/subscriber_timing.rs b/examples/local_video/src/subscriber_timing.rs index c633414ea..2a4fa1772 100644 --- a/examples/local_video/src/subscriber_timing.rs +++ b/examples/local_video/src/subscriber_timing.rs @@ -401,7 +401,7 @@ impl SubscriberTimingState { } } -const SUBSCRIBER_CSV_HEADER: &str = "sample,elapsed_ms,frame_id,capture_timestamp_us,webrtc_receive_timestamp_us,decoder_upload_timestamp_us,decoder_output_timestamp_us,frame_sink_timestamp_us,frame_selected_timestamp_us,frame_prepare_timestamp_us,frame_draw_encoded_timestamp_us,frame_gpu_complete_timestamp_us,exposure_to_receive_ms,receive_to_decode_ms,decode_to_sink_ms,sink_to_select_ms,select_to_prepare_ms,prepare_to_draw_encoded_ms,draw_encoded_to_gpu_complete_ms,receive_to_gpu_complete_ms,e2e_to_gpu_complete_ms,frame_id_gap,gpu_complete_interval_ms,packets_lost,frames_dropped,freeze_count,total_freeze_duration_ms"; +const SUBSCRIBER_CSV_HEADER: &str = "sample,elapsed_ms,frame_id,capture_timestamp_us,webrtc_receive_timestamp_us,decoder_upload_timestamp_us,decoder_output_timestamp_us,frame_sink_timestamp_us,frame_selected_timestamp_us,frame_prepare_timestamp_us,frame_draw_encoded_timestamp_us,frame_gpu_complete_timestamp_us,exposure_to_receive_ms,receive_and_assembly_ms,decode_ms,render_ms,receive_to_decode_ms,decode_to_sink_ms,sink_to_select_ms,select_to_prepare_ms,prepare_to_draw_encoded_ms,draw_encoded_to_gpu_complete_ms,receive_to_gpu_complete_ms,e2e_to_gpu_complete_ms,frame_id_gap,gpu_complete_interval_ms,packets_lost,frames_dropped,freeze_count,total_freeze_duration_ms"; #[derive(Clone, Copy)] struct InboundQualitySnapshot { @@ -479,7 +479,7 @@ impl SubscriberCsvLogger { let result = writeln!( self.writer, - "{},{:.3},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{}", + "{},{:.3},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{}", self.sample_count, frame_gpu_complete_timestamp_us.saturating_sub(first_gpu_complete_timestamp_us) as f64 / 1_000.0, @@ -497,6 +497,18 @@ impl SubscriberCsvLogger { Some(sample.sensor_exposure_timestamp_us), sample.webrtc_receive_timestamp_us, ), + CsvLatency::between( + sample.webrtc_receive_timestamp_us, + sample.decoder_upload_timestamp_us, + ), + CsvLatency::between( + sample.decoder_upload_timestamp_us, + sample.decoder_output_timestamp_us, + ), + CsvLatency::between( + sample.decoder_output_timestamp_us, + sample.frame_gpu_complete_timestamp_us, + ), CsvLatency::between( sample.webrtc_receive_timestamp_us, sample.decoder_output_timestamp_us, @@ -571,12 +583,9 @@ impl LatencyStats { #[derive(Default)] struct RenderLatencyWindow { - receive_to_decode: LatencyStats, - decode_to_sink: LatencyStats, - sink_to_select: LatencyStats, - select_to_prepare: LatencyStats, - prepare_to_draw_encoded: LatencyStats, - draw_encoded_to_gpu_complete: LatencyStats, + receive_and_assembly: LatencyStats, + decode: LatencyStats, + render: LatencyStats, receive_to_gpu_complete: LatencyStats, e2e_to_gpu_complete: LatencyStats, last_log: Option, @@ -588,39 +597,20 @@ impl RenderLatencyWindow { return; }; - if let (Some(webrtc_receive), Some(decoder_output)) = - (sample.webrtc_receive_timestamp_us, sample.decoder_output_timestamp_us) - { - self.receive_to_decode.record_delta(webrtc_receive, decoder_output); - } - - if let (Some(decoder_output), Some(frame_sink)) = - (sample.decoder_output_timestamp_us, sample.frame_sink_timestamp_us) - { - self.decode_to_sink.record_delta(decoder_output, frame_sink); - } - - if let (Some(frame_sink), Some(frame_selected)) = - (sample.frame_sink_timestamp_us, sample.frame_selected_timestamp_us) - { - self.sink_to_select.record_delta(frame_sink, frame_selected); - } - - if let (Some(frame_selected), Some(frame_prepare)) = - (sample.frame_selected_timestamp_us, sample.frame_prepare_timestamp_us) + if let (Some(webrtc_receive), Some(decoder_upload)) = + (sample.webrtc_receive_timestamp_us, sample.decoder_upload_timestamp_us) { - self.select_to_prepare.record_delta(frame_selected, frame_prepare); + self.receive_and_assembly.record_delta(webrtc_receive, decoder_upload); } - if let (Some(frame_prepare), Some(frame_draw_encoded)) = - (sample.frame_prepare_timestamp_us, sample.frame_draw_encoded_timestamp_us) + if let (Some(decoder_upload), Some(decoder_output)) = + (sample.decoder_upload_timestamp_us, sample.decoder_output_timestamp_us) { - self.prepare_to_draw_encoded.record_delta(frame_prepare, frame_draw_encoded); + self.decode.record_delta(decoder_upload, decoder_output); } - if let Some(frame_draw_encoded) = sample.frame_draw_encoded_timestamp_us { - self.draw_encoded_to_gpu_complete - .record_delta(frame_draw_encoded, frame_gpu_complete_timestamp_us); + if let Some(decoder_output) = sample.decoder_output_timestamp_us { + self.render.record_delta(decoder_output, frame_gpu_complete_timestamp_us); } if let Some(webrtc_receive) = sample.webrtc_receive_timestamp_us { @@ -646,26 +636,17 @@ impl RenderLatencyWindow { } info!( - "Subscriber GPU-completion latency: frames={}, receive_to_decode avg={} min={} max={}, decoder_to_sink avg={} min={} max={}, sink_to_select avg={} min={} max={}, select_to_prepare avg={} min={} max={}, prepare_to_draw_encoded avg={} min={} max={}, draw_encoded_to_gpu_complete avg={} min={} max={}, receive_to_gpu_complete avg={} min={} max={}, e2e_to_gpu_complete avg={} min={} max={}", + "Subscriber GPU-completion latency: frames={}, receive_and_assembly avg={} min={} max={}, decode avg={} min={} max={}, render avg={} min={} max={}, receive_to_gpu_complete avg={} min={} max={}, e2e_to_gpu_complete avg={} min={} max={}", self.e2e_to_gpu_complete.count, - latency_log_value(self.receive_to_decode.avg_us()), - latency_log_value(self.receive_to_decode.min_us), - latency_log_value(self.receive_to_decode.max_us), - latency_log_value(self.decode_to_sink.avg_us()), - latency_log_value(self.decode_to_sink.min_us), - latency_log_value(self.decode_to_sink.max_us), - latency_log_value(self.sink_to_select.avg_us()), - latency_log_value(self.sink_to_select.min_us), - latency_log_value(self.sink_to_select.max_us), - latency_log_value(self.select_to_prepare.avg_us()), - latency_log_value(self.select_to_prepare.min_us), - latency_log_value(self.select_to_prepare.max_us), - latency_log_value(self.prepare_to_draw_encoded.avg_us()), - latency_log_value(self.prepare_to_draw_encoded.min_us), - latency_log_value(self.prepare_to_draw_encoded.max_us), - latency_log_value(self.draw_encoded_to_gpu_complete.avg_us()), - latency_log_value(self.draw_encoded_to_gpu_complete.min_us), - latency_log_value(self.draw_encoded_to_gpu_complete.max_us), + latency_log_value(self.receive_and_assembly.avg_us()), + latency_log_value(self.receive_and_assembly.min_us), + latency_log_value(self.receive_and_assembly.max_us), + latency_log_value(self.decode.avg_us()), + latency_log_value(self.decode.min_us), + latency_log_value(self.decode.max_us), + latency_log_value(self.render.avg_us()), + latency_log_value(self.render.min_us), + latency_log_value(self.render.max_us), latency_log_value(self.receive_to_gpu_complete.avg_us()), latency_log_value(self.receive_to_gpu_complete.min_us), latency_log_value(self.receive_to_gpu_complete.max_us), @@ -684,7 +665,6 @@ struct SubscriberTimingDeltaValues { webrtc_receive: String, decoder_upload: String, decoder_output: String, - frame_draw_encoded: String, frame_gpu_complete: String, } @@ -705,13 +685,9 @@ impl SubscriberTimingDeltaValues { sample.decoder_output_timestamp_us, sample.decoder_upload_timestamp_us, ), - frame_draw_encoded: format_optional_timing_delta_ms( - sample.frame_draw_encoded_timestamp_us, - sample.decoder_output_timestamp_us, - ), frame_gpu_complete: format_optional_timing_delta_ms( sample.frame_gpu_complete_timestamp_us, - sample.frame_draw_encoded_timestamp_us, + sample.decoder_output_timestamp_us, ), } } @@ -808,33 +784,28 @@ fn build_timing_overlay_lines( timing_value_line("Frame ID", &frame_id), timing_line("sensor exposure", Some(base), &overlay_values.deltas.sensor_exposure), timing_line( - "webrtc receive", + "first packet receive", sample.webrtc_receive_timestamp_us, &overlay_values.deltas.webrtc_receive, ), timing_line( - "decoder upload", + "decode start", sample.decoder_upload_timestamp_us, &overlay_values.deltas.decoder_upload, ), timing_line( - "decoder output", + "decode complete", sample.decoder_output_timestamp_us, &overlay_values.deltas.decoder_output, ), timing_line( - "frame draw encoded", - sample.frame_draw_encoded_timestamp_us, - &overlay_values.deltas.frame_draw_encoded, - ), - timing_line( - "frame GPU complete", + "render complete", sample.frame_gpu_complete_timestamp_us, &overlay_values.deltas.frame_gpu_complete, ), ]; lines.extend([ - timing_value_line("Exposure to Receive", &overlay_values.exp2recv_latency), + timing_value_line("Exposure to First Pkt", &overlay_values.exp2recv_latency), timing_value_line("Receive to GPU", &overlay_values.receive_to_gpu_complete_latency), timing_value_line("e2e to GPU", &overlay_values.e2e_to_gpu_complete_latency), ]); @@ -900,12 +871,11 @@ mod tests { vec![ "Frame ID: 123", "sensor exposure: 01:02:03:456 0.0ms", - "webrtc receive: 01:02:03:488 +32.4ms", - "decoder upload: 01:02:03:491 +3.1ms", - "decoder output: 01:02:03:511 +19.8ms", - "frame draw encoded: 01:02:03:512 +1.6ms", - "frame GPU complete: 01:02:03:513 +0.7ms", - "Exposure to Receive: 32.4ms", + "first packet receive: 01:02:03:488 +32.4ms", + "decode start: 01:02:03:491 +3.1ms", + "decode complete: 01:02:03:511 +19.8ms", + "render complete: 01:02:03:513 +2.3ms", + "Exposure to First Pkt: 32.4ms", "Receive to GPU: 25.2ms", "e2e to GPU: 57.6ms", ] @@ -925,12 +895,11 @@ mod tests { vec![ "Frame ID: NA", "sensor exposure: 01:02:03:456 0.0ms", - "webrtc receive: --:--:--:--- +--.-ms", - "decoder upload: --:--:--:--- +--.-ms", - "decoder output: --:--:--:--- +--.-ms", - "frame draw encoded: --:--:--:--- +--.-ms", - "frame GPU complete: --:--:--:--- +--.-ms", - "Exposure to Receive: NA", + "first packet receive: --:--:--:--- +--.-ms", + "decode start: --:--:--:--- +--.-ms", + "decode complete: --:--:--:--- +--.-ms", + "render complete: --:--:--:--- +--.-ms", + "Exposure to First Pkt: NA", "Receive to GPU: NA", "e2e to GPU: NA", ] @@ -983,6 +952,30 @@ mod tests { assert_eq!(sample.frame_gpu_complete_timestamp_us, Some(1_800)); } + #[test] + fn subscriber_latency_window_separates_receive_decode_and_render() { + let now = Instant::now(); + let mut window = RenderLatencyWindow { last_log: Some(now), ..Default::default() }; + let sample = SubscriberTimingSample { + frame_id: Some(123), + sensor_exposure_timestamp_us: 1_000, + webrtc_receive_timestamp_us: Some(1_100), + decoder_upload_timestamp_us: Some(1_120), + decoder_output_timestamp_us: Some(1_200), + frame_sink_timestamp_us: Some(1_210), + frame_selected_timestamp_us: Some(1_220), + frame_prepare_timestamp_us: Some(1_230), + frame_draw_encoded_timestamp_us: Some(1_300), + frame_gpu_complete_timestamp_us: Some(1_500), + }; + + window.record(sample, now); + + assert_eq!(window.receive_and_assembly.avg_us(), Some(20)); + assert_eq!(window.decode.avg_us(), Some(80)); + assert_eq!(window.render.avg_us(), Some(300)); + } + #[test] fn subscriber_timing_ignores_stale_and_out_of_order_gpu_completions() { let mut state = SubscriberTimingState::default(); @@ -1031,13 +1024,12 @@ mod tests { let token = state.record_frame_draw_encoded(1_000, Some(1), 57_200, 57_900); state.record_frame_gpu_complete(token, 58_600); let lines = state.display_overlay_lines(now).expect("overlay should render"); - assert_eq!(lines[3], "decoder upload: 00:00:00:036 +3.1ms"); - assert_eq!(lines[4], "decoder output: 00:00:00:056 +19.8ms"); - assert_eq!(lines[5], "frame draw encoded: 00:00:00:057 +1.6ms"); - assert_eq!(lines[6], "frame GPU complete: 00:00:00:058 +0.7ms"); - assert_eq!(lines[7], "Exposure to Receive: 32.4ms"); - assert_eq!(lines[8], "Receive to GPU: 25.2ms"); - assert_eq!(lines[9], "e2e to GPU: 57.6ms"); + assert_eq!(lines[3], "decode start: 00:00:00:036 +3.1ms"); + assert_eq!(lines[4], "decode complete: 00:00:00:056 +19.8ms"); + assert_eq!(lines[5], "render complete: 00:00:00:058 +2.3ms"); + assert_eq!(lines[6], "Exposure to First Pkt: 32.4ms"); + assert_eq!(lines[7], "Receive to GPU: 25.2ms"); + assert_eq!(lines[8], "e2e to GPU: 57.6ms"); state.record_subscribe_event(subscribe_event( SubscribeTimingStage::WebrtcReceive, @@ -1060,24 +1052,22 @@ mod tests { let lines = state .display_overlay_lines(now + Duration::from_millis(99)) .expect("overlay should render"); - assert_eq!(lines[3], "decoder upload: 00:00:01:060 +3.1ms"); - assert_eq!(lines[4], "decoder output: 00:00:01:080 +19.8ms"); - assert_eq!(lines[5], "frame draw encoded: 00:00:01:100 +1.6ms"); - assert_eq!(lines[6], "frame GPU complete: 00:00:01:104 +0.7ms"); - assert_eq!(lines[7], "Exposure to Receive: 32.4ms"); - assert_eq!(lines[8], "Receive to GPU: 25.2ms"); - assert_eq!(lines[9], "e2e to GPU: 57.6ms"); + assert_eq!(lines[3], "decode start: 00:00:01:060 +3.1ms"); + assert_eq!(lines[4], "decode complete: 00:00:01:080 +19.8ms"); + assert_eq!(lines[5], "render complete: 00:00:01:104 +2.3ms"); + assert_eq!(lines[6], "Exposure to First Pkt: 32.4ms"); + assert_eq!(lines[7], "Receive to GPU: 25.2ms"); + assert_eq!(lines[8], "e2e to GPU: 57.6ms"); let lines = state .display_overlay_lines(now + Duration::from_millis(100)) .expect("overlay should render"); - assert_eq!(lines[3], "decoder upload: 00:00:01:060 +10.0ms"); - assert_eq!(lines[4], "decoder output: 00:00:01:080 +20.0ms"); - assert_eq!(lines[5], "frame draw encoded: 00:00:01:100 +20.0ms"); - assert_eq!(lines[6], "frame GPU complete: 00:00:01:104 +4.0ms"); - assert_eq!(lines[7], "Exposure to Receive: 50.0ms"); - assert_eq!(lines[8], "Receive to GPU: 54.0ms"); - assert_eq!(lines[9], "e2e to GPU: 104.0ms"); + assert_eq!(lines[3], "decode start: 00:00:01:060 +10.0ms"); + assert_eq!(lines[4], "decode complete: 00:00:01:080 +20.0ms"); + assert_eq!(lines[5], "render complete: 00:00:01:104 +24.0ms"); + assert_eq!(lines[6], "Exposure to First Pkt: 50.0ms"); + assert_eq!(lines[7], "Receive to GPU: 54.0ms"); + assert_eq!(lines[8], "e2e to GPU: 104.0ms"); } #[test] @@ -1166,6 +1156,9 @@ mod tests { assert_eq!(first[column("frame_prepare_timestamp_us")], "1220"); assert_eq!(first[column("frame_draw_encoded_timestamp_us")], "1250"); assert_eq!(first[column("frame_gpu_complete_timestamp_us")], "1300"); + assert_eq!(first[column("receive_and_assembly_ms")], "0.010"); + assert_eq!(first[column("decode_ms")], "0.090"); + assert_eq!(first[column("render_ms")], "0.100"); assert_eq!(first[column("sink_to_select_ms")], "0.005"); assert_eq!(first[column("select_to_prepare_ms")], "0.005"); assert_eq!(first[column("prepare_to_draw_encoded_ms")], "0.030"); diff --git a/libwebrtc/src/native/packet_trailer.rs b/libwebrtc/src/native/packet_trailer.rs index 476b01a9f..29710e2cf 100644 --- a/libwebrtc/src/native/packet_trailer.rs +++ b/libwebrtc/src/native/packet_trailer.rs @@ -50,11 +50,11 @@ pub enum PublishTimingStage { /// Stage reached by a native remote video frame in the subscribe pipeline. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SubscribeTimingStage { - /// WebRTC produced an encoded frame after RTP depacketization. + /// The frame's first RTP packet reached the receiver's network interface. WebrtcReceive, - /// The encoded frame was handed to WebRTC's decoder. + /// WebRTC started decoding after frame assembly and jitter-buffer scheduling. DecoderUpload, - /// WebRTC produced a decoded frame for the native video sink. + /// WebRTC finished decoding the frame. DecoderOutput, } @@ -236,12 +236,34 @@ impl PacketTrailerHandler { capture_timestamp_us: u64, frame_id: u32, ) { - let stage = match stage { - SubscribeTimingStage::WebrtcReceive => sys_pt::VideoSubscribeTimingStage::WebrtcReceive, - SubscribeTimingStage::DecoderUpload => sys_pt::VideoSubscribeTimingStage::DecoderUpload, - SubscribeTimingStage::DecoderOutput => sys_pt::VideoSubscribeTimingStage::DecoderOutput, - }; - self.sys_handle.emit_subscribe_timing(stage, capture_timestamp_us, frame_id); + self.sys_handle.emit_subscribe_timing( + sys_subscribe_timing_stage(stage), + capture_timestamp_us, + frame_id, + ); + } + + pub(crate) fn emit_subscribe_timing_at( + &self, + stage: SubscribeTimingStage, + capture_timestamp_us: u64, + frame_id: u32, + timestamp_us: u64, + ) { + self.sys_handle.emit_subscribe_timing_at( + sys_subscribe_timing_stage(stage), + capture_timestamp_us, + frame_id, + timestamp_us, + ); + } +} + +fn sys_subscribe_timing_stage(stage: SubscribeTimingStage) -> sys_pt::VideoSubscribeTimingStage { + match stage { + SubscribeTimingStage::WebrtcReceive => sys_pt::VideoSubscribeTimingStage::WebrtcReceive, + SubscribeTimingStage::DecoderUpload => sys_pt::VideoSubscribeTimingStage::DecoderUpload, + SubscribeTimingStage::DecoderOutput => sys_pt::VideoSubscribeTimingStage::DecoderOutput, } } diff --git a/libwebrtc/src/native/video_stream.rs b/libwebrtc/src/native/video_stream.rs index 61b6ab8bf..ff7c43a03 100644 --- a/libwebrtc/src/native/video_stream.rs +++ b/libwebrtc/src/native/video_stream.rs @@ -108,12 +108,31 @@ impl VideoTrackObserver { fn frame_metadata( &self, rtp_timestamp: u32, + decode_start_timestamp_us: u64, + decode_finish_timestamp_us: u64, handler: Option<&PacketTrailerHandler>, ) -> Option { handler .and_then(|handler| { handler.lookup_frame_metadata(rtp_timestamp).map(|(ts, fid, user_data)| { - handler.emit_subscribe_timing(SubscribeTimingStage::DecoderOutput, ts, fid); + if decode_start_timestamp_us != 0 { + handler.emit_subscribe_timing_at( + SubscribeTimingStage::DecoderUpload, + ts, + fid, + decode_start_timestamp_us, + ); + } + if decode_finish_timestamp_us != 0 { + handler.emit_subscribe_timing_at( + SubscribeTimingStage::DecoderOutput, + ts, + fid, + decode_finish_timestamp_us, + ); + } else { + handler.emit_subscribe_timing(SubscribeTimingStage::DecoderOutput, ts, fid); + } (ts, fid, user_data) }) }) @@ -128,8 +147,12 @@ impl VideoTrackObserver { impl sys_vt::VideoSink for VideoTrackObserver { fn on_frame(&self, frame: UniquePtr) { let packet_trailer_handler = self.packet_trailer_handler.lock().clone(); - let frame_metadata = - self.frame_metadata(frame.timestamp(), packet_trailer_handler.as_ref()); + let frame_metadata = self.frame_metadata( + frame.timestamp(), + frame.decode_start_timestamp_us(), + frame.decode_finish_timestamp_us(), + packet_trailer_handler.as_ref(), + ); self.frame_queue.push(VideoFrame { rotation: frame.rotation().into(), diff --git a/webrtc-sys/include/livekit/packet_trailer.h b/webrtc-sys/include/livekit/packet_trailer.h index 4aa1261e4..b5a00bc57 100644 --- a/webrtc-sys/include/livekit/packet_trailer.h +++ b/webrtc-sys/include/livekit/packet_trailer.h @@ -174,6 +174,12 @@ class PacketTrailerTransformer : public webrtc::FrameTransformerInterface { uint64_t user_timestamp, uint32_t frame_id) const; + /// Emit a receiver-side subscribe timing event at a supplied Unix timestamp. + void emit_subscribe_timing_at(VideoSubscribeTimingStage stage, + uint64_t user_timestamp, + uint32_t frame_id, + uint64_t timestamp_us) const; + private: void TransformSend( std::unique_ptr frame); @@ -302,6 +308,12 @@ class PacketTrailerHandler { uint64_t user_timestamp, uint32_t frame_id) const; + /// Emit a receiver-side subscribe timing event at a supplied Unix timestamp. + void emit_subscribe_timing_at(VideoSubscribeTimingStage stage, + uint64_t user_timestamp, + uint32_t frame_id, + uint64_t timestamp_us) const; + /// Access the underlying transformer for chaining. webrtc::scoped_refptr transformer() const; diff --git a/webrtc-sys/include/livekit/video_frame.h b/webrtc-sys/include/livekit/video_frame.h index 6d68f836c..92e00dab2 100644 --- a/webrtc-sys/include/livekit/video_frame.h +++ b/webrtc-sys/include/livekit/video_frame.h @@ -39,6 +39,8 @@ class VideoFrame { int64_t timestamp_us() const; int64_t ntp_time_ms() const; uint32_t timestamp() const; + uint64_t decode_start_timestamp_us() const; + uint64_t decode_finish_timestamp_us() const; VideoRotation rotation() const; std::unique_ptr video_frame_buffer() const; diff --git a/webrtc-sys/src/packet_trailer.cpp b/webrtc-sys/src/packet_trailer.cpp index 6dbeabaeb..4714ae19a 100644 --- a/webrtc-sys/src/packet_trailer.cpp +++ b/webrtc-sys/src/packet_trailer.cpp @@ -26,6 +26,7 @@ #include "livekit/rtp_receiver.h" #include "livekit/rtp_sender.h" #include "rtc_base/logging.h" +#include "rtc_base/time_utils.h" #include "webrtc-sys/src/packet_trailer.rs.h" namespace livekit_ffi { @@ -38,6 +39,30 @@ uint64_t CurrentUnixTimeMicros() { std::chrono::duration_cast(now).count()); } +constexpr int64_t kMaxFirstPacketAgeUs = 30'000'000; + +uint64_t FirstPacketReceiveUnixTimeMicros( + const webrtc::TransformableFrameInterface& frame) { + const uint64_t now_unix_us = CurrentUnixTimeMicros(); + const std::optional receive_time = frame.ReceiveTime(); + if (!receive_time.has_value() || !receive_time->IsFinite()) { + return now_unix_us; + } + + const int64_t now_monotonic_us = webrtc::TimeMicros(); + const int64_t receive_monotonic_us = receive_time->us(); + if (receive_monotonic_us > now_monotonic_us) { + return now_unix_us; + } + + const int64_t age_us = now_monotonic_us - receive_monotonic_us; + if (age_us > kMaxFirstPacketAgeUs || + static_cast(age_us) > now_unix_us) { + return now_unix_us; + } + return now_unix_us - static_cast(age_us); +} + std::vector BuildTrailerPayload(uint64_t user_timestamp, uint32_t frame_id, const std::vector& user_data) { @@ -354,8 +379,9 @@ void PacketTrailerTransformer::TransformReceive( // Update frame with stripped data frame->SetData(webrtc::ArrayView(stripped_data)); } - uint64_t receive_timestamp_us = - subscribe_timing_enabled() ? CurrentUnixTimeMicros() : 0; + uint64_t receive_timestamp_us = subscribe_timing_enabled() + ? FirstPacketReceiveUnixTimeMicros(*frame) + : 0; emit_subscribe_timing(VideoSubscribeTimingStage::WebrtcReceive, timing_meta.user_timestamp, timing_meta.frame_id, receive_timestamp_us); @@ -373,8 +399,6 @@ void PacketTrailerTransformer::TransformReceive( } if (cb) { - emit_subscribe_timing(VideoSubscribeTimingStage::DecoderUpload, - timing_meta.user_timestamp, timing_meta.frame_id); cb->OnTransformedFrame(std::move(frame)); } else { RTC_LOG(LS_WARNING) @@ -597,6 +621,14 @@ void PacketTrailerTransformer::emit_subscribe_timing( CurrentUnixTimeMicros()); } +void PacketTrailerTransformer::emit_subscribe_timing_at( + VideoSubscribeTimingStage stage, + uint64_t user_timestamp, + uint32_t frame_id, + uint64_t timestamp_us) const { + emit_subscribe_timing(stage, user_timestamp, frame_id, timestamp_us); +} + void PacketTrailerTransformer::emit_subscribe_timing( VideoSubscribeTimingStage stage, uint64_t user_timestamp, @@ -719,6 +751,15 @@ void PacketTrailerHandler::emit_subscribe_timing( transformer_->emit_subscribe_timing(stage, user_timestamp, frame_id); } +void PacketTrailerHandler::emit_subscribe_timing_at( + VideoSubscribeTimingStage stage, + uint64_t user_timestamp, + uint32_t frame_id, + uint64_t timestamp_us) const { + transformer_->emit_subscribe_timing_at(stage, user_timestamp, frame_id, + timestamp_us); +} + webrtc::scoped_refptr PacketTrailerHandler::transformer() const { return transformer_; } diff --git a/webrtc-sys/src/packet_trailer.rs b/webrtc-sys/src/packet_trailer.rs index ceabb5fa8..13c7c3648 100644 --- a/webrtc-sys/src/packet_trailer.rs +++ b/webrtc-sys/src/packet_trailer.rs @@ -121,6 +121,15 @@ pub mod ffi { frame_id: u32, ); + /// Emit a receiver-side subscribe timing event at a supplied Unix timestamp. + fn emit_subscribe_timing_at( + self: &PacketTrailerHandler, + stage: VideoSubscribeTimingStage, + user_timestamp: u64, + frame_id: u32, + timestamp_us: u64, + ); + /// Create a new packet trailer handler for a sender. fn new_packet_trailer_sender( peer_factory: SharedPtr, diff --git a/webrtc-sys/src/video_frame.cpp b/webrtc-sys/src/video_frame.cpp index dc6d25ed0..2c7ef481f 100644 --- a/webrtc-sys/src/video_frame.cpp +++ b/webrtc-sys/src/video_frame.cpp @@ -16,11 +16,45 @@ #include "livekit/video_frame.h" +#include #include +#include #include "api/video/video_frame.h" +#include "rtc_base/time_utils.h" namespace livekit_ffi { +namespace { + +constexpr int64_t kMaxDecodeTimestampAgeUs = 30'000'000; + +uint64_t MonotonicToUnixTimeMicros(webrtc::Timestamp timestamp) { + if (!timestamp.IsFinite()) { + return 0; + } + + const int64_t now_monotonic_us = webrtc::TimeMicros(); + const int64_t timestamp_monotonic_us = timestamp.us(); + if (timestamp_monotonic_us > now_monotonic_us) { + return 0; + } + + const int64_t age_us = now_monotonic_us - timestamp_monotonic_us; + if (age_us > kMaxDecodeTimestampAgeUs) { + return 0; + } + + const auto now = std::chrono::system_clock::now().time_since_epoch(); + const uint64_t now_unix_us = static_cast( + std::chrono::duration_cast(now).count()); + if (static_cast(age_us) > now_unix_us) { + return 0; + } + return now_unix_us - static_cast(age_us); +} + +} // namespace + VideoFrame::VideoFrame(const webrtc::VideoFrame& frame) : frame_(std::move(frame)) {} @@ -45,6 +79,20 @@ int64_t VideoFrame::ntp_time_ms() const { uint32_t VideoFrame::timestamp() const { return frame_.rtp_timestamp(); } +uint64_t VideoFrame::decode_start_timestamp_us() const { + const std::optional processing_time = + frame_.processing_time(); + return processing_time.has_value() + ? MonotonicToUnixTimeMicros(processing_time->start) + : 0; +} +uint64_t VideoFrame::decode_finish_timestamp_us() const { + const std::optional processing_time = + frame_.processing_time(); + return processing_time.has_value() + ? MonotonicToUnixTimeMicros(processing_time->finish) + : 0; +} VideoRotation VideoFrame::rotation() const { return static_cast(frame_.rotation()); diff --git a/webrtc-sys/src/video_frame.rs b/webrtc-sys/src/video_frame.rs index afd0382de..829df5ce3 100644 --- a/webrtc-sys/src/video_frame.rs +++ b/webrtc-sys/src/video_frame.rs @@ -43,6 +43,10 @@ pub mod ffi { fn timestamp_us(self: &VideoFrame) -> i64; fn ntp_time_ms(self: &VideoFrame) -> i64; fn timestamp(self: &VideoFrame) -> u32; + /// Actual decoder processing start, as Unix microseconds, or 0 when unavailable. + fn decode_start_timestamp_us(self: &VideoFrame) -> u64; + /// Actual decoder processing finish, as Unix microseconds, or 0 when unavailable. + fn decode_finish_timestamp_us(self: &VideoFrame) -> u64; fn rotation(self: &VideoFrame) -> VideoRotation; unsafe fn video_frame_buffer(self: &VideoFrame) -> UniquePtr; From 53b421bd74745ea70c4364d4f7c1d2e79a28a3b6 Mon Sep 17 00:00:00 2001 From: David Chen Date: Tue, 11 Aug 2026 13:51:57 -0700 Subject: [PATCH 9/9] fix color --- examples/local_video/scripts/generate_frame_report.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/local_video/scripts/generate_frame_report.py b/examples/local_video/scripts/generate_frame_report.py index d66e23048..91dbb82cd 100755 --- a/examples/local_video/scripts/generate_frame_report.py +++ b/examples/local_video/scripts/generate_frame_report.py @@ -40,7 +40,7 @@ HexColor("#4C78A8"), HexColor("#3A86FF"), HexColor("#2CB1BC"), - HexColor("#44FF33"), + HexColor("#04b034"), HexColor("#4ECDC4"), HexColor("#6C63FF"), )