use alloc::sync::Arc;
use core::sync::atomic::{AtomicBool, Ordering};
use crate::{Stop, StopReason};
pub(crate) struct StopperInner {
cancelled: AtomicBool,
}
impl Stop for StopperInner {
#[inline]
#[track_caller]
fn check(&self) -> Result<(), StopReason> {
if self.cancelled.load(Ordering::Relaxed) {
Err(StopReason::Cancelled)
} else {
Ok(())
}
}
#[inline]
#[track_caller]
fn should_stop(&self) -> bool {
self.cancelled.load(Ordering::Relaxed)
}
}
#[derive(Debug, Clone)]
pub struct Stopper {
pub(crate) inner: Arc<StopperInner>,
}
impl Stopper {
#[inline]
pub fn new() -> Self {
Self {
inner: Arc::new(StopperInner {
cancelled: AtomicBool::new(false),
}),
}
}
#[inline]
pub fn cancelled() -> Self {
Self {
inner: Arc::new(StopperInner {
cancelled: AtomicBool::new(true),
}),
}
}
#[inline]
pub fn cancel(&self) {
self.inner.cancelled.swap(true, Ordering::Release);
}
#[inline]
pub fn is_cancelled(&self) -> bool {
self.inner.cancelled.load(Ordering::Acquire)
}
}
impl Default for Stopper {
fn default() -> Self {
Self::new()
}
}
impl Stop for Stopper {
#[inline]
#[track_caller]
fn check(&self) -> Result<(), StopReason> {
self.inner.check()
}
#[inline]
#[track_caller]
fn should_stop(&self) -> bool {
self.inner.should_stop()
}
}
impl core::fmt::Debug for StopperInner {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("StopperInner")
.field("cancelled", &self.cancelled.load(Ordering::Relaxed))
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(feature = "std")]
#[test]
fn is_cancelled_after_the_stop_sees_writes_made_before_cancel() {
use core::sync::atomic::{AtomicU32, Ordering::Relaxed};
for _ in 0..if cfg!(miri) { 30 } else { 1_000 } {
let stop = Stopper::new();
let data = Arc::new(AtomicU32::new(0));
let reader = {
let (stop, data) = (stop.clone(), Arc::clone(&data));
std::thread::spawn(move || {
while !stop.should_stop() {
core::hint::spin_loop();
}
assert!(stop.is_cancelled());
data.load(Relaxed)
})
};
data.store(42, Relaxed);
stop.cancel();
assert_eq!(reader.join().unwrap(), 42);
}
}
#[cfg(feature = "std")]
#[test]
fn is_cancelled_after_a_second_cancel_sees_the_first_cancellers_writes() {
use core::sync::atomic::{AtomicU32, Ordering::Relaxed};
for _ in 0..if cfg!(miri) { 30 } else { 1_000 } {
let stop = Stopper::new();
let data = AtomicU32::new(0);
std::thread::scope(|scope| {
let reader = scope.spawn(|| {
while !stop.should_stop() {
core::hint::spin_loop();
}
assert!(stop.is_cancelled());
data.load(Relaxed)
});
scope.spawn(|| {
while !stop.should_stop() {
core::hint::spin_loop();
}
stop.cancel();
});
data.store(42, Relaxed);
stop.cancel();
assert_eq!(reader.join().unwrap(), 42);
});
}
}
#[test]
fn stopper_basic() {
let stop = Stopper::new();
assert!(!stop.is_cancelled());
assert!(!stop.should_stop());
assert!(stop.check().is_ok());
stop.cancel();
assert!(stop.is_cancelled());
assert!(stop.should_stop());
assert_eq!(stop.check(), Err(StopReason::Cancelled));
}
#[test]
fn stopper_cancelled_constructor() {
let stop = Stopper::cancelled();
assert!(stop.is_cancelled());
assert!(stop.should_stop());
}
#[test]
fn stopper_clone_shares_state() {
let stop1 = Stopper::new();
let stop2 = stop1.clone();
assert!(!stop1.should_stop());
assert!(!stop2.should_stop());
stop2.cancel();
assert!(stop1.should_stop());
assert!(stop2.should_stop());
}
#[test]
fn stopper_is_default() {
let stop: Stopper = Default::default();
assert!(!stop.is_cancelled());
}
#[test]
fn stopper_is_send_sync() {
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<Stopper>();
}
#[test]
fn stopper_can_outlive_original() {
let stop2 = {
let stop1 = Stopper::new();
stop1.clone()
};
assert!(!stop2.should_stop());
}
#[test]
fn cancel_is_idempotent() {
let stop = Stopper::new();
stop.cancel();
stop.cancel();
stop.cancel();
assert!(stop.is_cancelled());
}
}