use core::ffi::c_void;
use std::io;
use std::marker::PhantomData;
use std::os::windows::io::BorrowedHandle;
use std::ptr;
use std::sync::Mutex;
use std::time::{Duration, SystemTime};
use windows_sys::Win32::Foundation::{FALSE, TRUE};
use windows_sys::Win32::System::Threading::{
CloseThreadpoolCleanupGroup, CloseThreadpoolCleanupGroupMembers, CreateThreadpoolCleanupGroup,
IsThreadpoolTimerSet, PTP_CLEANUP_GROUP, PTP_TIMER, PTP_WAIT, PTP_WORK, SubmitThreadpoolWork,
WaitForThreadpoolTimerCallbacks, WaitForThreadpoolWaitCallbacks,
WaitForThreadpoolWorkCallbacks,
};
use crate::callback_env::CallbackEnviron;
use crate::timer::{
PeriodicTick, ThreadpoolPeriodicTimer, ThreadpoolTimer, TimerFiring, absolute_filetime,
arm_raw, disarm_raw, millis_u32, relative_filetime,
};
use crate::wait::{ThreadpoolWait, WaitActivation, WaitTarget, WaitableHandle};
use crate::work::ThreadpoolWork;
struct OwnedResource {
ptr: *mut c_void,
prepare_shutdown: unsafe fn(*mut c_void),
recover_repair: unsafe fn(*mut c_void) -> Option<crate::heal::Registration>,
free: unsafe fn(*mut c_void),
}
unsafe fn no_repair_recovery(_ptr: *mut c_void) -> Option<crate::heal::Registration> {
None
}
unsafe impl Send for OwnedResource {}
unsafe fn free_boxed<T>(ptr: *mut c_void) {
drop(unsafe { Box::from_raw(ptr.cast::<T>()) });
}
fn prepare_shutdown_noop(_ptr: *mut c_void) {}
pub struct CleanupGroup {
group: PTP_CLEANUP_GROUP,
resources: Mutex<Vec<OwnedResource>>,
#[cfg(test)]
before_release: Mutex<Option<Box<dyn Fn() + Send + Sync>>>,
}
unsafe impl Send for CleanupGroup {}
unsafe impl Sync for CleanupGroup {}
impl CleanupGroup {
pub fn new() -> io::Result<Self> {
let group = crate::trace_call!("CreateThreadpoolCleanupGroup", 0, 0, {
unsafe { CreateThreadpoolCleanupGroup() }
});
if group == 0 {
return Err(io::Error::last_os_error());
}
Ok(Self {
group,
resources: Mutex::new(Vec::new()),
#[cfg(test)]
before_release: Mutex::new(None),
})
}
#[cfg(all(test, feature = "self-heal"))]
pub(crate) fn on_before_release(&self, hook: impl Fn() + Send + Sync + 'static) {
*self
.before_release
.lock()
.unwrap_or_else(|poison| poison.into_inner()) = Some(Box::new(hook));
}
fn member_environment(&self, env: Option<&CallbackEnviron<'_>>) -> CallbackEnviron<'_> {
let mut member_env = match env {
Some(env) => CallbackEnviron::from_inner(*env.as_inner()),
None => CallbackEnviron::new(),
};
unsafe { member_env.set_cleanup_group(self.group, None) };
member_env
}
fn adopt(&self, resource: OwnedResource) {
self.resources
.lock()
.unwrap_or_else(|poison| poison.into_inner())
.push(resource);
}
pub fn create_work<F>(
&self,
callback: F,
env: Option<&CallbackEnviron>,
) -> io::Result<WorkMember<'_>>
where
F: Fn() + Send + Sync + 'static,
{
let mut member_env = self.member_environment(env);
let work = ThreadpoolWork::new(callback, Some(&mut member_env))?;
let (handle, context) = work.into_parts();
self.adopt(OwnedResource {
ptr: context,
prepare_shutdown: prepare_shutdown_noop,
recover_repair: no_repair_recovery,
free: ThreadpoolWork::drop_context,
});
Ok(WorkMember {
handle,
_group: PhantomData,
})
}
pub fn create_timer<F>(
&self,
callback: F,
env: Option<&CallbackEnviron>,
) -> io::Result<TimerMember<'_>>
where
F: Fn(&TimerFiring<'_>) + Send + Sync + 'static,
{
let mut member_env = self.member_environment(env);
let timer = ThreadpoolTimer::new(callback, Some(&mut member_env))?;
let (handle, context) = timer.into_parts();
self.adopt(OwnedResource {
ptr: context,
prepare_shutdown: ThreadpoolTimer::prepare_shutdown,
recover_repair: no_repair_recovery,
free: ThreadpoolTimer::drop_context,
});
Ok(TimerMember {
handle,
context,
_group: PhantomData,
})
}
pub fn create_periodic_timer<F>(
&self,
period: Duration,
callback: F,
env: Option<&CallbackEnviron>,
) -> io::Result<PeriodicTimerMember<'_>>
where
F: Fn(&PeriodicTick<'_>) + Send + Sync + 'static,
{
let mut member_env = self.member_environment(env);
let timer = ThreadpoolPeriodicTimer::new(period, callback, Some(&mut member_env))?;
let (handle, context, period) = timer.into_parts();
self.adopt(OwnedResource {
ptr: context,
prepare_shutdown: prepare_shutdown_noop,
recover_repair: no_repair_recovery,
free: ThreadpoolPeriodicTimer::drop_context,
});
Ok(PeriodicTimerMember {
handle,
period,
context,
_group: PhantomData,
})
}
pub fn create_wait<F>(
&self,
handle: WaitableHandle,
callback: F,
env: Option<&CallbackEnviron<'_>>,
) -> io::Result<WaitMember<'_>>
where
F: Fn(&WaitActivation<'_>) + Send + Sync + 'static,
{
let mut member_env = self.member_environment(env);
let wait = ThreadpoolWait::new(handle, callback, Some(&mut member_env))?;
let (raw, context, target) = wait.into_parts();
self.adopt(OwnedResource {
ptr: context,
prepare_shutdown: ThreadpoolWait::prepare_shutdown,
recover_repair: ThreadpoolWait::recover_repair,
free: ThreadpoolWait::drop_context,
});
let target = Box::into_raw(Box::new(target));
self.adopt(OwnedResource {
ptr: target.cast(),
prepare_shutdown: prepare_shutdown_noop,
recover_repair: no_repair_recovery,
free: free_boxed::<WaitTarget>,
});
Ok(WaitMember {
handle: raw,
watched: target,
context,
_group: PhantomData,
})
}
pub fn close_members(&mut self) {
self.release_members(false, false);
}
#[cfg(feature = "self-heal")]
pub fn close_members_cancelling(&mut self) {
self.release_members(true, true);
}
pub unsafe fn close_members_cancelling_no_heal_tracking(&mut self) {
self.release_members(true, false);
}
#[must_use]
pub fn owned_resources(&self) -> usize {
self.resources
.lock()
.unwrap_or_else(|poison| poison.into_inner())
.len()
}
fn release_members(&mut self, cancel_pending: bool, track_repairs: bool) {
let recovered: Vec<crate::heal::Registration> = {
let resources = self
.resources
.lock()
.unwrap_or_else(|poison| poison.into_inner());
for resource in resources.iter() {
unsafe { (resource.prepare_shutdown)(resource.ptr) };
}
if cancel_pending && track_repairs {
resources
.iter()
.filter_map(|resource| unsafe { (resource.recover_repair)(resource.ptr) })
.collect()
} else {
Vec::new()
}
};
#[cfg(test)]
{
let hook = self
.before_release
.lock()
.unwrap_or_else(|poison| poison.into_inner());
if let Some(hook) = hook.as_ref() {
hook();
}
}
crate::trace_call!(
"CloseThreadpoolCleanupGroupMembers",
self.group,
u32::from(cancel_pending),
{
unsafe {
CloseThreadpoolCleanupGroupMembers(
self.group,
if cancel_pending { TRUE } else { FALSE },
ptr::null_mut(),
);
}
}
);
let resources = std::mem::take(
&mut *self
.resources
.lock()
.unwrap_or_else(|poison| poison.into_inner()),
);
let mut tracked = true;
if cancel_pending && track_repairs {
for claim in recovered.iter() {
tracked &= claim.owe_repair_claimed();
}
}
drop(recovered);
for resource in resources {
unsafe { (resource.free)(resource.ptr) };
}
if track_repairs {
crate::obligation::fail_fast_if_untracked(tracked, "CleanupGroup");
}
}
}
impl Drop for CleanupGroup {
fn drop(&mut self) {
self.release_members(false, false);
crate::trace_call!("CloseThreadpoolCleanupGroup", self.group, 0, {
unsafe { CloseThreadpoolCleanupGroup(self.group) };
});
}
}
impl std::fmt::Debug for CleanupGroup {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CleanupGroup")
.field("owned_resources", &self.owned_resources())
.finish_non_exhaustive()
}
}
#[derive(Debug)]
pub struct WorkMember<'group> {
handle: PTP_WORK,
_group: PhantomData<&'group CleanupGroup>,
}
impl WorkMember<'_> {
pub fn submit(&self) {
crate::trace_call!("SubmitThreadpoolWork", self.handle, 0, {
unsafe { SubmitThreadpoolWork(self.handle) };
});
}
pub fn wait(&self) {
crate::trace_call!("WaitForThreadpoolWorkCallbacks", self.handle, 0, {
unsafe { WaitForThreadpoolWorkCallbacks(self.handle, FALSE) };
});
}
pub fn stop_and_drain(&self) {
self.wait();
}
pub fn cancel_pending(&self) {
crate::trace_call!("WaitForThreadpoolWorkCallbacks(cancel)", self.handle, 1, {
unsafe { WaitForThreadpoolWorkCallbacks(self.handle, TRUE) };
});
}
}
#[derive(Debug)]
pub struct TimerMember<'group> {
handle: PTP_TIMER,
context: *mut c_void,
_group: PhantomData<&'group CleanupGroup>,
}
unsafe impl Send for TimerMember<'_> {}
unsafe impl Sync for TimerMember<'_> {}
impl TimerMember<'_> {
pub fn set_after(&self, delay: Duration) {
unsafe { arm_raw(self.handle, relative_filetime(delay), 0, 0) };
}
pub fn set_at(&self, when: SystemTime) {
unsafe { arm_raw(self.handle, absolute_filetime(when), 0, 0) };
}
pub fn disarm(&self) {
unsafe { disarm_raw(self.handle) };
}
#[must_use]
pub fn is_set(&self) -> bool {
crate::trace_call!("IsThreadpoolTimerSet", self.handle, 0, {
unsafe { IsThreadpoolTimerSet(self.handle) != 0 }
})
}
pub fn wait(&self) {
crate::trace_call!("WaitForThreadpoolTimerCallbacks", self.handle, 0, {
unsafe { WaitForThreadpoolTimerCallbacks(self.handle, FALSE) };
});
}
pub fn stop_and_drain(&self) {
unsafe { ThreadpoolTimer::stop_and_drain_parts(self.context, self.handle) };
}
pub fn cancel_pending(&self) {
crate::trace_call!("WaitForThreadpoolTimerCallbacks(cancel)", self.handle, 1, {
unsafe { WaitForThreadpoolTimerCallbacks(self.handle, TRUE) };
});
}
}
#[derive(Debug)]
pub struct PeriodicTimerMember<'group> {
handle: PTP_TIMER,
period: Duration,
context: *mut c_void,
_group: PhantomData<&'group CleanupGroup>,
}
unsafe impl Send for PeriodicTimerMember<'_> {}
unsafe impl Sync for PeriodicTimerMember<'_> {}
impl PeriodicTimerMember<'_> {
#[must_use]
pub fn period(&self) -> Duration {
self.period
}
pub fn start(&self) {
self.start_after(self.period);
}
pub fn start_after(&self, first_delay: Duration) {
unsafe {
arm_raw(
self.handle,
relative_filetime(first_delay),
millis_u32(self.period),
0,
);
}
}
pub fn stop(&self) {
unsafe { disarm_raw(self.handle) };
}
#[must_use]
pub fn is_running(&self) -> bool {
crate::trace_call!("IsThreadpoolTimerSet", self.handle, 0, {
unsafe { IsThreadpoolTimerSet(self.handle) != 0 }
})
}
pub fn wait(&self) {
crate::trace_call!("WaitForThreadpoolTimerCallbacks", self.handle, 0, {
unsafe { WaitForThreadpoolTimerCallbacks(self.handle, FALSE) };
});
}
pub fn stop_and_drain(&self) {
unsafe { ThreadpoolPeriodicTimer::stop_and_drain_parts(self.context, self.handle) };
}
}
#[derive(Debug)]
pub struct WaitMember<'group> {
handle: PTP_WAIT,
watched: *mut WaitTarget,
context: *mut c_void,
_group: PhantomData<&'group CleanupGroup>,
}
unsafe impl Send for WaitMember<'_> {}
unsafe impl Sync for WaitMember<'_> {}
const _: () = {
const fn needs_send<T: Send>() {}
const fn needs_sync<T: Sync>() {}
needs_send::<TimerMember<'_>>();
needs_sync::<TimerMember<'_>>();
needs_send::<WorkMember<'_>>();
needs_sync::<WorkMember<'_>>();
needs_send::<PeriodicTimerMember<'_>>();
needs_sync::<PeriodicTimerMember<'_>>();
needs_send::<WaitMember<'_>>();
needs_sync::<WaitMember<'_>>();
};
impl WaitMember<'_> {
#[must_use]
pub fn handle(&self) -> BorrowedHandle<'_> {
unsafe { (*self.watched).borrow() }
}
pub fn arm(&self, timeout: Option<Duration>) {
unsafe { crate::wait::arm_member(self.handle, &*self.watched, timeout) };
}
pub fn disarm(&self) {
unsafe { crate::wait::disarm_raw(self.handle) };
}
pub fn wait(&self) {
crate::trace_call!("WaitForThreadpoolWaitCallbacks", self.handle, 0, {
unsafe { WaitForThreadpoolWaitCallbacks(self.handle, FALSE) };
});
}
pub fn stop_and_drain(&self) {
unsafe { ThreadpoolWait::stop_and_drain_parts(self.context, self.handle) };
}
#[cfg(feature = "self-heal")]
pub fn try_cancel_pending(&self) {
unsafe { self.try_cancel_pending_no_heal_tracking() };
let tracked = unsafe { ThreadpoolWait::owe_repair(self.context) };
crate::obligation::fail_fast_if_untracked(tracked, "WaitMember");
}
pub unsafe fn try_cancel_pending_no_heal_tracking(&self) {
crate::trace_call!("WaitForThreadpoolWaitCallbacks(cancel)", self.handle, 1, {
unsafe { WaitForThreadpoolWaitCallbacks(self.handle, TRUE) };
});
}
}
#[cfg(test)]
mod tests;