fix(stats): real encode thread timing via SwEncodeTiming + take_timing (#17)

encode_thread_loop hardcoded sws_us=0 and output_bytes=0, making the
stats dashboard show zeros for those columns. The previous encode_us
was measured around the entire encode_cpu_frame call, mixing sws +
encode work into one bucket.

Add SwEncodeTiming { sws_us, encode_us, output_bytes } to SwEncEncode.
- drain_encoder returns Result<usize> (total encoded bytes), accumulated
  before the Muxer/Channel match so branch duplication and multi-packet
  drain are both handled correctly. output_bytes = bytes produced by
  libavcodec even if downstream delivery drops them.
- encode_cpu_frame resets last_timing to Default at entry (so early
  returns from disconnect/pause/dedup never report stale prior values),
  measures sws_us around sws_scale, measures encode_us around
  avcodec_send_frame + drain, then stores the complete snapshot.
- take_timing() uses mem::take to return and clear in one step.
- flush() ignores the byte count from drain_encoder.
- state_portal encode_thread_loop calls take_timing() for real values.
This commit is contained in:
dailz
2026-06-14 08:50:47 +08:00
parent 36cee9d9dd
commit 0aba0e651e
2 changed files with 54 additions and 9 deletions
+50 -4
View File
@@ -7,6 +7,7 @@ use std::ptr;
use std::slice;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Instant;
use anyhow::{bail, Result};
use ffmpeg_next as ff;
@@ -36,6 +37,17 @@ pub struct ResolutionChange {
pub height: u32,
}
/// Per-frame timing snapshot for the software encoder, consumed by the stats
/// thread. `sws_us` measures NV12→YUV420P conversion, `encode_us` measures
/// `avcodec_send_frame` + drain, and `output_bytes` counts encoded bytes
/// produced by libavcodec (even if downstream delivery later drops them).
#[derive(Default, Clone, Copy, Debug)]
pub struct SwEncodeTiming {
pub sws_us: u64,
pub encode_us: u64,
pub output_bytes: usize,
}
// ---------------------------------------------------------------------------
// AvHwDevCtx
// ---------------------------------------------------------------------------
@@ -974,6 +986,10 @@ pub struct SwEncEncode {
/// `AV_PICTURE_TYPE_I` and bypasses the dedup hash check. Cleared only
/// after `avcodec_send_frame` accepts the forced frame.
force_keyframe_pending: bool,
/// Last per-frame timing snapshot. Reset to `Default` at the start of
/// every `encode_cpu_frame` call (even on early returns) so stale values
/// from a previous frame can never leak out.
last_timing: SwEncodeTiming,
}
const FNV1A_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
@@ -1037,6 +1053,7 @@ impl SwEncEncode {
bitrate,
gop_size,
force_keyframe_pending: false,
last_timing: SwEncodeTiming::default(),
})
}
@@ -1076,6 +1093,7 @@ impl SwEncEncode {
bitrate,
gop_size,
force_keyframe_pending: false,
last_timing: SwEncodeTiming::default(),
})
}
@@ -1089,12 +1107,18 @@ impl SwEncEncode {
}
}
let start_ts = self.starting_timestamp.unwrap_or(0);
self.drain_encoder(start_ts)?;
let _ = self.drain_encoder(start_ts)?;
Ok(())
}
pub fn take_timing(&mut self) -> SwEncodeTiming {
mem::take(&mut self.last_timing)
}
pub fn encode_cpu_frame(&mut self, frame: &CpuNv12Frame) -> Result<()> {
self.last_timing = SwEncodeTiming::default();
if self.webrtc_disconnected {
return Ok(());
}
@@ -1153,6 +1177,7 @@ impl SwEncEncode {
}
self.last_frame_hash = current_hash;
let sws_start = Instant::now();
// SAFETY: yuv_frame is an owned reusable YUV420P frame at the same dimensions as sw_nv12;
// sws_ctx was created for NV12 -> YUV420P with no resize, so sws_scale only converts format.
unsafe {
@@ -1180,6 +1205,7 @@ impl SwEncEncode {
bail!("sws_scale failed for software encoder: {scaled}");
}
}
let sws_us = sws_start.elapsed().as_micros() as u64;
let pts = frame.pts;
if self.starting_timestamp.is_none() {
@@ -1187,6 +1213,7 @@ impl SwEncEncode {
}
let start_ts = self.starting_timestamp.unwrap_or(0);
let enc_start = Instant::now();
// SAFETY: yuv_frame is initialized, writable, and matches the opened encoder format.
// pict_type is reset every frame: the AVFrame is reused, so without resetting to NONE
// a previously-forced I-type would leak into subsequent P-frames. With forced-idr=1
@@ -1211,7 +1238,16 @@ impl SwEncEncode {
self.force_keyframe_pending = false;
}
self.drain_encoder(start_ts)
let output_bytes = self.drain_encoder(start_ts)?;
let encode_us = enc_start.elapsed().as_micros() as u64;
self.last_timing = SwEncodeTiming {
sws_us,
encode_us,
output_bytes,
};
Ok(())
}
fn recreate_encoder(&mut self, width: u32, height: u32) -> Result<()> {
@@ -1258,7 +1294,8 @@ impl SwEncEncode {
Ok(())
}
fn drain_encoder(&mut self, start_ts: i64) -> Result<()> {
fn drain_encoder(&mut self, start_ts: i64) -> Result<usize> {
let mut total_bytes = 0usize;
loop {
let mut pkt = ff::Packet::empty();
// SAFETY: enc_video is an open encoder; pkt is writable packet storage.
@@ -1272,6 +1309,15 @@ impl SwEncEncode {
bail!("avcodec_receive_packet failed: {}", ff_err(ret));
}
// Count encoded bytes produced before the Muxer/Channel match to
// avoid branch duplication and handle multi-packet drain correctly.
// SAFETY: pkt was just filled by a successful avcodec_receive_packet;
// the size field is valid and initialized.
let pkt_size = unsafe { (*pkt.as_mut_ptr()).size };
if pkt_size > 0 {
total_bytes += pkt_size as usize;
}
match self.output {
Some(FrameOutput::Muxer(ref mut octx)) => {
let enc_tb = self.enc_video.time_base();
@@ -1334,7 +1380,7 @@ impl SwEncEncode {
None => {}
}
}
Ok(())
Ok(total_bytes)
}
}