From 8b4ca6d080af23d46e82fd1b440ad5df4295d768 Mon Sep 17 00:00:00 2001 From: mofeng-git Date: Thu, 1 Oct 2026 23:05:24 +0800 Subject: [PATCH] =?UTF-8?q?refactor(video):=20=E7=A7=BB=E9=99=A4=E6=80=A7?= =?UTF-8?q?=E8=83=BD=E5=AE=9E=E9=AA=8C=E5=BC=80=E5=85=B3=E5=92=8C=E9=80=90?= =?UTF-8?q?=E5=B8=A7=E7=BB=9F=E8=AE=A1=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- libs/hwcodec/README.md | 10 +- .../cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp | 41 ---- .../cpp/ffmpeg_ram/ffmpeg_ram_encode_stats.h | 17 -- libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_ffi.h | 2 - libs/hwcodec/src/ffmpeg_ram/encode.rs | 48 +---- src/video/codec/h264.rs | 4 - src/video/pipeline/delivery.rs | 7 +- src/video/pipeline/encoder_state.rs | 7 - src/video/pipeline/frame_mailbox.rs | 125 +---------- src/video/pipeline/mjpeg_admission.rs | 22 -- src/video/pipeline/shared.rs | 195 +----------------- src/webrtc/universal_session.rs | 3 +- src/webrtc/video_track.rs | 107 +--------- 13 files changed, 25 insertions(+), 563 deletions(-) delete mode 100644 libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode_stats.h diff --git a/libs/hwcodec/README.md b/libs/hwcodec/README.md index f9f899f6..4b87e5ba 100644 --- a/libs/hwcodec/README.md +++ b/libs/hwcodec/README.md @@ -49,13 +49,9 @@ input to FFmpeg V4L2 M2M encoders. It holds a `Bytes` owner in a read-only 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. +Encoded packet detachment remains in place to avoid retaining scarce V4L2 +capture buffers in downstream caches. FFmpeg still copies raw input into its +V4L2 MMAP buffers. ## System requirements diff --git a/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp b/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp index 856b751a..fb821554 100644 --- a/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp +++ b/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp @@ -5,7 +5,6 @@ extern "C" { #include #include #include -#include } #include @@ -15,7 +14,6 @@ extern "C" { #include #include "common.h" -#include "ffmpeg_ram_encode_stats.h" #include "owned_nv12_frame.h" #define LOG_MODULE "FFMPEG_RAM_ENC" @@ -149,8 +147,6 @@ 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; @@ -175,8 +171,6 @@ 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; @@ -305,7 +299,6 @@ 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)); @@ -324,15 +317,6 @@ 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); } @@ -344,16 +328,10 @@ public: 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; @@ -386,17 +364,8 @@ 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; } @@ -472,9 +441,7 @@ 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; } @@ -489,7 +456,6 @@ private: continue; } encoded = true; - const int64_t packet_start = timer_start(); if (packet_callback_) { AVPacket *owned_pkt = av_packet_clone(pkt_); if (!owned_pkt) { @@ -505,7 +471,6 @@ private: pkt_->flags & AV_PKT_FLAG_KEY, obj); } av_packet_unref(pkt_); - timer_add(stats_.packet_us, packet_start); } return 0; @@ -782,12 +747,6 @@ extern "C" int ffmpeg_ram_encode_owned_packet( } } -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 deleted file mode 100644 index 231a195b..00000000 --- a/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode_stats.h +++ /dev/null @@ -1,17 +0,0 @@ -#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 516ca69c..831fefbf 100644 --- a/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_ffi.h +++ b/libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_ffi.h @@ -2,7 +2,6 @@ #define FFMPEG_RAM_FFI_H #include -#include "ffmpeg_ram_encode_stats.h" #define AV_NUM_DATA_POINTERS 8 @@ -30,7 +29,6 @@ int ffmpeg_ram_encode_owned_packet(void *encoder, const uint8_t *data, 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/src/ffmpeg_ram/encode.rs b/libs/hwcodec/src/ffmpeg_ram/encode.rs index 6daf6458..a788cbd4 100644 --- a/libs/hwcodec/src/ffmpeg_ram/encode.rs +++ b/libs/hwcodec/src/ffmpeg_ram/encode.rs @@ -501,22 +501,6 @@ pub struct Encoder { 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 { @@ -565,8 +549,7 @@ impl Encoder { } 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"); + && ctx.pixfmt == AVPixelFormat::AV_PIX_FMT_NV12 as i32; Ok(Encoder { codec, frames: Box::into_raw(Box::new(Vec::::new())), @@ -575,9 +558,6 @@ impl Encoder { 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, }) } } @@ -626,28 +606,6 @@ impl Encoder { 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, @@ -688,7 +646,6 @@ impl Encoder { 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 { @@ -697,9 +654,6 @@ impl Encoder { key: frame.key, }) .collect(); - if let Some(start) = start { - self.output_copy_us += start.elapsed().as_micros() as u64; - } Ok(detached) } diff --git a/src/video/codec/h264.rs b/src/video/codec/h264.rs index 7465c4a7..e66b8919 100644 --- a/src/video/codec/h264.rs +++ b/src/video/codec/h264.rs @@ -378,10 +378,6 @@ impl H264Encoder { 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); diff --git a/src/video/pipeline/delivery.rs b/src/video/pipeline/delivery.rs index 7296a8a6..3859c78d 100644 --- a/src/video/pipeline/delivery.rs +++ b/src/video/pipeline/delivery.rs @@ -1,7 +1,6 @@ use std::collections::VecDeque; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; -use std::time::Duration; use parking_lot::Mutex; use tokio::sync::mpsc; @@ -16,7 +15,6 @@ pub(super) type FrameMailboxHandle = Arc 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); @@ -93,10 +90,6 @@ impl VideoEncoderTrait for H264EncoderWrapper { .collect()) } - fn input_stats(&self) -> Option { - self.0.input_stats() - } - fn codec_name(&self) -> &str { self.0.codec_name() } diff --git a/src/video/pipeline/frame_mailbox.rs b/src/video/pipeline/frame_mailbox.rs index 02316c1f..15bc05ec 100644 --- a/src/video/pipeline/frame_mailbox.rs +++ b/src/video/pipeline/frame_mailbox.rs @@ -1,21 +1,4 @@ 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 { @@ -67,21 +50,10 @@ impl FrameMailbox { self.receive_when(|| Some(())).map(|(frame, ())| frame) } - #[cfg(test)] pub(super) fn receive_when( - &self, - 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(); + ) -> Option<(Frame, Ready)> { let mut state = self.state.lock().expect("frame mailbox lock poisoned"); loop { if state.closed { @@ -89,18 +61,13 @@ impl FrameMailbox { } if state.pending.is_some() { if let Some(ready) = ready() { - return state.pending.take().map(|frame| (frame, ready, waits)); + return state.pending.take().map(|frame| (frame, ready)); } } - 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()); - } } } @@ -210,76 +177,7 @@ mod tests { } #[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() { + fn receiver_exits_when_closed_without_output_capacity() { let mailbox = Arc::new(FrameMailbox::new()); mailbox.publish(1, "pending"); let receiver = mailbox.clone(); @@ -287,16 +185,13 @@ mod tests { 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, - ); + let result = receiver.receive_when(|| { + if !reported { + waiting_sender.send(()).unwrap(); + reported = true; + } + None::<()> + }); result_sender.send(result).unwrap(); }); waiting_receiver diff --git a/src/video/pipeline/mjpeg_admission.rs b/src/video/pipeline/mjpeg_admission.rs index b274473d..9e71bd7e 100644 --- a/src/video/pipeline/mjpeg_admission.rs +++ b/src/video/pipeline/mjpeg_admission.rs @@ -20,14 +20,6 @@ pub(super) enum MjpegDropReason { } 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 } @@ -54,20 +46,6 @@ impl MjpegAdmissionPolicy { 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] { diff --git a/src/video/pipeline/shared.rs b/src/video/pipeline/shared.rs index 25a10fc9..31e6c75b 100644 --- a/src/video/pipeline/shared.rs +++ b/src/video/pipeline/shared.rs @@ -223,7 +223,6 @@ fn spawn_mjpeg_decode_workers( } let nv12_size = resolution.width as usize * resolution.height as usize * 3 / 2; let mut nv12 = worker_buffer_pool.take(nv12_size); - let decode_started = Instant::now(); let decode_result = decoder.decode_into(&job.data, &mut nv12); worker_buffer_pool.put(job.data); @@ -232,10 +231,7 @@ fn spawn_mjpeg_decode_workers( warn!("Dropping undecodable MJPEG frame: {}", error); continue; } - worker_pipeline.mjpeg_decode_time_ns.fetch_add( - decode_started.elapsed().as_nanos() as u64, - Ordering::Relaxed, - ); + worker_pipeline .mjpeg_decoded_frames .fetch_add(1, Ordering::Relaxed); @@ -496,94 +492,6 @@ 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); - let encoded_inputs = delta(stats.encoded_input_frames, previous.encoded_input_frames); - let encoded = delta(stats.encoded_frames, previous.encoded_frames); - let sent = delta(stats.sent_frames, previous.sent_frames); - let send_errors = delta(stats.send_errors, previous.send_errors); - let average_ms = |time_ns: u64, count: u64| { - if count == 0 { - 0.0 - } else { - time_ns as f64 / count as f64 / 1_000_000.0 - } - }; - info!( - "[VideoPerf] window_s={:.2} decoded_fps={:.2} encoded_fps={:.2} sent_fps={:.2} decode_ms={:.2} encode_ms={:.2} send_ms={:.2} encoded_mbps={:.2} dropped_before_decode={} decoded_unused={} backpressure_events={} slow_subscriber_skips={} send_errors={}", - elapsed, - decoded as f64 / elapsed, - encoded as f64 / elapsed, - sent as f64 / elapsed, - average_ms(delta(stats.mjpeg_decode_time_ns, previous.mjpeg_decode_time_ns), decoded), - average_ms(delta(stats.encode_time_ns, previous.encode_time_ns), encoded_inputs), - average_ms(delta(stats.send_time_ns, previous.send_time_ns), sent + send_errors), - delta(stats.encoded_bytes, previous.encoded_bytes) as f64 * 8.0 / elapsed / 1_000_000.0, - delta(stats.mjpeg_dropped_before_decode, previous.mjpeg_dropped_before_decode), - delta(stats.decoded_frames_not_encoded, previous.decoded_frames_not_encoded), - delta(stats.output_backpressure_events, previous.output_backpressure_events), - 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 #[derive(Debug, Clone, Default)] pub struct SharedVideoPipelineStats { @@ -596,17 +504,12 @@ pub struct SharedVideoPipelineStats { 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, - pub send_time_ns: u64, pub send_errors: u64, } @@ -633,13 +536,9 @@ pub struct SharedVideoPipeline { 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, output_backpressure_since_ns: AtomicU64, @@ -701,13 +600,9 @@ impl SharedVideoPipeline { 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), output_backpressure_since_ns: AtomicU64::new(0), @@ -881,19 +776,14 @@ impl SharedVideoPipeline { .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 .load(Ordering::Relaxed); stats.sent_frames = self.delivery_stats.sent_frames.load(Ordering::Relaxed); - stats.send_time_ns = self.delivery_stats.send_time_ns.load(Ordering::Relaxed); stats.send_errors = self.delivery_stats.send_errors.load(Ordering::Relaxed); stats } @@ -1291,9 +1181,7 @@ 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(), - ); + let admission_policy = MjpegAdmissionPolicy::BoundedPrefetch; if parallel_mjpeg_decode { encoder_config.input_format = PixelFormat::Nv12; info!("Using parallel libyuv MJPEG decode with hardware encoding"); @@ -1314,13 +1202,9 @@ impl SharedVideoPipeline { .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.output_backpressure_since_ns @@ -1328,7 +1212,6 @@ impl SharedVideoPipeline { self.encoded_frames_skipped_for_slow_subscribers .store(0, Ordering::Relaxed); self.delivery_stats.sent_frames.store(0, Ordering::Relaxed); - self.delivery_stats.send_time_ns.store(0, Ordering::Relaxed); self.delivery_stats.send_errors.store(0, Ordering::Relaxed); for subscriber in self.subscribers.read().iter() { subscriber.needs_keyframe.store(true, Ordering::Release); @@ -1357,17 +1240,12 @@ impl SharedVideoPipeline { let mut encoded_frame_count: u64 = 0; let mut last_fps_time = Instant::now(); let mut fps_frame_count: u64 = 0; - 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, waits)) = frame_mailbox - .receive_when_timed(|| pipeline.reserve_outputs(), perf_logging) + let Some((frame, reservations)) = + frame_mailbox.receive_when(|| pipeline.reserve_outputs()) else { break; }; @@ -1383,15 +1261,6 @@ 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) { error!("Failed to apply pipeline command: {}", e); @@ -1402,15 +1271,7 @@ impl SharedVideoPipeline { pipeline .encoded_input_frames .fetch_add(1, Ordering::Relaxed); - let encode_started = Instant::now(); let encode_result = pipeline.encode_frame_sync(&mut encoder_state, &frame); - 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) => { @@ -1449,28 +1310,6 @@ impl SharedVideoPipeline { pipeline.decoded_frames_not_encoded.load(Ordering::Relaxed), ); } - - 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(); - 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(); - } } pipeline.clear_cmd_tx(); @@ -1910,7 +1749,6 @@ impl SharedVideoPipeline { let nv12_size = resolution.width as usize * resolution.height as usize * 3 / 2; let mut nv12 = buffer_pool.take(nv12_size); - let decode_started = Instant::now(); if let Err(error) = decoder.decode_into(&owned, &mut nv12) { buffer_pool.put(owned); buffer_pool.put(nv12); @@ -1921,10 +1759,7 @@ impl SharedVideoPipeline { continue; } buffer_pool.put(owned); - pipeline.mjpeg_decode_time_ns.fetch_add( - decode_started.elapsed().as_nanos() as u64, - Ordering::Relaxed, - ); + pipeline .mjpeg_decoded_frames .fetch_add(1, Ordering::Relaxed); @@ -2582,10 +2417,6 @@ mod tests { ] { 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); @@ -2593,8 +2424,6 @@ mod tests { 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] @@ -2661,15 +2490,6 @@ mod tests { 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}; @@ -3003,12 +2823,11 @@ mod tests { fn send_statistics_include_successes_errors_and_duration() { let pipeline = delivery_test_pipeline(); let receiver = pipeline.subscribe(); - receiver.record_send(Duration::from_micros(100), true); - receiver.record_send(Duration::from_micros(200), false); + receiver.record_send(true); + receiver.record_send(false); let stats = pipeline.stats_snapshot(); assert_eq!(stats.sent_frames, 1); assert_eq!(stats.send_errors, 1); - assert_eq!(stats.send_time_ns, 300_000); } #[test] diff --git a/src/webrtc/universal_session.rs b/src/webrtc/universal_session.rs index 656aed20..e123bc4d 100644 --- a/src/webrtc/universal_session.rs +++ b/src/webrtc/universal_session.rs @@ -669,7 +669,6 @@ impl UniversalSession { } } - let send_started = Instant::now(); let send_result = video_track .write_frame_bytes_at( encoded_frame.data.clone(), @@ -677,7 +676,7 @@ impl UniversalSession { Some(encoded_frame.pts_ms), ) .await; - frame_rx.record_send(send_started.elapsed(), send_result.is_ok()); + frame_rx.record_send(send_result.is_ok()); match send_result { Ok(()) => { frames_sent += 1; diff --git a/src/webrtc/video_track.rs b/src/webrtc/video_track.rs index 4367e616..55bad84e 100644 --- a/src/webrtc/video_track.rs +++ b/src/webrtc/video_track.rs @@ -4,9 +4,9 @@ use bytes::Bytes; use rtp::codecs::h264::H264Payloader; use rtp::packetizer::Payloader; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::Duration; use tokio::sync::Mutex; -use tracing::{debug, info, trace, warn}; +use tracing::{debug, trace, warn}; use webrtc::media::Sample; use webrtc::rtp_transceiver::rtp_codec::RTCRtpCodecCapability; use webrtc::track::track_local::track_local_static_rtp::TrackLocalStaticRTP; @@ -191,63 +191,6 @@ struct H264TrackState { clock: H264RtpClock, sps: Option>, pps: Option>, - performance: H264SendPerformance, -} - -#[derive(Default)] -struct H264SendTiming { - prepare: Duration, - lock: Duration, - packetize: Duration, - write: Duration, -} - -struct H264SendPerformance { - last_report: Instant, - frames: u64, - packets: u64, - bytes: u64, - timing: H264SendTiming, -} - -impl H264SendPerformance { - fn new() -> Self { - Self { - last_report: Instant::now(), - frames: 0, - packets: 0, - bytes: 0, - timing: H264SendTiming::default(), - } - } - - fn record(&mut self, bytes: usize, packets: usize, timing: H264SendTiming) { - self.frames += 1; - self.packets += packets as u64; - self.bytes += bytes as u64; - self.timing.prepare += timing.prepare; - self.timing.lock += timing.lock; - self.timing.packetize += timing.packetize; - self.timing.write += timing.write; - let elapsed = self.last_report.elapsed(); - if elapsed < Duration::from_secs(5) { - return; - } - let average_ms = |duration: Duration| duration.as_secs_f64() * 1000.0 / self.frames as f64; - info!( - "[VideoSendPerf] window_s={:.2} frames={} packets_per_frame={:.2} frame_kib={:.2} prepare_ms={:.3} lock_ms={:.3} packetize_ms={:.3} write_ms={:.3} write_us_per_packet={:.3}", - elapsed.as_secs_f64(), - self.frames, - self.packets as f64 / self.frames as f64, - self.bytes as f64 / self.frames as f64 / 1024.0, - average_ms(self.timing.prepare), - average_ms(self.timing.lock), - average_ms(self.timing.packetize), - average_ms(self.timing.write), - self.timing.write.as_secs_f64() * 1_000_000.0 / self.packets.max(1) as f64, - ); - *self = Self::new(); - } } pub struct UniversalVideoTrack { @@ -256,7 +199,6 @@ pub struct UniversalVideoTrack { config: UniversalVideoTrackConfig, h265_state: Option>, h264_state: Mutex, - h264_perf_enabled: bool, } impl UniversalVideoTrack { @@ -311,7 +253,6 @@ impl UniversalVideoTrack { increment: 90000 / config.fps.max(1), started: false, }, - performance: H264SendPerformance::new(), }; Self { track, @@ -319,7 +260,6 @@ impl UniversalVideoTrack { config, h265_state, h264_state: Mutex::new(h264_state), - h264_perf_enabled: std::env::var("ONE_KVM_VIDEO_STATS").as_deref() == Ok("1"), } } @@ -369,18 +309,13 @@ impl UniversalVideoTrack { /// Keep the stack's H.264 STAP/FU payloader, with explicit RTP timestamps. async fn write_h264_frame(&self, data: Bytes, pts_ms: Option) -> Result<()> { - let prepare_started = self.h264_perf_enabled.then(Instant::now); let mut data = h264_bitstream::normalize_annex_b(data); let inspection = h264_bitstream::inspect_annex_b(data.as_ref()); let idr = inspection.is_idr; let has_parameter_sets = inspection.sps.is_some() && inspection.pps.is_some(); - let prepare_time = prepare_started.map(|started| started.elapsed()); - let lock_started = self.h264_perf_enabled.then(Instant::now); let mut state = self.h264_state.lock().await; - let lock_time = lock_started.map(|started| started.elapsed()); - let packetize_started = self.h264_perf_enabled.then(Instant::now); { if let Some(sps) = inspection.sps { state.sps = Some(sps.to_vec()); @@ -415,8 +350,6 @@ impl UniversalVideoTrack { } let timestamp = state.clock.next(pts_ms); let count = payloads.len(); - let packetize_time = packetize_started.map(|started| started.elapsed()); - let write_started = self.h264_perf_enabled.then(Instant::now); // Serialize complete access units, including their sequence allocation. for (index, payload) in payloads.into_iter().enumerate() { let sequence_number = state.sequence_number; @@ -436,18 +369,6 @@ impl UniversalVideoTrack { .await .map_err(|e| AppError::WebRtcError(format!("H264 RTP write failed: {e}")))?; } - if let Some(started) = write_started { - state.performance.record( - data.len(), - count, - H264SendTiming { - prepare: prepare_time.unwrap_or_default(), - lock: lock_time.unwrap_or_default(), - packetize: packetize_time.unwrap_or_default(), - write: started.elapsed(), - }, - ); - } Ok(()) } @@ -659,30 +580,6 @@ mod tests { } } - #[test] - fn h264_send_performance_accumulates_stage_timings() { - let mut performance = H264SendPerformance::new(); - for _ in 0..2 { - performance.record( - 1200, - 2, - H264SendTiming { - prepare: Duration::from_micros(100), - lock: Duration::from_micros(20), - packetize: Duration::from_micros(200), - write: Duration::from_micros(300), - }, - ); - } - assert_eq!(performance.frames, 2); - assert_eq!(performance.bytes, 2400); - assert_eq!(performance.packets, 4); - assert_eq!(performance.timing.prepare, Duration::from_micros(200)); - assert_eq!(performance.timing.lock, Duration::from_micros(40)); - assert_eq!(performance.timing.packetize, Duration::from_micros(400)); - assert_eq!(performance.timing.write, Duration::from_micros(600)); - } - #[test] fn srtp_backend_preserves_authentication_and_sequence_rollover() { use webrtc::srtp::context::Context;