nucleo_picker/
observer.rs

1//! # An observer channel
2use parking_lot::{Condvar, Mutex};
3use std::sync::{
4    Arc,
5    mpsc::{RecvError, SendError, TryRecvError},
6};
7
8type Channel<T> = Mutex<(Option<T>, bool)>;
9
10/// The 'notify' end of the single slot channel.
11pub(crate) struct Notifier<T> {
12    inner: Arc<(Channel<T>, Condvar)>,
13}
14
15#[inline]
16fn channel_inner<T>(msg: Option<T>) -> (Notifier<T>, Observer<T>) {
17    let inner = Arc::new((Mutex::new((msg, true)), Condvar::new()));
18
19    let observer = Observer {
20        inner: Arc::clone(&inner),
21    };
22
23    let notifier = Notifier { inner };
24
25    (notifier, observer)
26}
27
28pub(crate) fn occupied_channel<T>(msg: T) -> (Notifier<T>, Observer<T>) {
29    channel_inner(Some(msg))
30}
31
32pub(crate) fn channel<T>() -> (Notifier<T>, Observer<T>) {
33    channel_inner(None)
34}
35
36impl<T> Notifier<T> {
37    /// Push a message to the channel. This overwrites any pre-existing message already in the
38    /// channel.
39    pub fn push(&self, msg: T) -> Result<(), SendError<T>> {
40        // in principle, this is a TOCTOU error, since the observer could drop in between this
41        // check and the time that the message is pushed to the channel. however, this
42        // is not a serious concern here because this is indistinguishable from the case that
43        // the injector is dropped immediately (without sending any elements) even if it was
44        // successfully received
45        //
46        // this check should be thought of as 'best-effort', instead of actually being critical
47        // for logic.
48        if Arc::strong_count(&self.inner) == 1 {
49            // there are no observers so the channel is disconnected
50            Err(SendError(msg))
51        } else {
52            // overwrite the channel with the new message and notify an observer that a message
53            // is available
54            let (lock, cvar) = &*self.inner;
55            let mut channel = lock.lock();
56            channel.0 = Some(msg);
57            cvar.notify_one();
58            Ok(())
59        }
60    }
61}
62
63impl<T> Drop for Notifier<T> {
64    fn drop(&mut self) {
65        // when we drop the notifier, we need to inform all observers that are potentially waiting
66        // for a message that the channel is closed
67        let (lock, cvar) = &*self.inner;
68
69        let mut channel = lock.lock();
70        channel.1 = false;
71        cvar.notify_all();
72    }
73}
74
75/// An `Observer` watching for a single message `T`.
76///
77/// This is similar to the 'receiver' end of a channel of length 1, but instead of blocking, the
78/// 'sender' always overwrites any element in the channel. In particular, any message obtained by
79/// [`recv`](Observer::recv) or [`try_recv`](Observer::try_recv) is guaranteed to be the most
80/// up-to-date at the moment when the message is received.
81///
82/// The channel may be updated when not observed. Receiving a message moves it out of the observer.
83/// An observer can be cheaply cloned (a single [`Arc::clone`]) in order to watch for the message
84/// simultaneously from different threads.
85pub struct Observer<T> {
86    inner: Arc<(Channel<T>, Condvar)>,
87}
88
89impl<T> Clone for Observer<T> {
90    fn clone(&self) -> Self {
91        Self {
92            inner: Arc::clone(&self.inner),
93        }
94    }
95}
96
97impl<T> Observer<T> {
98    /// Receive a message, blocking until a message is available or the channel
99    /// disconnects.
100    pub fn recv(&self) -> Result<T, RecvError> {
101        let (lock, cvar) = &*self.inner;
102        let mut channel = lock.lock();
103        match channel.0.take() {
104            Some(msg) => Ok(msg),
105            None => {
106                if channel.1 {
107                    // the channel is active, so we wait for a notification
108                    // this uses `parking_lot::Condvar`, which is guaranteed not to wake up
109                    // spuriously
110                    cvar.wait(&mut channel);
111
112                    // we received a notification that there was a change
113                    match channel.0.take() {
114                        // the change was that a new message has been pushed, so we can return it
115                        Some(msg) => Ok(msg),
116                        // there is no message despite the notification, so the channel is
117                        // disconnected. this path is followed if the notifier is dropped while we
118                        // are waiting for a new message
119                        None => Err(RecvError),
120                    }
121                } else {
122                    Err(RecvError)
123                }
124            }
125        }
126    }
127
128    /// Optimistically receive a message if one is available without blocking the current thread.
129    ///
130    /// This operation will fail if there is no message or if there are are no remaining senders.
131    pub fn try_recv(&self) -> Result<T, TryRecvError> {
132        let (lock, _) = &*self.inner;
133        let mut channel = lock.lock();
134        channel.0.take().ok_or(if channel.1 {
135            TryRecvError::Empty
136        } else {
137            TryRecvError::Disconnected
138        })
139    }
140}
nucleo_picker/observer.rs

nucleo_picker/
observer.rs
