1 use crate::runtime::{Config, MetricsBatch, WorkerMetrics};
2 
3 use std::time::{Duration, Instant};
4 
5 /// Per-worker statistics. This is used for both tuning the scheduler and
6 /// reporting runtime-level metrics/stats.
7 pub(crate) struct Stats {
8     /// The metrics batch used to report runtime-level metrics/stats to the
9     /// user.
10     batch: MetricsBatch,
11 
12     /// Instant at which work last resumed (continued after park).
13     ///
14     /// This duplicates the value stored in `MetricsBatch`. We will unify
15     /// `Stats` and `MetricsBatch` when we stabilize metrics.
16     processing_scheduled_tasks_started_at: Instant,
17 
18     /// Number of tasks polled in the batch of scheduled tasks
19     tasks_polled_in_batch: usize,
20 
21     /// Exponentially-weighted moving average of time spent polling scheduled a
22     /// task.
23     ///
24     /// Tracked in nanoseconds, stored as a `f64` since that is what we use with
25     /// the EWMA calculations
26     task_poll_time_ewma: f64,
27 }
28 
29 /// How to weigh each individual poll time, value is plucked from thin air.
30 const TASK_POLL_TIME_EWMA_ALPHA: f64 = 0.1;
31 
32 /// Ideally, we wouldn't go above this, value is plucked from thin air.
33 const TARGET_GLOBAL_QUEUE_INTERVAL: f64 = Duration::from_micros(200).as_nanos() as f64;
34 
35 /// Max value for the global queue interval. This is 2x the previous default
36 const MAX_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL: u32 = 127;
37 
38 /// This is the previous default
39 const TARGET_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL: u32 = 61;
40 
41 impl Stats {
new(worker_metrics: &WorkerMetrics) -> Stats42     pub(crate) fn new(worker_metrics: &WorkerMetrics) -> Stats {
43         // Seed the value with what we hope to see.
44         let task_poll_time_ewma =
45             TARGET_GLOBAL_QUEUE_INTERVAL / TARGET_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL as f64;
46 
47         Stats {
48             batch: MetricsBatch::new(worker_metrics),
49             processing_scheduled_tasks_started_at: Instant::now(),
50             tasks_polled_in_batch: 0,
51             task_poll_time_ewma,
52         }
53     }
54 
tuned_global_queue_interval(&self, config: &Config) -> u3255     pub(crate) fn tuned_global_queue_interval(&self, config: &Config) -> u32 {
56         // If an interval is explicitly set, don't tune.
57         if let Some(configured) = config.global_queue_interval {
58             return configured;
59         }
60 
61         // As of Rust 1.45, casts from f64 -> u32 are saturating, which is fine here.
62         let tasks_per_interval = (TARGET_GLOBAL_QUEUE_INTERVAL / self.task_poll_time_ewma) as u32;
63 
64         // If we are using self-tuning, we don't want to return less than 2 as that would result in the
65         // global queue always getting checked first.
66         tasks_per_interval.clamp(2, MAX_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL)
67     }
68 
submit(&mut self, to: &WorkerMetrics)69     pub(crate) fn submit(&mut self, to: &WorkerMetrics) {
70         self.batch.submit(to, self.task_poll_time_ewma as u64);
71     }
72 
about_to_park(&mut self)73     pub(crate) fn about_to_park(&mut self) {
74         self.batch.about_to_park();
75     }
76 
unparked(&mut self)77     pub(crate) fn unparked(&mut self) {
78         self.batch.unparked();
79     }
80 
inc_local_schedule_count(&mut self)81     pub(crate) fn inc_local_schedule_count(&mut self) {
82         self.batch.inc_local_schedule_count();
83     }
84 
start_processing_scheduled_tasks(&mut self)85     pub(crate) fn start_processing_scheduled_tasks(&mut self) {
86         self.batch.start_processing_scheduled_tasks();
87 
88         self.processing_scheduled_tasks_started_at = Instant::now();
89         self.tasks_polled_in_batch = 0;
90     }
91 
end_processing_scheduled_tasks(&mut self)92     pub(crate) fn end_processing_scheduled_tasks(&mut self) {
93         self.batch.end_processing_scheduled_tasks();
94 
95         // Update the EWMA task poll time
96         if self.tasks_polled_in_batch > 0 {
97             let now = Instant::now();
98 
99             // If we "overflow" this conversion, we have bigger problems than
100             // slightly off stats.
101             let elapsed = (now - self.processing_scheduled_tasks_started_at).as_nanos() as f64;
102             let num_polls = self.tasks_polled_in_batch as f64;
103 
104             // Calculate the mean poll duration for a single task in the batch
105             let mean_poll_duration = elapsed / num_polls;
106 
107             // Compute the alpha weighted by the number of tasks polled this batch.
108             let weighted_alpha = 1.0 - (1.0 - TASK_POLL_TIME_EWMA_ALPHA).powf(num_polls);
109 
110             // Now compute the new weighted average task poll time.
111             self.task_poll_time_ewma = weighted_alpha * mean_poll_duration
112                 + (1.0 - weighted_alpha) * self.task_poll_time_ewma;
113         }
114     }
115 
start_poll(&mut self)116     pub(crate) fn start_poll(&mut self) {
117         self.batch.start_poll();
118 
119         self.tasks_polled_in_batch += 1;
120     }
121 
end_poll(&mut self)122     pub(crate) fn end_poll(&mut self) {
123         self.batch.end_poll();
124     }
125 
incr_steal_count(&mut self, by: u16)126     pub(crate) fn incr_steal_count(&mut self, by: u16) {
127         self.batch.incr_steal_count(by);
128     }
129 
incr_steal_operations(&mut self)130     pub(crate) fn incr_steal_operations(&mut self) {
131         self.batch.incr_steal_operations();
132     }
133 
incr_overflow_count(&mut self)134     pub(crate) fn incr_overflow_count(&mut self) {
135         self.batch.incr_overflow_count();
136     }
137 }
138