From db1e82d11afff521a6376eb47420cd0cb6ba5564 Mon Sep 17 00:00:00 2001 From: foxxcn Date: Sun, 11 Oct 2026 23:40:50 +0800 Subject: [PATCH 1/5] video: switch to a new session's track at once and request an IDR After a takeover or reconnect, the frame writer kept writing to the previous session's cached track until its one-second refresh, and nothing asked the encoder for a keyframe, so a new viewer waited up to a second for frames and then up to another GOP (1s) for a decodable one; this matches the 0.3-1s first-frame times measured in testing. Read the video and audio sinks per frame instead of caching them (an uncontended RwLock read at 60/100 Hz), and force a keyframe with headers when a new video track is attached. Co-Authored-By: Claude Opus 5.5 --- app/src/video.rs | 96 ++++++++++++++---------------------------------- 1 file changed, 27 insertions(+), 69 deletions(-) diff --git a/app/src/video.rs b/app/src/video.rs index e799299..23f5598 100644 --- a/app/src/video.rs +++ b/app/src/video.rs @@ -69,42 +69,23 @@ type AudioFrame = bytes::Bytes; static VIDEO_FRAME_TX: tokio::sync::OnceCell> = tokio::sync::OnceCell::const_new(); +/// Writes encoded frames to the current session's track. The sink is read per +/// frame rather than cached: a cached track kept feeding a taken-over session +/// for up to a second after the new one attached, delaying its first frame. fn video_writer_loop(mut rx: mpsc::Receiver) { tokio::spawn(async move { let mut last_pts: Option = None; - let mut current_track: Option> = None; - let mut track_refresh = interval(Duration::from_secs(1)); - track_refresh.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - - loop { - tokio::select! { - _ = track_refresh.tick() => { - current_track = VIDEO_SINK.read().await.clone(); - } - maybe_frame = rx.recv() => { - let Some((data, pts)) = maybe_frame else { break; }; - - let dur = match last_pts { - Some(prev) if pts > prev => Duration::from_micros(pts - prev), - _ => Duration::from_millis(16), - }; - last_pts = Some(pts); - - if current_track.is_none() { - current_track = VIDEO_SINK.read().await.clone(); - } - if let Some(track) = ¤t_track { - let sample = Sample { - data, - duration: dur, - ..Default::default() - }; - if let Err(e) = track.write_sample(&sample).await { - warn!("Failed to write video sample: {}", e); - current_track = None; - } - } - } + while let Some((data, pts)) = rx.recv().await { + let dur = match last_pts { + Some(prev) if pts > prev => Duration::from_micros(pts - prev), + _ => Duration::from_millis(16), + }; + last_pts = Some(pts); + + let Some(track) = VIDEO_SINK.read().await.clone() else { continue }; + let sample = Sample { data, duration: dur, ..Default::default() }; + if let Err(e) = track.write_sample(&sample).await { + warn!("Failed to write video sample: {}", e); } } }); @@ -147,8 +128,12 @@ pub async fn ensure_video_pipeline_started() -> anyhow::Result<()> { .map(|_| ()) } +/// Switches video output to a new session's track and asks the encoder for an +/// IDR, so the new viewer does not wait up to a GOP for its first decodable +/// frame. pub async fn attach_webrtc_sink(track: Arc) { *VIDEO_SINK.write().await = Some(track); + force_video_keyframe().await; } pub async fn attach_audio_sink(track: Arc) { @@ -163,42 +148,15 @@ pub async fn detach_webrtc_sink() { async fn audio_frame_writer(mut rx: mpsc::Receiver) { const AUDIO_FRAME_DURATION: Duration = Duration::from_millis(10); - let mut current_track: Option> = None; - let mut track_check_interval = interval(Duration::from_secs(1)); - track_check_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - - loop { - tokio::select! { - _ = track_check_interval.tick() => { - current_track = AUDIO_SINK.read().await.clone(); - } - frame = rx.recv() => { - match frame { - Some(data) => { - if !is_audio_enabled() { - continue; - } - if let Some(track) = ¤t_track { - let sample = Sample { - data, - duration: AUDIO_FRAME_DURATION, - ..Default::default() - }; - if let Err(e) = track.write_sample(&sample).await { - warn!("error writing audio sample: {}", e); - current_track = AUDIO_SINK.read().await.clone(); - } - } else { - current_track = AUDIO_SINK.read().await.clone(); - } - } - None => { - tokio::time::sleep(Duration::from_millis(500)).await; - current_track = AUDIO_SINK.read().await.clone(); - continue; - } - } - } + // Read the sink per frame for the same reason as the video writer. + while let Some(data) = rx.recv().await { + if !is_audio_enabled() { + continue; + } + let Some(track) = AUDIO_SINK.read().await.clone() else { continue }; + let sample = Sample { data, duration: AUDIO_FRAME_DURATION, ..Default::default() }; + if let Err(e) = track.write_sample(&sample).await { + warn!("error writing audio sample: {}", e); } } } From 7c630e3bea527d2932ee277dff655eb4dc2b19da Mon Sep 17 00:00:00 2001 From: foxxcn Date: Mon, 12 Oct 2026 00:29:39 +0800 Subject: [PATCH 2/5] video: request the new viewer's keyframe after DTLS, not at ICE connect Testing showed no first-frame gain: the first frame still arrived 0.1-1.1s after connecting, uniformly, i.e. at the next periodic IDR. The device log explains it: the keyframe was forced at ICE "connected", but the peer connection only reaches "connected" (DTLS done) afterwards, so the IDR was encoded and dropped before media could flow. Force the keyframe from the peer connection state handler on "connected", log "forced keyframe for new session ", and ask once more 300 ms later in case the RTP sender started slightly after. The handler captures only the session id, so it adds no reference cycle. Co-Authored-By: Claude Opus 5.5 --- app/src/video.rs | 6 ++---- app/src/webrtc.rs | 21 +++++++++++++++++++++ 2 files changed, 23 insertions(+), 4 deletions(-) diff --git a/app/src/video.rs b/app/src/video.rs index 23f5598..bb9bc37 100644 --- a/app/src/video.rs +++ b/app/src/video.rs @@ -128,12 +128,10 @@ pub async fn ensure_video_pipeline_started() -> anyhow::Result<()> { .map(|_| ()) } -/// Switches video output to a new session's track and asks the encoder for an -/// IDR, so the new viewer does not wait up to a GOP for its first decodable -/// frame. +/// Switches video output to a new session's track. The keyframe for the new +/// viewer is requested once its connection is fully up (see webrtc.rs). pub async fn attach_webrtc_sink(track: Arc) { *VIDEO_SINK.write().await = Some(track); - force_video_keyframe().await; } pub async fn attach_audio_sink(track: Arc) { diff --git a/app/src/webrtc.rs b/app/src/webrtc.rs index 93a3662..a9f6d65 100644 --- a/app/src/webrtc.rs +++ b/app/src/webrtc.rs @@ -13,6 +13,7 @@ use webrtc::ice_transport::ice_candidate::RTCIceCandidate; use webrtc::ice_transport::ice_connection_state::RTCIceConnectionState; use webrtc::interceptor::registry::Registry; use webrtc::peer_connection::configuration::RTCConfiguration; +use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState; use webrtc::rtp_transceiver::rtp_codec::{ RTCRtpCodecCapability, RTCRtpCodecParameters, RTPCodecType, }; @@ -309,6 +310,26 @@ impl WebRTCApi { }, )); + // A keyframe requested at ICE "connected" is encoded before DTLS has + // finished, so it never reaches the browser and the viewer waits for + // the next periodic IDR. Ask again once the whole connection is up, and + // once more shortly after in case the RTP sender started a little later. + let session_id_clone = session.id.clone(); + peer_connection.on_peer_connection_state_change(Box::new( + move |state: RTCPeerConnectionState| { + let session_id = session_id_clone.clone(); + Box::pin(async move { + if state != RTCPeerConnectionState::Connected { + return; + } + video::force_video_keyframe().await; + info!("forced keyframe for new session {}", session_id); + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + video::force_video_keyframe().await; + }) + }, + )); + let session_id_clone = session.id.clone(); let app_state_clone = config.app_state.clone(); peer_connection.on_ice_candidate(Box::new(move |candidate: Option| { From c9c96e9bd37d8bb71f04c981efc6b04d9463486a Mon Sep 17 00:00:00 2001 From: foxxcn Date: Mon, 12 Oct 2026 00:46:42 +0800 Subject: [PATCH 3/5] video: timestamp logs and measure forced keyframe latency The first-frame fix did not change measured first-frame times, and the log had no timestamps to tell why. Add timestamps to the log, log every keyframe request (and the follow-up one), and probe the encoder's output so the first keyframe after a request is logged with how many ms it took. A forced IDR shows up within a frame or two; a value near the GOP length means the request was ignored and a periodic keyframe answered. Co-Authored-By: Claude Opus 5.5 --- app/src/main.rs | 1 - app/src/pipeline.rs | 30 +++++++++++++++++++++++++++++- app/src/webrtc.rs | 1 + 3 files changed, 30 insertions(+), 2 deletions(-) diff --git a/app/src/main.rs b/app/src/main.rs index e4f2b0d..b7e7030 100644 --- a/app/src/main.rs +++ b/app/src/main.rs @@ -40,7 +40,6 @@ fn init_tracing(cli: &Cli) { let fmt_layer = tracing_subscriber::fmt::layer() .with_ansi(!cli.logging.log_no_color) - .without_time() .with_level(true) .with_target(true) .with_thread_ids(true) diff --git a/app/src/pipeline.rs b/app/src/pipeline.rs index bc26904..b8f1c31 100644 --- a/app/src/pipeline.rs +++ b/app/src/pipeline.rs @@ -90,6 +90,9 @@ pub struct VideoPipeline { running: Arc, raw_frame_waiters: Arc, raw_frame_requested: Arc, + /// When the oldest unanswered keyframe request was made; the encoder + /// output probe logs how long it took to produce the keyframe. + keyframe_requested_at: Arc>>, } impl VideoPipeline { @@ -167,6 +170,27 @@ impl VideoPipeline { encoder.set_property("gop", gop); } + let keyframe_requested_at: Arc>> = + Arc::default(); + let probe_requested_at = keyframe_requested_at.clone(); + encoder.static_pad("src").context("Encoder has no src pad")?.add_probe( + gst::PadProbeType::BUFFER, + move |_pad, info| { + if let Some(buffer) = info.buffer() + && !buffer.flags().contains(gst::BufferFlags::DELTA_UNIT) + { + match probe_requested_at.lock().take() { + Some(at) => info!( + elapsed_ms = at.elapsed().as_millis() as u64, + "encoder emitted keyframe after request" + ), + None => debug!("encoder emitted periodic keyframe"), + } + } + gst::PadProbeReturn::Ok + }, + ); + let parser_name = config.encoder.parser_element_name(); let parser = gst::ElementFactory::make(parser_name) .property("config-interval", -1i32) @@ -271,6 +295,7 @@ impl VideoPipeline { running, raw_frame_waiters, raw_frame_requested, + keyframe_requested_at, }) } @@ -781,8 +806,11 @@ impl PipelineManager { } pub fn force_keyframe(&self) { + self.video.keyframe_requested_at.lock().get_or_insert_with(std::time::Instant::now); let event = gst_video::UpstreamForceKeyUnitEvent::builder().all_headers(true).build(); - if !self.video.encoder.send_event(event) { + let handled = self.video.encoder.send_event(event); + info!(handled, "keyframe requested"); + if !handled { warn!("force-keyframe event was not handled by encoder"); } } diff --git a/app/src/webrtc.rs b/app/src/webrtc.rs index a9f6d65..9b76841 100644 --- a/app/src/webrtc.rs +++ b/app/src/webrtc.rs @@ -326,6 +326,7 @@ impl WebRTCApi { info!("forced keyframe for new session {}", session_id); tokio::time::sleep(std::time::Duration::from_millis(300)).await; video::force_video_keyframe().await; + info!("forced follow-up keyframe for session {}", session_id); }) }, )); From 6e0c19e71d5f384613739993db1c9a5013af4bd8 Mon Sep 17 00:00:00 2001 From: foxxcn Date: Mon, 12 Oct 2026 01:07:19 +0800 Subject: [PATCH 4/5] video: detect encoder IDRs from the bitstream The keyframe probe logged nothing in testing: every forced request was accepted, but no encoder output buffer lacked DELTA_UNIT, so the flag cannot tell IDRs apart. Check the first slice NAL type instead (H.264 only; the scan stops at the first slice) and log the delta flag along with the latency. Co-Authored-By: Claude Opus 5.5 --- app/src/pipeline.rs | 30 ++++++++++++++++++++++++++---- 1 file changed, 26 insertions(+), 4 deletions(-) diff --git a/app/src/pipeline.rs b/app/src/pipeline.rs index b8f1c31..868b7e0 100644 --- a/app/src/pipeline.rs +++ b/app/src/pipeline.rs @@ -173,18 +173,25 @@ impl VideoPipeline { let keyframe_requested_at: Arc>> = Arc::default(); let probe_requested_at = keyframe_requested_at.clone(); + let is_h264 = matches!(config.encoder, crate::cli::VideoEncoder::H264); encoder.static_pad("src").context("Encoder has no src pad")?.add_probe( gst::PadProbeType::BUFFER, move |_pad, info| { - if let Some(buffer) = info.buffer() - && !buffer.flags().contains(gst::BufferFlags::DELTA_UNIT) + // Decide by the bitstream, not the buffer flags: the MPP encoder + // does not reliably clear DELTA_UNIT on IDR frames. + if is_h264 + && let Some(buffer) = info.buffer() + && let Ok(map) = buffer.map_readable() + && contains_h264_idr(map.as_slice()) { + let delta = buffer.flags().contains(gst::BufferFlags::DELTA_UNIT); match probe_requested_at.lock().take() { Some(at) => info!( elapsed_ms = at.elapsed().as_millis() as u64, - "encoder emitted keyframe after request" + delta_flag = delta, + "encoder emitted IDR after request" ), - None => debug!("encoder emitted periodic keyframe"), + None => debug!(delta_flag = delta, "encoder emitted periodic IDR"), } } gst::PadProbeReturn::Ok @@ -816,6 +823,21 @@ impl PipelineManager { } } +/// Whether an Annex B H.264 access unit is an IDR: the first slice NAL found +/// is type 5. Stops at the first slice, so a P frame costs only its headers. +fn contains_h264_idr(data: &[u8]) -> bool { + for w in data.windows(4) { + if w[0] == 0 && w[1] == 0 && w[2] == 1 { + match w[3] & 0x1f { + 5 => return true, + 1 => return false, + _ => {} + } + } + } + false +} + pub fn encode_jpeg(sample: &gst::Sample) -> Result { let caps = sample.caps_owned().context("Raw frame has no caps")?; let encoder = gst::ElementFactory::make("mppjpegenc") From c50f9aed181f8dcac36c8320eb942d0e37817af0 Mon Sep 17 00:00:00 2001 From: foxxcn Date: Mon, 12 Oct 2026 09:48:58 +0800 Subject: [PATCH 5/5] video: make forced keyframes actually reach the MPP encoder The device log showed every keyframe request accepted, yet IDRs only ever came on the 1-second GOP boundary. Cause, in GstVideoEncoder (1.22): an "as soon as possible" force-key-unit request stays pending until the encoder outputs a frame marked as a sync point, and while one is pending, later ASAP requests are dropped. mpph264enc never marks sync points (its IDRs carry DELTA_UNIT), so after the first request per pipeline, none had any effect. Send the request with the current running time instead; timed requests are only compared with the last forced one, so each goes through. PLI and other requests are coalesced while one is outstanding (up to a GOP) to keep the encoder's never-cleared request queue small; the request for a newly connected viewer is never coalesced. The 300 ms follow-up request is no longer needed. Co-Authored-By: Claude Opus 5.5 --- app/src/pipeline.rs | 33 +++++++++++++++++++++++++++++++-- app/src/video.rs | 7 +++++++ app/src/webrtc.rs | 8 ++------ 3 files changed, 40 insertions(+), 8 deletions(-) diff --git a/app/src/pipeline.rs b/app/src/pipeline.rs index 868b7e0..52de25a 100644 --- a/app/src/pipeline.rs +++ b/app/src/pipeline.rs @@ -812,10 +812,39 @@ impl PipelineManager { self.video.request_raw_frame() } + /// Requests an IDR from the encoder. + /// + /// The request carries the current running time. An "as soon as possible" + /// request (no running time) only works once per pipeline: GstVideoEncoder + /// keeps it pending until the encoder outputs a frame flagged as a sync + /// point, which the MPP encoder never does, and while one is pending it + /// ignores every later ASAP request. Timed requests are only compared with + /// the last forced request, so each one goes through. + /// + /// Requests are skipped while an earlier one is still waiting for its IDR + /// (up to a GOP), so a burst of browser PLIs costs one keyframe, and so + /// the encoder's never-cleared request queue grows by a few entries per + /// session rather than per PLI. pub fn force_keyframe(&self) { + { + let requested_at = self.video.keyframe_requested_at.lock(); + if requested_at.is_some_and(|at| at.elapsed() < std::time::Duration::from_secs(1)) { + debug!("keyframe already requested, waiting for it"); + return; + } + } + self.force_keyframe_now(); + } + + /// Like [`Self::force_keyframe`] but never coalesced: a new viewer needs + /// an IDR encoded after its connection is up, even if one is in flight. + pub fn force_keyframe_now(&self) { self.video.keyframe_requested_at.lock().get_or_insert_with(std::time::Instant::now); - let event = gst_video::UpstreamForceKeyUnitEvent::builder().all_headers(true).build(); - let handled = self.video.encoder.send_event(event); + let mut builder = gst_video::UpstreamForceKeyUnitEvent::builder().all_headers(true); + if let Some(running_time) = self.video.encoder.current_running_time() { + builder = builder.running_time(running_time); + } + let handled = self.video.encoder.send_event(builder.build()); info!(handled, "keyframe requested"); if !handled { warn!("force-keyframe event was not handled by encoder"); diff --git a/app/src/video.rs b/app/src/video.rs index bb9bc37..a02de0d 100644 --- a/app/src/video.rs +++ b/app/src/video.rs @@ -317,6 +317,13 @@ pub async fn force_video_keyframe() { } } +/// Uncoalesced keyframe request for a viewer whose connection just came up. +pub async fn force_new_viewer_keyframe() { + if let Some(manager) = PIPELINE_MANAGER.lock().await.as_ref() { + manager.force_keyframe_now(); + } +} + pub async fn update_video_quality(quality: f32) -> anyhow::Result<()> { if !(0.0..=1.0).contains(&quality) { anyhow::bail!("Quality must be between 0.0 and 1.0"); diff --git a/app/src/webrtc.rs b/app/src/webrtc.rs index 9b76841..e2da470 100644 --- a/app/src/webrtc.rs +++ b/app/src/webrtc.rs @@ -312,8 +312,7 @@ impl WebRTCApi { // A keyframe requested at ICE "connected" is encoded before DTLS has // finished, so it never reaches the browser and the viewer waits for - // the next periodic IDR. Ask again once the whole connection is up, and - // once more shortly after in case the RTP sender started a little later. + // the next periodic IDR. Ask again once the whole connection is up. let session_id_clone = session.id.clone(); peer_connection.on_peer_connection_state_change(Box::new( move |state: RTCPeerConnectionState| { @@ -322,11 +321,8 @@ impl WebRTCApi { if state != RTCPeerConnectionState::Connected { return; } - video::force_video_keyframe().await; + video::force_new_viewer_keyframe().await; info!("forced keyframe for new session {}", session_id); - tokio::time::sleep(std::time::Duration::from_millis(300)).await; - video::force_video_keyframe().await; - info!("forced follow-up keyframe for session {}", session_id); }) }, ));