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