diff --git a/src/video/capture/linux.rs b/src/video/capture/linux.rs index ad6e7d5b..4135af44 100644 --- a/src/video/capture/linux.rs +++ b/src/video/capture/linux.rs @@ -42,6 +42,12 @@ pub struct CaptureMeta { pub sequence: u64, } +fn append_capture_plane(dst: &mut Vec, mapping: &[u8], offset: usize, bytes: usize) { + let start = offset.min(mapping.len()); + let end = offset.saturating_add(bytes).min(mapping.len()); + dst.extend_from_slice(&mapping[start..end]); +} + /// When set, DV ioctls use the subdev (rkcif: video node has no DV ioctls). #[derive(Debug, Clone, Default)] pub struct BridgeContext { @@ -629,12 +635,12 @@ impl CaptureStream { } let mapping = &self.mappings[index][plane_idx]; let start = data_offset.min(mapping.len()); - let end = (data_offset + bytes_used).min(mapping.len()); + let end = data_offset.saturating_add(bytes_used).min(mapping.len()); total += end.saturating_sub(start); } - dst.resize(total, 0); - let mut cursor = 0usize; + dst.clear(); + dst.reserve(total); for (plane_idx, plane) in dqbuf.planes_iter().enumerate() { let bytes_used = *plane.bytesused as usize; let data_offset = plane.data_offset.copied().unwrap_or(0) as usize; @@ -642,14 +648,7 @@ impl CaptureStream { continue; } let mapping = &self.mappings[index][plane_idx]; - let start = data_offset.min(mapping.len()); - let end = (data_offset + bytes_used).min(mapping.len()); - let len = end.saturating_sub(start); - if len == 0 { - continue; - } - dst[cursor..cursor + len].copy_from_slice(&mapping[start..end]); - cursor += len; + append_capture_plane(dst, mapping, data_offset, bytes_used); } self.queue_buffer(index as u32) @@ -1153,9 +1152,23 @@ fn set_fps(fd: &File, queue: QueueType, fps: u32) -> std::result::Result<(), ioc #[cfg(test)] mod tests { - use super::{open_capture_device, DvTimingsSignature, NativeHdmirxState}; + use super::{append_capture_plane, open_capture_device, DvTimingsSignature, NativeHdmirxState}; use crate::video::format::PixelFormat; + #[test] + fn capture_planes_append_without_stale_bytes_or_reallocation() { + let mut data = vec![99; 64]; + let pointer = data.as_ptr(); + data.clear(); + append_capture_plane(&mut data, &[1, 2, 3, 4], 1, 2); + append_capture_plane(&mut data, &[5, 6, 7], 0, 10); + assert_eq!(data, [2, 3, 5, 6, 7]); + assert_eq!(data.as_ptr(), pointer); + append_capture_plane(&mut data, &[8], usize::MAX, usize::MAX); + append_capture_plane(&mut data, &[8], 0, 0); + assert_eq!(data, [2, 3, 5, 6, 7]); + } + fn timing() -> DvTimingsSignature { DvTimingsSignature { width: 1920, diff --git a/src/video/codec/convert.rs b/src/video/codec/convert.rs index 305edd7e..2107c4fd 100644 --- a/src/video/codec/convert.rs +++ b/src/video/codec/convert.rs @@ -544,7 +544,7 @@ pub struct Nv12Converter { /// MJPEG decoder that writes NV12 directly using libyuv. pub struct MjpegToNv12Decoder { resolution: Resolution, - output_buffer: Nv12Buffer, + output_buffer: Option, size_checked: bool, } @@ -552,7 +552,7 @@ impl MjpegToNv12Decoder { pub fn new(resolution: Resolution) -> Self { Self { resolution, - output_buffer: Nv12Buffer::new(resolution), + output_buffer: None, size_checked: false, } } @@ -561,11 +561,14 @@ impl MjpegToNv12Decoder { self.check_size(input)?; let width = self.resolution.width as i32; let height = self.resolution.height as i32; + let output = self + .output_buffer + .get_or_insert_with(|| Nv12Buffer::new(self.resolution)); - libyuv::mjpg_to_nv12(input, self.output_buffer.as_bytes_mut(), width, height) + libyuv::mjpg_to_nv12(input, output.as_bytes_mut(), width, height) .map_err(|e| AppError::VideoError(format!("libyuv MJPEG->NV12 failed: {}", e)))?; - Ok(self.output_buffer.as_bytes()) + Ok(output.as_bytes()) } /// Decode into caller-owned storage so capture and encoding can run on @@ -899,6 +902,7 @@ mod tests { let mut output = Vec::with_capacity(output_size); let allocation = output.as_ptr(); let mut decoder = MjpegToNv12Decoder::new(resolution); + assert!(decoder.output_buffer.is_none()); decoder.decode_into(&jpeg, &mut output).unwrap(); assert_eq!(output.len(), output_size); @@ -907,5 +911,10 @@ mod tests { decoder.decode_into(&jpeg, &mut output).unwrap(); assert_eq!(output.len(), output_size); assert_eq!(output.as_ptr(), allocation); + + assert!(decoder.output_buffer.is_none()); + assert_eq!(decoder.decode(&jpeg).unwrap(), output.as_slice()); + let internal = decoder.decode(&jpeg).unwrap().as_ptr(); + assert_eq!(decoder.decode(&jpeg).unwrap().as_ptr(), internal); } } diff --git a/src/video/pipeline/shared.rs b/src/video/pipeline/shared.rs index 92cd33c4..25a10fc9 100644 --- a/src/video/pipeline/shared.rs +++ b/src/video/pipeline/shared.rs @@ -157,6 +157,11 @@ use crate::video::recovery::{wait_for_source_change, CaptureRecoveryPolicy}; use crate::video::signal::SignalStatus; const MIN_CAPTURE_FRAME_SIZE: usize = 128; +const OUTPUT_BACKPRESSURE_GRACE_NS: u64 = 10_000_000; + +fn sustained_output_backpressure(started_ns: u64, now_ns: u64) -> bool { + started_ns != 0 && now_ns.saturating_sub(started_ns - 1) >= OUTPUT_BACKPRESSURE_GRACE_NS +} type VideoFrameMailbox = FrameMailbox>; fn spawn_video_worker( @@ -637,6 +642,7 @@ pub struct SharedVideoPipeline { encoder_output_wait_ns: AtomicU64, output_backpressure_events: AtomicU64, output_backpressured: AtomicBool, + output_backpressure_since_ns: AtomicU64, encoded_frames_skipped_for_slow_subscribers: AtomicU64, delivery_stats: Arc, running: watch::Sender, @@ -704,6 +710,7 @@ impl SharedVideoPipeline { encoder_output_wait_ns: AtomicU64::new(0), output_backpressure_events: AtomicU64::new(0), output_backpressured: AtomicBool::new(false), + output_backpressure_since_ns: AtomicU64::new(0), encoded_frames_skipped_for_slow_subscribers: AtomicU64::new(0), delivery_stats: Arc::new(DeliveryStats::default()), running: running_tx, @@ -911,6 +918,13 @@ impl SharedVideoPipeline { stage: MjpegAdmissionStage, ) -> bool { let backpressured = self.output_backpressured.load(Ordering::Acquire); + if backpressured && policy == MjpegAdmissionPolicy::BoundedPrefetch { + let started = self.output_backpressure_since_ns.load(Ordering::Acquire); + let now = PROCESS_START.get_or_init(Instant::now).elapsed().as_nanos() as u64; + if !sustained_output_backpressure(started, now) { + return false; + } + } if !policy.check_pending(backpressured) { return false; } @@ -1081,6 +1095,11 @@ impl SharedVideoPipeline { } } if active_subscribers > 0 && reservations.is_empty() { + if !self.output_backpressured.load(Ordering::Acquire) { + let now = PROCESS_START.get_or_init(Instant::now).elapsed().as_nanos() as u64; + self.output_backpressure_since_ns + .store(now.saturating_add(1), Ordering::Release); + } if !self.output_backpressured.swap(true, Ordering::AcqRel) { self.output_backpressure_events .fetch_add(1, Ordering::Relaxed); @@ -1088,6 +1107,8 @@ impl SharedVideoPipeline { None } else { self.output_backpressured.store(false, Ordering::Release); + self.output_backpressure_since_ns + .store(0, Ordering::Release); Some(reservations) } } @@ -1302,6 +1323,8 @@ impl SharedVideoPipeline { self.encoder_output_wait_ns.store(0, Ordering::Relaxed); self.output_backpressure_events.store(0, Ordering::Relaxed); self.output_backpressured.store(false, Ordering::Release); + self.output_backpressure_since_ns + .store(0, Ordering::Release); self.encoded_frames_skipped_for_slow_subscribers .store(0, Ordering::Relaxed); self.delivery_stats.sent_frames.store(0, Ordering::Relaxed); @@ -2538,6 +2561,15 @@ mod tests { assert_eq!(stats.decoded_frames_not_encoded, 5); } + #[test] + fn brief_output_backpressure_keeps_bounded_prefetch_running() { + assert!(!sustained_output_backpressure(0, 100_000_000)); + assert!(!sustained_output_backpressure(1, 9_999_999)); + assert!(sustained_output_backpressure(1, 10_000_000)); + assert!(sustained_output_backpressure(1, 100_000_000)); + assert!(!sustained_output_backpressure(20_000_001, 10_000_000)); + } + #[test] fn mjpeg_drop_reasons_sum_to_total_without_double_counting() { let pipeline = delivery_test_pipeline(); @@ -2606,6 +2638,7 @@ mod tests { } mailbox.publish(1, delivery_test_raw_frame(1)); for stage in [MjpegAdmissionStage::Capture, MjpegAdmissionStage::Decode] { + std::thread::sleep(Duration::from_millis(11)); assert!(pipeline.drop_pending_mjpeg( &mailbox, MjpegAdmissionPolicy::BoundedPrefetch,