1use std::sync::atomic::{
17 AtomicBool, AtomicU32, AtomicU64,
18 Ordering::{Relaxed, SeqCst},
19};
20use std::time::{Duration, Instant};
21
22use crate::{Stop, StopReason};
23
24const DEFAULT_TARGET_NANOS: u64 = 100_000;
30
31const MAX_CHECKS_PER_CLOCK_READ: u32 = 64;
40
41#[inline]
43fn duration_to_nanos(d: Duration) -> u64 {
44 d.as_nanos().min(u64::MAX as u128) as u64
45}
46
47pub struct DebouncedTimeout<T> {
109 inner: T,
110 created: Instant,
112 deadline_nanos: u64,
114 target_nanos: u64,
116
117 schedule: Schedule,
120}
121
122struct Schedule {
124 countdown: AtomicU32,
129 skip_mod: AtomicU32,
131 last_measured_nanos: AtomicU64,
133 timed_out: AtomicBool,
135 calibrated: AtomicBool,
137}
138
139const NO_BASELINE: u64 = u64::MAX;
141
142impl Schedule {
143 fn new() -> Self {
145 Self {
146 countdown: AtomicU32::new(1),
147 skip_mod: AtomicU32::new(1),
148 last_measured_nanos: AtomicU64::new(NO_BASELINE),
149 timed_out: AtomicBool::new(false),
150 calibrated: AtomicBool::new(false),
151 }
152 }
153
154 #[inline(always)]
156 fn due(&self) -> bool {
157 !matches!(
158 self.countdown.fetch_sub(1, Relaxed),
159 2..=MAX_CHECKS_PER_CLOCK_READ
160 )
161 }
162}
163
164impl<T: Stop> DebouncedTimeout<T> {
165 #[inline]
170 pub fn new(inner: T, duration: Duration) -> Self {
171 let now = Instant::now();
172 Self {
173 inner,
174 created: now,
175 deadline_nanos: duration_to_nanos(duration),
176 target_nanos: DEFAULT_TARGET_NANOS,
177 schedule: Schedule::new(),
178 }
179 }
180
181 #[inline]
186 pub fn with_deadline(inner: T, deadline: Instant) -> Self {
187 let now = Instant::now();
188 Self {
189 inner,
190 created: now,
191 deadline_nanos: duration_to_nanos(deadline.saturating_duration_since(now)),
192 target_nanos: DEFAULT_TARGET_NANOS,
193 schedule: Schedule::new(),
194 }
195 }
196
197 #[inline]
206 pub fn with_target_interval(mut self, interval: Duration) -> Self {
207 self.target_nanos = duration_to_nanos(interval).max(1);
208 self
209 }
210
211 #[inline]
213 pub fn deadline(&self) -> Instant {
214 self.created + Duration::from_nanos(self.deadline_nanos)
215 }
216
217 #[inline]
221 pub fn remaining(&self) -> Duration {
222 self.deadline().saturating_duration_since(Instant::now())
223 }
224
225 #[inline]
227 pub fn inner(&self) -> &T {
228 &self.inner
229 }
230
231 #[inline]
233 pub fn into_inner(self) -> T {
234 self.inner
235 }
236
237 #[inline]
243 pub fn checks_per_clock_read(&self) -> u32 {
244 self.schedule.skip_mod.load(Relaxed)
245 }
246
247 pub fn clear_calibration(&self) {
254 let schedule = &self.schedule;
255 schedule.skip_mod.store(1, Relaxed);
256 schedule.calibrated.store(false, Relaxed);
257 schedule.last_measured_nanos.store(NO_BASELINE, Relaxed);
258 schedule.countdown.store(1, SeqCst);
260 }
261
262 #[cold]
264 #[inline(never)]
265 fn measure_and_recalibrate(&self) -> bool {
266 let schedule = &self.schedule;
267 let elapsed_nanos = self.created.elapsed().as_nanos() as u64;
268 if elapsed_nanos >= self.deadline_nanos {
269 schedule.timed_out.store(true, SeqCst);
271 schedule.countdown.store(1, SeqCst);
272 return true; }
274
275 let prev_nanos = schedule.last_measured_nanos.swap(elapsed_nanos, Relaxed);
278 let current_skip = schedule.skip_mod.load(Relaxed);
279 let delta_nanos = elapsed_nanos.saturating_sub(prev_nanos);
280 let nanos_per_call = delta_nanos / u64::from(current_skip);
281 let mut skip = current_skip;
282 if prev_nanos == NO_BASELINE {
283 } else if !schedule.calibrated.swap(true, Relaxed) {
285 let ideal = self
287 .target_nanos
288 .checked_div(nanos_per_call)
289 .unwrap_or(u64::MAX);
290 skip = ideal.clamp(1, u64::from(MAX_CHECKS_PER_CLOCK_READ)) as u32;
291 schedule.skip_mod.store(skip, Relaxed);
292 } else if let Some(ideal) = self.target_nanos.checked_div(nanos_per_call) {
293 let ideal_skip = ideal.clamp(1, u64::from(MAX_CHECKS_PER_CLOCK_READ)) as u32;
295 skip = if ideal_skip <= current_skip {
296 ideal_skip
298 } else {
299 current_skip + (ideal_skip - current_skip).div_ceil(8)
301 };
302 schedule.skip_mod.store(skip, Relaxed);
303 }
304 schedule.countdown.store(skip, SeqCst);
308 if schedule.timed_out.load(SeqCst) {
309 schedule.countdown.store(1, Relaxed);
310 }
311 false }
313}
314
315impl<T: Stop> Stop for DebouncedTimeout<T> {
316 #[inline]
317 #[track_caller]
318 fn check(&self) -> Result<(), StopReason> {
319 self.inner.check()?;
321
322 if !self.schedule.due() {
324 return Ok(());
325 }
326
327 if self.measure_and_recalibrate() {
329 Err(StopReason::TimedOut)
330 } else {
331 Ok(())
332 }
333 }
334
335 #[inline]
336 #[track_caller]
337 fn should_stop(&self) -> bool {
338 if self.inner.should_stop() {
339 return true;
340 }
341
342 if !self.schedule.due() {
343 return false;
344 }
345
346 self.measure_and_recalibrate()
347 }
348}
349
350impl<T: Stop> DebouncedTimeout<T> {
351 #[inline]
356 pub fn tighten(self, duration: Duration) -> Self {
357 let elapsed = duration_to_nanos(Instant::now().saturating_duration_since(self.created));
358 let new_deadline_nanos = elapsed.saturating_add(duration_to_nanos(duration));
359 let deadline_nanos = self.deadline_nanos.min(new_deadline_nanos);
360 Self {
361 inner: self.inner,
362 created: self.created,
363 deadline_nanos,
364 target_nanos: self.target_nanos,
365 schedule: Schedule::new(),
366 }
367 }
368
369 #[inline]
374 pub fn tighten_deadline(self, deadline: Instant) -> Self {
375 let new_deadline_nanos =
376 duration_to_nanos(deadline.saturating_duration_since(self.created));
377 let deadline_nanos = self.deadline_nanos.min(new_deadline_nanos);
378 Self {
379 inner: self.inner,
380 created: self.created,
381 deadline_nanos,
382 target_nanos: self.target_nanos,
383 schedule: Schedule::new(),
384 }
385 }
386}
387
388impl<T: Clone + Stop> Clone for DebouncedTimeout<T> {
389 fn clone(&self) -> Self {
391 Self {
392 inner: self.inner.clone(),
393 created: self.created,
394 deadline_nanos: self.deadline_nanos,
395 target_nanos: self.target_nanos,
396 schedule: Schedule::new(),
397 }
398 }
399}
400
401impl<T: core::fmt::Debug> core::fmt::Debug for DebouncedTimeout<T> {
402 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
403 let deadline = self.created + Duration::from_nanos(self.deadline_nanos);
404 f.debug_struct("DebouncedTimeout")
405 .field("inner", &self.inner)
406 .field("deadline", &deadline)
407 .field("target_interval_us", &(self.target_nanos / 1_000))
408 .field("skip_mod", &self.schedule.skip_mod.load(Relaxed))
409 .finish()
410 }
411}
412
413pub trait DebouncedTimeoutExt: Stop + Sized {
418 #[inline]
425 fn with_debounced_timeout(self, duration: Duration) -> DebouncedTimeout<Self> {
426 DebouncedTimeout::new(self, duration)
427 }
428
429 #[inline]
431 fn with_debounced_deadline(self, deadline: Instant) -> DebouncedTimeout<Self> {
432 DebouncedTimeout::with_deadline(self, deadline)
433 }
434}
435
436impl<T: Stop> DebouncedTimeoutExt for T {}
437
438#[cfg(test)]
439mod tests {
440 use super::*;
441 use crate::{StopSource, Unstoppable};
442
443 #[test]
444 fn basic_timeout() {
445 let source = StopSource::new();
446 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_millis(50));
447
448 assert!(!stop.should_stop());
449 assert!(stop.check().is_ok());
450
451 std::thread::sleep(Duration::from_millis(80));
452
453 for _ in 0..100 {
455 if stop.should_stop() {
456 return; }
458 }
459 panic!("should have detected timeout");
460 }
461
462 #[test]
463 fn cancel_before_timeout() {
464 let source = StopSource::new();
465 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60));
466
467 source.cancel();
468
469 assert!(stop.should_stop());
471 assert_eq!(stop.check(), Err(StopReason::Cancelled));
472 }
473
474 #[test]
475 fn calibration_ramps_up() {
476 let source = StopSource::new();
477 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60));
478
479 assert_eq!(stop.checks_per_clock_read(), 1);
481
482 for _ in 0..10_000 {
484 let _ = stop.check();
485 }
486
487 assert!(
489 stop.checks_per_clock_read() > 1,
490 "skip_mod should have increased, got {}",
491 stop.checks_per_clock_read()
492 );
493 }
494
495 #[test]
496 fn remaining_accuracy() {
497 let source = StopSource::new();
498 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
499
500 let remaining = stop.remaining();
501 assert!(remaining > Duration::from_secs(9));
502 assert!(remaining <= Duration::from_secs(10));
503 }
504
505 #[test]
506 fn tighten_works() {
507 let source = StopSource::new();
508 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60))
509 .tighten(Duration::from_secs(1));
510
511 let remaining = stop.remaining();
512 assert!(remaining < Duration::from_secs(2));
513 }
514
515 #[test]
516 fn clone_resets_calibration() {
517 let source = StopSource::new();
518 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60));
519
520 for _ in 0..10_000 {
522 let _ = stop.check();
523 }
524 assert!(stop.checks_per_clock_read() > 1);
525
526 let cloned = stop.clone();
528 assert_eq!(cloned.checks_per_clock_read(), 1);
529 }
530
531 #[test]
532 fn extension_trait() {
533 use super::DebouncedTimeoutExt;
534 let source = StopSource::new();
535 let stop = source
536 .as_ref()
537 .with_debounced_timeout(Duration::from_secs(10));
538 assert!(!stop.should_stop());
539 }
540
541 #[test]
542 fn is_send_sync() {
543 fn assert_send_sync<T: Send + Sync>() {}
544 assert_send_sync::<DebouncedTimeout<crate::StopRef<'_>>>();
545 }
546
547 #[test]
548 fn zero_duration_immediate_timeout() {
549 let source = StopSource::new();
550 let stop = DebouncedTimeout::new(source.as_ref(), Duration::ZERO);
551
552 assert_eq!(stop.check(), Err(StopReason::TimedOut));
554 }
555
556 #[test]
557 fn deadline_in_the_past() {
558 let source = StopSource::new();
559 let past = Instant::now() - Duration::from_secs(1);
560 let stop = DebouncedTimeout::with_deadline(source.as_ref(), past);
561
562 assert_eq!(stop.check(), Err(StopReason::TimedOut));
564 }
565
566 #[test]
567 fn with_deadline_basic() {
568 let source = StopSource::new();
569 let deadline = Instant::now() + Duration::from_millis(100);
570 let stop = DebouncedTimeout::with_deadline(source.as_ref(), deadline);
571
572 assert!(!stop.should_stop());
573
574 std::thread::sleep(Duration::from_millis(150));
575
576 for _ in 0..100 {
578 if stop.should_stop() {
579 return;
580 }
581 }
582 panic!("should have detected timeout");
583 }
584
585 #[test]
586 fn deadline_accessor() {
587 let source = StopSource::new();
588 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
589
590 let deadline = stop.deadline();
591 let remaining = stop.remaining();
592 assert!(remaining > Duration::from_secs(9));
593 assert!(remaining <= Duration::from_secs(10));
594 assert!(deadline > Instant::now() + Duration::from_secs(9));
596 }
597
598 #[test]
599 fn inner_access() {
600 let source = StopSource::new();
601 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
602
603 assert!(!stop.inner().should_stop());
604
605 source.cancel();
606
607 assert!(stop.inner().should_stop());
608 }
609
610 #[test]
611 fn into_inner_works() {
612 let source = StopSource::new();
613 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
614
615 let inner = stop.into_inner();
616 assert!(!inner.should_stop());
617 }
618
619 #[test]
620 fn tighten_deadline_works() {
621 let source = StopSource::new();
622 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60))
623 .tighten_deadline(Instant::now() + Duration::from_secs(1));
624
625 let remaining = stop.remaining();
626 assert!(remaining < Duration::from_secs(2));
627 }
628
629 #[test]
630 fn tighten_does_not_loosen() {
631 let source = StopSource::new();
632 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(1))
633 .tighten(Duration::from_secs(60));
634
635 let remaining = stop.remaining();
637 assert!(remaining < Duration::from_secs(2));
638 }
639
640 #[test]
641 fn debug_format() {
642 let source = StopSource::new();
643 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(10));
644 let debug = format!("{stop:?}");
645 assert!(debug.contains("DebouncedTimeout"));
646 assert!(debug.contains("skip_mod"));
647 assert!(debug.contains("target_interval_us"));
648 }
649
650 #[test]
651 fn with_target_interval_zero_clamps_to_one() {
652 let source = StopSource::new();
653 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60))
654 .with_target_interval(Duration::ZERO);
655
656 assert!(stop.check().is_ok());
658 }
659
660 #[test]
661 fn with_debounced_deadline_ext() {
662 let source = StopSource::new();
663 let deadline = Instant::now() + Duration::from_secs(10);
664 let stop = source.as_ref().with_debounced_deadline(deadline);
665
666 assert!(!stop.should_stop());
667 assert!(stop.remaining() > Duration::from_secs(9));
668 }
669
670 #[test]
671 fn check_and_should_stop_agree() {
672 let source = StopSource::new();
673 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60));
674
675 for _ in 0..1000 {
677 assert!(!stop.should_stop());
678 assert!(stop.check().is_ok());
679 }
680
681 source.cancel();
682
683 assert!(stop.should_stop());
685 assert_eq!(stop.check(), Err(StopReason::Cancelled));
686 }
687
688 #[test]
689 fn remaining_after_expiry_is_zero() {
690 let source = StopSource::new();
691 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_millis(1));
692
693 std::thread::sleep(Duration::from_millis(10));
694
695 assert_eq!(stop.remaining(), Duration::ZERO);
696 }
697
698 #[test]
699 fn adapts_to_slowdown() {
700 let source = StopSource::new();
701 let stop = DebouncedTimeout::new(source.as_ref(), Duration::from_secs(60))
703 .with_target_interval(Duration::from_micros(10));
704
705 for _ in 0..50_000 {
707 let _ = stop.check();
708 }
709 let fast_skip = stop.checks_per_clock_read();
710 assert!(fast_skip > 1, "should have ramped up, got {fast_skip}");
711
712 for _ in 0..(fast_skip as usize + 100) {
715 std::thread::sleep(Duration::from_micros(50));
716 let _ = stop.check();
717 }
718
719 let slow_skip = stop.checks_per_clock_read();
720 assert!(
721 slow_skip < fast_skip,
722 "should have reduced skip_mod from {fast_skip} to less, got {slow_skip}"
723 );
724 }
725
726 fn work_for(duration: Duration) {
728 let start = Instant::now();
729 while start.elapsed() < duration {
730 std::hint::spin_loop();
731 }
732 }
733
734 #[test]
735 fn reads_the_clock_at_least_every_64_checks() {
736 let stop = DebouncedTimeout::new(Unstoppable, Duration::from_secs(60));
737 for _ in 0..1_000_000 {
738 stop.check().unwrap();
739 assert!(stop.checks_per_clock_read() <= MAX_CHECKS_PER_CLOCK_READ);
740 }
741 }
742
743 fn calibrated_on_fast_checks() -> DebouncedTimeout<Unstoppable> {
746 let stop = DebouncedTimeout::new(Unstoppable, Duration::from_secs(3600));
747 let fast_until = Instant::now() + Duration::from_millis(20);
748 while Instant::now() < fast_until {
749 stop.check().unwrap();
750 }
751 stop
752 }
753
754 #[test]
755 fn a_passed_deadline_stops_within_64_checks() {
756 let mut stop = calibrated_on_fast_checks();
757 stop.deadline_nanos = 0;
758 let mut ok_checks = 0;
759 while stop.check().is_ok() {
760 ok_checks += 1;
761 assert!(ok_checks < MAX_CHECKS_PER_CLOCK_READ, "{ok_checks} checks");
762 }
763 }
764
765 #[test]
766 fn a_slowdown_after_calibration_stops_within_64_slow_checks() {
767 let mut stop = calibrated_on_fast_checks();
772 stop.deadline_nanos = duration_to_nanos(stop.created.elapsed()) + 20_000_000;
773 let mut slow_checks = 0;
774 while stop.check().is_ok() {
775 slow_checks += 1;
776 assert!(slow_checks < 1_000, "still running a second later");
777 work_for(Duration::from_millis(1));
778 }
779 assert!(slow_checks <= 20 + 64 + 8, "{slow_checks} slow checks");
783 }
784
785 #[test]
786 fn a_clock_read_just_before_another_threads_timeout_keeps_it_stopped() {
787 let stop = calibrated_on_fast_checks();
790 stop.schedule.timed_out.store(true, SeqCst);
791 stop.schedule.countdown.store(1, SeqCst);
792 assert!(!stop.measure_and_recalibrate());
793 assert_eq!(stop.schedule.countdown.load(Relaxed), 1);
794 }
795
796 #[test]
797 fn two_consecutive_checks_set_the_schedule() {
798 let stop = DebouncedTimeout::new(Unstoppable, Duration::from_secs(3600))
799 .with_target_interval(Duration::from_secs(60));
800 stop.check().unwrap();
801 assert_eq!(stop.checks_per_clock_read(), 1);
802 stop.check().unwrap();
803 assert_eq!(stop.checks_per_clock_read(), MAX_CHECKS_PER_CLOCK_READ);
804 }
805
806 #[test]
807 fn a_cleared_calibration_reads_the_clock_on_the_next_check() {
808 let mut stop = calibrated_on_fast_checks();
809 stop.clear_calibration();
810 assert_eq!(stop.checks_per_clock_read(), 1);
811 stop.deadline_nanos = 0;
812 assert_eq!(stop.check(), Err(StopReason::TimedOut));
813 }
814
815 #[test]
816 fn clearing_the_calibration_of_a_timed_out_timeout_keeps_it_stopped() {
817 let stop = DebouncedTimeout::new(Unstoppable, Duration::ZERO);
818 assert_eq!(stop.check(), Err(StopReason::TimedOut));
819 stop.clear_calibration();
820 for _ in 0..100 {
821 assert_eq!(stop.check(), Err(StopReason::TimedOut));
822 }
823 }
824
825 #[test]
826 fn once_timed_out_every_check_stops() {
827 let stop = DebouncedTimeout::new(Unstoppable, Duration::ZERO);
828 for _ in 0..10_000 {
829 assert_eq!(stop.check(), Err(StopReason::TimedOut));
830 assert!(stop.should_stop());
831 }
832 }
833
834 #[test]
835 fn threads_sharing_a_timeout_all_stop() {
836 let stop = DebouncedTimeout::new(Unstoppable, Duration::from_millis(20));
839 let start = Instant::now();
840 std::thread::scope(|scope| {
841 for _ in 0..4 {
842 scope.spawn(|| {
843 while stop.check().is_ok() {
844 assert!(start.elapsed() < Duration::from_secs(10));
845 }
846 });
847 }
848 });
849 assert_eq!(stop.check(), Err(StopReason::TimedOut));
850 }
851}