perf(video): 消除中间帧复制并优化 MJPEG 有界预取

通过 Bytes 所有权和只读 AVBufferRef 引用池化 NV12 输入，消除中间 AVFrame 的整帧复制，并保留 V4L2 编码包脱离硬件缓冲的处理。

允许输出健康时有界预取 MJPEG，在输出背压时优先解码前丢帧；增加等帧、输出等待、丢帧原因及编码分段耗时统计，提供预取和所有权输入的对照开关。

补充缓冲寿命、只读保护、软件编码回退和队列策略测试及接口说明，保留现有 PTS、解码 worker、音频和加密行为。

src/video/codec/h264.rs

@@ -370,7 +370,37 @@ impl H264Encoder {
 
         self.frame_count += 1;
 
-        match self.inner.encode_bytes(data, pts_ms) {
+        let result = self.inner.encode_bytes(data, pts_ms);
+        Self::finish_encode(result)
+    }
+
+    pub fn supports_owned_nv12_input(&self) -> bool {
+        self.inner.supports_owned_nv12_input()
+    }
+
+    pub fn input_stats(&self) -> Option<hwcodec::ffmpeg_ram::encode::EncodeInputStats> {
+        self.inner.input_stats()
+    }
+
+    pub fn encode_owned_nv12(&mut self, data: Bytes, pts_ms: i64) -> Result<Vec<HwEncodeFrame>> {
+        if !self.supports_owned_nv12_input() {
+            return self.encode_raw(data.as_ref(), pts_ms);
+        }
+        if data.len() < self.yuv_length as usize {
+            return Err(AppError::VideoError(format!(
+                "Frame data too small: {} < {}",
+                data.len(),
+                self.yuv_length
+            )));
+        }
+        self.frame_count += 1;
+        Self::finish_encode(self.inner.encode_owned_bytes(data, pts_ms))
+    }
+
+    fn finish_encode(
+        result: std::result::Result<Vec<hwcodec::ffmpeg_ram::encode::EncodeBytesFrame>, i32>,
+    ) -> Result<Vec<HwEncodeFrame>> {
+        match result {
             Ok(frames) => {
                 let owned_frames: Vec<HwEncodeFrame> = frames
                     .into_iter()
@@ -490,6 +520,39 @@ impl Encoder for H264Encoder {
 mod tests {
     use super::*;
 
+    #[test]
+    fn owned_input_fallback_preserves_software_packets_and_pts() {
+        let config = H264Config::low_latency(Resolution::new(64, 64), 500);
+        let mut copied = H264Encoder::with_codec(config.clone(), "libx264").unwrap();
+        let mut owned = H264Encoder::with_codec(config, "libx264").unwrap();
+        assert!(!owned.supports_owned_nv12_input());
+        let mut packets = 0;
+        for sequence in 0..12 {
+            let data = Bytes::from(vec![64 + sequence as u8; owned.yuv_length as usize]);
+            let pts = sequence * 33;
+            let expected = copied.encode_raw(data.as_ref(), pts).unwrap();
+            let actual = owned.encode_owned_nv12(data, pts).unwrap();
+            assert_eq!(actual.len(), expected.len());
+            packets += actual.len();
+            for (actual, expected) in actual.iter().zip(expected.iter()) {
+                assert_eq!(actual.data, expected.data);
+                assert_eq!(actual.pts, expected.pts);
+                assert_eq!(actual.key, expected.key);
+            }
+        }
+        assert!(packets > 0);
+    }
+
+    #[test]
+    fn owned_input_rejects_short_frame_before_encoding() {
+        let config = H264Config::low_latency(Resolution::new(64, 64), 500);
+        let mut encoder = H264Encoder::with_codec(config, "libx264").unwrap();
+        assert!(encoder
+            .encode_owned_nv12(Bytes::from_static(&[0]), 0)
+            .is_err());
+        assert_eq!(encoder.frame_count, 0);
+    }
+
     #[test]
     fn test_detect_encoder() {
         let (encoder_type, codec_name) = detect_best_encoder(1280, 720);

src/video/frame.rs

@@ -14,6 +14,14 @@ enum FrameData {
     Pooled(Arc<FrameBuffer>),
 }
 
+struct PooledFrameOwner(Arc<FrameBuffer>);
+
+impl AsRef<[u8]> for PooledFrameOwner {
+    fn as_ref(&self) -> &[u8] {
+        self.0.as_slice()
+    }
+}
+
 impl std::fmt::Debug for FrameData {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
@@ -198,6 +206,13 @@ impl VideoFrame {
         }
     }
 
+    pub(crate) fn owned_data_bytes(&self) -> Bytes {
+        match &self.data {
+            FrameData::Bytes(bytes) => bytes.clone(),
+            FrameData::Pooled(buffer) => Bytes::from_owner(PooledFrameOwner(buffer.clone())),
+        }
+    }
+
     /// Get data length
     pub fn len(&self) -> usize {
         self.data().len()
@@ -302,3 +317,38 @@ impl From<&VideoFrame> for FrameMeta {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn owned_pooled_bytes_preserve_pointer_and_return_buffer_after_last_reference() {
+        let pool = Arc::new(FrameBufferPool::new(1));
+        let buffer = Arc::new(FrameBuffer::new(vec![17; 24], Some(pool.clone())));
+        let pointer = buffer.as_slice().as_ptr();
+        let frame = VideoFrame::from_pooled(buffer, Resolution::new(4, 4), PixelFormat::Nv12, 4, 1);
+        let bytes = frame.owned_data_bytes();
+        assert_eq!(bytes.as_ptr(), pointer);
+        let retained = bytes.slice(4..);
+        drop(frame);
+        drop(bytes);
+        assert!(pool.pool.lock().is_empty());
+        assert_eq!(retained.as_ref(), &[17; 20]);
+        drop(retained);
+        assert_eq!(pool.pool.lock().len(), 1);
+        let recycled = pool.take(24);
+        assert_eq!(recycled.as_ptr(), pointer);
+    }
+
+    #[test]
+    fn owned_bytes_input_is_a_shared_reference() {
+        let bytes = Bytes::from(vec![29; 24]);
+        let pointer = bytes.as_ptr();
+        let frame = VideoFrame::new(bytes, Resolution::new(4, 4), PixelFormat::Nv12, 4, 1);
+        let owned = frame.owned_data_bytes();
+        drop(frame);
+        assert_eq!(owned.as_ptr(), pointer);
+        assert_eq!(owned.as_ref(), &[29; 24]);
+    }
+}

src/video/pipeline/encoder_state.rs

@@ -33,6 +33,15 @@ pub(super) struct EncoderThreadState {
 
 pub(super) trait VideoEncoderTrait: Send {
     fn encode_raw(&mut self, data: &[u8], pts_ms: i64) -> Result<Vec<EncodedFrame>>;
+    fn supports_owned_nv12_input(&self) -> bool {
+        false
+    }
+    fn encode_owned_nv12(&mut self, data: Bytes, pts_ms: i64) -> Result<Vec<EncodedFrame>> {
+        self.encode_raw(data.as_ref(), pts_ms)
+    }
+    fn input_stats(&self) -> Option<hwcodec::ffmpeg_ram::encode::EncodeInputStats> {
+        None
+    }
     fn set_bitrate(&mut self, bitrate_kbps: u32) -> Result<()>;
     fn codec_name(&self) -> &str;
     fn request_keyframe(&mut self);
@@ -67,6 +76,27 @@ impl VideoEncoderTrait for H264EncoderWrapper {
         self.0.set_bitrate(bitrate_kbps)
     }
 
+    fn supports_owned_nv12_input(&self) -> bool {
+        self.0.supports_owned_nv12_input()
+    }
+
+    fn encode_owned_nv12(&mut self, data: Bytes, pts_ms: i64) -> Result<Vec<EncodedFrame>> {
+        Ok(self
+            .0
+            .encode_owned_nv12(data, pts_ms)?
+            .into_iter()
+            .map(|frame| EncodedFrame {
+                data: frame.data,
+                key: frame.key,
+                pts: frame.pts,
+            })
+            .collect())
+    }
+
+    fn input_stats(&self) -> Option<hwcodec::ffmpeg_ram::encode::EncodeInputStats> {
+        self.0.input_stats()
+    }
+
     fn codec_name(&self) -> &str {
         self.0.codec_name()
     }
@@ -91,7 +121,11 @@ fn create_h264_encoder(
         },
         codec_name,
     )?;
-    info!("Created H264 encoder: {}", encoder.codec_name());
+    info!(
+        "Created H264 encoder: {} owned_nv12_input={}",
+        encoder.codec_name(),
+        encoder.supports_owned_nv12_input()
+    );
     Ok(Box::new(H264EncoderWrapper(encoder)))
 }
 

src/video/pipeline/frame_mailbox.rs

@@ -1,4 +1,21 @@
 use std::sync::{Condvar, Mutex};
+use std::time::{Duration, Instant};
+
+#[derive(Debug, Default, PartialEq, Eq)]
+pub(super) struct FrameWaitTimes {
+    pub(super) frame: Duration,
+    pub(super) output: Duration,
+}
+
+impl FrameWaitTimes {
+    fn record(&mut self, pending: bool, elapsed: Duration) {
+        if pending {
+            self.output += elapsed;
+        } else {
+            self.frame += elapsed;
+        }
+    }
+}
 
 #[derive(Debug, PartialEq, Eq)]
 pub(super) enum PublishResult {
@@ -50,10 +67,21 @@ impl<Frame> FrameMailbox<Frame> {
         self.receive_when(|| Some(())).map(|(frame, ())| frame)
     }
 
+    #[cfg(test)]
     pub(super) fn receive_when<Ready>(
         &self,
-        mut ready: impl FnMut() -> Option<Ready>,
+        ready: impl FnMut() -> Option<Ready>,
     ) -> Option<(Frame, Ready)> {
+        self.receive_when_timed(ready, false)
+            .map(|(frame, ready, _)| (frame, ready))
+    }
+
+    pub(super) fn receive_when_timed<Ready>(
+        &self,
+        mut ready: impl FnMut() -> Option<Ready>,
+        measure: bool,
+    ) -> Option<(Frame, Ready, FrameWaitTimes)> {
+        let mut waits = FrameWaitTimes::default();
         let mut state = self.state.lock().expect("frame mailbox lock poisoned");
         loop {
             if state.closed {
@@ -61,13 +89,18 @@ impl<Frame> FrameMailbox<Frame> {
             }
             if state.pending.is_some() {
                 if let Some(ready) = ready() {
-                    return state.pending.take().map(|frame| (frame, ready));
+                    return state.pending.take().map(|frame| (frame, ready, waits));
                 }
             }
+            let pending = state.pending.is_some();
+            let wait_started = measure.then(Instant::now);
             state = self
                 .available
                 .wait(state)
                 .expect("frame mailbox lock poisoned");
+            if let Some(started) = wait_started {
+                waits.record(pending, started.elapsed());
+            }
         }
     }
 
@@ -175,4 +208,107 @@ mod tests {
             worker.join().unwrap();
         }
     }
+
+    #[test]
+    fn wait_times_separate_missing_input_from_full_output() {
+        let mut waits = FrameWaitTimes::default();
+        waits.record(false, Duration::from_millis(3));
+        waits.record(true, Duration::from_millis(7));
+        waits.record(false, Duration::from_millis(2));
+        assert_eq!(waits.frame, Duration::from_millis(5));
+        assert_eq!(waits.output, Duration::from_millis(7));
+    }
+
+    #[test]
+    fn available_frame_has_no_wait_time() {
+        let mailbox = FrameMailbox::new();
+        mailbox.publish(1, "ready");
+        let result = mailbox.receive_when_timed(|| Some(42), true);
+        assert_eq!(result, Some(("ready", 42, FrameWaitTimes::default())));
+    }
+
+    #[test]
+    fn timed_output_wait_preserves_latest_frame_and_can_be_disabled() {
+        use std::sync::atomic::{AtomicBool, Ordering};
+
+        for measure in [false, true] {
+            let mailbox = Arc::new(FrameMailbox::new());
+            mailbox.publish(1, "initial");
+            let ready = Arc::new(AtomicBool::new(false));
+            let receiver_mailbox = mailbox.clone();
+            let receiver_ready = ready.clone();
+            let (waiting_sender, waiting_receiver) = mpsc::channel();
+            let (result_sender, result_receiver) = mpsc::channel();
+            let worker = std::thread::spawn(move || {
+                let mut reported = false;
+                let result = receiver_mailbox.receive_when_timed(
+                    || {
+                        if receiver_ready.load(Ordering::Acquire) {
+                            Some(42)
+                        } else {
+                            if !reported {
+                                waiting_sender.send(()).unwrap();
+                                reported = true;
+                            }
+                            None
+                        }
+                    },
+                    measure,
+                );
+                result_sender.send(result).unwrap();
+            });
+            waiting_receiver
+                .recv_timeout(Duration::from_secs(2))
+                .unwrap();
+            mailbox.publish(2, "latest");
+            ready.store(true, Ordering::Release);
+            mailbox.notify();
+            let (frame, permit, waits) = result_receiver
+                .recv_timeout(Duration::from_secs(2))
+                .unwrap()
+                .unwrap();
+            assert_eq!((frame, permit), ("latest", 42));
+            assert_eq!(waits.frame, Duration::ZERO);
+            if measure {
+                assert!(waits.output > Duration::ZERO);
+            } else {
+                assert_eq!(waits, FrameWaitTimes::default());
+            }
+            worker.join().unwrap();
+        }
+    }
+
+    #[test]
+    fn timed_receiver_exits_when_closed_without_output_capacity() {
+        let mailbox = Arc::new(FrameMailbox::new());
+        mailbox.publish(1, "pending");
+        let receiver = mailbox.clone();
+        let (waiting_sender, waiting_receiver) = mpsc::channel();
+        let (result_sender, result_receiver) = mpsc::channel();
+        let worker = std::thread::spawn(move || {
+            let mut reported = false;
+            let result = receiver.receive_when_timed(
+                || {
+                    if !reported {
+                        waiting_sender.send(()).unwrap();
+                        reported = true;
+                    }
+                    None::<()>
+                },
+                true,
+            );
+            result_sender.send(result).unwrap();
+        });
+        waiting_receiver
+            .recv_timeout(Duration::from_secs(2))
+            .unwrap();
+        mailbox.close();
+        assert_eq!(
+            result_receiver
+                .recv_timeout(Duration::from_secs(2))
+                .unwrap(),
+            None
+        );
+        worker.join().unwrap();
+    }
 }

src/video/pipeline/mjpeg_admission.rs

@@ -0,0 +1,125 @@
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum MjpegAdmissionPolicy {
+    PendingSlot,
+    BoundedPrefetch,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum MjpegAdmissionStage {
+    Capture,
+    Decode,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum MjpegDropReason {
+    PendingCapture,
+    PendingDecode,
+    BackpressureCapture,
+    BackpressureDecode,
+    WorkersBusy,
+}
+
+impl MjpegAdmissionPolicy {
+    pub(super) fn from_prefetch_setting(setting: Option<&str>) -> Self {
+        if setting == Some("0") {
+            Self::PendingSlot
+        } else {
+            Self::BoundedPrefetch
+        }
+    }
+
+    pub(super) fn check_pending(self, output_backpressured: bool) -> bool {
+        self == Self::PendingSlot || output_backpressured
+    }
+
+    pub(super) fn drop_reason(
+        self,
+        pending: bool,
+        output_backpressured: bool,
+        stage: MjpegAdmissionStage,
+    ) -> Option<MjpegDropReason> {
+        if !pending || !self.check_pending(output_backpressured) {
+            return None;
+        }
+        Some(match (output_backpressured, stage) {
+            (false, MjpegAdmissionStage::Capture) => MjpegDropReason::PendingCapture,
+            (false, MjpegAdmissionStage::Decode) => MjpegDropReason::PendingDecode,
+            (true, MjpegAdmissionStage::Capture) => MjpegDropReason::BackpressureCapture,
+            (true, MjpegAdmissionStage::Decode) => MjpegDropReason::BackpressureDecode,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn bounded_prefetch_is_default_and_zero_restores_pending_slot_policy() {
+        for setting in [None, Some("1"), Some(""), Some("invalid")] {
+            assert_eq!(
+                MjpegAdmissionPolicy::from_prefetch_setting(setting),
+                MjpegAdmissionPolicy::BoundedPrefetch
+            );
+        }
+        assert_eq!(
+            MjpegAdmissionPolicy::from_prefetch_setting(Some("0")),
+            MjpegAdmissionPolicy::PendingSlot
+        );
+    }
+
+    #[test]
+    fn prefetch_keeps_decoding_while_output_has_capacity() {
+        for stage in [MjpegAdmissionStage::Capture, MjpegAdmissionStage::Decode] {
+            for pending in [false, true] {
+                assert_eq!(
+                    MjpegAdmissionPolicy::BoundedPrefetch.drop_reason(pending, false, stage),
+                    None
+                );
+            }
+        }
+        assert!(!MjpegAdmissionPolicy::BoundedPrefetch.check_pending(false));
+    }
+
+    #[test]
+    fn output_backpressure_preserves_one_pending_frame() {
+        for policy in [
+            MjpegAdmissionPolicy::BoundedPrefetch,
+            MjpegAdmissionPolicy::PendingSlot,
+        ] {
+            for (stage, reason) in [
+                (
+                    MjpegAdmissionStage::Capture,
+                    MjpegDropReason::BackpressureCapture,
+                ),
+                (
+                    MjpegAdmissionStage::Decode,
+                    MjpegDropReason::BackpressureDecode,
+                ),
+            ] {
+                assert_eq!(policy.drop_reason(false, true, stage), None);
+                assert_eq!(policy.drop_reason(true, true, stage), Some(reason));
+            }
+        }
+    }
+
+    #[test]
+    fn legacy_policy_drops_when_pending_even_without_output_backpressure() {
+        for (stage, reason) in [
+            (
+                MjpegAdmissionStage::Capture,
+                MjpegDropReason::PendingCapture,
+            ),
+            (MjpegAdmissionStage::Decode, MjpegDropReason::PendingDecode),
+        ] {
+            assert_eq!(
+                MjpegAdmissionPolicy::PendingSlot.drop_reason(false, false, stage),
+                None
+            );
+            assert_eq!(
+                MjpegAdmissionPolicy::PendingSlot.drop_reason(true, false, stage),
+                Some(reason)
+            );
+        }
+    }
+}

src/video/pipeline/mod.rs

@@ -3,6 +3,7 @@
 mod delivery;
 mod encoder_state;
 mod frame_mailbox;
+mod mjpeg_admission;
 mod shared;
 
 pub use delivery::EncodedVideoFrameReceiver;

src/video/pipeline/shared.rs

@@ -33,6 +33,7 @@ use super::delivery::{
 };
 use super::encoder_state::{build_encoder_state, should_parallel_decode_mjpeg, EncoderThreadState};
 use super::frame_mailbox::{FrameMailbox, PublishResult};
+use super::mjpeg_admission::{MjpegAdmissionPolicy, MjpegAdmissionStage, MjpegDropReason};
 
 #[cfg(all(target_os = "linux", any(target_arch = "aarch64", target_arch = "arm")))]
 #[path = "dmabuf.rs"]
@@ -181,6 +182,7 @@ fn spawn_mjpeg_decode_workers(
     frame_mailbox: &Arc<VideoFrameMailbox>,
     buffer_pool: &Arc<FrameBufferPool>,
     resolution: Resolution,
+    admission_policy: MjpegAdmissionPolicy,
 ) -> Vec<SyncSender<MjpegDecodeJob>> {
     let available = std::thread::available_parallelism()
         .map(|count| count.get())
@@ -206,10 +208,11 @@ fn spawn_mjpeg_decode_workers(
                         worker_buffer_pool.put(job.data);
                         break;
                     }
-                    if worker_frame_mailbox.has_pending() {
-                        worker_pipeline
-                            .mjpeg_dropped_before_decode
-                            .fetch_add(1, Ordering::Relaxed);
+                    if worker_pipeline.drop_pending_mjpeg(
+                        &worker_frame_mailbox,
+                        admission_policy,
+                        MjpegAdmissionStage::Decode,
+                    ) {
                         worker_buffer_pool.put(job.data);
                         continue;
                     }
@@ -488,10 +491,47 @@ fn log_encoding_error(
     }
 }
 
+fn latency_percentile_ms(sorted_time_ns: &[u64], percentile: usize) -> f64 {
+    if sorted_time_ns.is_empty() {
+        return 0.0;
+    }
+    let index = (sorted_time_ns.len() * percentile.clamp(1, 100)).div_ceil(100) - 1;
+    sorted_time_ns[index] as f64 / 1_000_000.0
+}
+
+fn log_encoder_input_performance(
+    stats: hwcodec::ffmpeg_ram::encode::EncodeInputStats,
+    previous: &mut hwcodec::ffmpeg_ram::encode::EncodeInputStats,
+) {
+    let delta = |current: u64, before: u64| current.saturating_sub(before);
+    let inputs = delta(stats.input_frames, previous.input_frames);
+    let average_ms = |current: u64, before: u64| {
+        if inputs == 0 {
+            0.0
+        } else {
+            delta(current, before) as f64 / inputs as f64 / 1000.0
+        }
+    };
+    info!(
+        "[VideoInputPerf] inputs={} borrowed={} copied={} copied_bytes={} prepare_ms={:.3} send_frame_ms={:.3} receive_packet_ms={:.3} packet_callback_ms={:.3} output_copy_ms={:.3}",
+        inputs,
+        delta(stats.borrowed_frames, previous.borrowed_frames),
+        delta(stats.copied_frames, previous.copied_frames),
+        delta(stats.copied_bytes, previous.copied_bytes),
+        average_ms(stats.prepare_us, previous.prepare_us),
+        average_ms(stats.send_us, previous.send_us),
+        average_ms(stats.receive_us, previous.receive_us),
+        average_ms(stats.packet_us, previous.packet_us),
+        average_ms(stats.output_copy_us, previous.output_copy_us),
+    );
+    *previous = stats;
+}
+
 fn log_pipeline_performance(
     stats: &SharedVideoPipelineStats,
     previous: &SharedVideoPipelineStats,
     elapsed: f64,
+    sorted_encode_time_ns: &[u64],
 ) {
     let delta = |current: u64, previous: u64| current.saturating_sub(previous);
     let decoded = delta(stats.mjpeg_decoded_frames, previous.mjpeg_decoded_frames);
@@ -522,6 +562,21 @@ fn log_pipeline_performance(
         delta(stats.encoded_frames_skipped_for_slow_subscribers, previous.encoded_frames_skipped_for_slow_subscribers),
         send_errors,
     );
+    info!(
+        "[VideoQueuePerf] window_s={:.2} input_fps={:.2} frame_wait_ms={:.2} output_wait_ms={:.2} timed_inputs={} encode_p95_ms={:.2} encode_p99_ms={:.2} drop_pending_capture={} drop_pending_decode={} drop_backpressure_capture={} drop_backpressure_decode={} drop_workers_busy={}",
+        elapsed,
+        encoded_inputs as f64 / elapsed,
+        average_ms(delta(stats.encoder_frame_wait_ns, previous.encoder_frame_wait_ns), encoded_inputs),
+        average_ms(delta(stats.encoder_output_wait_ns, previous.encoder_output_wait_ns), encoded_inputs),
+        sorted_encode_time_ns.len(),
+        latency_percentile_ms(sorted_encode_time_ns, 95),
+        latency_percentile_ms(sorted_encode_time_ns, 99),
+        delta(stats.mjpeg_dropped_pending_capture, previous.mjpeg_dropped_pending_capture),
+        delta(stats.mjpeg_dropped_pending_decode, previous.mjpeg_dropped_pending_decode),
+        delta(stats.mjpeg_dropped_backpressure_capture, previous.mjpeg_dropped_backpressure_capture),
+        delta(stats.mjpeg_dropped_backpressure_decode, previous.mjpeg_dropped_backpressure_decode),
+        delta(stats.mjpeg_dropped_workers_busy, previous.mjpeg_dropped_workers_busy),
+    );
 }
 
 /// Pipeline statistics
@@ -530,12 +585,19 @@ pub struct SharedVideoPipelineStats {
     pub current_fps: f32,
     pub mjpeg_decoded_frames: u64,
     pub mjpeg_dropped_before_decode: u64,
+    pub mjpeg_dropped_pending_capture: u64,
+    pub mjpeg_dropped_pending_decode: u64,
+    pub mjpeg_dropped_backpressure_capture: u64,
+    pub mjpeg_dropped_backpressure_decode: u64,
+    pub mjpeg_dropped_workers_busy: u64,
     pub decoded_frames_not_encoded: u64,
     pub mjpeg_decode_time_ns: u64,
     pub encoded_input_frames: u64,
     pub encoded_frames: u64,
     pub encoded_bytes: u64,
     pub encode_time_ns: u64,
+    pub encoder_frame_wait_ns: u64,
+    pub encoder_output_wait_ns: u64,
     pub output_backpressure_events: u64,
     pub encoded_frames_skipped_for_slow_subscribers: u64,
     pub sent_frames: u64,
@@ -560,12 +622,19 @@ pub struct SharedVideoPipeline {
     frame_mailbox: FrameMailboxHandle,
     mjpeg_decoded_frames: AtomicU64,
     mjpeg_dropped_before_decode: AtomicU64,
+    mjpeg_dropped_pending_capture: AtomicU64,
+    mjpeg_dropped_pending_decode: AtomicU64,
+    mjpeg_dropped_backpressure_capture: AtomicU64,
+    mjpeg_dropped_backpressure_decode: AtomicU64,
+    mjpeg_dropped_workers_busy: AtomicU64,
     decoded_frames_not_encoded: AtomicU64,
     mjpeg_decode_time_ns: AtomicU64,
     encoded_input_frames: AtomicU64,
     encoded_frames: AtomicU64,
     encoded_bytes: AtomicU64,
     encode_time_ns: AtomicU64,
+    encoder_frame_wait_ns: AtomicU64,
+    encoder_output_wait_ns: AtomicU64,
     output_backpressure_events: AtomicU64,
     output_backpressured: AtomicBool,
     encoded_frames_skipped_for_slow_subscribers: AtomicU64,
@@ -620,12 +689,19 @@ impl SharedVideoPipeline {
             frame_mailbox: Arc::new(ParkingMutex::new(None)),
             mjpeg_decoded_frames: AtomicU64::new(0),
             mjpeg_dropped_before_decode: AtomicU64::new(0),
+            mjpeg_dropped_pending_capture: AtomicU64::new(0),
+            mjpeg_dropped_pending_decode: AtomicU64::new(0),
+            mjpeg_dropped_backpressure_capture: AtomicU64::new(0),
+            mjpeg_dropped_backpressure_decode: AtomicU64::new(0),
+            mjpeg_dropped_workers_busy: AtomicU64::new(0),
             decoded_frames_not_encoded: AtomicU64::new(0),
             mjpeg_decode_time_ns: AtomicU64::new(0),
             encoded_input_frames: AtomicU64::new(0),
             encoded_frames: AtomicU64::new(0),
             encoded_bytes: AtomicU64::new(0),
             encode_time_ns: AtomicU64::new(0),
+            encoder_frame_wait_ns: AtomicU64::new(0),
+            encoder_output_wait_ns: AtomicU64::new(0),
             output_backpressure_events: AtomicU64::new(0),
             output_backpressured: AtomicBool::new(false),
             encoded_frames_skipped_for_slow_subscribers: AtomicU64::new(0),
@@ -786,12 +862,25 @@ impl SharedVideoPipeline {
         stats.mjpeg_decoded_frames = self.mjpeg_decoded_frames.load(Ordering::Relaxed);
         stats.mjpeg_dropped_before_decode =
             self.mjpeg_dropped_before_decode.load(Ordering::Relaxed);
+        stats.mjpeg_dropped_pending_capture =
+            self.mjpeg_dropped_pending_capture.load(Ordering::Relaxed);
+        stats.mjpeg_dropped_pending_decode =
+            self.mjpeg_dropped_pending_decode.load(Ordering::Relaxed);
+        stats.mjpeg_dropped_backpressure_capture = self
+            .mjpeg_dropped_backpressure_capture
+            .load(Ordering::Relaxed);
+        stats.mjpeg_dropped_backpressure_decode = self
+            .mjpeg_dropped_backpressure_decode
+            .load(Ordering::Relaxed);
+        stats.mjpeg_dropped_workers_busy = self.mjpeg_dropped_workers_busy.load(Ordering::Relaxed);
         stats.decoded_frames_not_encoded = self.decoded_frames_not_encoded.load(Ordering::Relaxed);
         stats.mjpeg_decode_time_ns = self.mjpeg_decode_time_ns.load(Ordering::Relaxed);
         stats.encoded_input_frames = self.encoded_input_frames.load(Ordering::Relaxed);
         stats.encoded_frames = self.encoded_frames.load(Ordering::Relaxed);
         stats.encoded_bytes = self.encoded_bytes.load(Ordering::Relaxed);
         stats.encode_time_ns = self.encode_time_ns.load(Ordering::Relaxed);
+        stats.encoder_frame_wait_ns = self.encoder_frame_wait_ns.load(Ordering::Relaxed);
+        stats.encoder_output_wait_ns = self.encoder_output_wait_ns.load(Ordering::Relaxed);
         stats.output_backpressure_events = self.output_backpressure_events.load(Ordering::Relaxed);
         stats.encoded_frames_skipped_for_slow_subscribers = self
             .encoded_frames_skipped_for_slow_subscribers
@@ -802,6 +891,37 @@ impl SharedVideoPipeline {
         stats
     }
 
+    fn record_mjpeg_drop(&self, reason: MjpegDropReason) {
+        let counter = match reason {
+            MjpegDropReason::PendingCapture => &self.mjpeg_dropped_pending_capture,
+            MjpegDropReason::PendingDecode => &self.mjpeg_dropped_pending_decode,
+            MjpegDropReason::BackpressureCapture => &self.mjpeg_dropped_backpressure_capture,
+            MjpegDropReason::BackpressureDecode => &self.mjpeg_dropped_backpressure_decode,
+            MjpegDropReason::WorkersBusy => &self.mjpeg_dropped_workers_busy,
+        };
+        counter.fetch_add(1, Ordering::Relaxed);
+        self.mjpeg_dropped_before_decode
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    fn drop_pending_mjpeg(
+        &self,
+        mailbox: &VideoFrameMailbox,
+        policy: MjpegAdmissionPolicy,
+        stage: MjpegAdmissionStage,
+    ) -> bool {
+        let backpressured = self.output_backpressured.load(Ordering::Acquire);
+        if !policy.check_pending(backpressured) {
+            return false;
+        }
+        if let Some(reason) = policy.drop_reason(mailbox.has_pending(), backpressured, stage) {
+            self.record_mjpeg_drop(reason);
+            true
+        } else {
+            false
+        }
+    }
+
     /// Check if running
     pub fn is_running(&self) -> bool {
         *self.running_rx.borrow()
@@ -1150,21 +1270,36 @@ impl SharedVideoPipeline {
         }
 
         let mut encoder_config = config.clone();
+        let admission_policy = MjpegAdmissionPolicy::from_prefetch_setting(
+            std::env::var("ONE_KVM_MJPEG_PREFETCH").ok().as_deref(),
+        );
         if parallel_mjpeg_decode {
             encoder_config.input_format = PixelFormat::Nv12;
             info!("Using parallel libyuv MJPEG decode with hardware encoding");
+            info!("MJPEG admission policy: {:?}", admission_policy);
         }
         let mut encoder_state = build_encoder_state(&encoder_config)?;
         let frame_mailbox = Arc::new(VideoFrameMailbox::new());
         *self.frame_mailbox.lock() = Some(frame_mailbox.clone());
         self.mjpeg_decoded_frames.store(0, Ordering::Relaxed);
         self.mjpeg_dropped_before_decode.store(0, Ordering::Relaxed);
+        self.mjpeg_dropped_pending_capture
+            .store(0, Ordering::Relaxed);
+        self.mjpeg_dropped_pending_decode
+            .store(0, Ordering::Relaxed);
+        self.mjpeg_dropped_backpressure_capture
+            .store(0, Ordering::Relaxed);
+        self.mjpeg_dropped_backpressure_decode
+            .store(0, Ordering::Relaxed);
+        self.mjpeg_dropped_workers_busy.store(0, Ordering::Relaxed);
         self.decoded_frames_not_encoded.store(0, Ordering::Relaxed);
         self.mjpeg_decode_time_ns.store(0, Ordering::Relaxed);
         self.encoded_input_frames.store(0, Ordering::Relaxed);
         self.encoded_frames.store(0, Ordering::Relaxed);
         self.encoded_bytes.store(0, Ordering::Relaxed);
         self.encode_time_ns.store(0, Ordering::Relaxed);
+        self.encoder_frame_wait_ns.store(0, Ordering::Relaxed);
+        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.encoded_frames_skipped_for_slow_subscribers
@@ -1202,12 +1337,14 @@ impl SharedVideoPipeline {
                 let perf_logging = std::env::var("ONE_KVM_VIDEO_STATS").as_deref() == Ok("1");
                 let mut last_perf_time = Instant::now();
                 let mut last_perf_stats = pipeline.stats_snapshot();
+                let mut last_input_stats = Default::default();
+                let mut encode_latencies = Vec::with_capacity(if perf_logging { 256 } else { 0 });
                 let encode_error_throttler = LogThrottler::with_secs(ENCODE_ERROR_THROTTLE_SECS);
                 let mut suppressed_encode_errors: HashMap<String, u64> = HashMap::new();
 
                 while pipeline.running_flag.load(Ordering::Acquire) {
-                    let Some((frame, reservations)) =
-                        frame_mailbox.receive_when(|| pipeline.reserve_outputs())
+                    let Some((frame, reservations, waits)) = frame_mailbox
+                        .receive_when_timed(|| pipeline.reserve_outputs(), perf_logging)
                     else {
                         break;
                     };
@@ -1223,6 +1360,14 @@ impl SharedVideoPipeline {
                         }
                         continue;
                     }
+                    if perf_logging {
+                        pipeline
+                            .encoder_frame_wait_ns
+                            .fetch_add(waits.frame.as_nanos() as u64, Ordering::Relaxed);
+                        pipeline
+                            .encoder_output_wait_ns
+                            .fetch_add(waits.output.as_nanos() as u64, Ordering::Relaxed);
+                    }
 
                     while let Ok(cmd) = cmd_rx.try_recv() {
                         if let Err(e) = pipeline.apply_cmd(&mut encoder_state, cmd) {
@@ -1236,10 +1381,13 @@ impl SharedVideoPipeline {
                         .fetch_add(1, Ordering::Relaxed);
                     let encode_started = Instant::now();
                     let encode_result = pipeline.encode_frame_sync(&mut encoder_state, &frame);
-                    pipeline.encode_time_ns.fetch_add(
-                        encode_started.elapsed().as_nanos() as u64,
-                        Ordering::Relaxed,
-                    );
+                    let encode_elapsed_ns = encode_started.elapsed().as_nanos() as u64;
+                    pipeline
+                        .encode_time_ns
+                        .fetch_add(encode_elapsed_ns, Ordering::Relaxed);
+                    if perf_logging {
+                        encode_latencies.push(encode_elapsed_ns);
+                    }
 
                     match encode_result {
                         Ok(encoded_frames) => {
@@ -1282,7 +1430,21 @@ impl SharedVideoPipeline {
                     if perf_logging && last_perf_time.elapsed() >= Duration::from_secs(5) {
                         let elapsed = last_perf_time.elapsed().as_secs_f64();
                         let stats = pipeline.stats_snapshot();
-                        log_pipeline_performance(&stats, &last_perf_stats, elapsed);
+                        encode_latencies.sort_unstable();
+                        log_pipeline_performance(
+                            &stats,
+                            &last_perf_stats,
+                            elapsed,
+                            &encode_latencies,
+                        );
+                        if let Some(stats) = encoder_state
+                            .encoder
+                            .as_ref()
+                            .and_then(|encoder| encoder.input_stats())
+                        {
+                            log_encoder_input_performance(stats, &mut last_input_stats);
+                        }
+                        encode_latencies.clear();
                         last_perf_stats = stats;
                         last_perf_time = Instant::now();
                     }
@@ -1324,6 +1486,7 @@ impl SharedVideoPipeline {
                             &frame_mailbox,
                             &buffer_pool,
                             config.resolution,
+                            admission_policy,
                         )
                     })
                     .filter(|senders| !senders.is_empty());
@@ -1676,11 +1839,12 @@ impl SharedVideoPipeline {
 
                     if parallel_mjpeg_decode
                         && pixel_format == PixelFormat::Mjpeg
-                        && frame_mailbox.has_pending()
+                        && pipeline.drop_pending_mjpeg(
+                            &frame_mailbox,
+                            admission_policy,
+                            MjpegAdmissionStage::Capture,
+                        )
                     {
-                        pipeline
-                            .mjpeg_dropped_before_decode
-                            .fetch_add(1, Ordering::Relaxed);
                         buffer_pool.put(owned);
                         continue;
                     }
@@ -1712,9 +1876,7 @@ impl SharedVideoPipeline {
                             }
                         }
                         if let Some(job) = pending {
-                            pipeline
-                                .mjpeg_dropped_before_decode
-                                .fetch_add(1, Ordering::Relaxed);
+                            pipeline.record_mjpeg_drop(MjpegDropReason::WorkersBusy);
                             buffer_pool.put(job.data);
                         }
                         continue;
@@ -1895,7 +2057,15 @@ impl SharedVideoPipeline {
             encoder.encode_raw(nv12_data, pts_ms)
         } else {
             // Direct input (already in correct format)
-            encoder.encode_raw(raw_frame, pts_ms)
+            if decoded_buf.is_none()
+                && compacted_buf.is_none()
+                && frame.format == PixelFormat::Nv12
+                && encoder.supports_owned_nv12_input()
+            {
+                encoder.encode_owned_nv12(frame.owned_data_bytes(), pts_ms)
+            } else {
+                encoder.encode_raw(raw_frame, pts_ms)
+            }
         };
 
         match encode_result {
@@ -2368,6 +2538,105 @@ mod tests {
         assert_eq!(stats.decoded_frames_not_encoded, 5);
     }
 
+    #[test]
+    fn mjpeg_drop_reasons_sum_to_total_without_double_counting() {
+        let pipeline = delivery_test_pipeline();
+        for reason in [
+            MjpegDropReason::PendingCapture,
+            MjpegDropReason::PendingDecode,
+            MjpegDropReason::BackpressureCapture,
+            MjpegDropReason::BackpressureDecode,
+            MjpegDropReason::WorkersBusy,
+        ] {
+            pipeline.record_mjpeg_drop(reason);
+        }
+        pipeline.encoder_frame_wait_ns.store(123, Ordering::Relaxed);
+        pipeline
+            .encoder_output_wait_ns
+            .store(456, Ordering::Relaxed);
+        let stats = pipeline.stats_snapshot();
+        assert_eq!(stats.mjpeg_dropped_before_decode, 5);
+        assert_eq!(stats.mjpeg_dropped_pending_capture, 1);
+        assert_eq!(stats.mjpeg_dropped_pending_decode, 1);
+        assert_eq!(stats.mjpeg_dropped_backpressure_capture, 1);
+        assert_eq!(stats.mjpeg_dropped_backpressure_decode, 1);
+        assert_eq!(stats.mjpeg_dropped_workers_busy, 1);
+        assert_eq!(stats.encoder_frame_wait_ns, 123);
+        assert_eq!(stats.encoder_output_wait_ns, 456);
+    }
+
+    #[test]
+    fn bounded_prefetch_accepts_pending_frame_when_output_is_healthy() {
+        let pipeline = delivery_test_pipeline();
+        let mailbox = VideoFrameMailbox::new();
+        mailbox.publish(1, delivery_test_raw_frame(1));
+        for stage in [MjpegAdmissionStage::Capture, MjpegAdmissionStage::Decode] {
+            assert!(!pipeline.drop_pending_mjpeg(
+                &mailbox,
+                MjpegAdmissionPolicy::BoundedPrefetch,
+                stage,
+            ));
+            assert!(pipeline.drop_pending_mjpeg(
+                &mailbox,
+                MjpegAdmissionPolicy::PendingSlot,
+                stage,
+            ));
+        }
+        assert!(mailbox.has_pending());
+        let stats = pipeline.stats_snapshot();
+        assert_eq!(stats.mjpeg_dropped_before_decode, 2);
+        assert_eq!(stats.mjpeg_dropped_pending_capture, 1);
+        assert_eq!(stats.mjpeg_dropped_pending_decode, 1);
+    }
+
+    #[test]
+    fn bounded_prefetch_drops_only_with_pending_frame_and_output_backpressure() {
+        let pipeline = delivery_test_pipeline();
+        let mailbox = VideoFrameMailbox::new();
+        let mut receiver = pipeline.subscribe();
+        let reservations = pipeline.reserve_outputs().unwrap();
+        pipeline.broadcast_encoded_batch(vec![delivery_test_frame(1, true)].into(), reservations);
+        assert!(pipeline.reserve_outputs().is_none());
+        for stage in [MjpegAdmissionStage::Capture, MjpegAdmissionStage::Decode] {
+            assert!(!pipeline.drop_pending_mjpeg(
+                &mailbox,
+                MjpegAdmissionPolicy::BoundedPrefetch,
+                stage,
+            ));
+        }
+        mailbox.publish(1, delivery_test_raw_frame(1));
+        for stage in [MjpegAdmissionStage::Capture, MjpegAdmissionStage::Decode] {
+            assert!(pipeline.drop_pending_mjpeg(
+                &mailbox,
+                MjpegAdmissionPolicy::BoundedPrefetch,
+                stage,
+            ));
+        }
+        assert_eq!(receiver.try_recv().unwrap().sequence, 1);
+        let _reservations = pipeline.reserve_outputs().unwrap();
+        assert!(!pipeline.drop_pending_mjpeg(
+            &mailbox,
+            MjpegAdmissionPolicy::BoundedPrefetch,
+            MjpegAdmissionStage::Capture,
+        ));
+        assert!(mailbox.has_pending());
+        let stats = pipeline.stats_snapshot();
+        assert_eq!(stats.mjpeg_dropped_before_decode, 2);
+        assert_eq!(stats.mjpeg_dropped_backpressure_capture, 1);
+        assert_eq!(stats.mjpeg_dropped_backpressure_decode, 1);
+        assert_eq!(stats.mjpeg_dropped_pending_capture, 0);
+        assert_eq!(stats.mjpeg_dropped_pending_decode, 0);
+    }
+
+    #[test]
+    fn encoding_percentiles_use_nearest_rank_and_handle_empty_windows() {
+        assert_eq!(latency_percentile_ms(&[], 95), 0.0);
+        assert_eq!(latency_percentile_ms(&[3_000_000], 99), 3.0);
+        let sorted: Vec<u64> = (1..=100).map(|millis| millis * 1_000_000).collect();
+        assert_eq!(latency_percentile_ms(&sorted, 95), 95.0);
+        assert_eq!(latency_percentile_ms(&sorted, 99), 99.0);
+    }
+
     #[test]
     fn encoder_packet_pts_is_preserved() {
         use super::super::encoder_state::{EncodedFrame, VideoEncoderTrait};