1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
//! The builder's hooks and scheduler knobs: what the runtime calls on its
//! threads and around its tasks, how its workers run their loops, what it
//! does when a task panics, and the seed of its random choices.
use std::sync::Arc;
use super::{Builder, RngSeed, TaskMeta, UnhandledPanic};
impl Builder {
/// Called on each thread the runtime starts (workers and blocking-pool
/// threads), on that thread, before it does anything else.
pub fn on_thread_start(mut self, f: impl Fn() + Send + Sync + 'static) -> Self {
self.tuning.hooks.start = Some(Arc::new(f));
self
}
/// Called on each thread the runtime starts, on that thread, as it ends
/// (after a panic too).
pub fn on_thread_stop(mut self, f: impl Fn() + Send + Sync + 'static) -> Self {
self.tuning.hooks.stop = Some(Arc::new(f));
self
}
/// Called on a worker thread just before it sleeps with nothing to run.
pub fn on_thread_park(mut self, f: impl Fn() + Send + Sync + 'static) -> Self {
self.tuning.hooks.park = Some(Arc::new(f));
self
}
/// Called on a worker thread just after it wakes from a sleep.
pub fn on_thread_unpark(mut self, f: impl Fn() + Send + Sync + 'static) -> Self {
self.tuning.hooks.unpark = Some(Arc::new(f));
self
}
/// Called with each task this runtime spawns (not the blocking pool's
/// jobs), on the spawning thread, once the task exists.
pub fn on_task_spawn(mut self, f: impl Fn(&TaskMeta<'_>) + Send + Sync + 'static) -> Self {
self.task_hooks.spawn = Some(Arc::new(f));
self
}
/// Called just before each poll of a task of this runtime, on the thread
/// that polls it.
pub fn on_before_task_poll(
mut self,
f: impl Fn(&TaskMeta<'_>) + Send + Sync + 'static,
) -> Self {
self.task_hooks.before_poll = Some(Arc::new(f));
self
}
/// Called just after each poll of a task of this runtime, on the thread
/// that polled it.
pub fn on_after_task_poll(mut self, f: impl Fn(&TaskMeta<'_>) + Send + Sync + 'static) -> Self {
self.task_hooks.after_poll = Some(Arc::new(f));
self
}
/// Called when a task of this runtime is done with: its future completed
/// or was dropped by a cancel, on the thread that dropped it.
pub fn on_task_terminate(mut self, f: impl Fn(&TaskMeta<'_>) + Send + Sync + 'static) -> Self {
self.task_hooks.terminate = Some(Arc::new(f));
self
}
/// Called with each poll of a task of this runtime that its worker's
/// watch sees run past [`handoff_after`](Builder::handoff_after), on the
/// watch's own thread: the task's id and name say which one holds its
/// worker up. The task is the innermost one whose poll was running when
/// the watch looked; the poll may have ended by the time the hook runs.
/// A poll is told once, unless a `block_on` inside it ran a long poll of
/// its own and the shard stayed on that thread: the outer poll may then
/// be told again as it goes on. The shard is handed on first, so a slow
/// hook delays no rescue, but every other worker waits for its next look
/// while the hook runs: keep it short. A panic in it is caught (after the
/// panic hook reported it) and the watch goes on. Only workers are
/// watched; with `handoff_after(None)` [`build`](Builder::build) fails
/// with `InvalidInput`.
pub fn on_long_poll(mut self, f: impl Fn(&TaskMeta<'_>) + Send + Sync + 'static) -> Self {
self.long_poll = Some(Arc::new(f));
self
}
/// Task runs between a worker's looks at its timers and IO when it is
/// busy (default 61): lower answers IO sooner, higher runs tasks with
/// fewer interruptions. At least 1; [`build`](Builder::build) fails with
/// `InvalidInput` at 0.
pub fn event_interval(mut self, runs: u32) -> Self {
self.tuning.driver_interval = runs;
self
}
/// Task runs between the times a worker takes from its inbox (tasks
/// woken by other threads) before its own queue (default 31). At least 1;
/// [`build`](Builder::build) fails with `InvalidInput` at 0.
pub fn global_queue_interval(mut self, runs: u32) -> Self {
self.tuning.inbox_interval = runs;
self
}
/// The most IO events a worker takes from its driver in one look
/// (default: all that are ready). At least 1; [`build`](Builder::build)
/// fails with `InvalidInput` at 0.
pub fn max_io_events_per_tick(mut self, events: usize) -> Self {
self.tuning.max_io_events = events;
self
}
/// Turns off each worker's LIFO slot, where a task woken by the task
/// running on the worker goes to run next. Without it the woken task
/// goes to the back of the worker's queue: fairer when the queue is
/// long, slower for a task that hands work to another and waits for it.
pub fn disable_lifo_slot(mut self) -> Self {
self.tuning.lifo = false;
self
}
/// Seeds the runtime's random choices, so that they repeat run after
/// run; see [`RngSeed`].
pub fn rng_seed(mut self, seed: RngSeed) -> Self {
self.rng_seed = Some(seed);
self
}
/// What the runtime does when one of its tasks panics (default
/// [`UnhandledPanic::Ignore`]: the panic goes to the task's handle).
///
/// ```
/// use nuclei::runtime::{Runtime, UnhandledPanic};
///
/// let rt = Runtime::builder()
/// .unhandled_panic(UnhandledPanic::ShutdownRuntime)
/// .build()
/// .unwrap();
/// let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
/// rt.block_on(async {
/// let _task = nuclei::spawn(async { panic!("a task fails") });
/// core::future::pending::<()>().await
/// })
/// }));
/// assert!(ran.is_err(), "the block_on panics once the task did");
/// ```
pub fn unhandled_panic(mut self, policy: UnhandledPanic) -> Self {
self.unhandled_panic = policy;
self
}
}