use crate::loom::sync::Arc;
use crate::runtime::context;
use crate::runtime::scheduler::{self, current_thread, Inject};
use crate::task::Id;
use backtrace::BacktraceFrame;
use std::cell::Cell;
use std::collections::VecDeque;
use std::ffi::c_void;
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::ptr::{self, NonNull};
use std::task::{self, Poll};
mod symbol;
mod tree;
use symbol::Symbol;
use tree::Tree;
use super::{Notified, OwnedTasks, Schedule};
type Backtrace = Vec<BacktraceFrame>;
type SymbolTrace = Vec<Symbol>;
pub(crate) struct Context {
active_frame: Cell<Option<NonNull<Frame>>>,
collector: Cell<Option<Trace>>,
}
struct Frame {
inner_addr: *const c_void,
parent: Option<NonNull<Frame>>,
}
#[derive(Clone, Debug)]
pub(crate) struct Trace {
backtraces: Vec<Backtrace>,
}
pin_project_lite::pin_project! {
#[derive(Debug, Clone)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct Root<T> {
#[pin]
future: T,
}
}
const FAIL_NO_THREAD_LOCAL: &str = "The Tokio thread-local has been destroyed \
as part of shutting down the current \
thread, so collecting a taskdump is not \
possible.";
impl Context {
pub(crate) const fn new() -> Self {
Context {
active_frame: Cell::new(None),
collector: Cell::new(None),
}
}
unsafe fn try_with_current<F, R>(f: F) -> Option<R>
where
F: FnOnce(&Self) -> R,
{
crate::runtime::context::with_trace(f)
}
unsafe fn with_current_frame<F, R>(f: F) -> R
where
F: FnOnce(&Cell<Option<NonNull<Frame>>>) -> R,
{
Self::try_with_current(|context| f(&context.active_frame)).expect(FAIL_NO_THREAD_LOCAL)
}
fn with_current_collector<F, R>(f: F) -> R
where
F: FnOnce(&Cell<Option<Trace>>) -> R,
{
unsafe {
Self::try_with_current(|context| f(&context.collector)).expect(FAIL_NO_THREAD_LOCAL)
}
}
pub(crate) fn is_tracing() -> bool {
Self::with_current_collector(|maybe_collector| {
let collector = maybe_collector.take();
let result = collector.is_some();
maybe_collector.set(collector);
result
})
}
}
impl Trace {
#[inline(never)]
pub(crate) fn capture<F, R>(f: F) -> (R, Trace)
where
F: FnOnce() -> R,
{
let collector = Trace { backtraces: vec![] };
let previous = Context::with_current_collector(|current| current.replace(Some(collector)));
let result = f();
let collector =
Context::with_current_collector(|current| current.replace(previous)).unwrap();
(result, collector)
}
#[inline(never)]
pub(crate) fn root<F>(future: F) -> Root<F> {
Root { future }
}
pub(crate) fn backtraces(&self) -> &[Backtrace] {
&self.backtraces
}
}
#[inline(never)]
pub(crate) fn trace_leaf(cx: &mut task::Context<'_>) -> Poll<()> {
let did_trace = unsafe {
Context::try_with_current(|context_cell| {
if let Some(mut collector) = context_cell.collector.take() {
let mut frames = vec![];
let mut above_leaf = false;
if let Some(active_frame) = context_cell.active_frame.get() {
let active_frame = active_frame.as_ref();
backtrace::trace(|frame| {
let below_root = !ptr::eq(frame.symbol_address(), active_frame.inner_addr);
if above_leaf && below_root {
frames.push(frame.to_owned().into());
}
if ptr::eq(frame.symbol_address(), trace_leaf as *const _) {
above_leaf = true;
}
below_root
});
}
collector.backtraces.push(frames);
context_cell.collector.set(Some(collector));
true
} else {
false
}
})
.unwrap_or(false)
};
if did_trace {
context::with_scheduler(|scheduler| {
if let Some(scheduler) = scheduler {
match scheduler {
scheduler::Context::CurrentThread(s) => s.defer.defer(cx.waker()),
#[cfg(feature = "rt-multi-thread")]
scheduler::Context::MultiThread(s) => s.defer.defer(cx.waker()),
}
}
});
Poll::Pending
} else {
Poll::Ready(())
}
}
impl fmt::Display for Trace {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
Tree::from_trace(self.clone()).fmt(f)
}
}
fn defer<F: FnOnce() -> R, R>(f: F) -> impl Drop {
use std::mem::ManuallyDrop;
struct Defer<F: FnOnce() -> R, R>(ManuallyDrop<F>);
impl<F: FnOnce() -> R, R> Drop for Defer<F, R> {
#[inline(always)]
fn drop(&mut self) {
unsafe {
ManuallyDrop::take(&mut self.0)();
}
}
}
Defer(ManuallyDrop::new(f))
}
impl<T: Future> Future for Root<T> {
type Output = T::Output;
#[inline(never)]
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
unsafe {
let mut frame = Frame {
inner_addr: Self::poll as *const c_void,
parent: None,
};
Context::with_current_frame(|current| {
frame.parent = current.take();
current.set(Some(NonNull::from(&frame)));
});
let _restore = defer(|| {
Context::with_current_frame(|current| {
current.set(frame.parent);
});
});
let this = self.project();
this.future.poll(cx)
}
}
}
pub(in crate::runtime) fn trace_current_thread(
owned: &OwnedTasks<Arc<current_thread::Handle>>,
local: &mut VecDeque<Notified<Arc<current_thread::Handle>>>,
injection: &Inject<Arc<current_thread::Handle>>,
) -> Vec<(Id, Trace)> {
let mut dequeued = Vec::new();
while let Some(task) = local.pop_back() {
dequeued.push(task);
}
while let Some(task) = injection.pop() {
dequeued.push(task);
}
trace_owned(owned, dequeued)
}
cfg_rt_multi_thread! {
use crate::loom::sync::Mutex;
use crate::runtime::scheduler::multi_thread;
use crate::runtime::scheduler::multi_thread::Synced;
use crate::runtime::scheduler::inject::Shared;
pub(in crate::runtime) unsafe fn trace_multi_thread(
owned: &OwnedTasks<Arc<multi_thread::Handle>>,
local: &mut multi_thread::queue::Local<Arc<multi_thread::Handle>>,
synced: &Mutex<Synced>,
injection: &Shared<Arc<multi_thread::Handle>>,
) -> Vec<(Id, Trace)> {
let mut dequeued = Vec::new();
while let Some(notified) = local.pop() {
dequeued.push(notified);
}
let mut synced = synced.lock();
while let Some(notified) = injection.pop(&mut synced.inject) {
dequeued.push(notified);
}
drop(synced);
trace_owned(owned, dequeued)
}
}
fn trace_owned<S: Schedule>(owned: &OwnedTasks<S>, dequeued: Vec<Notified<S>>) -> Vec<(Id, Trace)> {
let mut tasks = dequeued;
owned.for_each(|task| {
if let Some(notified) = task.notify_for_tracing() {
tasks.push(notified);
}
});
tasks
.into_iter()
.map(|task| {
let local_notified = owned.assert_owner(task);
let id = local_notified.task.id();
let ((), trace) = Trace::capture(|| local_notified.run());
(id, trace)
})
.collect()
}