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 that the event is enabled first, so that accesses don't pay for the flag when it isn't.
838 if tracing::enabled!(tracing::Level::TRACE) && !TRACING_TRY_WITH.with(|emitting| emitting.replace(true)) {
839 struct Reset;
840 impl Drop for Reset {
841 fn drop(&mut self) {
842 TRACING_TRY_WITH.with(|emitting| emitting.set(false));
843 }
844 }
845 let _reset = Reset;
846 trace!(
847 "ExecutionState::try_with called from {:?}",
848 std::panic::Location::caller()
849 );
850 }
851 if EXECUTION_STATE.is_set() {
852 EXECUTION_STATE.with(|cell| {
853 if let Ok(mut state) = cell.try_borrow_mut() {
854 Ok(f(&mut state))
855 } else {
856 Err(ExecutionStateBorrowError::AlreadyBorrowed)
857 }
858 })
859 } else {
860 Err(ExecutionStateBorrowError::NotSet)
861 }
862 }
863
864 /// A shortcut to get the current task ID
865 pub fn me() -> TaskId {
866 Self::with(|s| s.current().id())
867 }
868
869 /// If there is only one attached, unfinished task and there is at least one detached, unfinished task
870 /// then exiting the attached task will cause the whole execution to exit. As a result, the unfinished
871 /// detached tasks are truncated -- their remaining events will not be executed because the program itself
872 /// has exited. This is relevant because it means that *exiting* a task can be a visible operation
873 /// in that it affects which events are executed.
874 pub fn exit_current_truncates_execution(&self) -> bool {
875 // Strictly speaking, this is only true if there are other runnable detached tasks, but always making the main thread
876 // exit a scheduling point is simpler conceptually
877 if self.current().id() == TaskId::from(0) {
878 return true;
879 }
880
881 // If the current task is detached, then it definitely doesn't truncate the execution
882 if self.current().is_detached() {
883 return false;
884 }
885
886 let mut single_unfinished_attached = false;
887 let mut has_unfinished_detached = false;
888 for t in self.tasks.iter() {
889 let unfinished_attached = !t.finished() && !t.detached;
890 if single_unfinished_attached && unfinished_attached {
891 // there are more than one unfinished attached tasks, so one exiting won't truncate
892 return false;
893 }
894
895 single_unfinished_attached |= unfinished_attached;
896 has_unfinished_detached |= !t.finished() && t.detached;
897 }
898 has_unfinished_detached && single_unfinished_attached
899 }
900
901 fn set_labels_for_new_task(state: &ExecutionState, task_id: TaskId, name: Option<String>) {
902 LABELS.with(|cell| {
903 let mut map = cell.borrow_mut();
904
905 // If parent has labels, inherit them
906 if let Some(parent_task_id) = state.try_current().map(|t| t.id()) {
907 let parent_map = map.get(&parent_task_id);
908 if let Some(parent_map) = parent_map {
909 let mut child_map = parent_map.clone();
910
911 // If the parent has a `ChildLabelFn` set, use that to update the child's Labels
912 if let Some(gen) = parent_map.get::<ChildLabelFn>() {
913 (gen.0)(task_id, &mut child_map);
914 }
915
916 map.insert(task_id, child_map);
917 }
918 }
919
920 // Add any name assigned to the task to its set of Labels
921 if let Some(name) = name {
922 let m = map.entry(task_id).or_default();
923 m.insert(TaskName::from(name));
924 }
925 });
926 }
927
928 // Note: `spawn_thread`, `spawn_main_thread`, and `spawn_future` share some similar logic.
929 // Changes to one of these functions likely need to be propagated to the other two as well.
930 pub fn spawn_main_thread(
931 f: Box<dyn FnOnce() + 'static>,
932 stack_size: usize,
933 caller: &'static Location<'static>,
934 ) -> TaskId {
935 let name = "main-thread".to_string();
936 let mut clock = VectorClock::new();
937
938 let task_id = Self::with(|state| {
939 let parent_span_id = state.top_level_span.id();
940 let task_id = TaskId(state.tasks.len());
941 let tag = state.get_tag_or_default_for_current_task();
942
943 Self::set_labels_for_new_task(state, task_id, Some(name.clone()));
944
945 clock.extend(task_id); // and extend it with an entry for the new thread
946
947 let schedule_len = CurrentSchedule::len();
948
949 let task = Task::from_closure(
950 f,
951 stack_size,
952 task_id,
953 Some(name),
954 clock,
955 parent_span_id,
956 schedule_len,
957 tag,
958 None,
959 TaskSignature::new_parentless(caller),
960 );
961 state.add_task(task);
962
963 task_id
964 });
965 crate::annotations::record_task_created(task_id, false);
966 task_id
967 }
968
969 // Note: `spawn_thread`, `spawn_main_thread`, and `spawn_future` share some similar logic.
970 // Changes to one of these functions likely need to be propagated to the other two as well.
971 /// Spawn a new task for a future. This doesn't create a yield point; the caller should do that
972 /// if it wants to give the new task a chance to run immediately.
973 pub fn spawn_future<F>(
974 future: F,
975 stack_size: usize,
976 name: Option<String>,
977 caller: &'static Location<'static>,
978 ) -> TaskId
979 where
980 F: Future<Output = ()> + 'static,
981 {
982 thread::switch();
983 let task_id = Self::with(|state| {
984 let schedule_len = CurrentSchedule::len();
985 let parent_span_id = state.top_level_span.id();
986
987 let task_id = TaskId(state.tasks.len());
988 let tag = state.get_tag_or_default_for_current_task();
989
990 Self::set_labels_for_new_task(state, task_id, name.clone());
991
992 let clock = state.increment_clock_mut(); // Increment the parent's clock
993 clock.extend(task_id); // and extend it with an entry for the new task
994
995 let task = Task::from_future(
996 future,
997 stack_size,
998 task_id,
999 name,
1000 clock.clone(),
1001 parent_span_id,
1002 schedule_len,
1003 tag,
1004 Some(state.current().id()),
1005 state.current_mut().signature.new_child(caller),
1006 );
1007
1008 state.add_task(task);
1009
1010 task_id
1011 });
1012 crate::annotations::record_task_created(task_id, true);
1013 task_id
1014 }
1015
1016 // Note: `spawn_thread`, `spawn_main_thread`, and `spawn_future` share some similar logic.
1017 // Changes to one of these functions likely need to be propagated to the other two as well.
1018 pub fn spawn_thread(
1019 f: Box<dyn FnOnce() + 'static>,
1020 stack_size: usize,
1021 name: Option<String>,
1022 mut initial_clock: Option<VectorClock>,
1023 caller: &'static Location<'static>,
1024 ) -> TaskId {
1025 thread::switch();
1026 let task_id = Self::with(|state| {
1027 let parent_span_id = state.top_level_span.id();
1028 let task_id = TaskId(state.tasks.len());
1029 let tag = state.get_tag_or_default_for_current_task();
1030
1031 Self::set_labels_for_new_task(state, task_id, name.clone());
1032
1033 let clock = if let Some(ref mut clock) = initial_clock {
1034 clock
1035 } else {
1036 // Inherit the clock of the parent thread (which spawned this task)
1037 state.increment_clock_mut()
1038 };
1039 clock.extend(task_id); // and extend it with an entry for the new thread
1040 let clock = clock.clone();
1041
1042 let task = Task::from_closure(
1043 f,
1044 stack_size,
1045 task_id,
1046 name,
1047 clock,
1048 parent_span_id,
1049 CurrentSchedule::len(),
1050 tag,
1051 Some(state.current().id()),
1052 state.current_mut().signature.new_child(caller),
1053 );
1054 state.add_task(task);
1055
1056 task_id
1057 });
1058 crate::annotations::record_task_created(task_id, false);
1059 task_id
1060 }
1061
1062 /// Tear down the execution, which ended as `kind` says, before it goes out of `EXECUTION_STATE`
1063 /// scope: drop what its unfinished tasks leave behind, the tasks' task-local values, and the
1064 /// execution's statics, while their destructors can still use Shuttle. Returns how that fails
1065 /// the test, if it does.
1066 ///
1067 /// A finished execution's unfinished tasks (detached future tasks, and tasks that destructors
1068 /// spawn during teardown) are torn down as if the execution cancelled them as it ended. While a
1069 /// task is torn down, it is the current task: its destructors can use Shuttle as the running
1070 /// task can, and what they do is attributed to it.
1071 ///
1072 /// * A task that never ran drops its function, and a future task that is parked between polls,
1073 /// the usual state of an unfinished future task, drops its future, the way an async runtime
1074 /// drops the futures of its tasks when it shuts down (see `Continuation::cancel`). They do
1075 /// this on their own stacks, and then drop their task-local values. So a destructor that
1076 /// blocks, on a lock that another unfinished task holds say, waits for teardown to drop that
1077 /// task. A panic is an ordinary panic, which fails the test once teardown is done.
1078 /// * A future task that stopped in the middle of `poll` has user code on its stack, which only
1079 /// unwinding the stack can drop. Teardown unwinds these stacks one at a time, once no other
1080 /// task can make progress, so that the locks that the other tasks held are free. Their
1081 /// destructors can use Shuttle too, but cannot wait: one that blocks, or panics, while a stack
1082 /// is unwound aborts the process, as any panic during unwinding does. A task that catches the
1083 /// unwind is unwound again at its next scheduling point.
1084 ///
1085 /// Nothing is scheduled. Teardown resumes the tasks it cancelled itself, in a fixed order that
1086 /// doesn't extend the schedule, so it is the same when the schedule is replayed (see
1087 /// `next_teardown_job`). A destructor that blocks waits until another destructor wakes it, and
1088 /// one that yields lets the others run; but if every destructor left blocks, nothing can wake
1089 /// them, and that fails the test like a deadlock. The step bound applies to destructors too (see
1090 /// `maybe_yield_in_teardown`). The execution's statics go last, as if the main thread dropped
1091 /// them, once nothing else is left that could still use them.
1092 ///
1093 /// A stopped or failed execution is abandoned instead. Shuttle's own destructors skip their
1094 /// bookkeeping (see `should_stop`), and panics are ignored, so that a failure is what gets
1095 /// reported. The stacks of its unfinished tasks are leaked. A failed execution's functions of
1096 /// tasks that never ran, and futures of parked future tasks, are leaked too, unless
1097 /// `UngracefulShutdownConfig::continuation_function_behavior` says to drop them, in which case
1098 /// they are cancelled as above. A stopped execution's functions of tasks that never ran are
1099 /// dropped that way. Every function that is dropped goes before any stack is freed, since a
1100 /// scoped thread's function borrows from its parent's stack. Task-local values and statics are
1101 /// dropped as in a finished execution.
1102 ///
1103 /// A task that was unwinding a panic when a stopped execution stopped finishes unwinding it first,
1104 /// and the panic fails the test (see `finish_unwinding`). A failed execution's is leaked like the
1105 /// rest, as running more of a failed execution risks a panic that aborts the process. If teardown
1106 /// can't tell which task it is (see `record_unwinding_task`), that fails the test too, and the
1107 /// rest of the execution is abandoned: unwinding that task's stack again would abort the process.
1108 fn tear_down(kind: TeardownKind) -> Option<TeardownFailure> {
1109 let mut plan = TeardownPlan::default();
1110 let mut report = TeardownReport::new();
1111 // Here, on the executor's stack, `std::thread::panicking()` says whether a task that switched
1112 // out is unwinding a panic still (see `record_unwinding_task`).
1113 let unwinding = std::thread::panicking();
1114 let lost_panic = Self::with(|state| {
1115 state.current_task = kind.final_state();
1116 state.next_task = ScheduledTask::None;
1117 state.teardown = Some(Teardown::new(kind, CurrentSchedule::len()));
1118 if kind != TeardownKind::Failed && unwinding {
1119 let unwinding_task = state.unwinding_task.take().map(|(id, _)| id);
1120 plan.unwinding_task = unwinding_task.filter(|&id| state.get(id).suspended());
1121 }
1122 let lost_panic = unwinding && plan.unwinding_task.is_none() && kind != TeardownKind::Failed;
1123 if lost_panic {
1124 // So that no stack is unwound.
1125 state.abandon_teardown();
1126 }
1127 lost_panic
1128 });
1129 if lost_panic {
1130 report.fail(
1131 "a task was still unwinding a panic when the execution ended, but it panicked while another task was \
1132 unwinding a panic, so Shuttle can't tell which task it is, to let it finish unwinding"
1133 .into(),
1134 );
1135 }
1136 // The panic hook goes back to reporting panics when teardown is done, however it ends.
1137 struct ReportPanicsAgain;
1138 impl Drop for ReportPanicsAgain {
1139 fn drop(&mut self) {
1140 TEARDOWN_IGNORES_PANICS.set(false);
1141 }
1142 }
1143 let report_panics_again = ReportPanicsAgain;
1144 TEARDOWN_IGNORES_PANICS.set(Self::with(|state| state.teardown().kind.abandons()));
1145
1146 loop {
1147 match Self::with(|state| state.next_teardown_job(&mut plan)) {
1148 TeardownJob::FinishUnwinding(id) => Self::finish_unwinding(id, &mut report),
1149 TeardownJob::OnOwnStack(id, resumption) => {
1150 Self::run_on_own_stack(id, resumption, &mut plan, &mut report)
1151 }
1152 TeardownJob::Unwind(id) => Self::unwind_task(id, &mut report),
1153 TeardownJob::Leak(id) => Self::leak_task(id, &mut report),
1154 TeardownJob::LeakFunction(id) => {
1155 Self::with(|state| {
1156 let task = state.get(id);
1157 let continuation = task
1158 .continuation
1159 .as_ref()
1160 .expect("a task that never ran has a continuation");
1161 continuation.borrow_mut().leak_function();
1162 });
1163 Self::finish_torn_down_task(id, &mut report);
1164 }
1165 TeardownJob::TaskLocals(id) => Self::drop_leftover_task_locals(id, &mut report),
1166 TeardownJob::Static(value) => Self::drop_static(value, &mut report),
1167 TeardownJob::Stuck => {
1168 let (deadlocked, blocked_tasks) = Self::with(|state| {
1169 let teardown = state.teardown();
1170 let deadlocked = !teardown.kind.abandons() && teardown.exceeded_step_bound.is_none();
1171 let blocked = plan
1172 .suspended
1173 .iter()
1174 .map(|&(id, _)| state.get(id).format_for_deadlock());
1175 let blocked = blocked.collect::<Vec<_>>();
1176 state.abandon_teardown();
1177 (deadlocked, blocked)
1178 });
1179 TEARDOWN_IGNORES_PANICS.set(true);
1180 if deadlocked {
1181 report.deadlocked(blocked_tasks);
1182 }
1183 for (id, _) in std::mem::take(&mut plan.suspended) {
1184 Self::leak_task(id, &mut report);
1185 }
1186 }
1187 TeardownJob::Abandon(max_steps) => {
1188 // The panic that stopped the destructor fails the test, unless the destructor
1189 // caught it.
1190 report.fail(format!(
1191 "a destructor exceeded the step bound ({max_steps}) while it was being dropped at the end of \
1192 the execution"
1193 ));
1194 Self::with(|state| state.abandon_teardown());
1195 TEARDOWN_IGNORES_PANICS.set(true);
1196 }
1197 TeardownJob::Done => {
1198 // What the tasks still hold goes last, as each task (see `drop_task_leftovers`).
1199 // Destructors can spawn tasks, which are torn down too.
1200 if !Self::drop_task_leftovers(&mut report) {
1201 break;
1202 }
1203 }
1204 }
1205 }
1206
1207 // The tasks' labels, and tags that no task holds, while `EXECUTION_STATE` is set. Their
1208 // destructors may look up tasks: formatting a `TaskId` looks up its name, say. In the order
1209 // the tasks were created, so that the destructors run in the same order every time.
1210 let tags = TASK_ID_TO_TAGS.with(|cell| std::mem::take(&mut *cell.borrow_mut()));
1211 let mut tags = tags.into_iter().collect::<Vec<_>>();
1212 tags.sort_unstable_by_key(|&(id, _)| id);
1213 for (_, tag) in tags {
1214 report.catch(move || drop(tag));
1215 }
1216 let labels = LABELS.with(|cell| std::mem::take(&mut *cell.borrow_mut()));
1217 let mut labels = labels.into_iter().collect::<Vec<_>>();
1218 labels.sort_unstable_by_key(|&(id, _)| id);
1219 for (_, labels) in labels {
1220 report.catch(move || drop(labels));
1221 }
1222 let tasks = Self::with(|state| std::mem::take(&mut state.tasks));
1223 report.catch(move || drop(tasks));
1224
1225 Self::with(|state| {
1226 state.teardown = None;
1227 #[cfg(debug_assertions)]
1228 {
1229 state.has_cleaned_up = true;
1230 }
1231 });
1232 drop(report_panics_again);
1233 report.finish()
1234 }
1235
1236 /// Choose what execution teardown does next (see `tear_down`).
1237 fn next_teardown_job(&mut self, plan: &mut TeardownPlan) -> TeardownJob {
1238 let teardown = self.teardown();
1239 let kind = teardown.kind;
1240 // Once a destructor has exceeded the step bound, which fails the test, the rest of the
1241 // execution is abandoned: the tasks that are left are leaked, as destructors that spawn tasks
1242 // whose destructors spawn tasks, say, would go on forever.
1243 let exceeded_step_bound = teardown.exceeded_step_bound;
1244 if let Some(max_steps) = exceeded_step_bound.filter(|_| !kind.abandons()) {
1245 return TeardownJob::Abandon(max_steps);
1246 }
1247 let abandoned = exceeded_step_bound.is_some();
1248 if abandoned && !plan.suspended.is_empty() {
1249 return TeardownJob::Stuck;
1250 }
1251
1252 // A task that was unwinding a panic when the execution stopped finishes unwinding it first.
1253 if let Some(id) = plan.unwinding_task.take() {
1254 return TeardownJob::FinishUnwinding(id);
1255 }
1256
1257 // A cancelled task that blocked, and has been woken since, goes first.
1258 if let Some(i) = plan
1259 .suspended
1260 .iter()
1261 .position(|&(id, blocked)| blocked && self.get(id).runnable())
1262 {
1263 plan.idle_yields = 0;
1264 return TeardownJob::OnOwnStack(plan.suspended.remove(i).0, Resumption::Resume);
1265 }
1266
1267 // Then the tasks that teardown hasn't looked at yet, in the order they were created.
1268 let leaks = UNGRACEFUL_SHUTDOWN_CONFIG.get().continuation_function_behavior.leaks();
1269 let (drop_functions, drop_parked_futures) = match kind {
1270 _ if abandoned => (false, false),
1271 TeardownKind::Finished => (true, true),
1272 TeardownKind::Stopped => (true, false),
1273 TeardownKind::Failed => (!leaks, !leaks),
1274 };
1275 while let Some(task) = self.tasks.get(plan.next_task) {
1276 plan.next_task += 1;
1277 let id = task.id();
1278 if task.finished() {
1279 // A future task's future can set task-local values after the task dropped its own
1280 // ones (see `Wrapper::finish`).
1281 if task.has_locals() {
1282 return TeardownJob::TaskLocals(id);
1283 }
1284 } else if task.never_ran() {
1285 plan.idle_yields = 0;
1286 return if drop_functions {
1287 TeardownJob::OnOwnStack(id, Resumption::Cancel)
1288 } else {
1289 TeardownJob::LeakFunction(id)
1290 };
1291 } else if drop_parked_futures && task.parked_between_polls() {
1292 plan.idle_yields = 0;
1293 return TeardownJob::OnOwnStack(id, Resumption::Cancel);
1294 } else {
1295 // The task is in the middle of user code.
1296 plan.stacks.push_back(id);
1297 }
1298 }
1299
1300 // A cancelled task that yielded runs again, in turn. While stacks are waiting to be unwound,
1301 // for a number of turns only, if no task finishes or blocks meanwhile (see
1302 // `TEARDOWN_YIELD_ROUNDS`).
1303 let unwinds_wait = kind == TeardownKind::Finished && !abandoned && !plan.stacks.is_empty();
1304 let yielded = |&(id, blocked): &(TaskId, bool)| !blocked && self.get(id).runnable();
1305 let num_yielded = plan.suspended.iter().filter(|task| yielded(task)).count();
1306 if num_yielded > 0 && (!unwinds_wait || plan.idle_yields < self.teardown_yield_budget(num_yielded)) {
1307 plan.idle_yields += 1;
1308 let i = plan.suspended.iter().position(yielded).unwrap();
1309 return TeardownJob::OnOwnStack(plan.suspended.remove(i).0, Resumption::Resume);
1310 }
1311
1312 // A finished execution's tasks in the middle of `poll` are unwound now, once no other task
1313 // can make progress.
1314 if unwinds_wait {
1315 plan.idle_yields = 0;
1316 return TeardownJob::Unwind(plan.stacks.pop_front().unwrap());
1317 }
1318
1319 // A static goes only once no cancelled task is left suspended, whose stack could still refer
1320 // to it.
1321 if !plan.suspended.is_empty() {
1322 return TeardownJob::Stuck;
1323 }
1324
1325 if let Some(value) = self.storage.pop() {
1326 return TeardownJob::Static(value);
1327 }
1328
1329 // An abandoned execution's stacks are leaked last, once the functions of the tasks that never
1330 // ran, which could borrow from them, are gone.
1331 if let Some(id) = plan.stacks.pop_front() {
1332 return TeardownJob::Leak(id);
1333 }
1334
1335 TeardownJob::Done
1336 }
1337
1338 /// How many turns in a row the tasks that yield get while stacks are waiting to be unwound (see
1339 /// `TEARDOWN_YIELD_ROUNDS`).
1340 fn teardown_yield_budget(&self, num_yielded: usize) -> usize {
1341 let rounds = TEARDOWN_YIELD_ROUNDS.saturating_mul(num_yielded);
1342 match self.teardown_step_bound() {
1343 Some(max_steps) => rounds.min(max_steps.saturating_sub(self.teardown().steps) / 4),
1344 None => rounds,
1345 }
1346 }
1347
1348 /// Abandon the rest of a finished execution's teardown, which has failed (see `tear_down`).
1349 fn abandon_teardown(&mut self) {
1350 let teardown = self.teardown_mut();
1351 if teardown.kind == TeardownKind::Finished {
1352 teardown.kind = TeardownKind::Failed;
1353 self.current_task = TeardownKind::Failed.final_state();
1354 }
1355 }
1356
1357 /// Resume a task on its own stack (see `Resumption`), until it finishes or switches out.
1358 fn run_on_own_stack(id: TaskId, resumption: Resumption, plan: &mut TeardownPlan, report: &mut TeardownReport) {
1359 let continuation = Self::with(|state| {
1360 state.begin_teardown_step(TeardownStep::Cancel(id));
1361 let task = state.get_mut(id);
1362 if resumption == Resumption::Cancel {
1363 // A parked future task is asleep, but it runs now, to drop its future.
1364 task.unblock();
1365 }
1366 task.continuation
1367 .clone()
1368 .expect("an unfinished task has a continuation")
1369 });
1370
1371 Execution::enter_task_span();
1372 let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
1373 let mut continuation = continuation.borrow_mut();
1374 // Only now can the task switch out (see `maybe_yield_in_teardown`).
1375 Self::with(|state| state.teardown_mut().on_own_stack = true);
1376 if resumption == Resumption::Cancel {
1377 continuation.cancel()
1378 } else {
1379 continuation.resume()
1380 }
1381 }));
1382 Self::with(|state| state.teardown_mut().on_own_stack = false);
1383 Execution::exit_task_span(matches!(result, Ok(false)));
1384 drop(continuation);
1385 // A panic's payload goes while the step is under way, so its destructor runs as the task.
1386 let finished = result.unwrap_or_else(|payload| {
1387 report.panicked(payload);
1388 true
1389 });
1390 let blocked = Self::with(|state| {
1391 state.end_teardown_step();
1392 !state.get(id).runnable()
1393 });
1394
1395 if finished {
1396 plan.idle_yields = 0;
1397 Self::finish_torn_down_task(id, report);
1398 } else {
1399 // A destructor blocked, or yielded (see `maybe_yield_in_teardown`).
1400 if blocked {
1401 plan.idle_yields = 0;
1402 }
1403 plan.suspended.push((id, blocked));
1404 }
1405 }
1406
1407 /// Let a task that switched out while it unwound a panic, and is unwinding it still when a
1408 /// stopped execution stopped, finish unwinding it, on its own stack (see `tear_down`). The panic
1409 /// then fails the test. The task runs until it has caught the panic, if it does, but no further.
1410 /// A task that blocks while the panic unwinds, or exceeds the step bound, waits for tasks that no
1411 /// longer run, but cannot be stopped with a panic, as that would abort the process. It switches
1412 /// out instead (see `maybe_yield_in_teardown`), and fails the test. Then its stack is leaked,
1413 /// which leaves the thread panicking.
1414 fn finish_unwinding(id: TaskId, report: &mut TeardownReport) {
1415 let (continuation, blocked) = Self::with(|state| {
1416 state.begin_teardown_step(TeardownStep::FinishUnwinding(id));
1417 let task = state.get(id);
1418 let continuation = task
1419 .continuation
1420 .clone()
1421 .expect("an unfinished task has a continuation");
1422 (continuation, !task.runnable())
1423 });
1424
1425 // A task that blocked already waits for a task that no longer runs.
1426 let result = if blocked {
1427 Ok(false)
1428 } else {
1429 Execution::enter_task_span();
1430 let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
1431 let mut continuation = continuation.borrow_mut();
1432 Self::with(|state| state.teardown_mut().on_own_stack = true);
1433 continuation.resume()
1434 }));
1435 Self::with(|state| state.teardown_mut().on_own_stack = false);
1436 Execution::exit_task_span(matches!(result, Ok(false)));
1437 result
1438 };
1439 drop(continuation);
1440
1441 // On the executor's stack, `std::thread::panicking()` says whether the task is unwinding the
1442 // panic still (see `record_unwinding_task`).
1443 let stuck = matches!(result, Ok(false)) && std::thread::panicking();
1444 // A panic's payload goes while the step is under way, so its destructor runs as the task.
1445 let finished = result.unwrap_or_else(|payload| {
1446 report.late_panic(payload);
1447 true
1448 });
1449 if stuck {
1450 let (name, exceeded_step_bound) =
1451 Self::with(|state| (state.get(id).display_name(), state.teardown().exceeded_step_bound));
1452 let why = match exceeded_step_bound {
1453 Some(max_steps) => format!("exceeded the step bound ({max_steps})"),
1454 None => "blocked".into(),
1455 };
1456 report.fail(format!(
1457 "{name} was unwinding a panic when the execution stopped, and {why} before it had finished unwinding it"
1458 ));
1459 }
1460 Self::with(|state| state.end_teardown_step());
1461
1462 if finished {
1463 Self::finish_torn_down_task(id, report);
1464 } else {
1465 // The task caught the panic, or is stuck.
1466 Self::leak_task(id, report);
1467 }
1468 }
1469
1470 /// Unwind the stack of a finished execution's task that is in the middle of `poll` (see
1471 /// `tear_down`).
1472 fn unwind_task(id: TaskId, report: &mut TeardownReport) {
1473 let continuation = Self::with(|state| {
1474 state.begin_teardown_step(TeardownStep::Unwind(id));
1475 let task = state.get_mut(id);
1476 // The task may have been blocked, but it runs now, to drop what is on its stack.
1477 task.unblock();
1478 task.take_continuation()
1479 });
1480 let continuation = continuation.expect("an unfinished task has a continuation");
1481
1482 // A panic while the stack unwinds aborts the process, so the panic hook reports every one.
1483 let ignoring_panics = TEARDOWN_IGNORES_PANICS.replace(false);
1484 // Among other things, this reinstates the task's default dispatcher, so that the guards on
1485 // its stack restore their priors in order as it unwinds (see `ParkedDefault`).
1486 Execution::enter_task_span();
1487 let result = panic::catch_unwind(panic::AssertUnwindSafe(move || {
1488 let continuation = Rc::try_unwrap(continuation).map_err(|_| ());
1489 let mut continuation = continuation.expect("teardown owns the continuation").into_inner();
1490 // Only now can the task switch, if it catches the unwind (see `maybe_yield_in_teardown`).
1491 Self::with(|state| state.teardown_mut().on_own_stack = true);
1492 continuation.unwind_stack();
1493 }));
1494 Self::with(|state| state.teardown_mut().on_own_stack = false);
1495 Execution::exit_task_span(false);
1496 TEARDOWN_IGNORES_PANICS.set(ignoring_panics);
1497 if let Err(payload) = result {
1498 report.panicked(payload);
1499 }
1500 Self::with(|state| {
1501 state.get_mut(id).lose_stack();
1502 state.end_teardown_step();
1503 });
1504
1505 Self::finish_torn_down_task(id, report);
1506 }
1507
1508 /// Leak the stack of an abandoned execution's task (see `tear_down`).
1509 fn leak_task(id: TaskId, report: &mut TeardownReport) {
1510 let (parked_default, continuation) = Self::with(|state| {
1511 let task = state.get_mut(id);
1512 task.lose_stack();
1513 (task.parked_default.take(), task.take_continuation())
1514 });
1515 // The parked default has to go before the task's stack does (see `ParkedDefault`).
1516 drop(parked_default);
1517 if let Some(continuation) = continuation {
1518 let continuation = Rc::try_unwrap(continuation).map_err(|_| ());
1519 let mut continuation = continuation.expect("teardown owns the continuation").into_inner();
1520 continuation.leak_stack();
1521 }
1522
1523 Self::finish_torn_down_task(id, report);
1524 }
1525
1526 /// Finish a task that execution teardown is done with, and drop the task-local values it left.
1527 fn finish_torn_down_task(id: TaskId, report: &mut TeardownReport) {
1528 crate::annotations::record_teardown_step(id);
1529 crate::annotations::record_task_terminated();
1530 Self::with(|state| {
1531 let task = state.get_mut(id);
1532 task.finish();
1533 // A task that joins this one can go on (see `JoinHandle::join`).
1534 if let Some(waiter) = task.take_waiter() {
1535 if !state.get(waiter).finished() {
1536 state.get_mut(waiter).unblock();
1537 }
1538 }
1539 });
1540 Self::drop_leftover_task_locals(id, report);
1541 }
1542
1543 /// Drop the task-local values that a task left, as the task, once its stack is gone (see
1544 /// `tear_down` and `Task::stand_in`).
1545 fn drop_leftover_task_locals(id: TaskId, report: &mut TeardownReport) {
1546 let state_before = Self::with(|state| {
1547 if !state.get(id).has_locals() {
1548 return None;
1549 }
1550 state.begin_teardown_step(TeardownStep::Leftovers(id));
1551 Some(state.get_mut(id).stand_in())
1552 });
1553 let Some(state_before) = state_before else {
1554 return;
1555 };
1556
1557 // See `pop_local` for why this loop looks slightly funky.
1558 while let Some(local) = Self::with(|state| state.get_mut(id).pop_local()) {
1559 report.catch(move || drop(local));
1560 }
1561
1562 Self::with(|state| {
1563 state.get_mut(id).stop_standing_in(state_before);
1564 state.end_teardown_step();
1565 });
1566 }
1567
1568 /// Drop what the tasks still hold whose destructors may run user code, at the end of teardown:
1569 /// their spans, which a `tracing` subscriber sees close, and their tags. Each task's go as the
1570 /// task (see `Task::stand_in`). Returns whether there was anything to drop.
1571 fn drop_task_leftovers(report: &mut TeardownReport) -> bool {
1572 let ids = Self::with(|state| {
1573 TASK_ID_TO_TAGS.with(|tags| {
1574 let tags = tags.borrow();
1575 let has_tag = |id| !tags.is_empty() && tags.contains_key(&id);
1576 let tasks = state
1577 .tasks
1578 .iter()
1579 .filter(|task| task.has_leftovers() || has_tag(task.id()));
1580 tasks.map(Task::id).collect::<Vec<_>>()
1581 })
1582 });
1583 for &id in &ids {
1584 let (state_before, leftovers) = Self::with(|state| {
1585 state.begin_teardown_step(TeardownStep::Leftovers(id));
1586 // `current::set_tag_for_task` keeps a reference to the tag too, which can be the last.
1587 let tag = TASK_ID_TO_TAGS.with(|tags| tags.borrow_mut().remove(&id));
1588 let task = state.get_mut(id);
1589 (task.stand_in(), (task.take_leftovers(), tag))
1590 });
1591 report.catch(move || drop(leftovers));
1592 Self::with(|state| {
1593 state.get_mut(id).stop_standing_in(state_before);
1594 state.end_teardown_step();
1595 });
1596 }
1597 !ids.is_empty()
1598 }
1599
1600 /// Drop one of the execution's statics, as the main thread (see `tear_down`).
1601 fn drop_static(value: Box<dyn Any>, report: &mut TeardownReport) {
1602 let main_thread = TaskId(0);
1603 let state_before = Self::with(|state| {
1604 state.begin_teardown_step(TeardownStep::Static);
1605 state.get_mut(main_thread).stand_in()
1606 });
1607
1608 report.catch(move || drop(value));
1609 // The static's destructor may have set task-local values of the main thread's, which go with
1610 // it.
1611 while let Some(local) = Self::with(|state| state.get_mut(main_thread).pop_local()) {
1612 report.catch(move || drop(local));
1613 }
1614
1615 Self::with(|state| {
1616 state.get_mut(main_thread).stop_standing_in(state_before);
1617 state.end_teardown_step();
1618 });
1619 }
1620
1621 /// Make `step` what execution teardown is doing, with its task as the current task (see
1622 /// `tear_down`).
1623 fn begin_teardown_step(&mut self, step: TeardownStep) {
1624 let id = match step {
1625 TeardownStep::Cancel(id)
1626 | TeardownStep::Unwind(id)
1627 | TeardownStep::FinishUnwinding(id)
1628 | TeardownStep::Leftovers(id) => id,
1629 // The main thread stands in for statics.
1630 TeardownStep::Static => TaskId(0),
1631 TeardownStep::Idle => unreachable!("teardown is always doing something during a step"),
1632 };
1633 self.current_task = ScheduledTask::Some(id);
1634 self.teardown_mut().step = step;
1635 TEARDOWN_STEP_PANICS.set(0);
1636 crate::annotations::record_teardown_step(id);
1637 }
1638
1639 /// End the step that `begin_teardown_step` began, if one is under way.
1640 fn end_teardown_step(&mut self) {
1641 let teardown = self.teardown_mut();
1642 teardown.step = TeardownStep::Idle;
1643 self.current_task = teardown.kind.final_state();
1644 }
1645
1646 fn teardown(&self) -> &Teardown {
1647 self.teardown.as_ref().expect("the execution is being torn down")
1648 }
1649
1650 fn teardown_mut(&mut self) -> &mut Teardown {
1651 self.teardown.as_mut().expect("the execution is being torn down")
1652 }
1653
1654 /// Whether the panic hook is to stay silent about a panic with `payload`, as execution teardown
1655 /// ignores the panics it catches now (see `tear_down`). A panic in a destructor while another
1656 /// panic unwinds aborts the process. So the hook reports the panic that says so, and any panic
1657 /// after the first in a step, or in what `TeardownReport::catch` drops, which may cause that.
1658 pub(crate) fn teardown_ignores_panic(payload: &dyn Any) -> bool {
1659 let panics = TEARDOWN_STEP_PANICS.get().saturating_add(1);
1660 TEARDOWN_STEP_PANICS.set(panics);
1661 let aborts = payload.downcast_ref::<&str>() == Some(&"panic in a destructor during cleanup");
1662 TEARDOWN_IGNORES_PANICS.get() && panics == 1 && !aborts
1663 }
1664
1665 /// Whether execution teardown is unwinding the current task's stack (see `tear_down`). That is
1666 /// not a panic, although `std::thread::panicking()` says otherwise, so it shouldn't poison a
1667 /// lock, say.
1668 pub fn unwinding_for_teardown() -> bool {
1669 Self::try_with(|state| {
1670 state
1671 .teardown
1672 .as_ref()
1673 .is_some_and(|teardown| matches!(teardown.step, TeardownStep::Unwind(_)))
1674 })
1675 .unwrap_or(false)
1676 }
1677
1678 /// Drop the current task's task-local values, as a task does when it finishes. A destructor can
1679 /// set another task-local value, which is dropped too (see `Task::pop_local`).
1680 pub fn drop_task_locals() {
1681 // See `pop_local` for why this loop looks slightly funky.
1682 while let Some(local) = Self::with(|state| state.current_mut().pop_local()) {
1683 tracing::trace!("dropping task-local value {:p}", local);
1684 drop(local);
1685 }
1686 }
1687
1688 /// Stop a destructor that execution teardown cannot let go on (see `maybe_yield_in_teardown`).
1689 #[cold]
1690 fn stop_destructor(stall: TeardownStall) -> ! {
1691 if std::thread::panicking() {
1692 // This aborts the process, so report it in any case.
1693 TEARDOWN_IGNORES_PANICS.set(false);
1694 }
1695 let (name, step) = Self::with(|state| (state.current().display_name(), state.teardown().step));
1696 let who = match step {
1697 TeardownStep::Static => format!("A static's destructor, dropped as {name},"),
1698 TeardownStep::Leftovers(_) => format!("A destructor of something that {name} left behind"),
1699 _ => name,
1700 };
1701 match stall {
1702 TeardownStall::Blocked => {
1703 let reason = if std::thread::panicking() || matches!(step, TeardownStep::Unwind(_)) {
1704 "while unwinding a stack, which cannot be suspended"
1705 } else {
1706 "outside any task's stack, where it cannot wait"
1707 };
1708 panic!(
1709 "{who} blocked while it was being dropped at the end of the execution, {reason}. A destructor made a \
1710 blocking call, such as locking a mutex that another task holds, receiving from an empty channel, \
1711 joining a task, or `block_on` on a future that cannot complete."
1712 )
1713 }
1714 TeardownStall::ExceededStepBound(max_steps) => panic!(
1715 "{who} exceeded the step bound ({max_steps}) while it was being dropped at the end of the execution. A \
1716 destructor may be spinning, waiting for something that no longer happens: once the execution is over, \
1717 only destructors run."
1718 ),
1719 }
1720 }
1721
1722 /// Determine whether the execution has finished.
1723 pub fn is_finished(&self) -> bool {
1724 self.current_task == ScheduledTask::Stopped || self.current_task == ScheduledTask::Finished
1725 }
1726
1727 /// Invoke the scheduler to decide which task to schedule next. Returns true if the chosen task
1728 /// is different from the currently running task, indicating that the current task should yield
1729 /// its execution.
1730 pub fn maybe_yield() -> bool {
1731 let decision = Self::with(|state| {
1732 if state.teardown.is_some() {
1733 return state.maybe_yield_in_teardown();
1734 }
1735
1736 if std::thread::panicking() {
1737 if !state.switched_out_unwinding {
1738 state.record_unwinding_task();
1739 }
1740 return Ok(true);
1741 }
1742 if state.switched_out_unwinding {
1743 // No task is unwinding a panic, as the tasks share the OS thread.
1744 state.forget_unwinding_task();
1745 }
1746
1747 debug_assert!(
1748 matches!(state.current_task, ScheduledTask::Some(_)) && state.next_task == ScheduledTask::None,
1749 "we're inside a task and scheduler should not yet have run"
1750 );
1751
1752 let result = state.schedule();
1753 // If scheduling failed, yield so that the outer scheduling loop can handle it.
1754 if result.is_err() {
1755 return Ok(true);
1756 }
1757
1758 // If the next task is the same as the current one, we can skip the context switch
1759 // and just advance to the next task immediately.
1760 if state.current_task == state.next_task {
1761 state.advance_to_next_task();
1762 Ok(false)
1763 } else {
1764 Ok(true)
1765 }
1766 });
1767 // Panic outside of `Self::with`, so the panic hook can name the task.
1768 decision.unwrap_or_else(|stall| Self::stop_destructor(stall))
1769 }
1770
1771 /// The current task switches out while it unwinds a panic, and no task was unwinding one when it
1772 /// was resumed (see `switched_out_unwinding`), so the panic is its own. (`std::thread::panicking()`
1773 /// can't tell which task panics: tasks share the OS thread. Between steps, on the executor's
1774 /// stack, it says exactly whether a task that switched out is unwinding a panic.) Until the task
1775 /// has finished unwinding the panic, the execution waits for it, even if it is detached.
1776 /// Otherwise the execution could end before the task resumes, and its panic would be lost. There
1777 /// is one such task at most: while it is unwinding, a task that is resumed can't tell a panic of
1778 /// its own from it.
1779 #[cold]
1780 fn record_unwinding_task(&mut self) {
1781 debug_assert!(self.unwinding_task.is_none());
1782 self.switched_out_unwinding = true;
1783 if let Some(me) = self.current_task.id() {
1784 let detached = std::mem::replace(&mut self.get_mut(me).detached, false);
1785 self.unwinding_task = Some((me, detached));
1786 }
1787 }
1788
1789 /// No task is unwinding a panic any more: the task that was recorded doing so (see
1790 /// `record_unwinding_task`) is detached again if it was before.
1791 #[cold]
1792 fn forget_unwinding_task(&mut self) {
1793 self.switched_out_unwinding = false;
1794 if let Some((task, detached)) = self.unwinding_task.take() {
1795 self.get_mut(task).detached = detached;
1796 }
1797 }
1798
1799 /// Detach a task, so that the execution doesn't wait for it to finish, as when its `JoinHandle`
1800 /// is dropped. A task that is unwinding a panic is detached once it has finished unwinding (see
1801 /// `record_unwinding_task`).
1802 pub fn detach(&mut self, id: TaskId) {
1803 match &mut self.unwinding_task {
1804 Some((task, detached)) if *task == id => *detached = true,
1805 _ => self.get_mut(id).detach(),
1806 }
1807 }
1808
1809 /// `maybe_yield` during execution teardown, which schedules nothing (see `tear_down`). Each
1810 /// scheduling point counts as a step, which the step bound limits as in a running execution.
1811 /// A task that teardown runs on its own stack switches out when it blocks or yields, so that
1812 /// teardown can run others. Anywhere else, or while a panic unwinds, nothing can be suspended,
1813 /// so a destructor that blocks is stopped with a panic. A task that finishes unwinding a panic
1814 /// switches out instead, and as soon as it has caught the panic (see `finish_unwinding`).
1815 #[cold]
1816 fn maybe_yield_in_teardown(&mut self) -> Result<bool, TeardownStall> {
1817 // Destructors do what tasks do, so `current::context_switches` counts their scheduling points
1818 // too.
1819 self.context_switches += 1;
1820 let panicking = std::thread::panicking();
1821 let yielded = std::mem::take(&mut self.has_yielded);
1822 let max_steps = self.teardown_step_bound();
1823
1824 let teardown = self.teardown_mut();
1825 teardown.steps += 1;
1826 let finishing_unwind = teardown.on_own_stack && matches!(teardown.step, TeardownStep::FinishUnwinding(_));
1827 if let Some(max_steps) = max_steps {
1828 // Stopping a destructor while a panic unwinds aborts the process, so give the unwind,
1829 // likely that of the panic that stopped the destructor, some more steps.
1830 let bound = if panicking && !finishing_unwind {
1831 max_steps.saturating_mul(2)
1832 } else {
1833 max_steps
1834 };
1835 if teardown.steps > bound {
1836 teardown.exceeded_step_bound = Some(max_steps);
1837 if finishing_unwind {
1838 return Ok(true);
1839 }
1840 return Err(TeardownStall::ExceededStepBound(max_steps));
1841 }
1842 }
1843 let on_own_stack = teardown.on_own_stack && !panicking;
1844 if on_own_stack && matches!(teardown.step, TeardownStep::Unwind(_)) {
1845 // The task caught the unwind of its stack, which starts again once the task switches
1846 // (see `Coroutine::force_unwind`). It has to run for that.
1847 self.current_mut().unblock();
1848 return Ok(true);
1849 }
1850 if finishing_unwind {
1851 // Once no panic is unwinding, the task has caught its panic.
1852 return Ok(!panicking || !self.current().runnable());
1853 }
1854
1855 // Between steps (dropping the payload of a panic, say), no task runs that could switch out.
1856 let Some(runnable) = self.try_current().map(Task::runnable) else {
1857 return Ok(false);
1858 };
1859 if !runnable && !on_own_stack {
1860 // A destructor blocked the task where it cannot wait. Make it runnable again, so that the
1861 // unwind of the panic that stops the destructor doesn't stall too.
1862 self.current_mut().unblock();
1863 return Err(TeardownStall::Blocked);
1864 }
1865 Ok(on_own_stack && (!runnable || yielded))
1866 }
1867
1868 /// The step bound for the destructors that execution teardown runs, if there is one (see
1869 /// `maybe_yield_in_teardown`).
1870 fn teardown_step_bound(&self) -> Option<usize> {
1871 match self.config.max_steps {
1872 MaxSteps::FailAfter(max_steps) => Some(max_steps),
1873 // A small bound that stops an execution is not meant to fail a test, and teardown has no
1874 // way to stop. So destructors get at least as many steps as an execution does by default.
1875 MaxSteps::ContinueAfter(max_steps) => Some(max_steps.max(DEFAULT_MAX_STEPS)),
1876 MaxSteps::None => None,
1877 }
1878 }
1879
1880 /// Tell the scheduler that the next context switch is an explicit yield requested by the
1881 /// current task. Some schedulers use this as a hint to influence scheduling.
1882 pub fn request_yield() {
1883 Self::with(|state| {
1884 state.has_yielded = true;
1885 });
1886 }
1887
1888 /// Check whether Shuttle's own `Drop` handlers should skip their bookkeeping, and early exit,
1889 /// because the execution has stopped, and so is being abandoned (see `tear_down`).
1890 ///
1891 /// We also stop if we are currently panicking (e.g., perhaps we're unwinding the stack for a
1892 /// panic triggered while someone held a Mutex, and so are executing the Drop handler for
1893 /// MutexGuard). This avoids calling back into the scheduler during a panic, because the state
1894 /// may be poisoned or otherwise invalid.
1895 ///
1896 /// While a finished execution is torn down, though, destructors run as in a running execution,
1897 /// also while a panic unwinds: teardown doesn't call the scheduler. A failed execution is
1898 /// abandoned like a stopped one.
1899 pub fn should_stop() -> bool {
1900 if std::thread::panicking() {
1901 // The state may be borrowed, so don't insist on it.
1902 return Self::try_with(|s| s.stops(true)).unwrap_or(true);
1903 }
1904 Self::with(|s| s.stops(false))
1905 }
1906
1907 /// `should_stop`, given whether the current task is panicking.
1908 pub(crate) fn stops(&self, panicking: bool) -> bool {
1909 match &self.teardown {
1910 Some(teardown) => teardown.kind.abandons(),
1911 None => {
1912 panicking || {
1913 assert_ne!(self.current_task, ScheduledTask::Finished);
1914 self.current_task == ScheduledTask::Stopped
1915 }
1916 }
1917 }
1918 }
1919
1920 /// Whether the execution stopped or failed, so that the stacks of its unfinished tasks are
1921 /// leaked rather than unwound (see `tear_down`).
1922 pub(crate) fn execution_stopped() -> bool {
1923 Self::try_with(|state| match &state.teardown {
1924 Some(teardown) => teardown.kind.abandons(),
1925 None => state.current_task == ScheduledTask::Stopped,
1926 })
1927 .unwrap_or(false)
1928 }
1929
1930 /// Generate some diagnostic information used when persisting failures.
1931 ///
1932 /// Because this method may be called from a panic hook, it must not panic.
1933 pub fn failing_task() -> String {
1934 Self::try_with(|state| {
1935 if let Some(task) = state.try_current() {
1936 let name = task.display_name();
1937 match state.teardown.as_ref().map(|teardown| teardown.step) {
1938 None | Some(TeardownStep::Idle) => name,
1939 Some(TeardownStep::Static) => format!("{name} (dropping a static at the end of the execution)"),
1940 Some(TeardownStep::FinishUnwinding(_)) => {
1941 format!("{name} (unwinding a panic at the end of the execution)")
1942 }
1943 Some(_) => format!("{name} (being dropped at the end of the execution)"),
1944 }
1945 } else if let Some(name) = &state.failed_task {
1946 // The task that failed the execution, which has been torn down since.
1947 name.clone()
1948 } else {
1949 "<unknown>".into()
1950 }
1951 })
1952 .unwrap_or_else(|e| format!("Tried to get ExecutionState, but got the following error: {e:?}"))
1953 }
1954
1955 /// Generate a random u64 from the current scheduler and return it.
1956 #[inline]
1957 pub fn next_u64() -> u64 {
1958 Self::with(|state| {
1959 if let Some(teardown) = &mut state.teardown {
1960 // Teardown doesn't extend the schedule (see `Teardown::rng`).
1961 return teardown.next_u64();
1962 }
1963 CurrentSchedule::push_random();
1964 state.scheduler.borrow_mut().next_u64()
1965 })
1966 }
1967
1968 pub fn current(&self) -> &Task {
1969 self.get(self.current_task.id().expect("there is no current task"))
1970 }
1971
1972 pub fn current_mut(&mut self) -> &mut Task {
1973 self.get_mut(self.current_task.id().expect("there is no current task"))
1974 }
1975
1976 pub fn try_current(&self) -> Option<&Task> {
1977 self.try_get(self.current_task.id()?)
1978 }
1979
1980 pub fn get(&self, id: TaskId) -> &Task {
1981 self.try_get(id).unwrap()
1982 }
1983
1984 /// Register a newly created task. Task ids are handed out sequentially as `tasks.len()`, so the
1985 /// new id is always greater than every existing one and `live_tasks` stays sorted.
1986 fn add_task(&mut self, task: Task) {
1987 debug_assert!(self.live_tasks.last().is_none_or(|last| *last < task.id()));
1988 self.live_tasks.push(task.id());
1989 self.tasks.push(task);
1990 }
1991
1992 /// Mark the task as finished and drop it from the set of live tasks.
1993 fn finish_task(&mut self, task_id: TaskId) {
1994 self.get_mut(task_id).finish();
1995 let idx = self
1996 .live_tasks
1997 .binary_search(&task_id)
1998 .expect("finished task must be live");
1999 self.live_tasks.remove(idx);
2000 if self.unwinding_task.is_some_and(|(task, _)| task == task_id) {
2001 self.unwinding_task = None;
2002 }
2003 }
2004
2005 /// Mark the current task as finished and drop it from the set of live tasks.
2006 fn finish_current_task(&mut self) {
2007 self.finish_task(self.current_task.id().unwrap());
2008 }
2009
2010 pub fn get_mut(&mut self, id: TaskId) -> &mut Task {
2011 self.tasks.get_mut(id.0).unwrap()
2012 }
2013
2014 pub fn try_get(&self, id: TaskId) -> Option<&Task> {
2015 self.tasks.get(id.0)
2016 }
2017
2018 pub fn try_get_mut(&mut self, id: TaskId) -> Option<&mut Task> {
2019 self.tasks.get_mut(id.0)
2020 }
2021
2022 /// Whether the execution is being torn down (see `tear_down`).
2023 pub fn in_cleanup(&self) -> bool {
2024 self.teardown.is_some()
2025 }
2026
2027 pub fn context_switches() -> usize {
2028 Self::with(|state| state.context_switches)
2029 }
2030
2031 #[track_caller]
2032 pub fn new_resource_signature(resource_type: ResourceType) -> ResourceSignature {
2033 ExecutionState::with(|s| s.current_mut().signature.new_resource(resource_type))
2034 }
2035
2036 pub fn get_storage<K: Into<StorageKey>, T: 'static>(&self, key: K) -> Option<&T> {
2037 self.storage
2038 .get(key.into())
2039 .map(|result| result.expect("global storage is never destructed"))
2040 }
2041
2042 pub fn init_storage<K: Into<StorageKey>, T: 'static>(&mut self, key: K, value: T) {
2043 self.storage.init(key.into(), value);
2044 }
2045
2046 pub fn get_clock(&self, id: TaskId) -> &VectorClock {
2047 &self.tasks.get(id.0).unwrap().clock
2048 }
2049
2050 pub fn get_clock_mut(&mut self, id: TaskId) -> &mut VectorClock {
2051 &mut self.tasks.get_mut(id.0).unwrap().clock
2052 }
2053
2054 /// Increment the current thread's clock entry and update its clock with the one provided.
2055 pub fn update_clock(&mut self, clock: &VectorClock) {
2056 let task = self.current_mut();
2057 task.clock.increment(task.id);
2058 task.clock.update(clock);
2059 }
2060
2061 /// Increment the current thread's clock and return a shared reference to it
2062 pub fn increment_clock(&mut self) -> &VectorClock {
2063 let task = self.current_mut();
2064 task.clock.increment(task.id);
2065 &task.clock
2066 }
2067
2068 /// Increment the current thread's clock and return a mutable reference to it
2069 pub fn increment_clock_mut(&mut self) -> &mut VectorClock {
2070 let task = self.current_mut();
2071 task.clock.increment(task.id);
2072 &mut task.clock
2073 }
2074
2075 /// Set the number of steps used, with respect to the step bound, to 0 (see
2076 /// `current::reset_step_count`). During execution teardown, these are the steps of destructors
2077 /// (see `maybe_yield_in_teardown`).
2078 pub(crate) fn reset_step_count(&mut self) {
2079 match &mut self.teardown {
2080 Some(teardown) => teardown.steps = 0,
2081 None => self.steps_reset_at = CurrentSchedule::len(),
2082 }
2083 }
2084
2085 /// Returns `true` if the test has exceeded the step bound, and `false` otherwise.
2086 fn is_step_bound_exceeded(&self, max_steps: usize) -> bool {
2087 CurrentSchedule::len() - self.steps_reset_at >= max_steps
2088 }
2089
2090 /// Run the scheduler to choose the next task to run. `has_yielded` should be false if the
2091 /// scheduler is being invoked from within a running task. If scheduling fails, returns an Err
2092 /// with a String describing the failure.
2093 fn schedule(&mut self) -> Result<(), StepError> {
2094 // Don't schedule twice. If `maybe_yield` ran the scheduler, we don't want to run it
2095 // again at the top of `step`.
2096 if self.next_task != ScheduledTask::None {
2097 return Ok(());
2098 }
2099
2100 self.context_switches += 1;
2101
2102 match self.config.max_steps {
2103 MaxSteps::FailAfter(max_steps) => {
2104 if self.is_step_bound_exceeded(max_steps) {
2105 return Err(StepError::StepBoundExceeded(max_steps));
2106 }
2107 }
2108 MaxSteps::ContinueAfter(max_steps) => {
2109 if self.is_step_bound_exceeded(max_steps) {
2110 // TODO: We have to set `Stopped` and return `Ok` here, else assertions will fail. This should probably be cleaned up.
2111 self.next_task = ScheduledTask::Stopped;
2112 return Ok(());
2113 }
2114 }
2115 MaxSteps::None => {}
2116 }
2117
2118 let mut unfinished_attached = false;
2119 let mut all_runnable_detached = true;
2120 let mut any_runnable = false;
2121
2122 // The loop below only looks at `live_tasks`, so a task missing from that set would silently
2123 // never be scheduled. Check that direction of the invariant here; the loop itself checks the
2124 // other direction (that no finished task is still in the set).
2125 debug_assert!(
2126 self.tasks
2127 .iter()
2128 .filter(|task| !task.finished() && task.runnable())
2129 .all(|task| self.live_tasks.binary_search(&task.id()).is_ok()),
2130 "live_tasks is missing a runnable unfinished task"
2131 );
2132
2133 for &task_id in &self.live_tasks {
2134 let task = &self.tasks[task_id.0];
2135 debug_assert!(!task.finished());
2136 unfinished_attached |= !task.detached;
2137 let is_runnable = task.runnable();
2138 any_runnable |= is_runnable;
2139
2140 if is_runnable {
2141 all_runnable_detached &= task.detached;
2142 self.runnable_tasks.push(task as *const Task);
2143 } else if task.can_spuriously_wakeup() {
2144 // Some blocked tasks can be woken up spuriously, even though the condition the task is
2145 // blocked on hasn't happened yet. We'll add such tasks to the list of runnable tasks, but
2146 // they won't contribute to the check on `any_runnable`; if the only runnable tasks
2147 // are ones that are waiting for a potential spurious wakeup, it should still be treated as
2148 // a deadlock since there's no guarantee that spurious wakeups will ever occur.
2149 self.runnable_tasks.push(task as *const Task);
2150 }
2151 }
2152
2153 // We should finish execution when either
2154 // (1) There are no runnable tasks, or
2155 // (2) All runnable tasks have been detached AND there are no unfinished attached tasks
2156 // If there are some unfinished attached tasks and all runnable tasks are detached, we must
2157 // run some detached task to give them a chance to unblock some unfinished attached task.
2158 if !any_runnable || (!unfinished_attached && all_runnable_detached) {
2159 self.next_task = ScheduledTask::Finished;
2160 return Ok(());
2161 }
2162
2163 let is_yielding = std::mem::replace(&mut self.has_yielded, false);
2164
2165 // Cast the slice of raw pointers to a slice of references in place to provide schedulers with a safe API
2166 //
2167 // SAFETY: This is safe because the tasks themselves are only being accessed through this shared reference by the
2168 // schedulers, and all references are always cleared from the runnable_tasks Vec at the end of this function.
2169 // The transmute itself is safe because *const and & have the same layout, and the pointer is created from a
2170 // reference earlier in this function.
2171 let task_refs = unsafe { std::mem::transmute::<&[*const Task], &[&Task]>(&self.runnable_tasks) };
2172
2173 self.next_task = self
2174 .scheduler
2175 .borrow_mut()
2176 .next_task(task_refs, self.current_task.id(), is_yielding)
2177 .map(ScheduledTask::Some)
2178 .unwrap_or(ScheduledTask::Stopped);
2179
2180 // Tracing this `in_scope` is purely a matter of taste. We do it because
2181 // 1) It is an action taken by the scheduler, and should thus be traced under the scheduler's span
2182 // 2) It creates a visual separation of scheduling decisions and `Task`-induced tracing.
2183 // Note that there is a case to be made for not `in_scope`-ing it, as that makes seeing the context
2184 // of the context switch clearer.
2185 //
2186 // Note also that changing this trace! statement requires changing the test `basic::labels::test_tracing_with_label_fn`
2187 // which relies on this trace reporting the `runnable` tasks.
2188 self.top_level_span.in_scope(|| {
2189 trace!(
2190 i=CurrentSchedule::len(),
2191 next_task=?self.next_task,
2192 runnable=?task_refs.iter().map(|task| task.id()).collect::<SmallVec<[_; DEFAULT_INLINE_TASKS]>>(),
2193 "scheduling decision"
2194 );
2195 });
2196
2197 // If the task chosen by the scheduler is blocked, then it should be one that can be
2198 // spuriously woken up, and we need to unblock it here so that it can execute.
2199 if let Some(tid) = self.next_task.id() {
2200 let task = self.get_mut(tid);
2201 assert!(task.runnable() || task.blocked());
2202 if task.blocked() {
2203 assert!(task.can_spuriously_wakeup());
2204 task.unblock();
2205 }
2206 }
2207
2208 // Retains the capacity of `runnable_tasks` for future calls of `schedule`
2209 self.runnable_tasks.clear();
2210
2211 Ok(())
2212 }
2213
2214 /// Set the next task as the current task
2215 fn advance_to_next_task(&mut self) {
2216 debug_assert_ne!(self.next_task, ScheduledTask::None);
2217 self.current_task = self.next_task.take();
2218
2219 if let ScheduledTask::Some(tid) = self.current_task {
2220 CurrentSchedule::push_task(tid);
2221 }
2222 }
2223
2224 // Sets the `tag` field of the current task.
2225 // Returns the `tag` which was there previously.
2226 #[allow(deprecated)]
2227 pub fn set_tag_for_current_task(tag: Arc<dyn Tag>) -> Option<Arc<dyn Tag>> {
2228 ExecutionState::with(|s| s.current_mut().set_tag(tag))
2229 }
2230
2231 #[allow(deprecated)]
2232 fn get_tag_or_default_for_current_task(&self) -> Option<Arc<dyn Tag>> {
2233 self.try_current().and_then(|current| current.get_tag())
2234 }
2235
2236 #[allow(deprecated)]
2237 pub fn get_tag_for_current_task() -> Option<Arc<dyn Tag>> {
2238 ExecutionState::with(|s| s.get_tag_or_default_for_current_task())
2239 }
2240
2241 #[allow(deprecated)]
2242 pub fn set_tag_for_task(task: TaskId, tag: Arc<dyn Tag>) -> Option<Arc<dyn Tag>> {
2243 ExecutionState::with(|s| s.get_mut(task).set_tag(tag))
2244 }
2245}
2246
2247#[cfg(debug_assertions)]
2248impl Drop for ExecutionState {
2249 fn drop(&mut self) {
2250 assert!(self.has_cleaned_up || std::thread::panicking());
2251 }
2252}
2253
2254/// Whether the running task's default dispatcher might not be the execution's.
2255fn task_may_have_own_default(state: &ExecutionState) -> bool {
2256 use tracing::level_filters::LevelFilter;
2257
2258 // No dispatcher enables anything, so whichever one is the default makes no difference. (If that
2259 // changes while the task is switched out, other tasks see its default, as before defaults were
2260 // parked.)
2261 if LevelFilter::current() == LevelFilter::OFF {
2262 return false;
2263 }
2264 tracing::dispatcher::get_default(|current| {
2265 if let Some(global) = state.top_level_dispatch_global_ptr {
2266 // `get_default` hands out the global default from where it lives, and anything else from
2267 // this thread's scoped default. So if it hands out the global default, the task has no
2268 // default of its own. Parking otherwise is sometimes unnecessary but always safe,
2269 // including if `tracing` ever handed out the global default from elsewhere.
2270 !std::ptr::eq(current, global)
2271 } else if current.is::<NoSubscriber>() {
2272 // Nothing is recorded either way if the execution's default is `Dispatch::none()` too.
2273 !state.top_level_dispatch.is::<NoSubscriber>()
2274 } else {
2275 // A task can only have a default of its own through a scoped default. `get_default` hands
2276 // nested calls `Dispatch::none()` only while some thread has one; otherwise they get the
2277 // global default, which is then `current` and everyone's default.
2278 tracing::dispatcher::get_default(|nested| nested.is::<NoSubscriber>())
2279 }
2280 })
2281}