use logwise::{ContextToken, Dispatch, EventRef, Interest, Metadata};
use some_executor::{
current_executor::current_executor,
task::{Configuration, Task},
};
use std::cell::Cell;
use std::future::Future;
use std::pin::Pin;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use std::task::{Context, Poll, Waker};
#[cfg(not(target_arch = "wasm32"))]
use std::thread;
#[cfg(target_arch = "wasm32")]
use wasm_lite_std as thread;
thread_local! {
static CURRENT: Cell<ContextToken> = const { Cell::new(ContextToken::NONE) };
}
struct Capture {
next: AtomicU64,
parents: Mutex<Vec<(ContextToken, ContextToken)>>,
events: Mutex<Vec<(&'static str, ContextToken)>>,
}
impl Dispatch for Capture {
fn generation(&self) -> usize {
0
}
fn interest(&self, _metadata: &'static Metadata) -> Interest {
Interest::CORE_SUPPORT.union(Interest::DETAIL_LOCAL)
}
fn emit(&self, event: EventRef<'_>) {
self.events
.lock()
.unwrap()
.push((event.metadata.event_name, event.context));
}
fn capture_context(&self) -> ContextToken {
CURRENT.with(Cell::get)
}
fn create_context(&self, parent: ContextToken, _name: &'static str) -> ContextToken {
let child = ContextToken::from_parts(self.next.fetch_add(1, Ordering::Relaxed) + 1, 0);
self.parents.lock().unwrap().push((child, parent));
child
}
fn enter_context(&self, context: ContextToken) -> ContextToken {
CURRENT.with(|current| current.replace(context))
}
fn exit_context(&self, previous: ContextToken) {
CURRENT.with(|current| current.set(previous));
}
}
static CAPTURE: Capture = Capture {
next: AtomicU64::new(0),
parents: Mutex::new(Vec::new()),
events: Mutex::new(Vec::new()),
};
struct PollTwice {
contexts: &'static Mutex<Vec<ContextToken>>,
first: bool,
}
impl Future for PollTwice {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.contexts
.lock()
.unwrap()
.push(logwise::context::capture());
if self.first {
self.first = false;
cx.waker().wake_by_ref();
Poll::Pending
} else {
Poll::Ready(())
}
}
}
static POLL_CONTEXTS: Mutex<Vec<ContextToken>> = Mutex::new(Vec::new());
#[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test(worker))]
#[cfg_attr(not(target_arch = "wasm32"), test)]
fn task_context_survives_thread_migration_and_restores_each_thread() {
logwise::install_dispatcher(&CAPTURE).unwrap();
let parent = ContextToken::from_parts(10_000, 0);
let parent_guard = logwise::context::enter(parent);
let task = Task::without_notifications(
"migrating".to_string(),
Configuration::default(),
PollTwice {
contexts: &POLL_CONTEXTS,
first: true,
},
);
let mut executor = current_executor();
let (spawned, observer) = task.spawn(&mut executor);
drop(parent_guard);
let mut spawned = thread::spawn(move || {
let mut spawned = Box::pin(spawned);
let mut cx = Context::from_waker(Waker::noop());
assert!(Future::poll(spawned.as_mut(), &mut cx).is_pending());
assert_eq!(logwise::context::capture(), ContextToken::NONE);
spawned
})
.join()
.unwrap();
let mut cx = Context::from_waker(Waker::noop());
assert!(Future::poll(spawned.as_mut(), &mut cx).is_ready());
assert_eq!(logwise::context::capture(), ContextToken::NONE);
drop(spawned);
drop(observer);
let parents = CAPTURE.parents.lock().unwrap();
assert_eq!(parents.len(), 1);
let child = parents[0].0;
assert_eq!(parents[0].1, parent);
drop(parents);
assert_eq!(*POLL_CONTEXTS.lock().unwrap(), vec![child, child]);
#[cfg(feature = "logwise-forensic")]
{
let events = CAPTURE.events.lock().unwrap();
let names: Vec<_> = events.iter().map(|event| event.0).collect();
assert!(names.contains(&"some_executor.task.spawned"));
assert!(names.contains(&"some_executor.task.first_poll"));
assert!(names.contains(&"some_executor.task.woken"));
assert!(names.contains(&"some_executor.task.completed"));
assert!(names.contains(&"some_executor.task.dropped"));
assert!(events.iter().all(|event| event.1 == child));
}
#[cfg(feature = "logwise-performance")]
{
let events = CAPTURE.events.lock().unwrap();
let names: Vec<_> = events.iter().map(|event| event.0).collect();
assert!(names.contains(&"some_executor.task.wall_lifetime"));
assert!(names.contains(&"some_executor.task.active_poll_time"));
assert!(names.contains(&"some_executor.task.wake_latency"));
}
let cancelled = Task::without_notifications(
"cancelled".to_string(),
Configuration::default(),
std::future::pending::<()>(),
);
let (cancelled, cancellation_observer) = cancelled.spawn(&mut executor);
drop(cancellation_observer);
let mut cancelled = Box::pin(cancelled);
assert!(Future::poll(cancelled.as_mut(), &mut cx).is_ready());
drop(cancelled);
#[cfg(feature = "logwise-forensic")]
{
let events = CAPTURE.events.lock().unwrap();
assert!(
events
.iter()
.any(|event| event.0 == "some_executor.task.cancelled")
);
}
#[cfg(not(target_arch = "wasm32"))]
{
let panicking =
Task::without_notifications("panicking".to_string(), Configuration::default(), async {
panic!("expected panic")
});
let (panicking, panic_observer) = panicking.spawn(&mut executor);
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let mut panicking = Box::pin(panicking);
let mut panic_context = Context::from_waker(Waker::noop());
let _ = Future::poll(panicking.as_mut(), &mut panic_context);
}));
assert!(result.is_err());
assert_eq!(logwise::context::capture(), ContextToken::NONE);
drop(panic_observer);
}
}