mirrors/gst-plugins-rs
mirrors/gst-plugins-rs
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git synced 2024-11-25 21:11:00 +00:00
Code Issues Projects Releases Wiki Activity

ts: add feature to add counters for performance evaluation

Browse source 
Add a `tuning` feature which adds counters that help with performance
evaluation. The only counter added so far accumulates the duration a
Scheduler has been parked, which is pretty accurate an indication of
CPU usage of the Scheduler.
This commit is contained in:
François Laignel 2022-09-07 23:10:29 +02:00 committed by Sebastian Dröge
parent 72acbebff0
commit 235ded35fd
5 changed files with 170 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
generic/threadshare/Cargo.toml
View file

@ -65,6 +65,8 @@ pkg-config = "0.3.15"
[features]
static = []
capi = []
# Adds performance counters used by benchmarking tools.
tuning = []
doc = ["gst/v1_18"]
[package.metadata.capi]

86
generic/threadshare/examples/benchmark.rs
View file

@ -39,29 +39,14 @@ pub static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
fn main() {
gst::init().unwrap();
#[cfg(debug_assertions)]
{
use std::path::Path;
let mut path = Path::new("target/debug");
if !path.exists() {
path = Path::new("../../target/debug");
}
gst::Registry::get().scan_path(path);
}
#[cfg(not(debug_assertions))]
{
use std::path::Path;
let mut path = Path::new("target/release");
if !path.exists() {
path = Path::new("../../target/release");
}
gst::Registry::get().scan_path(path);
}
// Register the plugins statically:
// - The executable can be run from anywhere.
// - No risk of running against a previous version.
// - `main` can use features that rely on `static`s or `thread_local`
// such as `Context::acquire` which otherwise don't point to
// the same `static` or `thread_local`, probably because
// the shared object uses its owns and the executable, others.
gstthreadshare::plugin_register_static().unwrap();
let args = env::args().collect::<Vec<_>>();
assert!(args.len() > 4);
@ -71,7 +56,7 @@ fn main() {
let wait: u32 = args[4].parse().unwrap();
// Nb buffers to await before stopping.
let max_buffers: Option<u64> = if args.len() > 5 {
let max_buffers: Option<f32> = if args.len() > 5 {
args[5].parse().ok()
} else {
None
@ -243,15 +228,24 @@ fn main() {
let l_clone = l.clone();
thread::spawn(move || {
let throughput_factor = 1_000f32 / (n_streams as f32);
let mut prev_reset_instant: Option<Instant> = None;
let mut count;
let mut reset_instant;
let n_streams_f32 = n_streams as f32;
let mut total_count = 0.0;
let mut ramp_up_complete_instant: Option<Instant> = None;
#[cfg(feature = "tuning")]
let ctx_0 = gstthreadshare::runtime::Context::acquire(
"context-0",
Duration::from_millis(wait as u64),
)
.unwrap();
#[cfg(feature = "tuning")]
let mut parked_init = Duration::ZERO;
loop {
count = counter.fetch_and(0, Ordering::SeqCst);
total_count += counter.fetch_and(0, Ordering::SeqCst) as f32 / n_streams_f32;
if let Some(max_buffers) = max_buffers {
if count > max_buffers {
if total_count > max_buffers {
gst::info!(CAT, "Stopping");
let stopping_instant = Instant::now();
pipeline.set_state(gst::State::Ready).unwrap();
@ -263,22 +257,34 @@ fn main() {
}
}
reset_instant = Instant::now();
if let Some(prev_reset_instant) = prev_reset_instant {
if let Some(init) = ramp_up_complete_instant {
let elapsed = init.elapsed();
gst::info!(
CAT,
"{:>6.2} / s / stream",
(count as f32) * throughput_factor
/ ((reset_instant - prev_reset_instant).as_millis() as f32)
total_count * 1_000.0 / elapsed.as_millis() as f32
);
#[cfg(feature = "tuning")]
gst::info!(
CAT,
"{:>6.2}% parked",
(ctx_0.parked_duration() - parked_init).as_nanos() as f32 * 100.0
/ elapsed.as_nanos() as f32
);
} else {
// Ramp up 30s worth of buffers before following parked
if total_count > 50.0 * 30.0 {
total_count = 0.0;
ramp_up_complete_instant = Some(Instant::now());
#[cfg(feature = "tuning")]
{
parked_init = ctx_0.parked_duration();
}
}
}
if let Some(sleep_duration) = THROUGHPUT_PERIOD.checked_sub(reset_instant.elapsed()) {
thread::sleep(sleep_duration);
}
prev_reset_instant = Some(reset_instant);
thread::sleep(THROUGHPUT_PERIOD);
}
});

34
generic/threadshare/examples/standalone/sink/imp.rs
View file

@ -189,6 +189,8 @@ struct Stats {
interval_late_warn: Duration,
interval_late_count: f32,
interval_late_count_delta: f32,
#[cfg(feature = "tuning")]
parked_duration_init: Duration,
}
impl Stats {
@ -238,6 +240,11 @@ impl Stats {
gst::info!(CAT, "Ramp up complete. Stats logs in {:2?}", LOG_PERIOD);
self.log_start_instant = Some(Instant::now());
self.last_delta_instant = self.log_start_instant;
#[cfg(feature = "tuning")]
{
self.parked_duration_init = Context::current().unwrap().parked_duration();
}
}
use std::cmp::Ordering::*;
@ -304,9 +311,7 @@ impl Stats {
self.last_delta_instant = Some(Instant::now());
if self.buffer_count_delta > 1.0 {
gst::info!(CAT, "Delta stats:");
let interval_mean = self.interval_sum_delta / self.buffer_count_delta;
let interval_std_dev = f32::sqrt(
self.interval_square_sum_delta / self.buffer_count_delta - interval_mean.powi(2),
@ -321,7 +326,7 @@ impl Stats {
self.interval_max_delta,
);
if self.interval_late_count_delta > 1.0 {
if self.interval_late_count_delta > f32::EPSILON {
gst::warning!(
CAT,
"o {:5.2}% late buffers",
@ -353,19 +358,35 @@ impl Stats {
self.latency_square_sum_delta = 0.0;
self.latency_min_delta = Duration::MAX;
self.latency_max_delta = Duration::ZERO;
}
self.buffer_count_delta = 0.0;
}
fn log_global(&mut self) {
if self.log_start_instant.is_none() {
if self.buffer_count < 1.0 {
return;
}
if self.buffer_count > 1.0 {
let _log_start = if let Some(log_start) = self.log_start_instant {
log_start
} else {
return;
};
gst::info!(CAT, "Global stats:");
#[cfg(feature = "tuning")]
{
let duration = _log_start.elapsed();
let parked_duration =
Context::current().unwrap().parked_duration() - self.parked_duration_init;
gst::info!(
CAT,
"o parked: {parked_duration:4.2?} ({:5.2?}%)",
(parked_duration.as_nanos() as f32 * 100.0 / duration.as_nanos() as f32)
);
}
let interval_mean = self.interval_sum / self.buffer_count;
let interval_std_dev =
f32::sqrt(self.interval_square_sum / self.buffer_count - interval_mean.powi(2));
@ -401,7 +422,6 @@ impl Stats {
);
}
}
}
struct TestSinkTask {
element: super::TestSink,

8
generic/threadshare/src/runtime/executor/context.rs
View file

@ -187,6 +187,14 @@ impl Context {
self.0.max_throttling()
}
/// Total duration the scheduler spent parked.
///
/// This is only useful for performance evaluation.
#[cfg(feature = "tuning")]
pub fn parked_duration(&self) -> Duration {
self.0.parked_duration()
}
/// Returns `true` if a `Context` is running on current thread.
pub fn is_context_thread() -> bool {
Scheduler::is_scheduler_thread()

19
generic/threadshare/src/runtime/executor/scheduler.rs
View file

@ -11,6 +11,8 @@ use gio::glib::clone::Downgrade;
use std::cell::RefCell;
use std::future::Future;
use std::panic;
#[cfg(feature = "tuning")]
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc as sync_mpsc;
use std::sync::{Arc, Condvar, Mutex, Weak};
use std::task::Poll;
@ -34,6 +36,8 @@ pub(super) struct Scheduler {
tasks: TaskQueue,
must_unpark: Mutex<bool>,
must_unpark_cvar: Condvar,
#[cfg(feature = "tuning")]
parked_duration: AtomicU64,
}
impl Scheduler {
@ -104,6 +108,8 @@ impl Scheduler {
tasks: TaskQueue::new(context_name),
must_unpark: Mutex::new(false),
must_unpark_cvar: Condvar::new(),
#[cfg(feature = "tuning")]
parked_duration: AtomicU64::new(0),
}));
*cur_scheduler = Some(handle.downgrade());
@ -201,12 +207,16 @@ impl Scheduler {
break;
}
if let Some(wait_duration) = self.max_throttling.checked_sub(last.elapsed()) {
if let Some(parking_duration) = self.max_throttling.checked_sub(last.elapsed()) {
let result = self
.must_unpark_cvar
.wait_timeout(must_unpark, wait_duration)
.wait_timeout(must_unpark, parking_duration)
.unwrap();
#[cfg(feature = "tuning")]
self.parked_duration
.fetch_add(parking_duration.subsec_nanos() as u64, Ordering::Relaxed);
must_unpark = result.0;
} else {
*must_unpark = false;
@ -360,6 +370,11 @@ impl Handle {
self.0.scheduler.max_throttling
}
#[cfg(feature = "tuning")]
pub fn parked_duration(&self) -> Duration {
Duration::from_nanos(self.0.scheduler.parked_duration.load(Ordering::Relaxed))
}
/// Executes the provided function relatively to this [`Scheduler`]'s [`Reactor`].
///
/// Usefull to initialze i/o sources and timers from outside