use std::sync::atomic::{
AtomicBool, AtomicU32, AtomicU64,
Ordering::{Relaxed, SeqCst},
};
use std::time::{Duration, Instant};
use crate::{Stop, StopReason};
const DEFAULT_TARGET_NANOS: u64 = 100_000;
const MAX_CHECKS_PER_CLOCK_READ: u32 = 64;
#[inline]
fn duration_to_nanos(d: Duration) -> u64 {
d.as_nanos().min(u64::MAX as u128) as u64
}
pub struct DebouncedTimeout<T> {
inner: T,
created: Instant,
deadline_nanos: u64,
target_nanos: u64,
schedule: Schedule,
}
struct Schedule {
countdown: AtomicU32,
skip_mod: AtomicU32,
last_measured_nanos: AtomicU64,
timed_out: AtomicBool,
calibrated: AtomicBool,
}
const NO_BASELINE: u64 = u64::MAX;
impl Schedule {
fn new() -> Self {
Self {
countdown: AtomicU32::new(1),
skip_mod: AtomicU32::new(1),
last_measured_nanos: AtomicU64::new(NO_BASELINE),
timed_out: AtomicBool::new(false),
calibrated: AtomicBool::new(false),
}
}
#[inline(always)]
fn due(&self) -> bool {
!matches!(
self.countdown.fetch_sub(1, Relaxed),
2..=MAX_CHECKS_PER_CLOCK_READ
)
}
}
impl<T: Stop> DebouncedTimeout<T> {
#[inline]
pub fn new(inner: T, duration: Duration) -> Self {
let now = Instant::now();
Self {
inner,
created: now,
deadline_nanos: duration_to_nanos(duration),
target_nanos: DEFAULT_TARGET_NANOS,
schedule: Schedule::new(),
}
}
#[inline]
pub fn with_deadline(inner: T, deadline: Instant) -> Self {
let now = Instant::now();
Self {
inner,
created: now,
deadline_nanos: duration_to_nanos(deadline.saturating_duration_since(now)),
target_nanos: DEFAULT_TARGET_NANOS,
schedule: Schedule::new(),
}
}
#[inline]
pub fn with_target_interval(mut self, interval: Duration) -> Self {
self.target_nanos = duration_to_nanos(interval).max(1);
self
}
#[inline]
pub fn deadline(&self) -> Instant {
self.created + Duration::from_nanos(self.deadline_nanos)
}
#[inline]
pub fn remaining(&self) -> Duration {
self.deadline().saturating_duration_since(Instant::now())
}
#[inline]
pub fn inner(&self) -> &T {
&self.inner
}
#[inline]
pub fn into_inner(self) -> T {
self.inner
}
#[inline]
pub fn checks_per_clock_read(&self) -> u32 {
self.schedule.skip_mod.load(Relaxed)
}
pub fn clear_calibration(&self) {
let schedule = &self.schedule;
schedule.skip_mod.store(1, Relaxed);
schedule.calibrated.store(false, Relaxed);
schedule.last_measured_nanos.store(NO_BASELINE, Relaxed);
schedule.countdown.store(1, SeqCst);
}
#[cold]
#[inline(never)]
fn measure_and_recalibrate(&self) -> bool {
let schedule = &self.schedule;
let elapsed_nanos = self.created.elapsed().as_nanos() as u64;
if elapsed_nanos >= self.deadline_nanos {
schedule.timed_out.store(true, SeqCst);
schedule.countdown.store(1, SeqCst);
return true; }
let prev_nanos = schedule.last_measured_nanos.swap(elapsed_nanos, Relaxed);
let current_skip = schedule.skip_mod.load(Relaxed);
let delta_nanos = elapsed_nanos.saturating_sub(prev_nanos);
let nanos_per_call = delta_nanos / u64::from(current_skip);
let mut skip = current_skip;
if prev_nanos == NO_BASELINE {
} else if !schedule.calibrated.swap(true, Relaxed) {
let ideal = self
.target_nanos
.checked_div(nanos_per_call)
.unwrap_or(u64::MAX);
skip = ideal.clamp(1, u64::from(MAX_CHECKS_PER_CLOCK_READ)) as u32;
schedule.skip_mod.store(skip, Relaxed);
} else if let Some(ideal) = self.target_nanos.checked_div(nanos_per_call) {
let ideal_skip = ideal.clamp(1, u64::from(MAX_CHECKS_PER_CLOCK_READ)) as u32;
skip = if ideal_skip <= current_skip {
ideal_skip
} else {
current_skip + (ideal_skip - current_skip).div_ceil(8)
};
schedule.skip_mod.store(skip, Relaxed);
}
schedule.countdown.store(skip, SeqCst);
if schedule.timed_out.load(SeqCst) {
schedule.countdown.store(1, Relaxed);
}
false }
}
impl<T: Stop> Stop for DebouncedTimeout<T> {
#[inline]
#[track_caller]
fn check(&self) -> Result<(), StopReason> {
self.inner.check()?;
if !self.schedule.due() {
return Ok(());
}
if self.measure_and_recalibrate() {
Err(StopReason::TimedOut)
} else {
Ok(())
}
}
#[inline]
#[track_caller]
fn should_stop(&self) -> bool {
if self.inner.should_stop() {
return true;
}
if !self.schedule.due() {
return false;
}
self.measure_and_recalibrate()
}
}
impl<T: Stop> DebouncedTimeout<T> {
#[inline]
pub fn tighten(self, duration: Duration) -> Self {
let elapsed = duration_to_nanos(Instant::now().saturating_duration_since(self.created));
let new_deadline_nanos = elapsed.saturating_add(duration_to_nanos(duration));
let deadline_nanos = self.deadline_nanos.min(new_deadline_nanos);
Self {
inner: self.inner,
created: self.created,
deadline_nanos,
target_nanos: self.target_nanos,
schedule: Schedule::new(),
}
}
#[inline]
pub fn tighten_deadline(self, deadline: Instant) -> Self {
let new_deadline_nanos =
duration_to_nanos(deadline.saturating_duration_since(self.created));
let deadline_nanos = self.deadline_nanos.min(new_deadline_nanos);
Self {
inner: self.inner,
created: self.created,
deadline_nanos,
target_nanos: self.target_nanos,
schedule: Schedule::new(),
}
}
}
impl<T: Clone + Stop> Clone for DebouncedTimeout<T> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
created: self.created,
deadline_nanos: self.deadline_nanos,
target_nanos: self.target_nanos,
schedule: Schedule::new(),
}
}
}
impl<T: core::fmt::Debug> core::fmt::Debug for DebouncedTimeout<T> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
let deadline = self.created + Duration::from_nanos(self.deadline_nanos);
f.debug_struct("DebouncedTimeout")
.field("inner", &self.inner)
.field("deadline", &deadline)
.field("target_interval_us", &(self.target_nanos / 1_000))
.field("skip_mod", &self.schedule.skip_mod.load(Relaxed))
.finish()
}
}
pub trait DebouncedTimeoutExt: Stop + Sized {
#[inline]
fn with_debounced_timeout(self, duration: Duration) -> DebouncedTimeout<Self> {
DebouncedTimeout::new(self, duration)
}
#[inline]
fn with_debounced_deadline(self, deadline: Instant) -> DebouncedTimeout<Self> {
DebouncedTimeout::with_deadline(self, deadline)
}
}
impl<T: Stop> DebouncedTimeoutExt for T {}
#[cfg(test)]
mod tests {
use super::*;
use crate::{StopSource, Unstoppable};
#[test]
fn basic_timeout() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_millis(50));
assert!(!stop.should_stop());
assert!(stop.check().is_ok());
std::thread::sleep(Duration::from_millis(80));
for _ in 0..100 {
if stop.should_stop() {
return; }
}
panic!("should have detected timeout");
}
#[test]
fn cancel_before_timeout() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60));
source.cancel();
assert!(stop.should_stop());
assert_eq!(stop.check(), Err(StopReason::Cancelled));
}
#[test]
fn calibration_ramps_up() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60));
assert_eq!(stop.checks_per_clock_read(), 1);
for _ in 0..10_000 {
let _ = stop.check();
}
assert!(
stop.checks_per_clock_read() > 1,
"skip_mod should have increased, got {}",
stop.checks_per_clock_read()
);
}
#[test]
fn remaining_accuracy() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
let remaining = stop.remaining();
assert!(remaining > Duration::from_secs(9));
assert!(remaining <= Duration::from_secs(10));
}
#[test]
fn tighten_works() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60))
.tighten(Duration::from_secs(1));
let remaining = stop.remaining();
assert!(remaining < Duration::from_secs(2));
}
#[test]
fn clone_resets_calibration() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60));
for _ in 0..10_000 {
let _ = stop.check();
}
assert!(stop.checks_per_clock_read() > 1);
let cloned = stop.clone();
assert_eq!(cloned.checks_per_clock_read(), 1);
}
#[test]
fn extension_trait() {
use super::DebouncedTimeoutExt;
let source = StopSource::new();
let stop = source
.as_ref()
.with_debounced_timeout(Duration::from_secs(10));
assert!(!stop.should_stop());
}
#[test]
fn is_send_sync() {
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<DebouncedTimeout<crate::StopRef<'_>>>();
}
#[test]
fn zero_duration_immediate_timeout() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::ZERO);
assert_eq!(stop.check(), Err(StopReason::TimedOut));
}
#[test]
fn deadline_in_the_past() {
let source = StopSource::new();
let past = Instant::now() - Duration::from_secs(1);
let stop = DebouncedTimeout::with_deadline(source.as_ref(), past);
assert_eq!(stop.check(), Err(StopReason::TimedOut));
}
#[test]
fn with_deadline_basic() {
let source = StopSource::new();
let deadline = Instant::now() + Duration::from_millis(100);
let stop = DebouncedTimeout::with_deadline(source.as_ref(), deadline);
assert!(!stop.should_stop());
std::thread::sleep(Duration::from_millis(150));
for _ in 0..100 {
if stop.should_stop() {
return;
}
}
panic!("should have detected timeout");
}
#[test]
fn deadline_accessor() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
let deadline = stop.deadline();
let remaining = stop.remaining();
assert!(remaining > Duration::from_secs(9));
assert!(remaining <= Duration::from_secs(10));
assert!(deadline > Instant::now() + Duration::from_secs(9));
}
#[test]
fn inner_access() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
assert!(!stop.inner().should_stop());
source.cancel();
assert!(stop.inner().should_stop());
}
#[test]
fn into_inner_works() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
let inner = stop.into_inner();
assert!(!inner.should_stop());
}
#[test]
fn tighten_deadline_works() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60))
.tighten_deadline(Instant::now() + Duration::from_secs(1));
let remaining = stop.remaining();
assert!(remaining < Duration::from_secs(2));
}
#[test]
fn tighten_does_not_loosen() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(1))
.tighten(Duration::from_secs(60));
let remaining = stop.remaining();
assert!(remaining < Duration::from_secs(2));
}
#[test]
fn debug_format() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
let debug = format!("{stop:?}");
assert!(debug.contains("DebouncedTimeout"));
assert!(debug.contains("skip_mod"));
assert!(debug.contains("target_interval_us"));
}
#[test]
fn with_target_interval_zero_clamps_to_one() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60))
.with_target_interval(Duration::ZERO);
assert!(stop.check().is_ok());
}
#[test]
fn with_debounced_deadline_ext() {
let source = StopSource::new();
let deadline = Instant::now() + Duration::from_secs(10);
let stop = source.as_ref().with_debounced_deadline(deadline);
assert!(!stop.should_stop());
assert!(stop.remaining() > Duration::from_secs(9));
}
#[test]
fn check_and_should_stop_agree() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60));
for _ in 0..1000 {
assert!(!stop.should_stop());
assert!(stop.check().is_ok());
}
source.cancel();
assert!(stop.should_stop());
assert_eq!(stop.check(), Err(StopReason::Cancelled));
}
#[test]
fn remaining_after_expiry_is_zero() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_millis(1));
std::thread::sleep(Duration::from_millis(10));
assert_eq!(stop.remaining(), Duration::ZERO);
}
#[test]
fn adapts_to_slowdown() {
let source = StopSource::new();
let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60))
.with_target_interval(Duration::from_micros(10));
for _ in 0..50_000 {
let _ = stop.check();
}
let fast_skip = stop.checks_per_clock_read();
assert!(fast_skip > 1, "should have ramped up, got {fast_skip}");
for _ in 0..(fast_skip as usize + 100) {
std::thread::sleep(Duration::from_micros(50));
let _ = stop.check();
}
let slow_skip = stop.checks_per_clock_read();
assert!(
slow_skip < fast_skip,
"should have reduced skip_mod from {fast_skip} to less, got {slow_skip}"
);
}
fn work_for(duration: Duration) {
let start = Instant::now();
while start.elapsed() < duration {
std::hint::spin_loop();
}
}
#[test]
fn reads_the_clock_at_least_every_64_checks() {
let stop = DebouncedTimeout::new(Unstoppable, Duration::from_secs(60));
for _ in 0..1_000_000 {
stop.check().unwrap();
assert!(stop.checks_per_clock_read() <= MAX_CHECKS_PER_CLOCK_READ);
}
}
fn calibrated_on_fast_checks() -> DebouncedTimeout<Unstoppable> {
let stop = DebouncedTimeout::new(Unstoppable, Duration::from_secs(3600));
let fast_until = Instant::now() + Duration::from_millis(20);
while Instant::now() < fast_until {
stop.check().unwrap();
}
stop
}
#[test]
fn a_passed_deadline_stops_within_64_checks() {
let mut stop = calibrated_on_fast_checks();
stop.deadline_nanos = 0;
let mut ok_checks = 0;
while stop.check().is_ok() {
ok_checks += 1;
assert!(ok_checks < MAX_CHECKS_PER_CLOCK_READ, "{ok_checks} checks");
}
}
#[test]
fn a_slowdown_after_calibration_stops_within_64_slow_checks() {
let mut stop = calibrated_on_fast_checks();
stop.deadline_nanos = duration_to_nanos(stop.created.elapsed()) + 20_000_000;
let mut slow_checks = 0;
while stop.check().is_ok() {
slow_checks += 1;
assert!(slow_checks < 1_000, "still running a second later");
work_for(Duration::from_millis(1));
}
assert!(slow_checks <= 20 + 64 + 8, "{slow_checks} slow checks");
}
#[test]
fn a_clock_read_just_before_another_threads_timeout_keeps_it_stopped() {
let stop = calibrated_on_fast_checks();
stop.schedule.timed_out.store(true, SeqCst);
stop.schedule.countdown.store(1, SeqCst);
assert!(!stop.measure_and_recalibrate());
assert_eq!(stop.schedule.countdown.load(Relaxed), 1);
}
#[test]
fn two_consecutive_checks_set_the_schedule() {
let stop = DebouncedTimeout::new(Unstoppable, Duration::from_secs(3600))
.with_target_interval(Duration::from_secs(60));
stop.check().unwrap();
assert_eq!(stop.checks_per_clock_read(), 1);
stop.check().unwrap();
assert_eq!(stop.checks_per_clock_read(), MAX_CHECKS_PER_CLOCK_READ);
}
#[test]
fn a_cleared_calibration_reads_the_clock_on_the_next_check() {
let mut stop = calibrated_on_fast_checks();
stop.clear_calibration();
assert_eq!(stop.checks_per_clock_read(), 1);
stop.deadline_nanos = 0;
assert_eq!(stop.check(), Err(StopReason::TimedOut));
}
#[test]
fn clearing_the_calibration_of_a_timed_out_timeout_keeps_it_stopped() {
let stop = DebouncedTimeout::new(Unstoppable, Duration::ZERO);
assert_eq!(stop.check(), Err(StopReason::TimedOut));
stop.clear_calibration();
for _ in 0..100 {
assert_eq!(stop.check(), Err(StopReason::TimedOut));
}
}
#[test]
fn once_timed_out_every_check_stops() {
let stop = DebouncedTimeout::new(Unstoppable, Duration::ZERO);
for _ in 0..10_000 {
assert_eq!(stop.check(), Err(StopReason::TimedOut));
assert!(stop.should_stop());
}
}
#[test]
fn threads_sharing_a_timeout_all_stop() {
let stop = DebouncedTimeout::new(Unstoppable, Duration::from_millis(20));
let start = Instant::now();
std::thread::scope(|scope| {
for _ in 0..4 {
scope.spawn(|| {
while stop.check().is_ok() {
assert!(start.elapsed() < Duration::from_secs(10));
}
});
}
});
assert_eq!(stop.check(), Err(StopReason::TimedOut));
}
}