diff --git a/libs/hwcodec/README.md b/libs/hwcodec/README.md index d81e614b..f9f899f6 100644 --- a/libs/hwcodec/README.md +++ b/libs/hwcodec/README.md @@ -41,6 +41,22 @@ Based on the information above, there are several optimizations and changes made * remove hevc_vaapi because of possible poor quality * amf: not tested, https://github.com/GPUOpen-LibrariesAndSDKs/AMF/issues/378 +#### Owned NV12 input + +The `bytes` feature provides `Encoder::encode_owned_bytes` for packed NV12 +input to FFmpeg V4L2 M2M encoders. It holds a `Bytes` owner in a read-only +`AVBufferRef` until FFmpeg releases its last reference, instead of copying the +input into an intermediate AVFrame. Other encoders retain the existing copy +path. Converted or stride-compacted One-KVM input also retains that path. + +`ONE_KVM_V4L2_OWNED_INPUT=0` disables this optimization when the encoder is +created, allowing comparisons with the same binary. `ONE_KVM_VIDEO_STATS=1` +enables preparation, send-frame, receive-packet, callback and output-copy +timings. One-KVM reports them in `[VideoInputPerf]` at five-second intervals; +`copied_bytes` counts only the intermediate raw-input copy, not FFmpeg's +remaining copy into V4L2 MMAP buffers. Encoded packet detachment remains in +place to avoid retaining scarce V4L2 capture buffers in downstream caches. + ## System requirements * intel diff --git a/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp b/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp index 19ffd84b..856b751a 100644 --- a/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp +++ b/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp @@ -5,6 +5,7 @@ extern "C" { #include #include #include +#include } #include @@ -14,6 +15,8 @@ extern "C" { #include #include "common.h" +#include "ffmpeg_ram_encode_stats.h" +#include "owned_nv12_frame.h" #define LOG_MODULE "FFMPEG_RAM_ENC" #include @@ -146,6 +149,8 @@ public: RamEncodePacketCallback packet_callback_ = NULL; int offset_[AV_NUM_DATA_POINTERS] = {0}; bool force_keyframe_ = false; // Force next frame to be a keyframe + bool measure_ = false; + RamEncodeStats stats_ = {}; AVHWDeviceType hw_device_type_ = AV_HWDEVICE_TYPE_NONE; AVPixelFormat hw_pixfmt_ = AV_PIX_FMT_NONE; @@ -170,6 +175,8 @@ public: thread_count_ = thread_count; gpu_ = gpu; callback_ = callback; + const char *stats_setting = getenv("ONE_KVM_VIDEO_STATS"); + measure_ = stats_setting && strcmp(stats_setting, "1") == 0; if (name_.find("vaapi") != std::string::npos) { hw_device_type_ = AV_HWDEVICE_TYPE_VAAPI; hw_pixfmt_ = AV_PIX_FMT_VAAPI; @@ -298,6 +305,7 @@ public: int encode(const uint8_t *data, int length, const void *obj, uint64_t ms) { int ret; + const int64_t prepare_start = timer_start(); if ((ret = av_frame_make_writable(frame_)) != 0) { LOG_ERROR(std::string("av_frame_make_writable failed, ret = ") + av_err2str(ret)); @@ -316,9 +324,42 @@ public: tmp_frame = frame_; } + timer_add(stats_.prepare_us, prepare_start); + if (measure_) { + stats_.input_frames++; + stats_.copied_frames++; + const int bytes = av_image_get_buffer_size(pixfmt_, width_, height_, 1); + if (bytes > 0) + stats_.copied_bytes += bytes; + } + return do_encode(tmp_frame, obj, ms); } + int encode_owned_packet(const uint8_t *data, int length, void *owner, + RamInputRelease release, const void *obj, + uint64_t ms, RamEncodePacketCallback callback) { + if (pixfmt_ != AV_PIX_FMT_NV12 || hw_device_type_ != AV_HWDEVICE_TYPE_NONE || + name_.find("v4l2m2m") == std::string::npos) { + release(owner, const_cast(data)); + return AVERROR(ENOSYS); + } + const int64_t prepare_start = timer_start(); + auto frame = make_owned_nv12_frame(data, length, width_, height_, owner, + release); + timer_add(stats_.prepare_us, prepare_start); + if (!frame) + return AVERROR(EINVAL); + if (measure_) { + stats_.input_frames++; + stats_.borrowed_frames++; + } + packet_callback_ = callback; + int ret = do_encode(frame.get(), obj, ms); + packet_callback_ = NULL; + return ret; + } + int encode_packet(const uint8_t *data, int length, const void *obj, uint64_t ms, RamEncodePacketCallback callback) { packet_callback_ = callback; @@ -345,6 +386,20 @@ public: } private: + int64_t timer_start() const { return measure_ ? av_gettime_relative() : 0; } + + void timer_add(uint64_t &elapsed, int64_t start) const { + if (measure_) + elapsed += av_gettime_relative() - start; + } + + int send_frame(AVFrame *frame) { + const int64_t start = timer_start(); + const int ret = avcodec_send_frame(c_, frame); + timer_add(stats_.send_us, start); + return ret; + } + int set_hwframe_ctx() { AVBufferRef *hw_frames_ref; AVHWFramesContext *frames_ctx = NULL; @@ -386,13 +441,13 @@ private: } force_keyframe_ = false; - ret = avcodec_send_frame(c_, frame); + ret = send_frame(frame); if (ret == AVERROR(EAGAIN)) { int drain_ret = receive_available_packets(obj, encoded); if (drain_ret < 0) { return drain_ret; } - ret = avcodec_send_frame(c_, frame); + ret = send_frame(frame); } if (ret == AVERROR(EAGAIN)) { return encoded ? 0 : AVERROR(EAGAIN); @@ -417,7 +472,9 @@ private: auto start = util::now(); while (util::elapsed_ms(start) < DECODE_TIMEOUT_MS) { + const int64_t receive_start = timer_start(); ret = avcodec_receive_packet(c_, pkt_); + timer_add(stats_.receive_us, receive_start); if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF) { return 0; } @@ -432,6 +489,7 @@ private: continue; } encoded = true; + const int64_t packet_start = timer_start(); if (packet_callback_) { AVPacket *owned_pkt = av_packet_clone(pkt_); if (!owned_pkt) { @@ -447,6 +505,7 @@ private: pkt_->flags & AV_PKT_FLAG_KEY, obj); } av_packet_unref(pkt_); + timer_add(stats_.packet_us, packet_start); } return 0; @@ -704,6 +763,31 @@ extern "C" void ffmpeg_ram_free_packet(void *packet) { } } +extern "C" int ffmpeg_ram_encode_owned_packet( + FFmpegRamEncoder *encoder, const uint8_t *data, int length, void *owner, + RamInputRelease release, const void *obj, uint64_t ms, + RamEncodePacketCallback callback) { + if (!encoder || !release) { + if (release) + release(owner, const_cast(data)); + return AVERROR(EINVAL); + } + try { + return encoder->encode_owned_packet(data, length, owner, release, obj, ms, + callback); + } catch (const std::exception &e) { + encoder->packet_callback_ = NULL; + LOG_ERROR(std::string("encode_owned_packet failed, ") + e.what()); + return -1; + } +} + +extern "C" void ffmpeg_ram_encoder_stats(FFmpegRamEncoder *encoder, + RamEncodeStats *stats) { + if (encoder && stats) + *stats = encoder->stats_; +} + extern "C" int ffmpeg_ram_set_bitrate(FFmpegRamEncoder *encoder, int kbs) { try { return encoder->set_bitrate(kbs); diff --git a/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode_stats.h b/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode_stats.h new file mode 100644 index 00000000..231a195b --- /dev/null +++ b/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode_stats.h @@ -0,0 +1,17 @@ +#ifndef FFMPEG_RAM_ENCODE_STATS_H +#define FFMPEG_RAM_ENCODE_STATS_H + +#include + +typedef struct RamEncodeStats { + uint64_t input_frames; + uint64_t borrowed_frames; + uint64_t copied_frames; + uint64_t copied_bytes; + uint64_t prepare_us; + uint64_t send_us; + uint64_t receive_us; + uint64_t packet_us; +} RamEncodeStats; + +#endif diff --git a/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_ffi.h b/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_ffi.h index b30de54c..516ca69c 100644 --- a/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_ffi.h +++ b/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_ffi.h @@ -2,6 +2,7 @@ #define FFMPEG_RAM_FFI_H #include +#include "ffmpeg_ram_encode_stats.h" #define AV_NUM_DATA_POINTERS 8 @@ -24,6 +25,12 @@ int ffmpeg_ram_encode(void *encoder, const uint8_t *data, int length, int ffmpeg_ram_encode_packet(void *encoder, const uint8_t *data, int length, const void *obj, int64_t ms, RamEncodePacketCallback callback); +int ffmpeg_ram_encode_owned_packet(void *encoder, const uint8_t *data, + int length, void *owner, + void (*release)(void *, uint8_t *), + const void *obj, int64_t ms, + RamEncodePacketCallback callback); +void ffmpeg_ram_encoder_stats(void *encoder, RamEncodeStats *stats); void ffmpeg_ram_free_encoder(void *encoder); void ffmpeg_ram_free_packet(void *packet); int ffmpeg_ram_get_linesize_offset_length(int pix_fmt, int width, int height, diff --git a/libs/hwcodec/cpp/ffmpeg_ram/owned_nv12_frame.h b/libs/hwcodec/cpp/ffmpeg_ram/owned_nv12_frame.h new file mode 100644 index 00000000..2b88b121 --- /dev/null +++ b/libs/hwcodec/cpp/ffmpeg_ram/owned_nv12_frame.h @@ -0,0 +1,51 @@ +#ifndef OWNED_NV12_FRAME_H +#define OWNED_NV12_FRAME_H + +extern "C" { +#include +} + +#include +#include + +using RamInputRelease = void (*)(void *, uint8_t *); + +struct RamFrameDeleter { + void operator()(AVFrame *frame) const { av_frame_free(&frame); } +}; + +using RamOwnedFrame = std::unique_ptr; + +static RamOwnedFrame make_owned_nv12_frame(const uint8_t *data, int length, + int width, int height, void *owner, + RamInputRelease release) { + const int64_t pixels = static_cast(width) * height; + if (!data || !release || width <= 0 || height <= 0 || width % 2 || + height % 2 || pixels > std::numeric_limits::max() / 3 * 2 || + length < pixels + pixels / 2) { + if (release) + release(owner, const_cast(data)); + return RamOwnedFrame(); + } + RamOwnedFrame frame(av_frame_alloc()); + if (!frame) { + release(owner, const_cast(data)); + return frame; + } + frame->buf[0] = av_buffer_create(const_cast(data), length, release, + owner, AV_BUFFER_FLAG_READONLY); + if (!frame->buf[0]) { + release(owner, const_cast(data)); + return RamOwnedFrame(); + } + frame->format = AV_PIX_FMT_NV12; + frame->width = width; + frame->height = height; + frame->data[0] = frame->buf[0]->data; + frame->data[1] = frame->data[0] + pixels; + frame->linesize[0] = width; + frame->linesize[1] = width; + return frame; +} + +#endif diff --git a/libs/hwcodec/src/ffmpeg_ram/encode.rs b/libs/hwcodec/src/ffmpeg_ram/encode.rs index 7f01dff0..6daf6458 100644 --- a/libs/hwcodec/src/ffmpeg_ram/encode.rs +++ b/libs/hwcodec/src/ffmpeg_ram/encode.rs @@ -1,5 +1,7 @@ #[cfg(feature = "bytes")] -use crate::ffmpeg_ram::{ffmpeg_ram_encode_packet, ffmpeg_ram_free_packet}; +use crate::ffmpeg_ram::{ + ffmpeg_ram_encode_owned_packet, ffmpeg_ram_encode_packet, ffmpeg_ram_free_packet, +}; use crate::{ common::DataFormat::{self, *}, ffmpeg::{init_av_log, AVPixelFormat}, @@ -498,6 +500,23 @@ pub struct Encoder { pub linesize: Vec, pub offset: Vec, pub length: i32, + owned_nv12_input: bool, + measure: bool, + #[cfg(feature = "bytes")] + output_copy_us: u64, +} + +#[derive(Clone, Copy, Debug, Default)] +pub struct EncodeInputStats { + pub input_frames: u64, + pub borrowed_frames: u64, + pub copied_frames: u64, + pub copied_bytes: u64, + pub prepare_us: u64, + pub send_us: u64, + pub receive_us: u64, + pub packet_us: u64, + pub output_copy_us: u64, } impl Encoder { @@ -545,6 +564,9 @@ impl Encoder { return Err(()); } + let owned_nv12_input = ctx.name.ends_with("_v4l2m2m") + && ctx.pixfmt == AVPixelFormat::AV_PIX_FMT_NV12 as i32 + && std::env::var("ONE_KVM_V4L2_OWNED_INPUT").as_deref() != Ok("0"); Ok(Encoder { codec, frames: Box::into_raw(Box::new(Vec::::new())), @@ -552,6 +574,10 @@ impl Encoder { linesize, offset, length: length[0], + owned_nv12_input, + measure: std::env::var("ONE_KVM_VIDEO_STATS").as_deref() == Ok("1"), + #[cfg(feature = "bytes")] + output_copy_us: 0, }) } } @@ -581,33 +607,107 @@ impl Encoder { #[cfg(feature = "bytes")] pub fn encode_bytes(&mut self, data: &[u8], ms: i64) -> Result, i32> { + let length = c_int::try_from(data.len()).map_err(|_| -22)?; unsafe { let mut frames = Vec::::new(); let result = ffmpeg_ram_encode_packet( self.codec, data.as_ptr(), - data.len() as _, + length, &mut frames as *mut _ as *const c_void, ms, Some(Encoder::packet_callback), ); - if result == -11 || result == 0 { - if self.ctx.name.contains("v4l2m2m") { - return Ok(frames - .into_iter() - .map(|frame| EncodeBytesFrame { - data: Bytes::copy_from_slice(frame.data.as_ref()), - pts: frame.pts, - key: frame.key, - }) - .collect()); - } - return Ok(frames); - } - Err(result) + self.finish_bytes_encode(frames, result) } } + pub fn supports_owned_nv12_input(&self) -> bool { + self.owned_nv12_input + } + + pub fn input_stats(&self) -> Option { + if !self.measure { + return None; + } + let mut stats = unsafe { std::mem::zeroed() }; + unsafe { crate::ffmpeg_ram::ffmpeg_ram_encoder_stats(self.codec, &mut stats) }; + Some(EncodeInputStats { + input_frames: stats.input_frames, + borrowed_frames: stats.borrowed_frames, + copied_frames: stats.copied_frames, + copied_bytes: stats.copied_bytes, + prepare_us: stats.prepare_us, + send_us: stats.send_us, + receive_us: stats.receive_us, + packet_us: stats.packet_us, + #[cfg(feature = "bytes")] + output_copy_us: self.output_copy_us, + #[cfg(not(feature = "bytes"))] + output_copy_us: 0, + }) + } + + #[cfg(feature = "bytes")] + pub fn encode_owned_bytes( + &mut self, + data: Bytes, + ms: i64, + ) -> Result, i32> { + if !self.owned_nv12_input { + return self.encode_bytes(data.as_ref(), ms); + } + let length = c_int::try_from(data.len()).map_err(|_| -22)?; + let owner = Box::new(data); + let pointer = owner.as_ptr(); + let mut frames = Vec::::new(); + let result = unsafe { + ffmpeg_ram_encode_owned_packet( + self.codec, + pointer, + length, + Box::into_raw(owner) as *mut c_void, + Some(Self::release_input), + &mut frames as *mut _ as *const c_void, + ms, + Some(Self::packet_callback), + ) + }; + self.finish_bytes_encode(frames, result) + } + + #[cfg(feature = "bytes")] + fn finish_bytes_encode( + &mut self, + frames: Vec, + result: i32, + ) -> Result, i32> { + if result != -11 && result != 0 { + return Err(result); + } + if !self.ctx.name.contains("v4l2m2m") { + return Ok(frames); + } + let start = self.measure.then(std::time::Instant::now); + let detached = frames + .into_iter() + .map(|frame| EncodeBytesFrame { + data: Bytes::copy_from_slice(frame.data.as_ref()), + pts: frame.pts, + key: frame.key, + }) + .collect(); + if let Some(start) = start { + self.output_copy_us += start.elapsed().as_micros() as u64; + } + Ok(detached) + } + + #[cfg(feature = "bytes")] + extern "C" fn release_input(owner: *mut c_void, _data: *mut u8) { + unsafe { drop(Box::from_raw(owner as *mut Bytes)) }; + } + extern "C" fn callback(data: *const u8, size: c_int, pts: i64, key: i32, obj: *const c_void) { unsafe { let frames = &mut *(obj as *mut Vec); diff --git a/src/video/codec/h264.rs b/src/video/codec/h264.rs index 8b0ac59a..7465c4a7 100644 --- a/src/video/codec/h264.rs +++ b/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 { + self.inner.input_stats() + } + + pub fn encode_owned_nv12(&mut self, data: Bytes, pts_ms: i64) -> Result> { + 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, i32>, + ) -> Result> { + match result { Ok(frames) => { let owned_frames: Vec = 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); diff --git a/src/video/frame.rs b/src/video/frame.rs index 9d0b43b0..404ffcdb 100644 --- a/src/video/frame.rs +++ b/src/video/frame.rs @@ -14,6 +14,14 @@ enum FrameData { Pooled(Arc), } +struct PooledFrameOwner(Arc); + +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]); + } +} diff --git a/src/video/pipeline/encoder_state.rs b/src/video/pipeline/encoder_state.rs index 346365b8..579665c7 100644 --- a/src/video/pipeline/encoder_state.rs +++ b/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>; + fn supports_owned_nv12_input(&self) -> bool { + false + } + fn encode_owned_nv12(&mut self, data: Bytes, pts_ms: i64) -> Result> { + self.encode_raw(data.as_ref(), pts_ms) + } + fn input_stats(&self) -> Option { + 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> { + 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 { + 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))) } diff --git a/src/video/pipeline/frame_mailbox.rs b/src/video/pipeline/frame_mailbox.rs index 2e0fa2c8..02316c1f 100644 --- a/src/video/pipeline/frame_mailbox.rs +++ b/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 FrameMailbox { self.receive_when(|| Some(())).map(|(frame, ())| frame) } + #[cfg(test)] pub(super) fn receive_when( &self, - mut ready: impl FnMut() -> Option, + ready: impl FnMut() -> Option, ) -> Option<(Frame, Ready)> { + self.receive_when_timed(ready, false) + .map(|(frame, ready, _)| (frame, ready)) + } + + pub(super) fn receive_when_timed( + &self, + mut ready: impl FnMut() -> Option, + 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 FrameMailbox { } 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(); + } } diff --git a/src/video/pipeline/mjpeg_admission.rs b/src/video/pipeline/mjpeg_admission.rs new file mode 100644 index 00000000..b274473d --- /dev/null +++ b/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 { + 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) + ); + } + } +} diff --git a/src/video/pipeline/mod.rs b/src/video/pipeline/mod.rs index cbd0f99d..27cf8f96 100644 --- a/src/video/pipeline/mod.rs +++ b/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; diff --git a/src/video/pipeline/shared.rs b/src/video/pipeline/shared.rs index 75b5f96e..92cd33c4 100644 --- a/src/video/pipeline/shared.rs +++ b/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, buffer_pool: &Arc, resolution: Resolution, + admission_policy: MjpegAdmissionPolicy, ) -> Vec> { 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 = 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 = (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};