almost_enough/

stopper.rs

//! The default cancellation primitive.
//!
//! [`Stopper`] is the recommended type for most use cases. It's a simple,
//! Arc-based cancellation flag with unified clone semantics.
//!
//! # Example
//!
//! ```rust
//! use almost_enough::{Stopper, Stop};
//!
//! let stop = Stopper::new();
//! let stop2 = stop.clone();  // Both share the same flag
//!
//! assert!(!stop.should_stop());
//!
//! stop2.cancel();  // Any clone can cancel
//! assert!(stop.should_stop());
//! ```
//!
//! # Design
//!
//! `Stopper` uses a unified clone model (like tokio's `CancellationToken`):
//! - Just clone to share - no separate "token" type
//! - Any clone can call `cancel()`
//! - Any clone can check `should_stop()`
//!
//! This is simpler than source/token split but means you can't prevent
//! a recipient from cancelling. If you need that, use
//! [`StopSource`](crate::StopSource)/[`StopRef`](crate::StopRef).
//!
//! # Memory Ordering
//!
//! `cancel()` is a Release swap and checks are Relaxed loads, so a check
//! costs one plain load. To read what the cancelling thread wrote before
//! `cancel()`, call [`Stopper::is_cancelled`], an Acquire load, after the
//! work reports the cancellation. [`SyncStopper`](crate::SyncStopper) makes every
//! check an Acquire load instead.

use alloc::sync::Arc;
use core::sync::atomic::{AtomicBool, Ordering};

use crate::{Stop, StopReason};

/// Inner state for [`Stopper`] — implements [`Stop`] directly so that
/// `Arc<StopperInner>` can be widened to `Arc<dyn Stop>` without double-wrapping.
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)
    }
}

/// A cancellation primitive with unified clone semantics.
///
/// This is the recommended default for most use cases. Clone it to share
/// the cancellation state - any clone can cancel or check status.
///
/// Converts to [`StopToken`](crate::StopToken) via `From`/`Into` with zero
/// overhead — the existing `Arc` is reused, not double-wrapped.
///
/// # Example
///
/// ```rust
/// use almost_enough::{Stopper, Stop};
///
/// let stop = Stopper::new();
///
/// // Pass a clone to another thread
/// let stop2 = stop.clone();
/// std::thread::spawn(move || {
///     while !stop2.should_stop() {
///         // do work
///         break;
///     }
/// }).join().unwrap();
///
/// // Cancel from original
/// stop.cancel();
/// ```
///
/// # Performance
///
/// - Size: 8 bytes (one pointer)
/// - `check()`: ~1-2ns (single atomic load with Relaxed ordering)
/// - `clone()`: atomic increment
/// - `cancel()`: atomic store
/// - `into() -> StopToken`: zero-cost (Arc pointer widening)
#[derive(Debug, Clone)]
pub struct Stopper {
    pub(crate) inner: Arc<StopperInner>,
}

impl Stopper {
    /// Create a new stopper.
    #[inline]
    pub fn new() -> Self {
        Self {
            inner: Arc::new(StopperInner {
                cancelled: AtomicBool::new(false),
            }),
        }
    }

    /// Create a stopper that is already cancelled.
    ///
    /// Useful for testing or when you want to signal immediate stop.
    #[inline]
    pub fn cancelled() -> Self {
        Self {
            inner: Arc::new(StopperInner {
                cancelled: AtomicBool::new(true),
            }),
        }
    }

    /// Signal all clones to stop.
    ///
    /// A Release swap: writes made before a `cancel()` are visible to a
    /// thread whose [`is_cancelled`](Self::is_cancelled) returns `true` after
    /// it, even when several threads cancel. Calling it again changes nothing
    /// else.
    #[inline]
    pub fn cancel(&self) {
        // A swap, not a store: a second canceller's plain store would end the
        // first one's release sequence, and a reader that saw only the second
        // could miss the first canceller's writes.
        self.inner.cancelled.swap(true, Ordering::Release);
    }

    /// Whether cancellation has been requested, as an Acquire load.
    ///
    /// When it returns `true`, this thread also sees every write the
    /// cancelling thread made before [`cancel`](Self::cancel), such as a
    /// reason code kept in an atomic. Call it after the work reports the
    /// stop. In a loop, use [`check`](Stop::check) or
    /// [`should_stop`](Stop::should_stop): one Relaxed load each, which see
    /// the stop just as soon but promise nothing about other writes.
    #[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::*;

    /// The writer cancels after a Relaxed store; the reader waits with
    /// Relaxed checks, then calls `is_cancelled()`. Miri's weak-memory
    /// emulation fails this if `cancel()` isn't Release or `is_cancelled()`
    /// isn't Acquire; CI runs it over many seeds.
    #[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() {
        // A second canceller (a guard dropping, a watchdog) cancels again
        // after seeing the first cancel through a Relaxed check.
        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());

        // Either clone can cancel
        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()
        };
        // Original is dropped, but clone still works
        assert!(!stop2.should_stop());
    }

    #[test]
    fn cancel_is_idempotent() {
        let stop = Stopper::new();
        stop.cancel();
        stop.cancel();
        stop.cancel();
        assert!(stop.is_cancelled());
    }
}