#![doc = include_str!("../README.md")]
#![cfg_attr(docsrs, feature(doc_cfg))]
#![cfg_attr(coverage_nightly, feature(coverage_attribute))]
#![forbid(unsafe_code)]
pub mod sync;
mod primitives;
#[cfg(feature = "crossbeam")]
mod timers;
#[cfg(feature = "crossbeam")]
#[cfg_attr(docsrs, doc(cfg(feature = "crossbeam")))]
pub use crossbeam_channel;
#[cfg(any(test, feature = "test-clock"))]
mod paused;
#[cfg(feature = "test-clock")]
pub use paused::TestClock;
#[cfg(all(test, not(feature = "test-clock")))]
use paused::TestClock;
use std::fmt;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};
use primitives::{Condvar, Mutex, MutexGuard};
#[derive(Clone)]
pub struct Clock {
#[cfg(any(test, feature = "test-clock"))]
paused: Option<Arc<paused::Paused>>,
}
impl Clock {
pub const fn real() -> Self {
Self {
#[cfg(any(test, feature = "test-clock"))]
paused: None,
}
}
pub fn now(&self) -> Instant {
#[cfg(any(test, feature = "test-clock"))]
if let Some(paused) = &self.paused {
return paused.now();
}
Instant::now()
}
pub fn elapsed(&self, since: Instant) -> Duration {
self.now().saturating_duration_since(since)
}
pub fn system_time(&self) -> SystemTime {
#[cfg(any(test, feature = "test-clock"))]
if let Some(paused) = &self.paused {
return paused.system_time();
}
SystemTime::now()
}
pub fn sleep(&self, duration: Duration) {
#[cfg(any(test, feature = "test-clock"))]
if self.paused.is_some() {
self.sleep_until(self.now() + duration);
return;
}
std::thread::sleep(duration);
}
pub fn sleep_until(&self, deadline: Instant) {
self.waiter().wait_until(Some(deadline), || None::<()>);
}
fn identity(&self) -> Option<*const ()> {
#[cfg(any(test, feature = "test-clock"))]
if let Some(paused) = &self.paused {
return Some(Arc::as_ptr(paused).cast());
}
None
}
fn waiter(&self) -> Waiter {
let signal = Arc::new(Signal::default());
#[cfg(any(test, feature = "test-clock"))]
if let Some(paused) = &self.paused {
paused.register(&signal);
}
Waiter {
clock: self.clone(),
signal,
}
}
}
#[cfg(not(any(test, feature = "test-clock")))]
const _: () = assert!(size_of::<Clock>() == 0);
impl PartialEq for Clock {
fn eq(&self, other: &Self) -> bool {
self.identity() == other.identity()
}
}
impl Eq for Clock {}
impl fmt::Debug for Clock {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let mut clock = f.debug_struct("Clock");
#[cfg(any(test, feature = "test-clock"))]
if let Some(paused) = &self.paused {
return clock
.field("paused", &true)
.field("advanced", &paused.advanced())
.field("system_time", &paused.system_time())
.finish();
}
clock.field("paused", &false).finish()
}
}
struct Waiter {
clock: Clock,
signal: Arc<Signal>,
}
impl Waiter {
fn wait_until<T>(
&self,
deadline: Option<Instant>,
mut ready: impl FnMut() -> Option<T>,
) -> Option<T> {
let timer = self.timer(deadline);
loop {
let seen = self.signal.generation();
if let Some(value) = ready() {
return Some(value);
}
if deadline.is_some_and(|deadline| self.clock.now() >= deadline) {
return None;
}
drop(self.park(self.signal.lock(), seen, deadline, timer));
}
}
fn timer(&self, deadline: Option<Instant>) -> Option<Instant> {
#[cfg(any(test, feature = "test-clock"))]
if self.clock.paused.is_some() {
return None;
}
deadline
}
fn park<'a>(
&'a self,
mut state: MutexGuard<'a, SignalState>,
seen: u64,
deadline: Option<Instant>,
timer: Option<Instant>,
) -> MutexGuard<'a, SignalState> {
#[cfg(any(test, feature = "test-clock"))]
let mut blocked = None;
loop {
let now = self.clock.now();
if state.generation != seen || deadline.is_some_and(|deadline| now >= deadline) {
break;
}
#[cfg(test)]
if let Some(hook) = self
.clock
.paused
.as_ref()
.and_then(|paused| paused.take_hook(blocked.is_some()))
{
drop(state);
hook(timer);
state = self.signal.lock();
continue;
}
#[cfg(any(test, feature = "test-clock"))]
if blocked.is_none() {
blocked = self
.clock
.paused
.as_ref()
.map(|paused| paused.block(deadline));
}
#[cfg(test)]
{
state.parks += 1;
self.signal.parked.notify_all();
}
state = match timer {
None => self
.signal
.changed
.wait(state)
.expect("waiter signal not poisoned"),
Some(timer) => {
self.signal
.changed
.wait_timeout(state, timer - now)
.expect("waiter signal not poisoned")
.0
}
};
}
#[cfg(any(test, feature = "test-clock"))]
drop(blocked);
state
}
}
#[cfg(any(test, feature = "test-clock"))]
impl Drop for Waiter {
fn drop(&mut self) {
if let Some(paused) = &self.clock.paused {
paused.unregister(&self.signal);
}
}
}
impl fmt::Debug for Waiter {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Waiter")
.field("clock", &self.clock)
.finish_non_exhaustive()
}
}
#[derive(Default)]
struct Signal {
state: Mutex<SignalState>,
changed: Condvar,
#[cfg(test)]
parked: Condvar,
}
#[derive(Default)]
struct SignalState {
generation: u64,
notifications: u64,
#[cfg(test)]
parks: usize,
}
impl Signal {
fn lock(&self) -> MutexGuard<'_, SignalState> {
self.state.lock().expect("waiter signal not poisoned")
}
fn generation(&self) -> u64 {
self.lock().generation
}
fn notify_one(&self) {
let mut state = self.lock();
state.generation = state.generation.wrapping_add(1);
state.notifications = state.notifications.wrapping_add(1);
self.changed.notify_one();
}
fn notify_all(&self) {
let mut state = self.lock();
state.generation = state.generation.wrapping_add(1);
state.notifications = state.notifications.wrapping_add(1);
self.changed.notify_all();
}
#[cfg(any(test, feature = "test-clock"))]
fn advance(&self) {
let mut state = self.lock();
state.generation = state.generation.wrapping_add(1);
self.changed.notify_all();
}
}
#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests;