rings_core/lifecycle/
stop.rs

1use std::sync::atomic::AtomicBool;
2use std::sync::atomic::Ordering;
3use std::sync::Arc;
4
5use event_listener::Event;
6
7#[derive(Default)]
8struct StopState {
9    requested: AtomicBool,
10    changed: Event,
11}
12
13/// Authority that can request cooperative shutdown for one lifecycle scope.
14#[derive(Clone, Default)]
15pub struct StopSource {
16    state: Arc<StopState>,
17}
18
19impl StopSource {
20    /// Create a fresh lifecycle source in the running state.
21    pub fn new() -> Self {
22        Self::default()
23    }
24
25    /// Create a read-only token linked to this source.
26    pub fn token(&self) -> StopToken {
27        StopToken {
28            state: Arc::clone(&self.state),
29        }
30    }
31
32    /// Request shutdown for every token linked to this source.
33    ///
34    /// This operation is idempotent and monotonic; there is no resume state.
35    pub fn request_stop(&self) {
36        if !self.state.requested.swap(true, Ordering::AcqRel) {
37            self.state.changed.notify(usize::MAX);
38        }
39    }
40
41    /// Return whether this source has requested shutdown.
42    pub fn is_stop_requested(&self) -> bool {
43        self.state.requested.load(Ordering::Acquire)
44    }
45}
46
47impl std::fmt::Debug for StopSource {
48    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
49        formatter
50            .debug_struct("StopSource")
51            .field("requested", &self.is_stop_requested())
52            .finish()
53    }
54}
55
56/// Read-only cooperative shutdown capability for long-running loops.
57#[derive(Clone, Default)]
58pub struct StopToken {
59    state: Arc<StopState>,
60}
61
62impl StopToken {
63    /// Create a token that is never stopped by an external source.
64    pub fn never() -> Self {
65        Self::default()
66    }
67
68    /// Return whether the owner has requested cooperative shutdown.
69    pub fn should_stop(&self) -> bool {
70        self.state.requested.load(Ordering::Acquire)
71    }
72
73    /// Wait until the linked source requests shutdown.
74    ///
75    /// Registration happens before the second predicate check, so a request racing with this
76    /// method cannot be missed. Calling this method on [`Self::never`] waits indefinitely.
77    pub async fn stopped(&self) {
78        loop {
79            if self.should_stop() {
80                return;
81            }
82            let changed = self.state.changed.listen();
83            if self.should_stop() {
84                return;
85            }
86            changed.await;
87        }
88    }
89}
90
91impl std::fmt::Debug for StopToken {
92    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
93        formatter
94            .debug_struct("StopToken")
95            .field("requested", &self.should_stop())
96            .finish()
97    }
98}
rings_core/lifecycle/stop.rs

rings_core/lifecycle/
stop.rs
