shuttle_engine/runtime/execution.rs
1use crate::config::DEFAULT_MAX_STEPS;
2use crate::runtime::failure::{init_panic_hook, persist_failure};
3use crate::runtime::storage::{StorageKey, StorageMap};
4use crate::runtime::task::clock::VectorClock;
5use crate::runtime::task::labels::Labels;
6use crate::runtime::task::{ChildLabelFn, ParkedDefault, Task, TaskId, TaskName, TaskSignature, DEFAULT_INLINE_TASKS};
7use crate::runtime::thread;
8use crate::runtime::thread::continuation::PooledContinuation;
9use crate::scheduler::{Schedule, Scheduler};
10use crate::sync_types::{ResourceSignature, ResourceType};
11use crate::thread_support::thread_fn;
12use crate::{backtrace_enabled, Config, MaxSteps, UNGRACEFUL_SHUTDOWN_CONFIG};
13use scoped_tls::scoped_thread_local;
14use smallvec::SmallVec;
15use std::any::Any;
16use std::cell::{Cell, RefCell};
17use std::collections::{HashMap, VecDeque};
18use std::fmt::Debug;
19use std::future::Future;
20use std::panic::{self, Location};
21use std::rc::Rc;
22use std::sync::Arc;
23use tracing::subscriber::NoSubscriber;
24use tracing::{trace, Span};
25
26#[allow(deprecated)]
27use super::task::Tag;
28
29// We use this scoped TLS to smuggle the ExecutionState, which is not 'static, across tasks that
30// need access to it (to spawn new tasks, interrogate task status, etc).
31scoped_thread_local! {
32 static EXECUTION_STATE: RefCell<ExecutionState>
33}
34
35thread_local! {
36 // Whether the panic hook stays silent, because execution teardown ignores the panics it catches
37 // now (see `ExecutionState::tear_down`). Outside `EXECUTION_STATE`, so that the hook can read it
38 // while the state is borrowed.
39 static TEARDOWN_IGNORES_PANICS: Cell<bool> = const { Cell::new(false) };
40 // How many panics there have been in the current step of execution teardown, for the panic
41 // hook (see `ExecutionState::teardown_ignores_panic`).
42 static TEARDOWN_STEP_PANICS: Cell<usize> = const { Cell::new(0) };
43}
44
45thread_local! {
46 // Whether this thread is emitting the `ExecutionState::try_with` event. A `tracing` subscriber
47 // may access the state while it handles an event (to call `current::clock()`, say), and that
48 // access must not emit the event again: `tracing` only refuses re-entrant calls for scoped
49 // subscribers, and with a global one this would recurse until the stack overflows.
50 static TRACING_TRY_WITH: Cell<bool> = const { Cell::new(false) };
51}
52
53// The reason this is separated out from `ExecutionState` is to ensure that we're always able to persist the schedule.
54// If we don't do this, then we may panic while borrowing `ExecutionState`, and then not be able to emit the schedule.
55// If we then panic again while trying to handle the panic, such that the panic becomes an abort, we will never log
56// the schedule.
57//
58// It is expected that if the `ExecutionState` exists, then this will exist, and any usage of this happens through the
59// `ExecutionState`, or at a point where it is known that the `ExecutionState` must exist (eg. when serializing on a panic).
60thread_local! {
61 static CURRENT_SCHEDULE: CurrentSchedule = CurrentSchedule::default();
62}
63
64#[derive(Debug, Default)]
65pub struct CurrentSchedule {
66 current_schedule: RefCell<Schedule>,
67}
68
69impl CurrentSchedule {
70 fn init(schedule: Schedule) {
71 CURRENT_SCHEDULE.with(|cs| *cs.current_schedule.borrow_mut() = schedule)
72 }
73
74 /// Add the given task ID as the next step of the schedule.
75 fn push_task(tid: TaskId) {
76 CURRENT_SCHEDULE.with(|cs| cs.current_schedule.borrow_mut().push_task(tid))
77 }
78
79 /// Add a choice of a random u64 value as the next step of the schedule
80 fn push_random() {
81 CURRENT_SCHEDULE.with(|cs| cs.current_schedule.borrow_mut().push_random())
82 }
83
84 /// Return the number of steps in the schedule
85 pub fn len() -> usize {
86 CURRENT_SCHEDULE.with(|cs| (*cs.current_schedule.borrow()).len())
87 }
88
89 /// Returns a clone of the inner schedule
90 pub fn get_schedule() -> Schedule {
91 CURRENT_SCHEDULE.with(|cs| (*cs.current_schedule.borrow()).clone())
92 }
93}
94
95thread_local! {
96 #[allow(clippy::complexity)]
97 #[allow(deprecated)]
98 pub static TASK_ID_TO_TAGS: RefCell<HashMap<TaskId, Arc<dyn Tag>>> = RefCell::new(HashMap::new());
99}
100
101thread_local! {
102 pub static LABELS: RefCell<HashMap<TaskId, Labels>> = RefCell::new(HashMap::new());
103}
104
105/// An `Execution` encapsulates a single run of a function under test against a chosen scheduler.
106/// Its only useful method is `Execution::run`, which executes the function to completion.
107///
108/// The key thing that an `Execution` manages is the `ExecutionState`, which contains all the
109/// mutable state a test's tasks might need access to during execution (to block/unblock tasks,
110/// spawn new tasks, etc). The `Execution` makes this state available through the `EXECUTION_STATE`
111/// static variable, but clients get access to it by calling `ExecutionState::with`.
112pub struct Execution {
113 scheduler: Rc<RefCell<dyn Scheduler>>,
114 initial_schedule: Schedule,
115}
116
117impl std::fmt::Debug for Execution {
118 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
119 f.debug_struct("Execution").finish_non_exhaustive()
120 }
121}
122
123impl Execution {
124 /// Construct a new execution that will use the given scheduler. The execution should then be
125 /// invoked via its `run` method, which takes as input the closure for task 0.
126 pub fn new(scheduler: Rc<RefCell<dyn Scheduler>>, initial_schedule: Schedule) -> Self {
127 Self {
128 scheduler,
129 initial_schedule,
130 }
131 }
132}
133
134#[derive(Debug)]
135enum StepError {
136 // Contains the panic payload of the task that failed.
137 TaskFailure(Box<dyn Any + Send>),
138 // The scheduler didn't make a decision. Indicates a scheduler error.
139 SchedulingError,
140 // Scheduling deetected a deadlock.
141 Deadlock,
142 // We exceeded the step bound of `MaxSteps::FailAfter`, which this is.
143 StepBoundExceeded(usize),
144 // Task panic and `config.immediately_return_on_panic` is set to `true`.
145 TaskPanicEarlyReturn,
146}
147
148/// How a failed execution fails the test, once it has been torn down.
149enum Failure {
150 /// Resume unwinding with the panic payload.
151 Unwind(Box<dyn Any + Send>),
152 /// Panic with the message.
153 Panic(String),
154 /// Panic with the message, as a `&'static str` payload.
155 StaticPanic(&'static str),
156}
157
158/// How tearing down an execution that had not failed fails the test (see `ExecutionState::tear_down`).
159enum TeardownFailure {
160 /// A destructor panicked: resume unwinding with its payload.
161 Unwind(Box<dyn Any + Send>),
162 /// A task that was unwinding a panic when the execution stopped has finished unwinding it:
163 /// resume unwinding with its payload.
164 LatePanic(Box<dyn Any + Send>),
165 /// Panic with the message: destructors blocked, and nothing could wake them, say.
166 Panic(String),
167}
168
169impl Execution {
170 /// Run a function to be tested, taking control of scheduling it and any tasks it might spawn.
171 /// This function runs until `f` and all tasks spawned by `f` have terminated, or until the
172 /// scheduler returns `None`, indicating the execution should not be explored any further.
173 pub fn run<F>(mut self, config: &Config, f: F, caller: &'static Location<'static>)
174 where
175 F: FnOnce() + Send + 'static,
176 {
177 let state = RefCell::new(ExecutionState::new(config.clone(), Rc::clone(&self.scheduler)));
178
179 init_panic_hook(config.clone());
180 CurrentSchedule::init(self.initial_schedule.clone());
181 UNGRACEFUL_SHUTDOWN_CONFIG.set(config.ungraceful_shutdown_config);
182
183 EXECUTION_STATE.set(&state, move || {
184 // Spawn `f` as the first task
185 ExecutionState::spawn_main_thread(
186 Box::new(move || thread_fn(f, true, Default::default())),
187 config.stack_size,
188 caller,
189 );
190
191 // Run the test to completion. A panic out of the executor (from a scheduler, say) fails
192 // the execution like a task's panic does, so that the execution is still torn down while
193 // Shuttle is there to call into.
194 let immediately_return_on_panic = UNGRACEFUL_SHUTDOWN_CONFIG.get().immediately_return_on_panic;
195 let failure = match panic::catch_unwind(panic::AssertUnwindSafe(|| {
196 self.run_to_completion(immediately_return_on_panic)
197 })) {
198 Ok(Ok(())) => None,
199 Ok(Err(e)) => Some(Self::failure(e, config)),
200 // The panic hook has reported it.
201 Err(payload) => Some(Failure::Unwind(payload)),
202 };
203
204 // Tear down the execution before it goes out of `EXECUTION_STATE` scope. A failed
205 // execution is torn down before its failure is raised, so that nothing that the failure
206 // leaves to unwind can call into Shuttle once `EXECUTION_STATE` is gone.
207 let kind = if failure.is_some() {
208 TeardownKind::Failed
209 } else if ExecutionState::execution_stopped() {
210 TeardownKind::Stopped
211 } else {
212 TeardownKind::Finished
213 };
214 let teardown_failure = ExecutionState::tear_down(kind);
215 match failure {
216 Some(Failure::Unwind(payload)) => panic::resume_unwind(payload),
217 Some(Failure::Panic(message)) => panic::panic_any(message),
218 Some(Failure::StaticPanic(message)) => panic::panic_any(message),
219 None => match teardown_failure {
220 Some(TeardownFailure::Unwind(payload)) => {
221 persist_failure(config);
222 eprintln!("test panicked while dropping the tasks that were unfinished at the end of the execution");
223 panic::resume_unwind(payload);
224 }
225 Some(TeardownFailure::LatePanic(payload)) => {
226 persist_failure(config);
227 eprintln!("test panicked in a task that was still unwinding the panic when the execution stopped");
228 panic::resume_unwind(payload);
229 }
230 Some(TeardownFailure::Panic(message)) => panic::panic_any(message),
231 None => {}
232 },
233 }
234 });
235 }
236
237 /// Report the failure of the execution, and return how to fail the test with it once the
238 /// execution has been torn down.
239 fn failure(e: StepError, config: &Config) -> Failure {
240 persist_failure(config);
241 // Teardown changes the current task, so remember which task failed for the panic hook.
242 ExecutionState::with(|state| state.failed_task = state.try_current().map(Task::display_name));
243
244 match e {
245 StepError::TaskFailure(payload) => {
246 eprintln!("test panicked in task '{}'", ExecutionState::failing_task());
247
248 Failure::Unwind(payload)
249 }
250 StepError::Deadlock => {
251 let blocked_tasks = ExecutionState::with(|state| {
252 state
253 .tasks
254 .iter()
255 .filter(|t| !t.finished())
256 .map(|t| t.format_for_deadlock())
257 .collect::<Vec<_>>()
258 });
259
260 // Collecting backtraces is expensive, so we only want to do it if the user opts in to collecting them.
261 if !backtrace_enabled() {
262 eprintln!("Test deadlocked, and {} is not set. If either of those are set then the backtrace of each task will be collected and printed as part of the panic message.", crate::CAPTURE_BACKTRACE)
263 }
264
265 Failure::Panic(format!("deadlock! blocked tasks: [{}]", blocked_tasks.join(", ")))
266 }
267 StepError::SchedulingError => {
268 Failure::StaticPanic("no task was scheduled\nThis indicates an issue with the scheduler.")
269 }
270 StepError::StepBoundExceeded(max_steps) => Failure::Panic(format!(
271 "exceeded max_steps bound {max_steps}. this might be caused by an unfair schedule (e.g., a spin loop)?"
272 )),
273 StepError::TaskPanicEarlyReturn => Failure::Unwind(Box::new("Task panicked, and early return is enabled.")),
274 }
275 }
276
277 fn enter_task_span() {
278 // Enter the Task's span
279 // (Note that if any issues arise with spans and tracing, then
280 // 1) calling `exit` until `None` before entering the `Task`s `Span`,
281 // 2) storing the entirety of the `span_stack` when creating the `Task`, and
282 // 3) storing `top_level_span` as a stack
283 // should be tried.)
284 let parked = ExecutionState::with(|state| {
285 // Go through `Span::with_subscriber` rather than calling `Span::current()` inside
286 // `tracing::dispatcher::get_default`; see `exit_task_span` for why.
287 tracing::Span::current().with_subscriber(|(id, subscriber)| subscriber.exit(id));
288
289 // The `span_stack` stores `Span`s such that the top of the stack is the outermost `Span`,
290 // meaning that parents (left-most when printed) are entered first.
291 while let Some(span) = state.current_mut().span_stack.pop() {
292 span.with_subscriber(|(id, subscriber)| subscriber.enter(id));
293 }
294
295 if state.config.record_steps_in_span {
296 state.current().step_span.record("i", CurrentSchedule::len());
297 }
298
299 state.current_mut().parked_default.take()
300 });
301
302 // Reinstate the default dispatcher the task had when it last switched out. Outside
303 // `ExecutionState::with`, as that drops the dispatcher it displaces.
304 if let Some(parked) = parked {
305 parked.reinstate();
306 }
307 }
308
309 fn exit_task_span(yielded: bool) {
310 // Leave the Task's span and store the exited `Span` stack in order to restore it the next time the Task is run
311 ExecutionState::with(|state| {
312 // Before the task's spans are exited, so they are looked up in the dispatcher they were entered in.
313 if yielded {
314 Execution::park_task_default(state);
315 }
316
317 debug_assert!(state.current().span_stack.is_empty());
318 // Note that `Span::current()` must not be called from inside a
319 // `tracing::dispatcher::get_default` callback: `get_default` marks the thread's
320 // dispatcher state as in use for the duration of the callback, and while any thread in
321 // the process holds a scoped default subscriber, a nested `Span::current()` then
322 // returns `Span::none()`. This loop would then exit nothing while we still enter
323 // `top_level_span` below, leaking one entry per scheduling step until a later
324 // `Span::current()` resolved to a closed span and panicked. So each exit goes through
325 // `Span::with_subscriber`, which hands us the span's own dispatcher directly.
326 // `Span::current()` returns a disabled span once no span is entered, and a disabled span
327 // is exactly one `with_subscriber` does nothing for, so every iteration exits a span.
328 while let Some(current) = Some(tracing::Span::current()).filter(|span| !span.is_disabled()) {
329 current.with_subscriber(|(id, subscriber)| subscriber.exit(id));
330 state.current_mut().span_stack.push(current);
331 }
332
333 state
334 .top_level_span
335 .with_subscriber(|(id, subscriber)| subscriber.enter(id));
336 });
337 }
338
339 /// Saves the running task's default `tracing` dispatcher and reinstates the execution's.
340 ///
341 /// `tracing`'s default dispatcher is per OS thread, and every task runs on this one. Without
342 /// this, a task that yields inside `tracing::subscriber::with_default` would leave its dispatcher
343 /// installed for the scheduler and every other task, and `exit_task_span` would look for the
344 /// task's spans in that dispatcher instead of the one they were entered in.
345 fn park_task_default(state: &mut ExecutionState) {
346 // Installing a guard when there is nothing to park would put `tracing` on its slower scoped
347 // path for the whole process.
348 if !task_may_have_own_default(state) {
349 return;
350 }
351 // The guard remembers the task's current default, and `enter_task_span` drops it to
352 // reinstate exactly that.
353 let guard = tracing::dispatcher::set_default(&state.top_level_dispatch);
354 state.current_mut().parked_default = Some(ParkedDefault::new(guard));
355 }
356
357 /// Run the execution to completion.
358 #[inline]
359 fn run_to_completion(&mut self, immediately_return_on_panic: bool) -> Result<(), StepError> {
360 loop {
361 let next_step: Option<Rc<RefCell<PooledContinuation>>> = ExecutionState::with(|state| {
362 state.schedule()?;
363 state.advance_to_next_task();
364
365 match state.current_task {
366 ScheduledTask::Some(tid) => {
367 let task = state.get(tid);
368 Ok(Some(
369 task.continuation
370 .clone()
371 .expect("only execution teardown takes a task's continuation"),
372 ))
373 }
374 ScheduledTask::Finished => {
375 // The scheduler decided we're finished, so there are either no runnable tasks,
376 // or all runnable tasks are detached and there are no unfinished attached
377 // tasks. Therefore, it's a deadlock if there are unfinished attached tasks.
378 if state.tasks.iter().any(|t| !t.finished() && !t.detached) {
379 Err(StepError::Deadlock)
380 } else {
381 Ok(None)
382 }
383 }
384 ScheduledTask::Stopped => Ok(None),
385 ScheduledTask::None => Err(StepError::SchedulingError),
386 }
387 })?;
388
389 // Run a single step of the chosen task.
390 let ret = match next_step {
391 Some(continuation) => {
392 Execution::enter_task_span();
393
394 let result = panic::catch_unwind(panic::AssertUnwindSafe(|| continuation.borrow_mut().resume()));
395
396 Execution::exit_task_span(matches!(result, Ok(false)));
397
398 result
399 }
400 None => return Ok(()),
401 };
402
403 match ret {
404 // Task finished
405 Ok(true) => {
406 crate::annotations::record_task_terminated();
407 ExecutionState::with(|state| {
408 state.finish_current_task();
409 // The task may have caught the panic that was unwinding last, and finished
410 // without another scheduling point. Here, between steps,
411 // `std::thread::panicking()` says exactly whether a task that switched out is
412 // unwinding a panic still (see `record_unwinding_task`).
413 if state.switched_out_unwinding && !std::thread::panicking() {
414 state.forget_unwinding_task();
415 }
416 });
417 }
418 // Task yielded
419 Ok(false) => {
420 // We may have `switch`ed out of the task before we finished unwinding the stack (ie. a `drop` handler calls `switch`).
421 // If `immediately_return_on_panic` is set, we will then return. If we don't do this, then we run the risk of panicking
422 // again in some other task, which would result in the test aborting.
423 if immediately_return_on_panic && std::thread::panicking() {
424 ExecutionState::with(|state| state.current_task = ScheduledTask::Stopped);
425 return Err(StepError::TaskPanicEarlyReturn);
426 }
427 }
428 // Task failed
429 Err(e) => return Err(StepError::TaskFailure(e)),
430 }
431 }
432 }
433}
434
435/// `ExecutionState` contains the portion of a single execution's state that needs to be reachable
436/// from within a task's execution. It tracks which tasks exist and their states, as well as which
437/// tasks are pending spawn.
438pub struct ExecutionState {
439 pub config: Config,
440 // invariant: tasks are never removed from this list, until execution teardown drops them all
441 tasks: SmallVec<[Task; DEFAULT_INLINE_TASKS]>,
442 // invariant: if this transitions to Stopped or Finished, it can never change again, except that
443 // execution teardown makes a failed execution Stopped, and makes each task that it tears down the
444 // current task while it does (see `tear_down`)
445 current_task: ScheduledTask,
446 // the task the scheduler has chosen to run next
447 next_task: ScheduledTask,
448 // whether the current task has asked to yield
449 has_yielded: bool,
450 // the number of scheduling decisions made so far
451 context_switches: usize,
452 // the schedule length last time `reset_stop_bound()` was called
453 pub steps_reset_at: usize,
454
455 // static values for the current execution
456 storage: StorageMap,
457
458 scheduler: Rc<RefCell<dyn Scheduler>>,
459
460 // the state of execution teardown, while the execution is being torn down (see `tear_down`)
461 teardown: Option<Teardown>,
462
463 // the name of the task that failed the execution, if a task did, for the panic hook to report
464 // once the execution has been torn down (see `failing_task`)
465 failed_task: Option<String>,
466
467 // the task that switched out while unwinding a panic, if it may be unwinding it still, with
468 // whether it is detached otherwise: it is attached until it has finished unwinding (see
469 // `record_unwinding_task`)
470 unwinding_task: Option<(TaskId, bool)>,
471
472 // whether a task has switched out while unwinding a panic since `std::thread::panicking()` was
473 // last false: if not, no task but the current one can be unwinding a panic (see
474 // `record_unwinding_task`)
475 switched_out_unwinding: bool,
476
477 #[cfg(debug_assertions)]
478 has_cleaned_up: bool,
479
480 // The `Span` which the `ExecutionState` was created under. Will be the parent of all `Task` `Span`s
481 pub top_level_span: Span,
482
483 // The default `tracing` dispatcher the `ExecutionState` was created under. Tasks start out with it,
484 // and it is reinstated whenever a task with a different default switches out.
485 top_level_dispatch: tracing::Dispatch,
486
487 // If `top_level_dispatch` is the global default, where `get_default` hands it out from (see
488 // `task_may_have_own_default`).
489 top_level_dispatch_global_ptr: Option<*const tracing::Dispatch>,
490
491 // Persistent Vec used as a bump allocator for references to runnable tasks to avoid slow allocation
492 // on each scheduling decision. Should not be used outside of the `schedule` function
493 runnable_tasks: Vec<*const Task>,
494
495 // Ids of all tasks that have not yet finished, kept sorted in ascending order.
496 //
497 // `tasks` never shrinks, so it accumulates every task ever created by the execution. Scanning it
498 // on every scheduling decision therefore costs O(tasks ever created), even though only the
499 // unfinished ones can ever be scheduled. This set lets `schedule` iterate just the live tasks.
500 //
501 // invariant: contains exactly the ids of the tasks in `tasks` that are not `Finished`, in
502 // ascending order. Maintained by pushing on task creation (ids are handed out sequentially, so
503 // pushing keeps this sorted) and removing in `finish_current_task`. This is sound because
504 // `Finished` is a terminal state: it is only ever set by `Task::finish`, and `block`, `sleep`,
505 // and `unblock` all assert that they are never applied to a finished task. (Execution teardown
506 // schedules nothing, and doesn't maintain this: see `tear_down`. It also lets a finished task
507 // stand in for a while: see `Task::stand_in`.)
508 live_tasks: Vec<TaskId>,
509}
510
511impl std::fmt::Debug for ExecutionState {
512 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
513 f.debug_struct("ExecutionState").finish_non_exhaustive()
514 }
515}
516
517#[derive(Debug, PartialEq, Eq, Clone, Copy)]
518enum ScheduledTask {
519 None, // no task has ever been scheduled
520 Some(TaskId), // this task is running
521 Stopped, // the scheduler asked us to stop running
522 Finished, // all tasks have finished running
523}
524
525impl ScheduledTask {
526 fn id(&self) -> Option<TaskId> {
527 match self {
528 ScheduledTask::Some(tid) => Some(*tid),
529 _ => None,
530 }
531 }
532
533 fn take(&mut self) -> Self {
534 std::mem::replace(self, ScheduledTask::None)
535 }
536}
537
538/// How an execution ended, which decides how it is torn down (see `ExecutionState::tear_down`).
539#[derive(Debug, PartialEq, Eq, Clone, Copy)]
540enum TeardownKind {
541 /// Every attached task finished.
542 Finished,
543 /// The scheduler stopped the execution before that.
544 Stopped,
545 /// The execution failed. Its failure is raised once it has been torn down.
546 Failed,
547}
548
549impl TeardownKind {
550 /// Whether the execution is abandoned, rather than torn down as if its tasks were cancelled.
551 fn abandons(self) -> bool {
552 self != Self::Finished
553 }
554
555 /// What `ExecutionState::current_task` is between the steps of teardown.
556 fn final_state(self) -> ScheduledTask {
557 match self {
558 Self::Finished => ScheduledTask::Finished,
559 Self::Stopped | Self::Failed => ScheduledTask::Stopped,
560 }
561 }
562}
563
564/// What execution teardown is doing (see `ExecutionState::tear_down`). In each step, the task named
565/// is the current task.
566#[derive(Debug, PartialEq, Eq, Clone, Copy)]
567enum TeardownStep {
568 /// Nothing, between two steps.
569 Idle,
570 /// The task runs on its own stack, cancelled: it drops its future, or the function it never ran.
571 Cancel(TaskId),
572 /// The task's stack is unwound.
573 Unwind(TaskId),
574 /// The task runs on its own stack to finish unwinding a panic (see
575 /// `ExecutionState::finish_unwinding`).
576 FinishUnwinding(TaskId),
577 /// What the task left behind, its task-local values say, is dropped on the executor's stack.
578 Leftovers(TaskId),
579 /// A static is dropped on the executor's stack, with the main thread standing in.
580 Static,
581}
582
583/// The state of execution teardown (see `ExecutionState::tear_down`).
584#[derive(Debug)]
585struct Teardown {
586 kind: TeardownKind,
587 step: TeardownStep,
588 /// The scheduling points that destructors have reached, which the step bound limits.
589 steps: usize,
590 /// The step bound, once a destructor has exceeded it. Then the rest of the execution is
591 /// abandoned (see `next_teardown_job`).
592 exceeded_step_bound: Option<usize>,
593 /// Whether the current task runs on its own stack now, so that it can switch out (see
594 /// `maybe_yield_in_teardown`).
595 on_own_stack: bool,
596 /// The state of the generator that destructors draw random numbers from, instead of the
597 /// scheduler, so that teardown doesn't extend the schedule but is the same when it is replayed.
598 rng: u64,
599}
600
601impl Teardown {
602 fn new(kind: TeardownKind, schedule_len: usize) -> Self {
603 Self {
604 kind,
605 step: TeardownStep::Idle,
606 steps: 0,
607 exceeded_step_bound: None,
608 on_own_stack: false,
609 rng: schedule_len as u64,
610 }
611 }
612
613 /// The next number from the generator (SplitMix64).
614 fn next_u64(&mut self) -> u64 {
615 self.rng = self.rng.wrapping_add(0x9E37_79B9_7F4A_7C15);
616 let mut z = self.rng;
617 z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
618 z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
619 z ^ (z >> 31)
620 }
621}
622
623/// Where execution teardown has got to (see `ExecutionState::next_teardown_job`).
624#[derive(Debug, Default)]
625struct TeardownPlan {
626 /// The task that switched out while it unwound a panic, and is unwinding it still (see
627 /// `ExecutionState::finish_unwinding`).
628 unwinding_task: Option<TaskId>,
629 /// The index in `ExecutionState::tasks` of the next task that teardown hasn't looked at.
630 next_task: usize,
631 /// The tasks that teardown cancelled and that switched out, in the order they did, each with
632 /// whether it blocked (rather than yielded).
633 suspended: Vec<(TaskId, bool)>,
634 /// The unfinished tasks in the middle of user code, whose stacks teardown unwinds or leaks.
635 stacks: VecDeque<TaskId>,
636 /// How many times in a row a task that yielded has been resumed without any task finishing or
637 /// blocking since, while stacks are waiting to be unwound.
638 idle_yields: usize,
639}
640
641/// How many turns each task that yields gets before a stack is unwound, if no task finishes or
642/// blocks meanwhile (see `ExecutionState::next_teardown_job`), unless that would take more than a
643/// quarter of the steps that the step bound leaves. The tasks that yield may wait for a task whose
644/// stack teardown is going to unwind, or that task's destructors may wait for a lock that they
645/// hold; unwinding a stack can't wait.
646const TEARDOWN_YIELD_ROUNDS: usize = 1000;
647
648/// How execution teardown resumes a task on the task's own stack (see
649/// `ExecutionState::run_on_own_stack`).
650#[derive(Debug, PartialEq, Eq, Clone, Copy)]
651enum Resumption {
652 /// Cancel the task (see `Continuation::cancel`).
653 Cancel,
654 /// Resume a cancelled task that switched out.
655 Resume,
656}
657
658/// What execution teardown does next (see `ExecutionState::next_teardown_job`).
659enum TeardownJob {
660 /// Let the task finish unwinding a panic.
661 FinishUnwinding(TaskId),
662 /// Resume the task on its own stack.
663 OnOwnStack(TaskId, Resumption),
664 /// Unwind the task's stack.
665 Unwind(TaskId),
666 /// Leak the task's stack.
667 Leak(TaskId),
668 /// Leak the function of a task that never ran.
669 LeakFunction(TaskId),
670 /// Drop task-local values that a finished task left.
671 TaskLocals(TaskId),
672 /// Drop a static.
673 Static(Box<dyn Any>),
674 /// Every cancelled task that hasn't finished blocked, and nothing can wake them, or teardown is
675 /// abandoning them.
676 Stuck,
677 /// A destructor exceeded the step bound, which this is: abandon the rest of the execution.
678 Abandon(usize),
679 Done,
680}
681
682/// Why execution teardown stops a destructor (see `ExecutionState::maybe_yield_in_teardown`).
683enum TeardownStall {
684 /// It blocked where it cannot wait.
685 Blocked,
686 /// It exceeded the step bound, which this is.
687 ExceededStepBound(usize),
688}
689
690/// How execution teardown is going to fail the test, if it does (see `ExecutionState::tear_down`).
691struct TeardownReport {
692 failure: Option<TeardownFailure>,
693 ignored_panics: usize,
694}
695
696impl TeardownReport {
697 fn new() -> Self {
698 Self {
699 failure: None,
700 ignored_panics: 0,
701 }
702 }
703
704 /// Run `f`, which drops something, and deal with a panic from it.
705 fn catch(&mut self, f: impl FnOnce()) {
706 TEARDOWN_STEP_PANICS.set(0);
707 if let Err(payload) = panic::catch_unwind(panic::AssertUnwindSafe(f)) {
708 self.panicked(payload);
709 }
710 }
711
712 /// A destructor panicked. The first panic while tearing down a finished execution fails the
713 /// test, and teardown ignores any other.
714 fn panicked(&mut self, payload: Box<dyn Any + Send>) {
715 let abandoned = ExecutionState::with(|state| state.teardown().kind.abandons());
716 if !abandoned && self.failure.is_none() {
717 self.failure = Some(TeardownFailure::Unwind(payload));
718 // The panic hook has reported this one, but stays silent for those ignored from now on.
719 TEARDOWN_IGNORES_PANICS.set(true);
720 } else {
721 self.ignored_panics += 1;
722 // Dropping the payload can panic too.
723 if let Err(payload) = panic::catch_unwind(panic::AssertUnwindSafe(move || drop(payload))) {
724 std::mem::forget(payload);
725 }
726 }
727 }
728
729 /// A task that was unwinding a panic when the execution stopped has finished unwinding it. That
730 /// panic fails the test, unless something else already does.
731 fn late_panic(&mut self, payload: Box<dyn Any + Send>) {
732 if self.failure.is_none() {
733 self.failure = Some(TeardownFailure::LatePanic(payload));
734 } else {
735 self.panicked(payload);
736 }
737 }
738
739 /// Fail the test with `message`, unless something else already does.
740 fn fail(&mut self, message: String) {
741 if self.failure.is_none() {
742 self.failure = Some(TeardownFailure::Panic(message));
743 }
744 }
745
746 /// The destructors that teardown of a finished execution has left all blocked, and nothing can
747 /// wake them.
748 fn deadlocked(&mut self, blocked_tasks: Vec<String>) {
749 self.fail(format!(
750 "deadlock while dropping the tasks that were unfinished at the end of the execution! blocked tasks: [{}]",
751 blocked_tasks.join(", ")
752 ));
753 }
754
755 fn finish(self) -> Option<TeardownFailure> {
756 if self.failure.is_some() && self.ignored_panics > 0 {
757 eprintln!(
758 "{} more panics while dropping the tasks that were unfinished at the end of the execution were ignored",
759 self.ignored_panics
760 );
761 }
762 self.failure
763 }
764}
765
766/// Error type for when an `ExecutionState::with` fails
767#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
768pub enum ExecutionStateBorrowError {
769 /// `ExecutionState` is currently not set
770 NotSet,
771 /// We are trying to borrow `ExecutionState` while it is already borrowed
772 AlreadyBorrowed,
773}
774
775impl ExecutionState {
776 fn new(config: Config, scheduler: Rc<RefCell<dyn Scheduler>>) -> Self {
777 let (top_level_dispatch, top_level_dispatch_global_ptr) = tracing::dispatcher::get_default(|dispatch| {
778 // While no thread has a scoped default, `get_default` hands out the global default, and
779 // nested calls get it too rather than `Dispatch::none()`.
780 let is_global = tracing::dispatcher::get_default(|nested| !nested.is::<NoSubscriber>());
781 (dispatch.clone(), is_global.then_some(dispatch as *const _))
782 });
783 Self {
784 config,
785 tasks: SmallVec::new(),
786 current_task: ScheduledTask::None,
787 next_task: ScheduledTask::None,
788 has_yielded: false,
789 context_switches: 0,
790 steps_reset_at: 0,
791 storage: StorageMap::new(),
792 scheduler,
793 teardown: None,
794 failed_task: None,
795 unwinding_task: None,
796 switched_out_unwinding: false,
797 #[cfg(debug_assertions)]
798 has_cleaned_up: false,
799 top_level_span: tracing::Span::current(),
800 top_level_dispatch,
801 top_level_dispatch_global_ptr,
802 runnable_tasks: Vec::with_capacity(DEFAULT_INLINE_TASKS),
803 live_tasks: Vec::with_capacity(DEFAULT_INLINE_TASKS),
804 }
805 }
806
807 /// Invoke a closure with access to the current execution state. Library code uses this to gain
808 /// access to the state of the execution to influence scheduling (e.g. to register a task as
809 /// blocked).
810 #[inline]
811 #[track_caller]
812 pub fn with<F, T>(f: F) -> T
813 where
814 F: FnOnce(&mut ExecutionState) -> T,
815 {
816 Self::try_with(f).unwrap_or_else(|e| {
817 eprintln!("`ExecutionState::try_with` failed with error: {e:?}");
818 eprintln!(
819 "Backtrace for `with`: {:#?}",
820 std::backtrace::Backtrace::force_capture()
821 );
822 match e {
823 ExecutionStateBorrowError::AlreadyBorrowed => panic!("`ExecutionState::with` panicked because `ExecutionState` is already borrowed."),
824 ExecutionStateBorrowError::NotSet => panic!("`ExecutionState::with` panicked because `ExecutionState` is not set. Are you accessing a Shuttle primitive outside of a Shuttle test?"),
825 }
826 })
827 }
828
829 /// Like `with`, but returns None instead of panicking if there is no current ExecutionState or
830 /// if the current ExecutionState is already borrowed.
831 #[inline]
832 #[track_caller]
833 pub fn try_with<F, T>(f: F) -> Result<T, ExecutionStateBorrowError>
834 where
835 F: FnOnce(&mut ExecutionState) -> T,
836 {
837 // Check the level first, which only reads the most verbose level that any subscriber enables,
838 // so that accesses don't pay for the flag when TRACE is off. Asking whether the event is
839 // enabled runs the subscribers' filters, which can access the state too (to read a
840 // task-local, say), so it has to wait until the flag is set, which `trace!` below does.
841 if tracing::level_enabled!(tracing::Level::TRACE) && !TRACING_TRY_WITH.with(|emitting| emitting.replace(true)) {
842 struct Reset;
843 impl Drop for Reset {
844 fn drop(&mut self) {
845 TRACING_TRY_WITH.with(|emitting| emitting.set(false));
846 }
847 }
848 let _reset = Reset;
849 trace!(
850 "ExecutionState::try_with called from {:?}",
851 std::panic::Location::caller()
852 );
853 }
854 if EXECUTION_STATE.is_set() {
855 EXECUTION_STATE.with(|cell| {
856 if let Ok(mut state) = cell.try_borrow_mut() {
857 Ok(f(&mut state))
858 } else {
859 Err(ExecutionStateBorrowError::AlreadyBorrowed)
860 }
861 })
862 } else {
863 Err(ExecutionStateBorrowError::NotSet)
864 }
865 }
866
867 /// A shortcut to get the current task ID
868 pub fn me() -> TaskId {
869 Self::with(|s| s.current().id())
870 }
871
872 /// If there is only one attached, unfinished task and there is at least one detached, unfinished task
873 /// then exiting the attached task will cause the whole execution to exit. As a result, the unfinished
874 /// detached tasks are truncated -- their remaining events will not be executed because the program itself
875 /// has exited. This is relevant because it means that *exiting* a task can be a visible operation
876 /// in that it affects which events are executed.
877 pub fn exit_current_truncates_execution(&self) -> bool {
878 // Strictly speaking, this is only true if there are other runnable detached tasks, but always making the main thread
879 // exit a scheduling point is simpler conceptually
880 if self.current().id() == TaskId::from(0) {
881 return true;
882 }
883
884 // If the current task is detached, then it definitely doesn't truncate the execution
885 if self.current().is_detached() {
886 return false;
887 }
888
889 let mut single_unfinished_attached = false;
890 let mut has_unfinished_detached = false;
891 for t in self.tasks.iter() {
892 let unfinished_attached = !t.finished() && !t.detached;
893 if single_unfinished_attached && unfinished_attached {
894 // there are more than one unfinished attached tasks, so one exiting won't truncate
895 return false;
896 }
897
898 single_unfinished_attached |= unfinished_attached;
899 has_unfinished_detached |= !t.finished() && t.detached;
900 }
901 has_unfinished_detached && single_unfinished_attached
902 }
903
904 fn set_labels_for_new_task(state: &ExecutionState, task_id: TaskId, name: Option<String>) {
905 LABELS.with(|cell| {
906 let mut map = cell.borrow_mut();
907
908 // If parent has labels, inherit them
909 if let Some(parent_task_id) = state.try_current().map(|t| t.id()) {
910 let parent_map = map.get(&parent_task_id);
911 if let Some(parent_map) = parent_map {
912 let mut child_map = parent_map.clone();
913
914 // If the parent has a `ChildLabelFn` set, use that to update the child's Labels
915 if let Some(gen) = parent_map.get::<ChildLabelFn>() {
916 (gen.0)(task_id, &mut child_map);
917 }
918
919 map.insert(task_id, child_map);
920 }
921 }
922
923 // Add any name assigned to the task to its set of Labels
924 if let Some(name) = name {
925 let m = map.entry(task_id).or_default();
926 m.insert(TaskName::from(name));
927 }
928 });
929 }
930
931 // Note: `spawn_thread`, `spawn_main_thread`, and `spawn_future` share some similar logic.
932 // Changes to one of these functions likely need to be propagated to the other two as well.
933 pub fn spawn_main_thread(
934 f: Box<dyn FnOnce() + 'static>,
935 stack_size: usize,
936 caller: &'static Location<'static>,
937 ) -> TaskId {
938 let name = "main-thread".to_string();
939 let mut clock = VectorClock::new();
940
941 let task_id = Self::with(|state| {
942 let parent_span_id = state.top_level_span.id();
943 let task_id = TaskId(state.tasks.len());
944 let tag = state.get_tag_or_default_for_current_task();
945
946 Self::set_labels_for_new_task(state, task_id, Some(name.clone()));
947
948 clock.extend(task_id); // and extend it with an entry for the new thread
949
950 let schedule_len = CurrentSchedule::len();
951
952 let task = Task::from_closure(
953 f,
954 stack_size,
955 task_id,
956 Some(name),
957 clock,
958 parent_span_id,
959 schedule_len,
960 tag,
961 None,
962 TaskSignature::new_parentless(caller),
963 );
964 state.add_task(task);
965
966 task_id
967 });
968 crate::annotations::record_task_created(task_id, false);
969 task_id
970 }
971
972 // Note: `spawn_thread`, `spawn_main_thread`, and `spawn_future` share some similar logic.
973 // Changes to one of these functions likely need to be propagated to the other two as well.
974 /// Spawn a new task for a future. This doesn't create a yield point; the caller should do that
975 /// if it wants to give the new task a chance to run immediately.
976 pub fn spawn_future<F>(
977 future: F,
978 stack_size: usize,
979 name: Option<String>,
980 caller: &'static Location<'static>,
981 ) -> TaskId
982 where
983 F: Future<Output = ()> + 'static,
984 {
985 thread::switch();
986 let task_id = Self::with(|state| {
987 let schedule_len = CurrentSchedule::len();
988 let parent_span_id = state.top_level_span.id();
989
990 let task_id = TaskId(state.tasks.len());
991 let tag = state.get_tag_or_default_for_current_task();
992
993 Self::set_labels_for_new_task(state, task_id, name.clone());
994
995 let clock = state.increment_clock_mut(); // Increment the parent's clock
996 clock.extend(task_id); // and extend it with an entry for the new task
997
998 let task = Task::from_future(
999 future,
1000 stack_size,
1001 task_id,
1002 name,
1003 clock.clone(),
1004 parent_span_id,
1005 schedule_len,
1006 tag,
1007 Some(state.current().id()),
1008 state.current_mut().signature.new_child(caller),
1009 );
1010
1011 state.add_task(task);
1012
1013 task_id
1014 });
1015 crate::annotations::record_task_created(task_id, true);
1016 task_id
1017 }
1018
1019 // Note: `spawn_thread`, `spawn_main_thread`, and `spawn_future` share some similar logic.
1020 // Changes to one of these functions likely need to be propagated to the other two as well.
1021 pub fn spawn_thread(
1022 f: Box<dyn FnOnce() + 'static>,
1023 stack_size: usize,
1024 name: Option<String>,
1025 mut initial_clock: Option<VectorClock>,
1026 caller: &'static Location<'static>,
1027 ) -> TaskId {
1028 thread::switch();
1029 let task_id = Self::with(|state| {
1030 let parent_span_id = state.top_level_span.id();
1031 let task_id = TaskId(state.tasks.len());
1032 let tag = state.get_tag_or_default_for_current_task();
1033
1034 Self::set_labels_for_new_task(state, task_id, name.clone());
1035
1036 let clock = if let Some(ref mut clock) = initial_clock {
1037 clock
1038 } else {
1039 // Inherit the clock of the parent thread (which spawned this task)
1040 state.increment_clock_mut()
1041 };
1042 clock.extend(task_id); // and extend it with an entry for the new thread
1043 let clock = clock.clone();
1044
1045 let task = Task::from_closure(
1046 f,
1047 stack_size,
1048 task_id,
1049 name,
1050 clock,
1051 parent_span_id,
1052 CurrentSchedule::len(),
1053 tag,
1054 Some(state.current().id()),
1055 state.current_mut().signature.new_child(caller),
1056 );
1057 state.add_task(task);
1058
1059 task_id
1060 });
1061 crate::annotations::record_task_created(task_id, false);
1062 task_id
1063 }
1064
1065 /// Tear down the execution, which ended as `kind` says, before it goes out of `EXECUTION_STATE`
1066 /// scope: drop what its unfinished tasks leave behind, the tasks' task-local values, and the
1067 /// execution's statics, while their destructors can still use Shuttle. Returns how that fails
1068 /// the test, if it does.
1069 ///
1070 /// A finished execution's unfinished tasks (detached future tasks, and tasks that destructors
1071 /// spawn during teardown) are torn down as if the execution cancelled them as it ended. While a
1072 /// task is torn down, it is the current task: its destructors can use Shuttle as the running
1073 /// task can, and what they do is attributed to it.
1074 ///
1075 /// * A task that never ran drops its function, and a future task that is parked between polls,
1076 /// the usual state of an unfinished future task, drops its future, the way an async runtime
1077 /// drops the futures of its tasks when it shuts down (see `Continuation::cancel`). They do
1078 /// this on their own stacks, and then drop their task-local values. So a destructor that
1079 /// blocks, on a lock that another unfinished task holds say, waits for teardown to drop that
1080 /// task. A panic is an ordinary panic, which fails the test once teardown is done.
1081 /// * A future task that stopped in the middle of `poll` has user code on its stack, which only
1082 /// unwinding the stack can drop. Teardown unwinds these stacks one at a time, once no other
1083 /// task can make progress, so that the locks that the other tasks held are free. Their
1084 /// destructors can use Shuttle too, but cannot wait: one that blocks, or panics, while a stack
1085 /// is unwound aborts the process, as any panic during unwinding does. A task that catches the
1086 /// unwind is unwound again at its next scheduling point.
1087 ///
1088 /// Nothing is scheduled. Teardown resumes the tasks it cancelled itself, in a fixed order that
1089 /// doesn't extend the schedule, so it is the same when the schedule is replayed (see
1090 /// `next_teardown_job`). A destructor that blocks waits until another destructor wakes it, and
1091 /// one that yields lets the others run; but if every destructor left blocks, nothing can wake
1092 /// them, and that fails the test like a deadlock. The step bound applies to destructors too (see
1093 /// `maybe_yield_in_teardown`). The execution's statics go last, as if the main thread dropped
1094 /// them, once nothing else is left that could still use them.
1095 ///
1096 /// A stopped or failed execution is abandoned instead. Shuttle's own destructors skip their
1097 /// bookkeeping (see `should_stop`), and panics are ignored, so that a failure is what gets
1098 /// reported. The stacks of its unfinished tasks are leaked. A failed execution's functions of
1099 /// tasks that never ran, and futures of parked future tasks, are leaked too, unless
1100 /// `UngracefulShutdownConfig::continuation_function_behavior` says to drop them, in which case
1101 /// they are cancelled as above. A stopped execution's functions of tasks that never ran are
1102 /// dropped that way. Every function that is dropped goes before any stack is freed, since a
1103 /// scoped thread's function borrows from its parent's stack. Task-local values and statics are
1104 /// dropped as in a finished execution.
1105 ///
1106 /// A task that was unwinding a panic when a stopped execution stopped finishes unwinding it first,
1107 /// and the panic fails the test (see `finish_unwinding`). A failed execution's is leaked like the
1108 /// rest, as running more of a failed execution risks a panic that aborts the process. If teardown
1109 /// can't tell which task it is (see `record_unwinding_task`), that fails the test too, and the
1110 /// rest of the execution is abandoned: unwinding that task's stack again would abort the process.
1111 fn tear_down(kind: TeardownKind) -> Option<TeardownFailure> {
1112 let mut plan = TeardownPlan::default();
1113 let mut report = TeardownReport::new();
1114 // Here, on the executor's stack, `std::thread::panicking()` says whether a task that switched
1115 // out is unwinding a panic still (see `record_unwinding_task`).
1116 let unwinding = std::thread::panicking();
1117 let lost_panic = Self::with(|state| {
1118 state.current_task = kind.final_state();
1119 state.next_task = ScheduledTask::None;
1120 state.teardown = Some(Teardown::new(kind, CurrentSchedule::len()));
1121 if kind != TeardownKind::Failed && unwinding {
1122 let unwinding_task = state.unwinding_task.take().map(|(id, _)| id);
1123 plan.unwinding_task = unwinding_task.filter(|&id| state.get(id).suspended());
1124 }
1125 let lost_panic = unwinding && plan.unwinding_task.is_none() && kind != TeardownKind::Failed;
1126 if lost_panic {
1127 // So that no stack is unwound.
1128 state.abandon_teardown();
1129 }
1130 lost_panic
1131 });
1132 if lost_panic {
1133 report.fail(
1134 "a task was still unwinding a panic when the execution ended, but it panicked while another task was \
1135 unwinding a panic, so Shuttle can't tell which task it is, to let it finish unwinding"
1136 .into(),
1137 );
1138 }
1139 // The panic hook goes back to reporting panics when teardown is done, however it ends.
1140 struct ReportPanicsAgain;
1141 impl Drop for ReportPanicsAgain {
1142 fn drop(&mut self) {
1143 TEARDOWN_IGNORES_PANICS.set(false);
1144 }
1145 }
1146 let report_panics_again = ReportPanicsAgain;
1147 TEARDOWN_IGNORES_PANICS.set(Self::with(|state| state.teardown().kind.abandons()));
1148
1149 loop {
1150 match Self::with(|state| state.next_teardown_job(&mut plan)) {
1151 TeardownJob::FinishUnwinding(id) => Self::finish_unwinding(id, &mut report),
1152 TeardownJob::OnOwnStack(id, resumption) => {
1153 Self::run_on_own_stack(id, resumption, &mut plan, &mut report)
1154 }
1155 TeardownJob::Unwind(id) => Self::unwind_task(id, &mut report),
1156 TeardownJob::Leak(id) => Self::leak_task(id, &mut report),
1157 TeardownJob::LeakFunction(id) => {
1158 Self::with(|state| {
1159 let task = state.get(id);
1160 let continuation = task
1161 .continuation
1162 .as_ref()
1163 .expect("a task that never ran has a continuation");
1164 continuation.borrow_mut().leak_function();
1165 });
1166 Self::finish_torn_down_task(id, &mut report);
1167 }
1168 TeardownJob::TaskLocals(id) => Self::drop_leftover_task_locals(id, &mut report),
1169 TeardownJob::Static(value) => Self::drop_static(value, &mut report),
1170 TeardownJob::Stuck => {
1171 let (deadlocked, blocked_tasks) = Self::with(|state| {
1172 let teardown = state.teardown();
1173 let deadlocked = !teardown.kind.abandons() && teardown.exceeded_step_bound.is_none();
1174 let blocked = plan
1175 .suspended
1176 .iter()
1177 .map(|&(id, _)| state.get(id).format_for_deadlock());
1178 let blocked = blocked.collect::<Vec<_>>();
1179 state.abandon_teardown();
1180 (deadlocked, blocked)
1181 });
1182 TEARDOWN_IGNORES_PANICS.set(true);
1183 if deadlocked {
1184 report.deadlocked(blocked_tasks);
1185 }
1186 for (id, _) in std::mem::take(&mut plan.suspended) {
1187 Self::leak_task(id, &mut report);
1188 }
1189 }
1190 TeardownJob::Abandon(max_steps) => {
1191 // The panic that stopped the destructor fails the test, unless the destructor
1192 // caught it.
1193 report.fail(format!(
1194 "a destructor exceeded the step bound ({max_steps}) while it was being dropped at the end of \
1195 the execution"
1196 ));
1197 Self::with(|state| state.abandon_teardown());
1198 TEARDOWN_IGNORES_PANICS.set(true);
1199 }
1200 TeardownJob::Done => {
1201 // What the tasks still hold goes last, as each task (see `drop_task_leftovers`).
1202 // Destructors can spawn tasks, which are torn down too.
1203 if !Self::drop_task_leftovers(&mut report) {
1204 break;
1205 }
1206 }
1207 }
1208 }
1209
1210 // The tasks' labels, and tags that no task holds, while `EXECUTION_STATE` is set. Their
1211 // destructors may look up tasks: formatting a `TaskId` looks up its name, say. In the order
1212 // the tasks were created, so that the destructors run in the same order every time.
1213 let tags = TASK_ID_TO_TAGS.with(|cell| std::mem::take(&mut *cell.borrow_mut()));
1214 let mut tags = tags.into_iter().collect::<Vec<_>>();
1215 tags.sort_unstable_by_key(|&(id, _)| id);
1216 for (_, tag) in tags {
1217 report.catch(move || drop(tag));
1218 }
1219 let labels = LABELS.with(|cell| std::mem::take(&mut *cell.borrow_mut()));
1220 let mut labels = labels.into_iter().collect::<Vec<_>>();
1221 labels.sort_unstable_by_key(|&(id, _)| id);
1222 for (_, labels) in labels {
1223 report.catch(move || drop(labels));
1224 }
1225 let tasks = Self::with(|state| std::mem::take(&mut state.tasks));
1226 report.catch(move || drop(tasks));
1227
1228 Self::with(|state| {
1229 state.teardown = None;
1230 #[cfg(debug_assertions)]
1231 {
1232 state.has_cleaned_up = true;
1233 }
1234 });
1235 drop(report_panics_again);
1236 report.finish()
1237 }
1238
1239 /// Choose what execution teardown does next (see `tear_down`).
1240 fn next_teardown_job(&mut self, plan: &mut TeardownPlan) -> TeardownJob {
1241 let teardown = self.teardown();
1242 let kind = teardown.kind;
1243 // Once a destructor has exceeded the step bound, which fails the test, the rest of the
1244 // execution is abandoned: the tasks that are left are leaked, as destructors that spawn tasks
1245 // whose destructors spawn tasks, say, would go on forever.
1246 let exceeded_step_bound = teardown.exceeded_step_bound;
1247 if let Some(max_steps) = exceeded_step_bound.filter(|_| !kind.abandons()) {
1248 return TeardownJob::Abandon(max_steps);
1249 }
1250 let abandoned = exceeded_step_bound.is_some();
1251 if abandoned && !plan.suspended.is_empty() {
1252 return TeardownJob::Stuck;
1253 }
1254
1255 // A task that was unwinding a panic when the execution stopped finishes unwinding it first.
1256 if let Some(id) = plan.unwinding_task.take() {
1257 return TeardownJob::FinishUnwinding(id);
1258 }
1259
1260 // A cancelled task that blocked, and has been woken since, goes first.
1261 if let Some(i) = plan
1262 .suspended
1263 .iter()
1264 .position(|&(id, blocked)| blocked && self.get(id).runnable())
1265 {
1266 plan.idle_yields = 0;
1267 return TeardownJob::OnOwnStack(plan.suspended.remove(i).0, Resumption::Resume);
1268 }
1269
1270 // Then the tasks that teardown hasn't looked at yet, in the order they were created.
1271 let leaks = UNGRACEFUL_SHUTDOWN_CONFIG.get().continuation_function_behavior.leaks();
1272 let (drop_functions, drop_parked_futures) = match kind {
1273 _ if abandoned => (false, false),
1274 TeardownKind::Finished => (true, true),
1275 TeardownKind::Stopped => (true, false),
1276 TeardownKind::Failed => (!leaks, !leaks),
1277 };
1278 while let Some(task) = self.tasks.get(plan.next_task) {
1279 plan.next_task += 1;
1280 let id = task.id();
1281 if task.finished() {
1282 // A task can set task-local values after it dropped its own ones: the result it
1283 // keeps for its `JoinHandle` is dropped after them if the handle is dropped while
1284 // they are (see `Wrapper::finish` and `thread_fn`).
1285 if task.has_locals() {
1286 return TeardownJob::TaskLocals(id);
1287 }
1288 } else if task.never_ran() {
1289 plan.idle_yields = 0;
1290 return if drop_functions {
1291 TeardownJob::OnOwnStack(id, Resumption::Cancel)
1292 } else {
1293 TeardownJob::LeakFunction(id)
1294 };
1295 } else if drop_parked_futures && task.parked_between_polls() {
1296 plan.idle_yields = 0;
1297 return TeardownJob::OnOwnStack(id, Resumption::Cancel);
1298 } else {
1299 // The task is in the middle of user code.
1300 plan.stacks.push_back(id);
1301 }
1302 }
1303
1304 // A cancelled task that yielded runs again, in turn. While stacks are waiting to be unwound,
1305 // for a number of turns only, if no task finishes or blocks meanwhile (see
1306 // `TEARDOWN_YIELD_ROUNDS`).
1307 let unwinds_wait = kind == TeardownKind::Finished && !abandoned && !plan.stacks.is_empty();
1308 let yielded = |&(id, blocked): &(TaskId, bool)| !blocked && self.get(id).runnable();
1309 let num_yielded = plan.suspended.iter().filter(|task| yielded(task)).count();
1310 if num_yielded > 0 && (!unwinds_wait || plan.idle_yields < self.teardown_yield_budget(num_yielded)) {
1311 plan.idle_yields += 1;
1312 let i = plan.suspended.iter().position(yielded).unwrap();
1313 return TeardownJob::OnOwnStack(plan.suspended.remove(i).0, Resumption::Resume);
1314 }
1315
1316 // A finished execution's tasks in the middle of `poll` are unwound now, once no other task
1317 // can make progress.
1318 if unwinds_wait {
1319 plan.idle_yields = 0;
1320 return TeardownJob::Unwind(plan.stacks.pop_front().unwrap());
1321 }
1322
1323 // A static goes only once no cancelled task is left suspended, whose stack could still refer
1324 // to it.
1325 if !plan.suspended.is_empty() {
1326 return TeardownJob::Stuck;
1327 }
1328
1329 if let Some(value) = self.storage.pop() {
1330 return TeardownJob::Static(value);
1331 }
1332
1333 // An abandoned execution's stacks are leaked last, once the functions of the tasks that never
1334 // ran, which could borrow from them, are gone.
1335 if let Some(id) = plan.stacks.pop_front() {
1336 return TeardownJob::Leak(id);
1337 }
1338
1339 TeardownJob::Done
1340 }
1341
1342 /// How many turns in a row the tasks that yield get while stacks are waiting to be unwound (see
1343 /// `TEARDOWN_YIELD_ROUNDS`).
1344 fn teardown_yield_budget(&self, num_yielded: usize) -> usize {
1345 let rounds = TEARDOWN_YIELD_ROUNDS.saturating_mul(num_yielded);
1346 match self.teardown_step_bound() {
1347 Some(max_steps) => rounds.min(max_steps.saturating_sub(self.teardown().steps) / 4),
1348 None => rounds,
1349 }
1350 }
1351
1352 /// Abandon the rest of a finished execution's teardown, which has failed (see `tear_down`).
1353 fn abandon_teardown(&mut self) {
1354 let teardown = self.teardown_mut();
1355 if teardown.kind == TeardownKind::Finished {
1356 teardown.kind = TeardownKind::Failed;
1357 self.current_task = TeardownKind::Failed.final_state();
1358 }
1359 }
1360
1361 /// Resume a task on its own stack (see `Resumption`), until it finishes or switches out.
1362 fn run_on_own_stack(id: TaskId, resumption: Resumption, plan: &mut TeardownPlan, report: &mut TeardownReport) {
1363 let continuation = Self::with(|state| {
1364 state.begin_teardown_step(TeardownStep::Cancel(id));
1365 let task = state.get_mut(id);
1366 if resumption == Resumption::Cancel {
1367 // A parked future task is asleep, but it runs now, to drop its future.
1368 task.unblock();
1369 }
1370 task.continuation
1371 .clone()
1372 .expect("an unfinished task has a continuation")
1373 });
1374
1375 Execution::enter_task_span();
1376 let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
1377 let mut continuation = continuation.borrow_mut();
1378 // Only now can the task switch out (see `maybe_yield_in_teardown`).
1379 Self::with(|state| state.teardown_mut().on_own_stack = true);
1380 if resumption == Resumption::Cancel {
1381 continuation.cancel()
1382 } else {
1383 continuation.resume()
1384 }
1385 }));
1386 Self::with(|state| state.teardown_mut().on_own_stack = false);
1387 Execution::exit_task_span(matches!(result, Ok(false)));
1388 drop(continuation);
1389 // A panic's payload goes while the step is under way, so its destructor runs as the task.
1390 let finished = result.unwrap_or_else(|payload| {
1391 report.panicked(payload);
1392 true
1393 });
1394 let blocked = Self::with(|state| {
1395 state.end_teardown_step();
1396 !state.get(id).runnable()
1397 });
1398
1399 if finished {
1400 plan.idle_yields = 0;
1401 Self::finish_torn_down_task(id, report);
1402 } else {
1403 // A destructor blocked, or yielded (see `maybe_yield_in_teardown`).
1404 if blocked {
1405 plan.idle_yields = 0;
1406 }
1407 plan.suspended.push((id, blocked));
1408 }
1409 }
1410
1411 /// Let a task that switched out while it unwound a panic, and is unwinding it still when a
1412 /// stopped execution stopped, finish unwinding it, on its own stack (see `tear_down`). The panic
1413 /// then fails the test. The task runs until it has caught the panic, if it does, but no further.
1414 /// A task that blocks while the panic unwinds, or exceeds the step bound, waits for tasks that no
1415 /// longer run, but cannot be stopped with a panic, as that would abort the process. It switches
1416 /// out instead (see `maybe_yield_in_teardown`), and fails the test. Then its stack is leaked,
1417 /// which leaves the thread panicking.
1418 fn finish_unwinding(id: TaskId, report: &mut TeardownReport) {
1419 let (continuation, blocked) = Self::with(|state| {
1420 state.begin_teardown_step(TeardownStep::FinishUnwinding(id));
1421 let task = state.get(id);
1422 let continuation = task
1423 .continuation
1424 .clone()
1425 .expect("an unfinished task has a continuation");
1426 (continuation, !task.runnable())
1427 });
1428
1429 // A task that blocked already waits for a task that no longer runs.
1430 let result = if blocked {
1431 Ok(false)
1432 } else {
1433 Execution::enter_task_span();
1434 let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
1435 let mut continuation = continuation.borrow_mut();
1436 Self::with(|state| state.teardown_mut().on_own_stack = true);
1437 continuation.resume()
1438 }));
1439 Self::with(|state| state.teardown_mut().on_own_stack = false);
1440 Execution::exit_task_span(matches!(result, Ok(false)));
1441 result
1442 };
1443 drop(continuation);
1444
1445 // On the executor's stack, `std::thread::panicking()` says whether the task is unwinding the
1446 // panic still (see `record_unwinding_task`).
1447 let stuck = matches!(result, Ok(false)) && std::thread::panicking();
1448 // A panic's payload goes while the step is under way, so its destructor runs as the task.
1449 let finished = result.unwrap_or_else(|payload| {
1450 report.late_panic(payload);
1451 true
1452 });
1453 if stuck {
1454 let (name, exceeded_step_bound) =
1455 Self::with(|state| (state.get(id).display_name(), state.teardown().exceeded_step_bound));
1456 let why = match exceeded_step_bound {
1457 Some(max_steps) => format!("exceeded the step bound ({max_steps})"),
1458 None => "blocked".into(),
1459 };
1460 report.fail(format!(
1461 "{name} was unwinding a panic when the execution stopped, and {why} before it had finished unwinding it"
1462 ));
1463 }
1464 Self::with(|state| state.end_teardown_step());
1465
1466 if finished {
1467 Self::finish_torn_down_task(id, report);
1468 } else {
1469 // The task caught the panic, or is stuck.
1470 Self::leak_task(id, report);
1471 }
1472 }
1473
1474 /// Unwind the stack of a finished execution's task that is in the middle of `poll` (see
1475 /// `tear_down`).
1476 fn unwind_task(id: TaskId, report: &mut TeardownReport) {
1477 let continuation = Self::with(|state| {
1478 state.begin_teardown_step(TeardownStep::Unwind(id));
1479 let task = state.get_mut(id);
1480 // The task may have been blocked, but it runs now, to drop what is on its stack.
1481 task.unblock();
1482 task.take_continuation()
1483 });
1484 let continuation = continuation.expect("an unfinished task has a continuation");
1485
1486 // A panic while the stack unwinds aborts the process, so the panic hook reports every one.
1487 let ignoring_panics = TEARDOWN_IGNORES_PANICS.replace(false);
1488 // Among other things, this reinstates the task's default dispatcher, so that the guards on
1489 // its stack restore their priors in order as it unwinds (see `ParkedDefault`).
1490 Execution::enter_task_span();
1491 let result = panic::catch_unwind(panic::AssertUnwindSafe(move || {
1492 let continuation = Rc::try_unwrap(continuation).map_err(|_| ());
1493 let mut continuation = continuation.expect("teardown owns the continuation").into_inner();
1494 // Only now can the task switch, if it catches the unwind (see `maybe_yield_in_teardown`).
1495 Self::with(|state| state.teardown_mut().on_own_stack = true);
1496 continuation.unwind_stack();
1497 }));
1498 Self::with(|state| state.teardown_mut().on_own_stack = false);
1499 Execution::exit_task_span(false);
1500 TEARDOWN_IGNORES_PANICS.set(ignoring_panics);
1501 if let Err(payload) = result {
1502 report.panicked(payload);
1503 }
1504 Self::with(|state| {
1505 state.get_mut(id).lose_stack();
1506 state.end_teardown_step();
1507 });
1508
1509 Self::finish_torn_down_task(id, report);
1510 }
1511
1512 /// Leak the stack of an abandoned execution's task (see `tear_down`).
1513 fn leak_task(id: TaskId, report: &mut TeardownReport) {
1514 let (parked_default, continuation) = Self::with(|state| {
1515 let task = state.get_mut(id);
1516 task.lose_stack();
1517 (task.parked_default.take(), task.take_continuation())
1518 });
1519 // The parked default has to go before the task's stack does (see `ParkedDefault`).
1520 drop(parked_default);
1521 if let Some(continuation) = continuation {
1522 let continuation = Rc::try_unwrap(continuation).map_err(|_| ());
1523 let mut continuation = continuation.expect("teardown owns the continuation").into_inner();
1524 continuation.leak_stack();
1525 }
1526
1527 Self::finish_torn_down_task(id, report);
1528 }
1529
1530 /// Finish a task that execution teardown is done with, and drop the task-local values it left.
1531 fn finish_torn_down_task(id: TaskId, report: &mut TeardownReport) {
1532 crate::annotations::record_teardown_step(id);
1533 crate::annotations::record_task_terminated();
1534 Self::with(|state| {
1535 let task = state.get_mut(id);
1536 task.finish();
1537 // A task that joins this one can go on (see `JoinHandle::join`).
1538 if let Some(waiter) = task.take_waiter() {
1539 if !state.get(waiter).finished() {
1540 state.get_mut(waiter).unblock();
1541 }
1542 }
1543 });
1544 Self::drop_leftover_task_locals(id, report);
1545 }
1546
1547 /// Drop the task-local values that a task left, as the task, once its stack is gone (see
1548 /// `tear_down` and `Task::stand_in`).
1549 fn drop_leftover_task_locals(id: TaskId, report: &mut TeardownReport) {
1550 let state_before = Self::with(|state| {
1551 if !state.get(id).has_locals() {
1552 return None;
1553 }
1554 state.begin_teardown_step(TeardownStep::Leftovers(id));
1555 Some(state.get_mut(id).stand_in())
1556 });
1557 let Some(state_before) = state_before else {
1558 return;
1559 };
1560
1561 // See `pop_local` for why this loop looks slightly funky.
1562 while let Some(local) = Self::with(|state| state.get_mut(id).pop_local()) {
1563 report.catch(move || drop(local));
1564 }
1565
1566 Self::with(|state| {
1567 state.get_mut(id).stop_standing_in(state_before);
1568 state.end_teardown_step();
1569 });
1570 }
1571
1572 /// Drop what the tasks still hold whose destructors may run user code, at the end of teardown:
1573 /// their spans, which a `tracing` subscriber sees close, and their tags. Each task's go as the
1574 /// task (see `Task::stand_in`). Returns whether there was anything to drop.
1575 fn drop_task_leftovers(report: &mut TeardownReport) -> bool {
1576 let ids = Self::with(|state| {
1577 TASK_ID_TO_TAGS.with(|tags| {
1578 let tags = tags.borrow();
1579 let has_tag = |id| !tags.is_empty() && tags.contains_key(&id);
1580 let tasks = state
1581 .tasks
1582 .iter()
1583 .filter(|task| task.has_leftovers() || has_tag(task.id()));
1584 tasks.map(Task::id).collect::<Vec<_>>()
1585 })
1586 });
1587 for &id in &ids {
1588 let (state_before, leftovers) = Self::with(|state| {
1589 state.begin_teardown_step(TeardownStep::Leftovers(id));
1590 // `current::set_tag_for_task` keeps a reference to the tag too, which can be the last.
1591 let tag = TASK_ID_TO_TAGS.with(|tags| tags.borrow_mut().remove(&id));
1592 let task = state.get_mut(id);
1593 (task.stand_in(), (task.take_leftovers(), tag))
1594 });
1595 report.catch(move || drop(leftovers));
1596 Self::with(|state| {
1597 state.get_mut(id).stop_standing_in(state_before);
1598 state.end_teardown_step();
1599 });
1600 }
1601 !ids.is_empty()
1602 }
1603
1604 /// Drop one of the execution's statics, as the main thread (see `tear_down`).
1605 fn drop_static(value: Box<dyn Any>, report: &mut TeardownReport) {
1606 let main_thread = TaskId(0);
1607 let state_before = Self::with(|state| {
1608 state.begin_teardown_step(TeardownStep::Static);
1609 state.get_mut(main_thread).stand_in()
1610 });
1611
1612 report.catch(move || drop(value));
1613 // The static's destructor may have set task-local values of the main thread's, which go with
1614 // it.
1615 while let Some(local) = Self::with(|state| state.get_mut(main_thread).pop_local()) {
1616 report.catch(move || drop(local));
1617 }
1618
1619 Self::with(|state| {
1620 state.get_mut(main_thread).stop_standing_in(state_before);
1621 state.end_teardown_step();
1622 });
1623 }
1624
1625 /// Make `step` what execution teardown is doing, with its task as the current task (see
1626 /// `tear_down`).
1627 fn begin_teardown_step(&mut self, step: TeardownStep) {
1628 let id = match step {
1629 TeardownStep::Cancel(id)
1630 | TeardownStep::Unwind(id)
1631 | TeardownStep::FinishUnwinding(id)
1632 | TeardownStep::Leftovers(id) => id,
1633 // The main thread stands in for statics.
1634 TeardownStep::Static => TaskId(0),
1635 TeardownStep::Idle => unreachable!("teardown is always doing something during a step"),
1636 };
1637 self.current_task = ScheduledTask::Some(id);
1638 self.teardown_mut().step = step;
1639 TEARDOWN_STEP_PANICS.set(0);
1640 crate::annotations::record_teardown_step(id);
1641 }
1642
1643 /// End the step that `begin_teardown_step` began, if one is under way.
1644 fn end_teardown_step(&mut self) {
1645 let teardown = self.teardown_mut();
1646 teardown.step = TeardownStep::Idle;
1647 self.current_task = teardown.kind.final_state();
1648 }
1649
1650 fn teardown(&self) -> &Teardown {
1651 self.teardown.as_ref().expect("the execution is being torn down")
1652 }
1653
1654 fn teardown_mut(&mut self) -> &mut Teardown {
1655 self.teardown.as_mut().expect("the execution is being torn down")
1656 }
1657
1658 /// Whether the panic hook is to stay silent about a panic with `payload`, as execution teardown
1659 /// ignores the panics it catches now (see `tear_down`). A panic in a destructor while another
1660 /// panic unwinds aborts the process. So the hook reports the panic that says so, and any panic
1661 /// after the first in a step, or in what `TeardownReport::catch` drops, which may cause that.
1662 pub(crate) fn teardown_ignores_panic(payload: &dyn Any) -> bool {
1663 let panics = TEARDOWN_STEP_PANICS.get().saturating_add(1);
1664 TEARDOWN_STEP_PANICS.set(panics);
1665 let aborts = payload.downcast_ref::<&str>() == Some(&"panic in a destructor during cleanup");
1666 TEARDOWN_IGNORES_PANICS.get() && panics == 1 && !aborts
1667 }
1668
1669 /// Whether execution teardown is unwinding the current task's stack (see `tear_down`). That is
1670 /// not a panic, although `std::thread::panicking()` says otherwise, so it shouldn't poison a
1671 /// lock, say.
1672 pub fn unwinding_for_teardown() -> bool {
1673 Self::try_with(|state| {
1674 state
1675 .teardown
1676 .as_ref()
1677 .is_some_and(|teardown| matches!(teardown.step, TeardownStep::Unwind(_)))
1678 })
1679 .unwrap_or(false)
1680 }
1681
1682 /// Drop the current task's task-local values, as a task does when it finishes. A destructor can
1683 /// set another task-local value, which is dropped too (see `Task::pop_local`).
1684 pub fn drop_task_locals() {
1685 // See `pop_local` for why this loop looks slightly funky.
1686 while let Some(local) = Self::with(|state| state.current_mut().pop_local()) {
1687 tracing::trace!("dropping task-local value {:p}", local);
1688 drop(local);
1689 }
1690 }
1691
1692 /// Stop a destructor that execution teardown cannot let go on (see `maybe_yield_in_teardown`).
1693 #[cold]
1694 fn stop_destructor(stall: TeardownStall) -> ! {
1695 if std::thread::panicking() {
1696 // This aborts the process, so report it in any case.
1697 TEARDOWN_IGNORES_PANICS.set(false);
1698 }
1699 let (name, step) = Self::with(|state| (state.current().display_name(), state.teardown().step));
1700 let who = match step {
1701 TeardownStep::Static => format!("A static's destructor, dropped as {name},"),
1702 TeardownStep::Leftovers(_) => format!("A destructor of something that {name} left behind"),
1703 _ => name,
1704 };
1705 match stall {
1706 TeardownStall::Blocked => {
1707 let reason = if std::thread::panicking() || matches!(step, TeardownStep::Unwind(_)) {
1708 "while unwinding a stack, which cannot be suspended"
1709 } else {
1710 "outside any task's stack, where it cannot wait"
1711 };
1712 panic!(
1713 "{who} blocked while it was being dropped at the end of the execution, {reason}. A destructor made a \
1714 blocking call, such as locking a mutex that another task holds, receiving from an empty channel, \
1715 joining a task, or `block_on` on a future that cannot complete."
1716 )
1717 }
1718 TeardownStall::ExceededStepBound(max_steps) => panic!(
1719 "{who} exceeded the step bound ({max_steps}) while it was being dropped at the end of the execution. A \
1720 destructor may be spinning, waiting for something that no longer happens: once the execution is over, \
1721 only destructors run."
1722 ),
1723 }
1724 }
1725
1726 /// Determine whether the execution has finished.
1727 pub fn is_finished(&self) -> bool {
1728 self.current_task == ScheduledTask::Stopped || self.current_task == ScheduledTask::Finished
1729 }
1730
1731 /// Invoke the scheduler to decide which task to schedule next. Returns true if the chosen task
1732 /// is different from the currently running task, indicating that the current task should yield
1733 /// its execution.
1734 pub fn maybe_yield() -> bool {
1735 let decision = Self::with(|state| {
1736 if state.teardown.is_some() {
1737 return state.maybe_yield_in_teardown();
1738 }
1739
1740 if std::thread::panicking() {
1741 if !state.switched_out_unwinding {
1742 state.record_unwinding_task();
1743 }
1744 return Ok(true);
1745 }
1746 if state.switched_out_unwinding {
1747 // No task is unwinding a panic, as the tasks share the OS thread.
1748 state.forget_unwinding_task();
1749 }
1750
1751 debug_assert!(
1752 matches!(state.current_task, ScheduledTask::Some(_)) && state.next_task == ScheduledTask::None,
1753 "we're inside a task and scheduler should not yet have run"
1754 );
1755
1756 let result = state.schedule();
1757 // If scheduling failed, yield so that the outer scheduling loop can handle it.
1758 if result.is_err() {
1759 return Ok(true);
1760 }
1761
1762 // If the next task is the same as the current one, we can skip the context switch
1763 // and just advance to the next task immediately.
1764 if state.current_task == state.next_task {
1765 state.advance_to_next_task();
1766 Ok(false)
1767 } else {
1768 Ok(true)
1769 }
1770 });
1771 // Panic outside of `Self::with`, so the panic hook can name the task.
1772 decision.unwrap_or_else(|stall| Self::stop_destructor(stall))
1773 }
1774
1775 /// The current task switches out while it unwinds a panic, and no task was unwinding one when it
1776 /// was resumed (see `switched_out_unwinding`), so the panic is its own. (`std::thread::panicking()`
1777 /// can't tell which task panics: tasks share the OS thread. Between steps, on the executor's
1778 /// stack, it says exactly whether a task that switched out is unwinding a panic.) Until the task
1779 /// has finished unwinding the panic, the execution waits for it, even if it is detached.
1780 /// Otherwise the execution could end before the task resumes, and its panic would be lost. There
1781 /// is one such task at most: while it is unwinding, a task that is resumed can't tell a panic of
1782 /// its own from it.
1783 #[cold]
1784 fn record_unwinding_task(&mut self) {
1785 debug_assert!(self.unwinding_task.is_none());
1786 self.switched_out_unwinding = true;
1787 if let Some(me) = self.current_task.id() {
1788 let detached = std::mem::replace(&mut self.get_mut(me).detached, false);
1789 self.unwinding_task = Some((me, detached));
1790 }
1791 }
1792
1793 /// No task is unwinding a panic any more: the task that was recorded doing so (see
1794 /// `record_unwinding_task`) is detached again if it was before.
1795 #[cold]
1796 fn forget_unwinding_task(&mut self) {
1797 self.switched_out_unwinding = false;
1798 if let Some((task, detached)) = self.unwinding_task.take() {
1799 self.get_mut(task).detached = detached;
1800 }
1801 }
1802
1803 /// Detach a task, so that the execution doesn't wait for it to finish, as when its `JoinHandle`
1804 /// is dropped. A task that is unwinding a panic is detached once it has finished unwinding (see
1805 /// `record_unwinding_task`).
1806 pub fn detach(&mut self, id: TaskId) {
1807 match &mut self.unwinding_task {
1808 Some((task, detached)) if *task == id => *detached = true,
1809 _ => self.get_mut(id).detach(),
1810 }
1811 }
1812
1813 /// `maybe_yield` during execution teardown, which schedules nothing (see `tear_down`). Each
1814 /// scheduling point counts as a step, which the step bound limits as in a running execution.
1815 /// A task that teardown runs on its own stack switches out when it blocks or yields, so that
1816 /// teardown can run others. Anywhere else, or while a panic unwinds, nothing can be suspended,
1817 /// so a destructor that blocks is stopped with a panic. A task that finishes unwinding a panic
1818 /// switches out instead, and as soon as it has caught the panic (see `finish_unwinding`).
1819 #[cold]
1820 fn maybe_yield_in_teardown(&mut self) -> Result<bool, TeardownStall> {
1821 // Destructors do what tasks do, so `current::context_switches` counts their scheduling points
1822 // too.
1823 self.context_switches += 1;
1824 let panicking = std::thread::panicking();
1825 let yielded = std::mem::take(&mut self.has_yielded);
1826 let max_steps = self.teardown_step_bound();
1827
1828 let teardown = self.teardown_mut();
1829 teardown.steps += 1;
1830 let finishing_unwind = teardown.on_own_stack && matches!(teardown.step, TeardownStep::FinishUnwinding(_));
1831 if let Some(max_steps) = max_steps {
1832 // Stopping a destructor while a panic unwinds aborts the process, so give the unwind,
1833 // likely that of the panic that stopped the destructor, some more steps.
1834 let bound = if panicking && !finishing_unwind {
1835 max_steps.saturating_mul(2)
1836 } else {
1837 max_steps
1838 };
1839 if teardown.steps > bound {
1840 teardown.exceeded_step_bound = Some(max_steps);
1841 if finishing_unwind {
1842 return Ok(true);
1843 }
1844 return Err(TeardownStall::ExceededStepBound(max_steps));
1845 }
1846 }
1847 let on_own_stack = teardown.on_own_stack && !panicking;
1848 if on_own_stack && matches!(teardown.step, TeardownStep::Unwind(_)) {
1849 // The task caught the unwind of its stack, which starts again once the task switches
1850 // (see `Coroutine::force_unwind`). It has to run for that.
1851 self.current_mut().unblock();
1852 return Ok(true);
1853 }
1854 if finishing_unwind {
1855 // Once no panic is unwinding, the task has caught its panic.
1856 return Ok(!panicking || !self.current().runnable());
1857 }
1858
1859 // Between steps (dropping the payload of a panic, say), no task runs that could switch out.
1860 let Some(runnable) = self.try_current().map(Task::runnable) else {
1861 return Ok(false);
1862 };
1863 if !runnable && !on_own_stack {
1864 // A destructor blocked the task where it cannot wait. Make it runnable again, so that the
1865 // unwind of the panic that stops the destructor doesn't stall too.
1866 self.current_mut().unblock();
1867 return Err(TeardownStall::Blocked);
1868 }
1869 Ok(on_own_stack && (!runnable || yielded))
1870 }
1871
1872 /// The step bound for the destructors that execution teardown runs, if there is one (see
1873 /// `maybe_yield_in_teardown`).
1874 fn teardown_step_bound(&self) -> Option<usize> {
1875 match self.config.max_steps {
1876 MaxSteps::FailAfter(max_steps) => Some(max_steps),
1877 // A small bound that stops an execution is not meant to fail a test, and teardown has no
1878 // way to stop. So destructors get at least as many steps as an execution does by default.
1879 MaxSteps::ContinueAfter(max_steps) => Some(max_steps.max(DEFAULT_MAX_STEPS)),
1880 MaxSteps::None => None,
1881 }
1882 }
1883
1884 /// Tell the scheduler that the next context switch is an explicit yield requested by the
1885 /// current task. Some schedulers use this as a hint to influence scheduling.
1886 pub fn request_yield() {
1887 Self::with(|state| {
1888 state.has_yielded = true;
1889 });
1890 }
1891
1892 /// Check whether Shuttle's own `Drop` handlers should skip their bookkeeping, and early exit,
1893 /// because the execution has stopped, and so is being abandoned (see `tear_down`).
1894 ///
1895 /// We also stop if we are currently panicking (e.g., perhaps we're unwinding the stack for a
1896 /// panic triggered while someone held a Mutex, and so are executing the Drop handler for
1897 /// MutexGuard). This avoids calling back into the scheduler during a panic, because the state
1898 /// may be poisoned or otherwise invalid.
1899 ///
1900 /// While a finished execution is torn down, though, destructors run as in a running execution,
1901 /// also while a panic unwinds: teardown doesn't call the scheduler. A failed execution is
1902 /// abandoned like a stopped one.
1903 pub fn should_stop() -> bool {
1904 if std::thread::panicking() {
1905 // The state may be borrowed, so don't insist on it.
1906 return Self::try_with(|s| s.stops(true)).unwrap_or(true);
1907 }
1908 Self::with(|s| s.stops(false))
1909 }
1910
1911 /// `should_stop`, given whether the current task is panicking.
1912 pub(crate) fn stops(&self, panicking: bool) -> bool {
1913 match &self.teardown {
1914 Some(teardown) => teardown.kind.abandons(),
1915 None => {
1916 panicking || {
1917 assert_ne!(self.current_task, ScheduledTask::Finished);
1918 self.current_task == ScheduledTask::Stopped
1919 }
1920 }
1921 }
1922 }
1923
1924 /// Whether the execution stopped or failed, so that the stacks of its unfinished tasks are
1925 /// leaked rather than unwound (see `tear_down`).
1926 pub(crate) fn execution_stopped() -> bool {
1927 Self::try_with(|state| match &state.teardown {
1928 Some(teardown) => teardown.kind.abandons(),
1929 None => state.current_task == ScheduledTask::Stopped,
1930 })
1931 .unwrap_or(false)
1932 }
1933
1934 /// Generate some diagnostic information used when persisting failures.
1935 ///
1936 /// Because this method may be called from a panic hook, it must not panic.
1937 pub fn failing_task() -> String {
1938 Self::try_with(|state| {
1939 if let Some(task) = state.try_current() {
1940 let name = task.display_name();
1941 match state.teardown.as_ref().map(|teardown| teardown.step) {
1942 None | Some(TeardownStep::Idle) => name,
1943 Some(TeardownStep::Static) => format!("{name} (dropping a static at the end of the execution)"),
1944 Some(TeardownStep::FinishUnwinding(_)) => {
1945 format!("{name} (unwinding a panic at the end of the execution)")
1946 }
1947 Some(_) => format!("{name} (being dropped at the end of the execution)"),
1948 }
1949 } else if let Some(name) = &state.failed_task {
1950 // The task that failed the execution, which has been torn down since.
1951 name.clone()
1952 } else {
1953 "<unknown>".into()
1954 }
1955 })
1956 .unwrap_or_else(|e| format!("Tried to get ExecutionState, but got the following error: {e:?}"))
1957 }
1958
1959 /// Generate a random u64 from the current scheduler and return it.
1960 #[inline]
1961 pub fn next_u64() -> u64 {
1962 Self::with(|state| {
1963 if let Some(teardown) = &mut state.teardown {
1964 // Teardown doesn't extend the schedule (see `Teardown::rng`).
1965 return teardown.next_u64();
1966 }
1967 CurrentSchedule::push_random();
1968 state.scheduler.borrow_mut().next_u64()
1969 })
1970 }
1971
1972 pub fn current(&self) -> &Task {
1973 self.get(self.current_task.id().expect("there is no current task"))
1974 }
1975
1976 pub fn current_mut(&mut self) -> &mut Task {
1977 self.get_mut(self.current_task.id().expect("there is no current task"))
1978 }
1979
1980 pub fn try_current(&self) -> Option<&Task> {
1981 self.try_get(self.current_task.id()?)
1982 }
1983
1984 pub fn get(&self, id: TaskId) -> &Task {
1985 self.try_get(id).unwrap()
1986 }
1987
1988 /// Register a newly created task. Task ids are handed out sequentially as `tasks.len()`, so the
1989 /// new id is always greater than every existing one and `live_tasks` stays sorted.
1990 fn add_task(&mut self, task: Task) {
1991 debug_assert!(self.live_tasks.last().is_none_or(|last| *last < task.id()));
1992 self.live_tasks.push(task.id());
1993 self.tasks.push(task);
1994 }
1995
1996 /// Mark the task as finished and drop it from the set of live tasks.
1997 fn finish_task(&mut self, task_id: TaskId) {
1998 self.get_mut(task_id).finish();
1999 let idx = self
2000 .live_tasks
2001 .binary_search(&task_id)
2002 .expect("finished task must be live");
2003 self.live_tasks.remove(idx);
2004 if self.unwinding_task.is_some_and(|(task, _)| task == task_id) {
2005 self.unwinding_task = None;
2006 }
2007 }
2008
2009 /// Mark the current task as finished and drop it from the set of live tasks.
2010 fn finish_current_task(&mut self) {
2011 self.finish_task(self.current_task.id().unwrap());
2012 }
2013
2014 pub fn get_mut(&mut self, id: TaskId) -> &mut Task {
2015 self.tasks.get_mut(id.0).unwrap()
2016 }
2017
2018 pub fn try_get(&self, id: TaskId) -> Option<&Task> {
2019 self.tasks.get(id.0)
2020 }
2021
2022 pub fn try_get_mut(&mut self, id: TaskId) -> Option<&mut Task> {
2023 self.tasks.get_mut(id.0)
2024 }
2025
2026 /// Whether the execution is being torn down (see `tear_down`).
2027 pub fn in_cleanup(&self) -> bool {
2028 self.teardown.is_some()
2029 }
2030
2031 pub fn context_switches() -> usize {
2032 Self::with(|state| state.context_switches)
2033 }
2034
2035 #[track_caller]
2036 pub fn new_resource_signature(resource_type: ResourceType) -> ResourceSignature {
2037 ExecutionState::with(|s| s.current_mut().signature.new_resource(resource_type))
2038 }
2039
2040 pub fn get_storage<K: Into<StorageKey>, T: 'static>(&self, key: K) -> Option<&T> {
2041 self.storage
2042 .get(key.into())
2043 .map(|result| result.expect("global storage is never destructed"))
2044 }
2045
2046 pub fn init_storage<K: Into<StorageKey>, T: 'static>(&mut self, key: K, value: T) {
2047 self.storage.init(key.into(), value);
2048 }
2049
2050 pub fn get_clock(&self, id: TaskId) -> &VectorClock {
2051 &self.tasks.get(id.0).unwrap().clock
2052 }
2053
2054 pub fn get_clock_mut(&mut self, id: TaskId) -> &mut VectorClock {
2055 &mut self.tasks.get_mut(id.0).unwrap().clock
2056 }
2057
2058 /// Increment the current thread's clock entry and update its clock with the one provided.
2059 pub fn update_clock(&mut self, clock: &VectorClock) {
2060 let task = self.current_mut();
2061 task.clock.increment(task.id);
2062 task.clock.update(clock);
2063 }
2064
2065 /// Increment the current thread's clock and return a shared reference to it
2066 pub fn increment_clock(&mut self) -> &VectorClock {
2067 let task = self.current_mut();
2068 task.clock.increment(task.id);
2069 &task.clock
2070 }
2071
2072 /// Increment the current thread's clock and return a mutable reference to it
2073 pub fn increment_clock_mut(&mut self) -> &mut VectorClock {
2074 let task = self.current_mut();
2075 task.clock.increment(task.id);
2076 &mut task.clock
2077 }
2078
2079 /// Set the number of steps used, with respect to the step bound, to 0 (see
2080 /// `current::reset_step_count`). During execution teardown, these are the steps of destructors
2081 /// (see `maybe_yield_in_teardown`).
2082 pub(crate) fn reset_step_count(&mut self) {
2083 match &mut self.teardown {
2084 Some(teardown) => teardown.steps = 0,
2085 None => self.steps_reset_at = CurrentSchedule::len(),
2086 }
2087 }
2088
2089 /// Returns `true` if the test has exceeded the step bound, and `false` otherwise.
2090 fn is_step_bound_exceeded(&self, max_steps: usize) -> bool {
2091 CurrentSchedule::len() - self.steps_reset_at >= max_steps
2092 }
2093
2094 /// Run the scheduler to choose the next task to run. `has_yielded` should be false if the
2095 /// scheduler is being invoked from within a running task. If scheduling fails, returns an Err
2096 /// with a String describing the failure.
2097 fn schedule(&mut self) -> Result<(), StepError> {
2098 // Don't schedule twice. If `maybe_yield` ran the scheduler, we don't want to run it
2099 // again at the top of `step`.
2100 if self.next_task != ScheduledTask::None {
2101 return Ok(());
2102 }
2103
2104 self.context_switches += 1;
2105
2106 match self.config.max_steps {
2107 MaxSteps::FailAfter(max_steps) => {
2108 if self.is_step_bound_exceeded(max_steps) {
2109 return Err(StepError::StepBoundExceeded(max_steps));
2110 }
2111 }
2112 MaxSteps::ContinueAfter(max_steps) => {
2113 if self.is_step_bound_exceeded(max_steps) {
2114 // TODO: We have to set `Stopped` and return `Ok` here, else assertions will fail. This should probably be cleaned up.
2115 self.next_task = ScheduledTask::Stopped;
2116 return Ok(());
2117 }
2118 }
2119 MaxSteps::None => {}
2120 }
2121
2122 let mut unfinished_attached = false;
2123 let mut all_runnable_detached = true;
2124 let mut any_runnable = false;
2125
2126 // The loop below only looks at `live_tasks`, so a task missing from that set would silently
2127 // never be scheduled. Check that direction of the invariant here; the loop itself checks the
2128 // other direction (that no finished task is still in the set).
2129 debug_assert!(
2130 self.tasks
2131 .iter()
2132 .filter(|task| !task.finished() && task.runnable())
2133 .all(|task| self.live_tasks.binary_search(&task.id()).is_ok()),
2134 "live_tasks is missing a runnable unfinished task"
2135 );
2136
2137 for &task_id in &self.live_tasks {
2138 let task = &self.tasks[task_id.0];
2139 debug_assert!(!task.finished());
2140 unfinished_attached |= !task.detached;
2141 let is_runnable = task.runnable();
2142 any_runnable |= is_runnable;
2143
2144 if is_runnable {
2145 all_runnable_detached &= task.detached;
2146 self.runnable_tasks.push(task as *const Task);
2147 } else if task.can_spuriously_wakeup() {
2148 // Some blocked tasks can be woken up spuriously, even though the condition the task is
2149 // blocked on hasn't happened yet. We'll add such tasks to the list of runnable tasks, but
2150 // they won't contribute to the check on `any_runnable`; if the only runnable tasks
2151 // are ones that are waiting for a potential spurious wakeup, it should still be treated as
2152 // a deadlock since there's no guarantee that spurious wakeups will ever occur.
2153 self.runnable_tasks.push(task as *const Task);
2154 }
2155 }
2156
2157 // We should finish execution when either
2158 // (1) There are no runnable tasks, or
2159 // (2) All runnable tasks have been detached AND there are no unfinished attached tasks
2160 // If there are some unfinished attached tasks and all runnable tasks are detached, we must
2161 // run some detached task to give them a chance to unblock some unfinished attached task.
2162 if !any_runnable || (!unfinished_attached && all_runnable_detached) {
2163 self.next_task = ScheduledTask::Finished;
2164 return Ok(());
2165 }
2166
2167 let is_yielding = std::mem::replace(&mut self.has_yielded, false);
2168
2169 // Cast the slice of raw pointers to a slice of references in place to provide schedulers with a safe API
2170 //
2171 // SAFETY: This is safe because the tasks themselves are only being accessed through this shared reference by the
2172 // schedulers, and all references are always cleared from the runnable_tasks Vec at the end of this function.
2173 // The transmute itself is safe because *const and & have the same layout, and the pointer is created from a
2174 // reference earlier in this function.
2175 let task_refs = unsafe { std::mem::transmute::<&[*const Task], &[&Task]>(&self.runnable_tasks) };
2176
2177 self.next_task = self
2178 .scheduler
2179 .borrow_mut()
2180 .next_task(task_refs, self.current_task.id(), is_yielding)
2181 .map(ScheduledTask::Some)
2182 .unwrap_or(ScheduledTask::Stopped);
2183
2184 // Tracing this `in_scope` is purely a matter of taste. We do it because
2185 // 1) It is an action taken by the scheduler, and should thus be traced under the scheduler's span
2186 // 2) It creates a visual separation of scheduling decisions and `Task`-induced tracing.
2187 // Note that there is a case to be made for not `in_scope`-ing it, as that makes seeing the context
2188 // of the context switch clearer.
2189 //
2190 // Note also that changing this trace! statement requires changing the test `basic::labels::test_tracing_with_label_fn`
2191 // which relies on this trace reporting the `runnable` tasks.
2192 self.top_level_span.in_scope(|| {
2193 trace!(
2194 i=CurrentSchedule::len(),
2195 next_task=?self.next_task,
2196 runnable=?task_refs.iter().map(|task| task.id()).collect::<SmallVec<[_; DEFAULT_INLINE_TASKS]>>(),
2197 "scheduling decision"
2198 );
2199 });
2200
2201 // If the task chosen by the scheduler is blocked, then it should be one that can be
2202 // spuriously woken up, and we need to unblock it here so that it can execute.
2203 if let Some(tid) = self.next_task.id() {
2204 let task = self.get_mut(tid);
2205 assert!(task.runnable() || task.blocked());
2206 if task.blocked() {
2207 assert!(task.can_spuriously_wakeup());
2208 task.unblock();
2209 }
2210 }
2211
2212 // Retains the capacity of `runnable_tasks` for future calls of `schedule`
2213 self.runnable_tasks.clear();
2214
2215 Ok(())
2216 }
2217
2218 /// Set the next task as the current task
2219 fn advance_to_next_task(&mut self) {
2220 debug_assert_ne!(self.next_task, ScheduledTask::None);
2221 self.current_task = self.next_task.take();
2222
2223 if let ScheduledTask::Some(tid) = self.current_task {
2224 CurrentSchedule::push_task(tid);
2225 }
2226 }
2227
2228 // Sets the `tag` field of the current task.
2229 // Returns the `tag` which was there previously.
2230 #[allow(deprecated)]
2231 pub fn set_tag_for_current_task(tag: Arc<dyn Tag>) -> Option<Arc<dyn Tag>> {
2232 ExecutionState::with(|s| s.current_mut().set_tag(tag))
2233 }
2234
2235 #[allow(deprecated)]
2236 fn get_tag_or_default_for_current_task(&self) -> Option<Arc<dyn Tag>> {
2237 self.try_current().and_then(|current| current.get_tag())
2238 }
2239
2240 #[allow(deprecated)]
2241 pub fn get_tag_for_current_task() -> Option<Arc<dyn Tag>> {
2242 ExecutionState::with(|s| s.get_tag_or_default_for_current_task())
2243 }
2244
2245 #[allow(deprecated)]
2246 pub fn set_tag_for_task(task: TaskId, tag: Arc<dyn Tag>) -> Option<Arc<dyn Tag>> {
2247 ExecutionState::with(|s| s.get_mut(task).set_tag(tag))
2248 }
2249}
2250
2251#[cfg(debug_assertions)]
2252impl Drop for ExecutionState {
2253 fn drop(&mut self) {
2254 assert!(self.has_cleaned_up || std::thread::panicking());
2255 }
2256}
2257
2258/// Whether the running task's default dispatcher might not be the execution's.
2259fn task_may_have_own_default(state: &ExecutionState) -> bool {
2260 use tracing::level_filters::LevelFilter;
2261
2262 // No dispatcher enables anything, so whichever one is the default makes no difference. (If that
2263 // changes while the task is switched out, other tasks see its default, as before defaults were
2264 // parked.)
2265 if LevelFilter::current() == LevelFilter::OFF {
2266 return false;
2267 }
2268 tracing::dispatcher::get_default(|current| {
2269 if let Some(global) = state.top_level_dispatch_global_ptr {
2270 // `get_default` hands out the global default from where it lives, and anything else from
2271 // this thread's scoped default. So if it hands out the global default, the task has no
2272 // default of its own. Parking otherwise is sometimes unnecessary but always safe,
2273 // including if `tracing` ever handed out the global default from elsewhere.
2274 !std::ptr::eq(current, global)
2275 } else if current.is::<NoSubscriber>() {
2276 // Nothing is recorded either way if the execution's default is `Dispatch::none()` too.
2277 !state.top_level_dispatch.is::<NoSubscriber>()
2278 } else {
2279 // A task can only have a default of its own through a scoped default. `get_default` hands
2280 // nested calls `Dispatch::none()` only while some thread has one; otherwise they get the
2281 // global default, which is then `current` and everyone's default.
2282 tracing::dispatcher::get_default(|nested| nested.is::<NoSubscriber>())
2283 }
2284 })
2285}