diff --git a/src/state_portal.rs b/src/state_portal.rs index 26049ce..fad2075 100644 --- a/src/state_portal.rs +++ b/src/state_portal.rs @@ -34,6 +34,7 @@ struct EncodeThread { handle: Option>, input_tx: crossbeam_channel::Sender, timing_rx: crossbeam_channel::Receiver, + duplicate_count: std::sync::Arc, } struct WebrtcThread { @@ -244,14 +245,26 @@ impl StatePortal { bitrate_rx, encoder_resolution_rx, )?; + let duplicate_count = std::sync::Arc::new( + std::sync::atomic::AtomicU64::new(0), + ); + let duplicate_count_for_thread = duplicate_count.clone(); let handle = std::thread::Builder::new() .name("wl-webrtc-encode".into()) - .spawn(move || encode_thread_loop(encode, input_rx, timing_tx))?; + .spawn(move || { + encode_thread_loop( + encode, + input_rx, + timing_tx, + duplicate_count_for_thread, + ) + })?; self.enc_import = Some(import); self.enc_thread = Some(EncodeThread { handle: Some(handle), input_tx, timing_rx, + duplicate_count, }); let wrtc = self.webrtc.take().ok_or_else(|| { @@ -331,6 +344,11 @@ impl StatePortal { timing.output_bytes, ); } + // Read duplicate counter (delta computed in setter) + let total = enc_thread + .duplicate_count + .load(std::sync::atomic::Ordering::Relaxed); + self.stats.set_duplicate_frames_skipped(total); } if let Some(ref webrtc_thread) = self.webrtc_thread { while let Ok((gap_ms, age_ms)) = webrtc_thread.sent_gap_rx.try_recv() { @@ -634,6 +652,7 @@ fn encode_thread_loop( mut encode: SwEncEncode, input_rx: crossbeam_channel::Receiver, timing_tx: crossbeam_channel::Sender, + duplicate_count: std::sync::Arc, ) { loop { match input_rx.recv() { @@ -647,10 +666,11 @@ fn encode_thread_loop( output_bytes: t.output_bytes, }); } + Ok(EncodeOutcome::SkippedDuplicate) => { + duplicate_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } Ok(_) => { - // SkippedPaused / SkippedDisconnected / SkippedDuplicate - // Do not report timing; do not tick encoded_fps. - // take_timing() intentionally NOT called — last_timing stays default. + // SkippedPaused / SkippedDisconnected — no counter needed } Err(e) => { tracing::error!("Encode thread error: {e}"); diff --git a/src/stats.rs b/src/stats.rs index bc87515..2c6bfb6 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -39,6 +39,12 @@ pub struct PipelineStats { sent_frames: u64, pipewire_dropped: u64, over_budget_count: u64, + /// Count of frames dropped by encode thread due to Y-plane hash dedup + /// (EncodeOutcome::SkippedDuplicate). Read from atomic counter set by + /// encode thread, computed as delta since previous snapshot. + duplicate_frames_skipped: u64, + // Running total from the encode-thread atomic; NOT reset between windows. + prev_duplicate_frames_skipped: u64, // --- queue depth at last observation --- capture_queue_depth: usize, @@ -75,6 +81,8 @@ impl PipelineStats { sent_frames: 0, pipewire_dropped: 0, over_budget_count: 0, + duplicate_frames_skipped: 0, + prev_duplicate_frames_skipped: 0, capture_queue_depth: 0, encoded_queue_depth: 0, capture_gaps_ms: Vec::new(), @@ -186,6 +194,13 @@ impl PipelineStats { self.pipewire_dropped = total_dropped.saturating_sub(prev_dropped); } + /// Update duplicate frames skipped counter (absolute value from atomic). + /// Computes delta from previous value, like set_pipewire_dropped. + pub fn set_duplicate_frames_skipped(&mut self, total_skipped: u64) { + self.duplicate_frames_skipped = total_skipped.saturating_sub(self.prev_duplicate_frames_skipped); + self.prev_duplicate_frames_skipped = total_skipped; + } + /// Update queue depth observations. pub fn set_queue_depths(&mut self, capture: usize, encoded: usize) { self.capture_queue_depth = capture; @@ -216,6 +231,7 @@ impl PipelineStats { sent_frames: self.sent_frames, pipewire_dropped: self.pipewire_dropped, over_budget_count: self.over_budget_count, + duplicate_frames_skipped: self.duplicate_frames_skipped, capture_queue_depth: self.capture_queue_depth, encoded_queue_depth: self.encoded_queue_depth, capture_gap_avg_ms: avg_f64(&self.capture_gaps_ms), @@ -254,6 +270,7 @@ impl PipelineStats { self.sent_frames = 0; self.pipewire_dropped = 0; self.over_budget_count = 0; + self.duplicate_frames_skipped = 0; self.capture_queue_depth = 0; self.encoded_queue_depth = 0; self.capture_gaps_ms.clear(); @@ -288,6 +305,7 @@ pub struct StatsSnapshot { pub sent_frames: u64, pub pipewire_dropped: u64, pub over_budget_count: u64, + pub duplicate_frames_skipped: u64, // Queue depths pub capture_queue_depth: usize, pub encoded_queue_depth: usize, @@ -330,8 +348,8 @@ impl std::fmt::Display for StatsSnapshot { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!( f, - "capture_fps={:.1} encoded_fps={:.1} sent_fps={:.1} \ - pw_dropped={} over_budget={} \ + "capture_fps={:.1} encoded_fps={:.1} sent_fps={:.1} \ + pw_dropped={} over_budget={} duplicate_frames_skipped={} \ cap_q={} enc_q={} \ cap_gap_p95={:.1}ms cap_gap_max={:.1}ms \ enc_gap_p95={:.1}ms enc_gap_max={:.1}ms \ @@ -346,6 +364,7 @@ impl std::fmt::Display for StatsSnapshot { self.sent_fps, self.pipewire_dropped, self.over_budget_count, + self.duplicate_frames_skipped, self.capture_queue_depth, self.encoded_queue_depth, self.capture_gap_p95_ms,