#[cfg(any(feature = "trace", test))]
#[cfg_attr(not(feature = "trace"), allow(dead_code))]
#[derive(Clone, Copy)]
struct Record {
at: std::time::Duration,
thread: u32,
target: &'static str,
event: &'static str,
a: u64,
b: u64,
}
#[cfg(any(feature = "trace", test))]
struct Buffer {
records: Vec<Record>,
head: usize,
dropped: usize,
announced: bool,
}
#[cfg(any(feature = "trace", test))]
impl Buffer {
fn with_capacity(capacity: usize) -> Self {
Self {
records: Vec::with_capacity(capacity),
head: 0,
dropped: 0,
announced: false,
}
}
fn push(&mut self, record: Record, capacity: usize) -> bool {
if capacity == 0 {
self.dropped += 1;
let first = !self.announced;
self.announced = true;
return first;
}
if self.records.len() < capacity {
self.records.push(record);
return false;
}
let size = self.records.len();
self.records[self.head] = record;
self.head = (self.head + 1) % size;
self.dropped += 1;
let first = !self.announced;
self.announced = true;
first
}
fn iter(&self) -> impl Iterator<Item = &Record> {
let (wrapped, oldest_first) = self.records.split_at(self.head);
oldest_first.iter().chain(wrapped.iter())
}
fn clear(&mut self) {
self.records.clear();
self.head = 0;
self.dropped = 0;
self.announced = false;
}
}
#[cfg(feature = "trace")]
#[used]
#[unsafe(link_section = ".CRT$XCU")]
static ARM_BEFORE_MAIN: extern "C" fn() = {
extern "C" fn arm() {
ARMED_BEFORE_MAIN.store(true, std::sync::atomic::Ordering::Relaxed);
let _ = imp::enabled();
}
arm
};
#[cfg(feature = "trace")]
static ARMED_BEFORE_MAIN: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
#[cfg(all(test, feature = "trace"))]
pub(crate) fn armed_before_main() -> bool {
ARMED_BEFORE_MAIN.load(std::sync::atomic::Ordering::Relaxed)
}
#[cfg(all(test, feature = "trace"))]
pub(crate) fn in_a_trace_armed_child(name: &str, filter: &str, body: impl FnOnce()) {
const CHILD_VAR: &str = "WTPS_TRACE_ARMED_CHILD";
const BODY_RAN: i32 = 7;
if std::env::var(CHILD_VAR).as_deref() == Ok(name) {
body();
std::process::exit(BODY_RAN);
}
let exe = std::env::current_exe().expect("locate the test binary");
let status = std::process::Command::new(exe)
.env(CHILD_VAR, name)
.env("WINDOWS_THREADPOOL_TRACE", filter)
.args(["--exact", name, "--test-threads", "1"])
.status()
.expect("run the trace-armed child");
assert_eq!(
status.code(),
Some(BODY_RAN),
"the trace-armed child did not reach the end of `{name}` (exit {:?}). Exit 0 means \
`--exact {name}` matched nothing -- the test was renamed and the string passed here \
was not -- and any other code means the body's own assertions failed; its output is \
above",
status.code()
);
}
#[cfg(feature = "trace")]
mod imp {
use std::sync::Mutex;
use std::sync::atomic::{AtomicU8, Ordering};
use std::time::Instant;
use super::{Buffer, Record};
const CAPACITY: usize = 262_144;
static ARMED: AtomicU8 = AtomicU8::new(ARMED_UNKNOWN);
const ARMED_UNKNOWN: u8 = 0;
const ARMED_OFF: u8 = 1;
const ARMED_ON: u8 = 2;
static STATE: std::sync::OnceLock<Mutex<Buffer>> = std::sync::OnceLock::new();
static STARTED: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
static FILTERS: std::sync::OnceLock<Vec<String>> = std::sync::OnceLock::new();
fn state() -> &'static Mutex<Buffer> {
STATE.get_or_init(|| Mutex::new(Buffer::with_capacity(CAPACITY)))
}
fn started() -> Instant {
*STARTED.get_or_init(Instant::now)
}
fn filters() -> &'static Vec<String> {
FILTERS.get_or_init(|| {
std::env::var("WINDOWS_THREADPOOL_TRACE")
.unwrap_or_default()
.split(',')
.map(|piece| piece.trim().to_owned())
.filter(|piece| !piece.is_empty())
.collect()
})
}
#[inline(always)]
pub fn enabled() -> bool {
match ARMED.load(Ordering::Relaxed) {
ARMED_ON => true,
ARMED_OFF => false,
_ => {
let on = !filters().is_empty();
ARMED.store(if on { ARMED_ON } else { ARMED_OFF }, Ordering::Relaxed);
if on {
observe_exceptions();
}
on
}
}
}
#[inline(always)]
pub fn wants(target: &str) -> bool {
enabled()
&& filters()
.iter()
.any(|filter| filter == "*" || target.contains(filter.as_str()))
}
#[inline]
pub fn record(target: &'static str, event: &'static str, a: u64, b: u64) {
if !wants(target) {
return;
}
let at = started().elapsed();
let thread = unsafe { windows_sys::Win32::System::Threading::GetCurrentThreadId() };
let first_eviction = {
let Ok(mut state) = state().lock() else {
return;
};
state.push(
Record {
at,
thread,
target,
event,
a,
b,
},
CAPACITY,
)
};
if first_eviction {
eprintln!(
"windows-threadpool-sys trace: the {CAPACITY}-record buffer is full and is now \
evicting the oldest records, so this capture no longer reaches back to the \
start of the run. Narrow WINDOWS_THREADPOOL_TRACE, or raise CAPACITY."
);
}
}
pub fn dump() -> String {
let Ok(state) = state().lock() else {
return "<trace lock poisoned>".to_owned();
};
let mut out = String::with_capacity(state.records.len() * 64);
if state.dropped > 0 {
out.push_str(&format!(
" ... {} earlier record(s) dropped; raise CAPACITY to keep them\n",
state.dropped
));
}
for record in state.iter() {
out.push_str(&format!(
" {:>12.6}s t{:<6} {:<22} {:<36} {:>6} {:>6}\n",
record.at.as_secs_f64(),
record.thread,
record.target,
record.event,
record.a,
record.b
));
}
out
}
pub fn clear() {
if let Ok(mut state) = state().lock() {
state.clear();
}
}
const EXCEPTION_TARGET: &str = "exception";
const EXCEPTION_CONTINUE_SEARCH: i32 = 0;
const CALL_FIRST: u32 = 1;
fn record_without_blocking(event: &'static str, a: u64, b: u64) {
if ARMED.load(Ordering::Relaxed) != ARMED_ON {
return;
}
let Some(filters) = FILTERS.get() else {
return;
};
if !filters
.iter()
.any(|filter| filter == "*" || EXCEPTION_TARGET.contains(filter.as_str()))
{
return;
}
let (Some(started), Some(state)) = (STARTED.get(), STATE.get()) else {
return;
};
let at = started.elapsed();
let thread = unsafe { windows_sys::Win32::System::Threading::GetCurrentThreadId() };
let Ok(mut state) = state.try_lock() else {
return;
};
let _ = state.push(
Record {
at,
thread,
target: EXCEPTION_TARGET,
event,
a,
b,
},
CAPACITY,
);
}
unsafe extern "system" fn exception_observer(
info: *mut windows_sys::Win32::System::Diagnostics::Debug::EXCEPTION_POINTERS,
) -> i32 {
if !info.is_null() {
let record = unsafe { (*info).ExceptionRecord };
if !record.is_null() {
let (code, address) =
unsafe { ((*record).ExceptionCode, (*record).ExceptionAddress) };
record_without_blocking("raised", code as u32 as u64, address as u64);
}
}
EXCEPTION_CONTINUE_SEARCH
}
pub fn observe_exceptions() {
use std::sync::atomic::AtomicBool;
static INSTALLED: AtomicBool = AtomicBool::new(false);
if INSTALLED.swap(true, Ordering::SeqCst) {
return;
}
let _ = started();
let _ = state();
unsafe {
windows_sys::Win32::System::Diagnostics::Debug::AddVectoredExceptionHandler(
CALL_FIRST,
Some(exception_observer),
)
};
}
}
#[cfg(not(feature = "trace"))]
mod imp {
#[inline(always)]
pub fn enabled() -> bool {
false
}
#[inline(always)]
pub fn wants(_target: &str) -> bool {
false
}
#[inline(always)]
pub fn record(_target: &'static str, _event: &'static str, _a: u64, _b: u64) {}
pub fn dump() -> String {
"<built without the `trace` feature>".to_owned()
}
pub fn clear() {}
pub fn observe_exceptions() {}
}
pub use imp::{clear, dump, enabled, observe_exceptions, record, wants};
#[cfg(all(test, feature = "trace"))]
pub(crate) fn counted(target: &str, event: &str) -> usize {
dump()
.lines()
.filter(|line| {
let mut fields = line.split_whitespace();
fields.next();
fields.next();
fields.next() == Some(target) && fields.next() == Some(event)
})
.count()
}
#[macro_export]
macro_rules! trace_record {
($target:expr, $event:expr) => {
$crate::trace::record($target, $event, 0, 0)
};
($target:expr, $event:expr, $a:expr) => {
if $crate::trace::wants($target) {
$crate::trace::record($target, $event, $a as u64, 0);
}
};
($target:expr, $event:expr, $a:expr, $b:expr) => {
if $crate::trace::wants($target) {
$crate::trace::record($target, $event, $a as u64, $b as u64);
}
};
}
#[macro_export]
macro_rules! trace_call {
($name:literal, $a:expr, $b:expr, $call:block) => {{
$crate::trace_record!("syscall-enter", $name, $a, $b);
let result = $call;
$crate::trace_record!("syscall-leave", $name, $a, $b);
result
}};
}
#[cfg(test)]
mod tests;