nuclei/rt/
tuning.rs

1//! How a worker shard runs its loop ([`Tuning`]: the builder's scheduler
2//! knobs) and the functions a runtime calls on its threads ([`ThreadHooks`]):
3//! when one starts and stops (workers and blocking-pool threads), and when a
4//! worker parks and wakes.
5
6use std::sync::Arc;
7
8/// Task runs between reads of the coarse clock; a power of two, so the run
9/// loop tells the runs apart with a mask.
10pub(crate) const QUARTZ_EVERY: u32 = 16;
11/// Nanoseconds of task running after which a worker shard looks at its driver
12/// (1 ms), whatever the count of runs.
13pub(crate) const DRIVE_BOUND: u64 = 1_000_000;
14
15/// How a worker shard runs its loop, and the hooks it calls when it parks;
16/// a shard that is not a worker keeps the defaults.
17#[derive(Clone, Debug)]
18pub(crate) struct Tuning {
19    /// Task runs between looks at the driver (timers and IO).
20    pub(crate) driver_interval: u32,
21    /// Nanoseconds of task running after which the driver gets a look, however
22    /// few runs that was (read from [`Quartz`](super::quartz::Quartz) every
23    /// [`QUARTZ_EVERY`] runs).
24    pub(crate) drive_bound: u64,
25    /// Task runs between looks at the inbox first.
26    pub(crate) inbox_interval: u32,
27    /// The most IO events one look at the driver takes.
28    pub(crate) max_io_events: usize,
29    /// A task woken by the running task runs next.
30    pub(crate) lifo: bool,
31    pub(crate) hooks: ThreadHooks,
32    /// Jobs in flight on the runtime's blocking pool: while any is, a shard that
33    /// runs dry parks, and a park one of them cuts short does not make it spin.
34    pub(crate) blocking: Option<Arc<crate::sync::internal::AtomicUsize>>,
35    /// The least a zero-copy write must have left for a send of it to go
36    /// zero-copy on the worker's ring.
37    #[cfg(ring)]
38    pub(crate) zero_copy_min: usize,
39    /// The shard times its IO operations, by kind (`Builder::record_latencies`).
40    pub(crate) record_latencies: bool,
41    /// The shard counts its IO operations (`Builder::count_io`).
42    #[cfg(feature = "io-stats")]
43    pub(crate) count_io: bool,
44}
45
46impl Default for Tuning {
47    fn default() -> Self {
48        Self {
49            driver_interval: super::shard::DRIVER_INTERVAL,
50            drive_bound: DRIVE_BOUND,
51            inbox_interval: super::shard::INBOX_INTERVAL,
52            max_io_events: usize::MAX,
53            lifo: true,
54            hooks: ThreadHooks::default(),
55            blocking: None,
56            #[cfg(ring)]
57            zero_copy_min: crate::io::source::ring::SEND_ZC_MIN,
58            record_latencies: false,
59            #[cfg(feature = "io-stats")]
60            count_io: false,
61        }
62    }
63}
64
65/// A hook: called with nothing, on the thread it is about.
66pub(crate) type Hook = Arc<dyn Fn() + Send + Sync>;
67
68/// A runtime's thread hooks; each unset one costs nothing.
69#[derive(Clone, Default)]
70pub(crate) struct ThreadHooks {
71    pub(crate) start: Option<Hook>,
72    pub(crate) stop: Option<Hook>,
73    pub(crate) park: Option<Hook>,
74    pub(crate) unpark: Option<Hook>,
75}
76
77impl ThreadHooks {
78    /// Calls the start hook now and the stop hook when the guard drops, even
79    /// when the thread ends by a panic.
80    pub(crate) fn started(&self) -> Stopping {
81        if let Some(start) = &self.start {
82            start();
83        }
84        Stopping(self.stop.clone())
85    }
86}
87
88/// Calls a thread's stop hook when it drops.
89pub(crate) struct Stopping(Option<Hook>);
90
91impl Drop for Stopping {
92    fn drop(&mut self) {
93        if let Some(stop) = &self.0 {
94            stop();
95        }
96    }
97}
98
99impl core::fmt::Debug for ThreadHooks {
100    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
101        f.debug_struct("ThreadHooks")
102            .field("start", &self.start.is_some())
103            .field("stop", &self.stop.is_some())
104            .field("park", &self.park.is_some())
105            .field("unpark", &self.unpark.is_some())
106            .finish()
107    }
108}
109
110/// What a task hook is told about the task.
111#[derive(Debug)]
112pub struct TaskMeta<'a> {
113    id: crate::task::Id,
114    name: Option<Arc<str>>,
115    _task: core::marker::PhantomData<&'a ()>,
116}
117
118impl TaskMeta<'_> {
119    pub(crate) fn new(id: crate::task::Id) -> Self {
120        Self {
121            id,
122            name: super::task::names::get(id),
123            _task: core::marker::PhantomData,
124        }
125    }
126
127    /// The task's id.
128    pub fn id(&self) -> crate::task::Id {
129        self.id
130    }
131
132    /// The task's name, as [`task::Builder::name`](crate::task::Builder::name)
133    /// gave it; `None` for a task spawned without one.
134    pub fn name(&self) -> Option<&str> {
135        self.name.as_deref()
136    }
137}
138
139/// A task hook.
140pub(crate) type TaskHook = Arc<dyn Fn(&TaskMeta<'_>) + Send + Sync>;
141
142/// A runtime's task hooks, and the runtime to shut down when a task panics
143/// if it does that; a task of a runtime with none carries none.
144#[derive(Clone, Default)]
145pub(crate) struct TaskHooks {
146    pub(crate) spawn: Option<TaskHook>,
147    pub(crate) before_poll: Option<TaskHook>,
148    pub(crate) after_poll: Option<TaskHook>,
149    pub(crate) terminate: Option<TaskHook>,
150    /// Weak: the runtime holds its tasks' hooks.
151    pub(crate) shutdown: Option<std::sync::Weak<super::runtime::Inner>>,
152}
153
154impl TaskHooks {
155    pub(crate) fn any(&self) -> bool {
156        self.spawn.is_some()
157            || self.before_poll.is_some()
158            || self.after_poll.is_some()
159            || self.terminate.is_some()
160            || self.shutdown.is_some()
161    }
162}
163
164/// Whether any runtime of the process has task hooks: while none does, a
165/// spawn that does not know its runtime looks for none.
166pub(crate) static ANY_TASK_HOOKS: core::sync::atomic::AtomicBool =
167    core::sync::atomic::AtomicBool::new(false);
168
169/// What a runtime does when one of its tasks panics
170/// ([`Builder::unhandled_panic`](crate::runtime::Builder::unhandled_panic)).
171#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
172#[non_exhaustive]
173pub enum UnhandledPanic {
174    /// The panic is the task's outcome, for whoever awaits it; the runtime
175    /// runs on.
176    #[default]
177    Ignore,
178    /// The runtime shuts down at once: its workers stop and drop their
179    /// tasks, and every `block_on` of it panics, those running and every
180    /// later one. The panicked task's handle still reports the panic.
181    ShutdownRuntime,
182}
183
184/// Calls `hook`, if set, for the task `id`. A hook that panics aborts the
185/// process: it runs in the middle of a task's change of state.
186pub(crate) fn call(hook: Option<&TaskHook>, id: crate::task::Id) {
187    if let Some(hook) = hook {
188        crate::rt::task::abort_on_panic(|| hook(&TaskMeta::new(id)));
189    }
190}
nuclei/rt/tuning.rs

nuclei/rt/
tuning.rs
