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 |