GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-10-27 16:00:49 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor(video): 移除性能实验开关和逐帧统计代码

Browse Source
This commit is contained in:
mofeng-git
2026-10-01 23:05:24 +08:00
parent 9a1f4ff4b5
commit 8b4ca6d080
13 changed files with 25 additions and 563 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
libs/hwcodec/README.md
View File

@@ -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

41
libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode.cpp
View File

@@ -5,7 +5,6 @@ extern "C" {
#include <libavutil/imgutils.h>
#include <libavutil/log.h>
#include <libavutil/opt.h>
#include <libavutil/time.h>
}
#include <stdbool.h>
@@ -15,7 +14,6 @@ extern "C" {
#include <string>
#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<uint8_t *>(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);

17
libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_encode_stats.h
View File

@@ -1,17 +0,0 @@
#ifndef FFMPEG_RAM_ENCODE_STATS_H
#define FFMPEG_RAM_ENCODE_STATS_H
#include <stdint.h>
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

2
libs/hwcodec/cpp/ffmpeg_ram/ffmpeg_ram_ffi.h
View File

@@ -2,7 +2,6 @@
#define FFMPEG_RAM_FFI_H
#include <stdint.h>
#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,

48
libs/hwcodec/src/ffmpeg_ram/encode.rs
View File

@@ -501,22 +501,6 @@ pub struct Encoder {
pub offset: Vec<i32>,
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::<EncodeFrame>::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<EncodeInputStats> {
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)
}

4
src/video/codec/h264.rs
View File

@@ -378,10 +378,6 @@ impl H264Encoder {
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);

7
src/video/pipeline/delivery.rs
View File

@@ -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<Mutex<Option<Arc<FrameMailbox<Arc<Video
#[derive(Default)]
pub(super) struct DeliveryStats {
pub sent_frames: AtomicU64,
pub send_time_ns: AtomicU64,
pub send_errors: AtomicU64,
}
@@ -80,10 +78,7 @@ impl EncodedVideoFrameReceiver {
}
}
pub(crate) fn record_send(&self, elapsed: Duration, succeeded: bool) {
self.stats
.send_time_ns
.fetch_add(elapsed.as_nanos() as u64, Ordering::Relaxed);
pub(crate) fn record_send(&self, succeeded: bool) {
if succeeded {
self.stats.sent_frames.fetch_add(1, Ordering::Relaxed);
} else {

7
src/video/pipeline/encoder_state.rs
View File

@@ -39,9 +39,6 @@ pub(super) trait VideoEncoderTrait: Send {
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);
@@ -93,10 +90,6 @@ impl VideoEncoderTrait for H264EncoderWrapper {
.collect())
}
fn input_stats(&self) -> Option<hwcodec::ffmpeg_ram::encode::EncodeInputStats> {
self.0.input_stats()
}
fn codec_name(&self) -> &str {
self.0.codec_name()
}

125
src/video/pipeline/frame_mailbox.rs
View File

@@ -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<Frame> FrameMailbox<Frame> {
self.receive_when(|| Some(())).map(|(frame, ())| frame)
}
#[cfg(test)]
pub(super) fn receive_when<Ready>(
&self,
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();
) -> Option<(Frame, Ready)> {
let mut state = self.state.lock().expect("frame mailbox lock poisoned");
loop {
if state.closed {
@@ -89,18 +61,13 @@ impl<Frame> FrameMailbox<Frame> {
}
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

22
src/video/pipeline/mjpeg_admission.rs
View File

@@ -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] {

195
src/video/pipeline/shared.rs
View File

@@ -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<String, u64> = 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<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};
@@ -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]

3
src/webrtc/universal_session.rs
View File

@@ -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;

107
src/webrtc/video_track.rs
View File

@@ -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<Vec<u8>>,
pps: Option<Vec<u8>>,
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<Mutex<H265RtpState>>,
h264_state: Mutex<H264TrackState>,
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<i64>) -> 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;