mod periodic;
pub use periodic::{PeriodicTick, ThreadpoolPeriodicTimer};
use std::cell::Cell;
use std::io;
use std::ptr;
use std::sync::Mutex;
use std::sync::atomic::{AtomicIsize, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use windows_sys::Win32::Foundation::{FALSE, FILETIME, TRUE};
use windows_sys::Win32::System::Threading::{
CloseThreadpoolTimer, CreateThreadpoolTimer, IsThreadpoolTimerSet, PTP_CALLBACK_INSTANCE,
PTP_TIMER, SetThreadpoolTimer, WaitForThreadpoolTimerCallbacks,
};
use crate::callback_env::CallbackEnviron;
mod filetime {
pub const TICKS_PER_SECOND: u64 = 10_000_000;
pub const NANOS_PER_TICK: u32 = 100;
pub const SECONDS_1601_TO_1970: u64 = 11_644_473_600;
}
fn filetime_from_ticks(ticks: i64) -> FILETIME {
let bits = ticks as u64;
FILETIME {
dwLowDateTime: bits as u32,
dwHighDateTime: (bits >> 32) as u32,
}
}
pub(crate) fn relative_filetime(delay: Duration) -> FILETIME {
let ticks = delay
.as_secs()
.saturating_mul(filetime::TICKS_PER_SECOND)
.saturating_add(u64::from(delay.subsec_nanos() / filetime::NANOS_PER_TICK));
let ticks = i64::try_from(ticks).unwrap_or(i64::MAX);
filetime_from_ticks(-ticks)
}
pub(crate) fn absolute_filetime(when: SystemTime) -> FILETIME {
let since_unix = when.duration_since(UNIX_EPOCH).unwrap_or(Duration::ZERO);
let ticks = since_unix
.as_secs()
.saturating_add(filetime::SECONDS_1601_TO_1970)
.saturating_mul(filetime::TICKS_PER_SECOND)
.saturating_add(u64::from(
since_unix.subsec_nanos() / filetime::NANOS_PER_TICK,
));
filetime_from_ticks(i64::try_from(ticks).unwrap_or(i64::MAX))
}
pub(crate) fn millis_u32(duration: Duration) -> u32 {
u32::try_from(duration.as_millis()).unwrap_or(u32::MAX)
}
pub(crate) unsafe fn arm_raw(timer: PTP_TIMER, due: FILETIME, period_ms: u32, window_ms: u32) {
unsafe { SetThreadpoolTimer(timer, &due, period_ms, window_ms) };
}
pub(crate) unsafe fn disarm_raw(timer: PTP_TIMER) {
unsafe { SetThreadpoolTimer(timer, ptr::null(), 0, 0) };
}
pub(crate) struct TimerContext {
pub(crate) timer: AtomicIsize,
suppress_rearm: Mutex<u32>,
#[cfg(test)]
rearm_observer: Mutex<Option<std::sync::Arc<Mutex<Vec<bool>>>>>,
callback: Box<dyn Fn(&TimerFiring<'_>) + Send + Sync + 'static>,
}
impl TimerContext {
fn suppression(&self) -> std::sync::MutexGuard<'_, u32> {
self.suppress_rearm
.lock()
.unwrap_or_else(|poison| poison.into_inner())
}
fn suppress_and_disarm(&self) {
let mut suppressed = self.suppression();
*suppressed = suppressed.saturating_add(1);
let timer = self.timer.load(Ordering::Acquire);
if timer != 0 {
unsafe { disarm_raw(timer) };
}
}
fn release_suppression(&self) {
let mut suppressed = self.suppression();
*suppressed = suppressed.saturating_sub(1);
}
}
#[derive(Clone, Copy)]
enum PendingRearm {
After(Duration),
At(SystemTime),
}
pub struct TimerFiring<'ctx> {
ctx: &'ctx TimerContext,
pending: Cell<Option<PendingRearm>>,
}
impl std::fmt::Debug for TimerFiring<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("TimerFiring").finish_non_exhaustive()
}
}
impl TimerFiring<'_> {
pub fn rearm_after(&self, delay: Duration) {
self.pending.set(Some(PendingRearm::After(delay)));
}
pub fn rearm_at(&self, when: SystemTime) {
self.pending.set(Some(PendingRearm::At(when)));
}
fn apply_pending(&self) {
let applied = self.apply_pending_reporting();
#[cfg(test)]
if let Some(applied) = applied {
let observer = self
.ctx
.rearm_observer
.lock()
.unwrap_or_else(|poison| poison.into_inner())
.clone();
if let Some(observer) = observer {
observer
.lock()
.unwrap_or_else(|poison| poison.into_inner())
.push(applied);
}
}
let _ = applied;
}
fn apply_pending_reporting(&self) -> Option<bool> {
let pending = self.pending.get()?;
let suppressed = self.ctx.suppression();
if *suppressed > 0 {
return Some(false);
}
let timer = self.ctx.timer.load(Ordering::Acquire);
debug_assert_ne!(timer, 0, "the timer must be published before callbacks");
let due = match pending {
PendingRearm::After(delay) => relative_filetime(delay),
PendingRearm::At(when) => absolute_filetime(when),
};
unsafe { arm_raw(timer, due, 0, 0) };
drop(suppressed);
Some(true)
}
}
unsafe extern "system" fn timer_trampoline(
_instance: PTP_CALLBACK_INSTANCE,
context: *mut core::ffi::c_void,
_timer: PTP_TIMER,
) {
let ctx = unsafe { &*(context as *const TimerContext) };
let firing = TimerFiring {
ctx,
pending: Cell::new(None),
};
(ctx.callback)(&firing);
firing.apply_pending();
}
pub struct ThreadpoolTimer {
timer: PTP_TIMER,
context: *mut TimerContext,
}
unsafe impl Send for ThreadpoolTimer {}
unsafe impl Sync for ThreadpoolTimer {}
impl ThreadpoolTimer {
pub fn new<F>(callback: F, env: Option<&mut CallbackEnviron>) -> io::Result<Self>
where
F: Fn(&TimerFiring<'_>) + Send + Sync + 'static,
{
let context = Box::into_raw(Box::new(TimerContext {
timer: AtomicIsize::new(0),
suppress_rearm: Mutex::new(0),
#[cfg(test)]
rearm_observer: Mutex::new(None),
callback: Box::new(callback),
}));
let env_ptr = env.map_or(ptr::null_mut(), |e| e.as_mut_ptr());
let timer = unsafe {
CreateThreadpoolTimer(Some(timer_trampoline), context.cast(), env_ptr.cast_const())
};
if timer == 0 {
let error = io::Error::last_os_error();
unsafe { drop(Box::from_raw(context)) };
return Err(error);
}
unsafe { (*context).timer.store(timer, Ordering::Release) };
Ok(Self { timer, context })
}
pub fn set_after(&self, delay: Duration) {
unsafe { arm_raw(self.timer, relative_filetime(delay), 0, 0) };
}
pub fn set_at(&self, when: SystemTime) {
unsafe { arm_raw(self.timer, absolute_filetime(when), 0, 0) };
}
pub fn set_after_with_window(&self, delay: Duration, window: Duration) {
unsafe { arm_raw(self.timer, relative_filetime(delay), 0, millis_u32(window)) };
}
pub fn disarm(&self) {
unsafe { disarm_raw(self.timer) };
}
#[cfg(test)]
pub(crate) fn observe_rearms(&self, observer: &std::sync::Arc<Mutex<Vec<bool>>>) {
let ctx = unsafe { &*self.context };
*ctx.rearm_observer
.lock()
.unwrap_or_else(|poison| poison.into_inner()) = Some(std::sync::Arc::clone(observer));
}
#[must_use]
pub fn is_set(&self) -> bool {
unsafe { IsThreadpoolTimerSet(self.timer) != 0 }
}
pub fn wait(&self) {
unsafe { WaitForThreadpoolTimerCallbacks(self.timer, FALSE) };
}
pub fn cancel_pending(&self) {
unsafe { WaitForThreadpoolTimerCallbacks(self.timer, TRUE) };
}
pub fn stop_and_drain(&self) {
let ctx = unsafe { &*self.context };
ctx.suppress_and_disarm();
self.cancel_pending();
ctx.release_suppression();
}
pub(crate) fn into_parts(self) -> (PTP_TIMER, *mut core::ffi::c_void) {
let this = std::mem::ManuallyDrop::new(self);
(this.timer, this.context.cast())
}
pub(crate) unsafe fn drop_context(context: *mut core::ffi::c_void) {
drop(unsafe { Box::from_raw(context.cast::<TimerContext>()) });
}
pub(crate) unsafe fn prepare_shutdown(context: *mut core::ffi::c_void) {
let ctx = unsafe { &*context.cast::<TimerContext>() };
ctx.suppress_and_disarm();
}
}
impl Drop for ThreadpoolTimer {
fn drop(&mut self) {
let ctx = unsafe { &*self.context };
ctx.suppress_and_disarm();
self.cancel_pending();
unsafe {
CloseThreadpoolTimer(self.timer);
drop(Box::from_raw(self.context));
}
}
}
impl std::fmt::Debug for ThreadpoolTimer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ThreadpoolTimer")
.field("is_set", &self.is_set())
.finish_non_exhaustive()
}
}
#[cfg(test)]
mod tests;