From 41651c527a3753622df11b3bdeb61d7cdd4f764c Mon Sep 17 00:00:00 2001 From: foxxcn Date: Sun, 11 Oct 2026 23:41:51 +0800 Subject: [PATCH] video: pause encoding while nobody is watching The pipeline encoded 4K60 continuously even with no session attached; with no viewer rustkvm_app used ~56% of a core, ~46% of it in the encoder's output thread, all for frames that were thrown away. Put a valve in front of the encoder. It starts closed, opens when a session's video track is attached (with a forced keyframe so the viewer can decode immediately) and closes when the last one detaches; a restarted pipeline picks up the current state. The valve forwards sticky events, so the encoder stays negotiated and video state, the dynamic GOP and the screenshot probe (moved to the valve's input) keep working while paused. rustkvm_video_frames_total now only counts frames encoded for a viewer. Co-Authored-By: Claude Opus 5.5 --- app/src/pipeline.rs | 37 ++++++++++++++++++++++++++++++++++++- app/src/video.rs | 10 ++++++++++ app/src/webrtc.rs | 1 + docs/api.md | 2 +- 4 files changed, 48 insertions(+), 2 deletions(-) diff --git a/app/src/pipeline.rs b/app/src/pipeline.rs index bc26904..b8d55d2 100644 --- a/app/src/pipeline.rs +++ b/app/src/pipeline.rs @@ -87,6 +87,7 @@ pub struct VideoPipeline { pipeline: gst::Pipeline, appsink: gst_app::AppSink, encoder: gst::Element, + valve: gst::Element, running: Arc, raw_frame_waiters: Arc, raw_frame_requested: Arc, @@ -145,6 +146,17 @@ impl VideoPipeline { .build() .context("Failed to create queue")?; + // Gates the encoder: while nobody is watching it drops raw frames so the + // hardware encoder and its output thread sit idle. Sticky events (caps) + // still pass, so the encoder stays negotiated and video state, dynamic + // GOP and the screenshot probe keep working. Starts closed. + let valve = gst::ElementFactory::make("valve") + .name("encode_valve") + .property_from_str("drop-mode", "forward-sticky-events") + .property("drop", true) + .build() + .context("Failed to create valve")?; + let encoder_name = config.encoder.gst_element_name(); info!("Creating {} encoder for {}", encoder_name, config.encoder.codec_name()); @@ -190,6 +202,7 @@ impl VideoPipeline { &videorate, &rate_caps, &queue, + &valve, &encoder, &parser, appsink.upcast_ref(), @@ -203,6 +216,7 @@ impl VideoPipeline { &videorate, &rate_caps, &queue, + &valve, &encoder, &parser, appsink.upcast_ref(), @@ -215,7 +229,8 @@ impl VideoPipeline { // steady-state cost per frame is a single relaxed atomic load. let probe_waiters = raw_frame_waiters.clone(); let probe_requested = raw_frame_requested.clone(); - encoder.static_pad("sink").context("Encoder has no sink pad")?.add_probe( + // On the valve's input, so snapshots work while encoding is paused. + valve.static_pad("sink").context("Valve has no sink pad")?.add_probe( gst::PadProbeType::BUFFER, move |pad, info| { if !probe_requested.swap(false, Ordering::AcqRel) { @@ -268,6 +283,7 @@ impl VideoPipeline { pipeline, appsink, encoder: encoder_for_state, + valve, running, raw_frame_waiters, raw_frame_requested, @@ -302,6 +318,17 @@ impl VideoPipeline { self.running.load(Ordering::SeqCst) } + /// Opens or closes the encoder valve. Returns whether this changed it. + pub fn set_encoding_active(&self, active: bool) -> bool { + let drop = !active; + if self.valve.property::("drop") == drop { + return false; + } + self.valve.set_property("drop", drop); + info!(active, "video encoding {}", if active { "resumed" } else { "paused" }); + true + } + pub fn stop(&self) -> Result<()> { self.running.store(false, Ordering::SeqCst); @@ -786,6 +813,14 @@ impl PipelineManager { warn!("force-keyframe event was not handled by encoder"); } } + + /// Resumes encoding for a viewer (with a keyframe so it can decode at once) + /// or pauses it when nobody is watching. + pub fn set_encoding_active(&self, active: bool) { + if self.video.set_encoding_active(active) && active { + self.force_keyframe(); + } + } } pub fn encode_jpeg(sample: &gst::Sample) -> Result { diff --git a/app/src/video.rs b/app/src/video.rs index e799299..735905d 100644 --- a/app/src/video.rs +++ b/app/src/video.rs @@ -158,6 +158,14 @@ pub async fn attach_audio_sink(track: Arc) { pub async fn detach_webrtc_sink() { *VIDEO_SINK.write().await = None; *AUDIO_SINK.write().await = None; + set_encoding_active(false).await; +} + +/// Encoding only runs while a session has a video track attached. +pub async fn set_encoding_active(active: bool) { + if let Some(manager) = PIPELINE_MANAGER.lock().await.as_ref() { + manager.set_encoding_active(active); + } } async fn audio_frame_writer(mut rx: mpsc::Receiver) { @@ -252,6 +260,8 @@ async fn start_pipeline( } manager.start()?; + // A restarted pipeline comes up paused; resume it if a viewer is attached. + manager.set_encoding_active(VIDEO_SINK.read().await.is_some()); *PIPELINE_MANAGER.lock().await = Some(manager); Ok(()) diff --git a/app/src/webrtc.rs b/app/src/webrtc.rs index 93a3662..65aa68e 100644 --- a/app/src/webrtc.rs +++ b/app/src/webrtc.rs @@ -258,6 +258,7 @@ impl WebRTCApi { increment_session_counter().await; if let Some(track) = video_track.clone() { video::attach_webrtc_sink(track).await; + video::set_encoding_active(true).await; info!("Video track attached to WebRTC session {}", session_id); } if let Some(track) = audio_track.clone() { diff --git a/docs/api.md b/docs/api.md index 21f5d81..ca26316 100644 --- a/docs/api.md +++ b/docs/api.md @@ -213,7 +213,7 @@ enabled the device announces itself under `homeassistant/…/rustkvm_ | Metric | Type | | --- | --- | | `rustkvm_app_info` | gauge with build labels | -| `rustkvm_video_frames_total` | counter | +| `rustkvm_video_frames_total` | counter (encoded frames; encoding pauses while no session is watching) | | `rustkvm_audio_frames_total` | counter | | `rustkvm_video_pipeline_restarts_total` | counter | | `rustkvm_rpc_calls_total{method}` | counter |