tokio/runtime/scheduler/multi_thread/
stats.rsuse crate::runtime::{Config, MetricsBatch, WorkerMetrics};
use std::time::{Duration, Instant};
pub(crate) struct Stats {
batch: MetricsBatch,
processing_scheduled_tasks_started_at: Instant,
tasks_polled_in_batch: usize,
task_poll_time_ewma: f64,
}
const TASK_POLL_TIME_EWMA_ALPHA: f64 = 0.1;
const TARGET_GLOBAL_QUEUE_INTERVAL: f64 = Duration::from_micros(200).as_nanos() as f64;
const MAX_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL: u32 = 127;
const TARGET_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL: u32 = 61;
impl Stats {
pub(crate) fn new(worker_metrics: &WorkerMetrics) -> Stats {
let task_poll_time_ewma =
TARGET_GLOBAL_QUEUE_INTERVAL / TARGET_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL as f64;
Stats {
batch: MetricsBatch::new(worker_metrics),
processing_scheduled_tasks_started_at: Instant::now(),
tasks_polled_in_batch: 0,
task_poll_time_ewma,
}
}
pub(crate) fn tuned_global_queue_interval(&self, config: &Config) -> u32 {
if let Some(configured) = config.global_queue_interval {
return configured;
}
let tasks_per_interval = (TARGET_GLOBAL_QUEUE_INTERVAL / self.task_poll_time_ewma) as u32;
tasks_per_interval.clamp(2, MAX_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL)
}
pub(crate) fn submit(&mut self, to: &WorkerMetrics) {
self.batch.submit(to, self.task_poll_time_ewma as u64);
}
pub(crate) fn about_to_park(&mut self) {
self.batch.about_to_park();
}
pub(crate) fn unparked(&mut self) {
self.batch.unparked();
}
pub(crate) fn inc_local_schedule_count(&mut self) {
self.batch.inc_local_schedule_count();
}
pub(crate) fn start_processing_scheduled_tasks(&mut self) {
self.batch.start_processing_scheduled_tasks();
self.processing_scheduled_tasks_started_at = Instant::now();
self.tasks_polled_in_batch = 0;
}
pub(crate) fn end_processing_scheduled_tasks(&mut self) {
self.batch.end_processing_scheduled_tasks();
if self.tasks_polled_in_batch > 0 {
let now = Instant::now();
let elapsed = (now - self.processing_scheduled_tasks_started_at).as_nanos() as f64;
let num_polls = self.tasks_polled_in_batch as f64;
let mean_poll_duration = elapsed / num_polls;
let weighted_alpha = 1.0 - (1.0 - TASK_POLL_TIME_EWMA_ALPHA).powf(num_polls);
self.task_poll_time_ewma = weighted_alpha * mean_poll_duration
+ (1.0 - weighted_alpha) * self.task_poll_time_ewma;
}
}
pub(crate) fn start_poll(&mut self) {
self.batch.start_poll();
self.tasks_polled_in_batch += 1;
}
pub(crate) fn end_poll(&mut self) {
self.batch.end_poll();
}
pub(crate) fn incr_steal_count(&mut self, by: u16) {
self.batch.incr_steal_count(by);
}
pub(crate) fn incr_steal_operations(&mut self) {
self.batch.incr_steal_operations();
}
pub(crate) fn incr_overflow_count(&mut self) {
self.batch.incr_overflow_count();
}
}