tor_events/
lib.rs

Help
1#![cfg_attr(docsrs, feature(doc_auto_cfg, doc_cfg))]
2#![doc = include_str!("../README.md")]
3// @@ begin lint list maintained by maint/add_warning @@
4#![allow(renamed_and_removed_lints)] // @@REMOVE_WHEN(ci_arti_stable)
5#![allow(unknown_lints)] // @@REMOVE_WHEN(ci_arti_nightly)
6#![warn(missing_docs)]
7#![warn(noop_method_call)]
8#![warn(unreachable_pub)]
9#![warn(clippy::all)]
10#![deny(clippy::await_holding_lock)]
11#![deny(clippy::cargo_common_metadata)]
12#![deny(clippy::cast_lossless)]
13#![deny(clippy::checked_conversions)]
14#![warn(clippy::cognitive_complexity)]
15#![deny(clippy::debug_assert_with_mut_call)]
16#![deny(clippy::exhaustive_enums)]
17#![deny(clippy::exhaustive_structs)]
18#![deny(clippy::expl_impl_clone_on_copy)]
19#![deny(clippy::fallible_impl_from)]
20#![deny(clippy::implicit_clone)]
21#![deny(clippy::large_stack_arrays)]
22#![warn(clippy::manual_ok_or)]
23#![deny(clippy::missing_docs_in_private_items)]
24#![warn(clippy::needless_borrow)]
25#![warn(clippy::needless_pass_by_value)]
26#![warn(clippy::option_option)]
27#![deny(clippy::print_stderr)]
28#![deny(clippy::print_stdout)]
29#![warn(clippy::rc_buffer)]
30#![deny(clippy::ref_option_ref)]
31#![warn(clippy::semicolon_if_nothing_returned)]
32#![warn(clippy::trait_duplication_in_bounds)]
33#![deny(clippy::unchecked_duration_subtraction)]
34#![deny(clippy::unnecessary_wraps)]
35#![warn(clippy::unseparated_literal_suffix)]
36#![deny(clippy::unwrap_used)]
37#![deny(clippy::mod_module_files)]
38#![allow(clippy::let_unit_value)] // This can reasonably be done for explicitness
39#![allow(clippy::uninlined_format_args)]
40#![allow(clippy::significant_drop_in_scrutinee)] // arti/-/merge_requests/588/#note_2812945
41#![allow(clippy::result_large_err)] // temporary workaround for arti#587
42#![allow(clippy::needless_raw_string_hashes)] // complained-about code is fine, often best
43#![allow(clippy::needless_lifetimes)] // See arti#1765
44#![allow(mismatched_lifetime_syntaxes)] // temporary workaround for arti#2060
45//! <!-- @@ end lint list maintained by maint/add_warning @@ -->
46
47pub mod events;
48
49use crate::events::{TorEvent, TorEventKind};
50use async_broadcast::{InactiveReceiver, Receiver, Sender, TrySendError};
51use futures::channel::mpsc;
52use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
53use futures::future::Either;
54use futures::StreamExt;
55use std::pin::Pin;
56use std::sync::atomic::{AtomicUsize, Ordering};
57use std::sync::OnceLock;
58use std::task::{Context, Poll};
59use thiserror::Error;
60use tracing::{error, warn};
61
62/// Pointer to an `UnboundedSender`, used to send events into the `EventReactor`.
63static EVENT_SENDER: OnceLock<UnboundedSender<TorEvent>> = OnceLock::new();
64/// An inactive receiver for the currently active broadcast channel, if there is one.
65static CURRENT_RECEIVER: OnceLock<InactiveReceiver<TorEvent>> = OnceLock::new();
66/// The number of `TorEventKind`s there are.
67const EVENT_KIND_COUNT: usize = 1;
68/// An array containing one `AtomicUsize` for each `TorEventKind`, used to track subscriptions.
69///
70/// When a `TorEventReceiver` subscribes to a `TorEventKind`, it uses its `usize` value to index
71/// into this array and increment the associated `AtomicUsize` (and decrements it to unsubscribe).
72/// This lets event emitters check whether there are any subscribers, and avoid emitting events
73/// if there aren't.
74static EVENT_SUBSCRIBERS: [AtomicUsize; EVENT_KIND_COUNT] = [AtomicUsize::new(0); EVENT_KIND_COUNT];
75
76/// The size of the internal broadcast channel used to implement event subscription.
77pub static BROADCAST_CAPACITY: usize = 512;
78
79/// A reactor used to forward events to make the event reporting system work.
80///
81/// # Note
82///
83/// Currently, this type is a singleton; there is one event reporting system used for the entire
84/// program. This is not stable, and may change in future.
85pub struct EventReactor {
86    /// A receiver that the reactor uses to learn about incoming events.
87    ///
88    /// This is unbounded so that event publication doesn't have to be async.
89    receiver: UnboundedReceiver<TorEvent>,
90    /// A sender that the reactor uses to publish events.
91    ///
92    /// Events are only sent here if at least one subscriber currently wants them.
93    broadcast: Sender<TorEvent>,
94}
95
96impl EventReactor {
97    /// Initialize the event reporting system, returning a reactor that must be run for it to work,
98    /// and a `TorEventReceiver` that can be used to extract events from the system. If the system
99    /// has already been initialized, returns `None` instead of a reactor.
100    ///
101    /// # Warnings
102    ///
103    /// The returned reactor *must* be run with `EventReactor::run`, in a background async task.
104    /// If it is not, the event system might consume unbounded amounts of memory.
105    pub fn new() -> Option<Self> {
106        let (tx, rx) = mpsc::unbounded();
107        if EVENT_SENDER.set(tx).is_ok() {
108            let (btx, brx) = async_broadcast::broadcast(BROADCAST_CAPACITY);
109            CURRENT_RECEIVER
110                .set(brx.deactivate())
111                .expect("CURRENT_RECEIVER can't be set if EVENT_SENDER is unset!");
112            Some(Self {
113                receiver: rx,
114                broadcast: btx,
115            })
116        } else {
117            None
118        }
119    }
120    /// Get a `TorEventReceiver` to receive events from, assuming an `EventReactor` is already
121    /// running somewhere. (If it isn't, returns `None`.)
122    ///
123    /// As noted in the type-level documentation, this function might not always work this way.
124    pub fn receiver() -> Option<TorEventReceiver> {
125        CURRENT_RECEIVER
126            .get()
127            .map(|rx| TorEventReceiver::wrap(rx.clone()))
128    }
129    /// Run the event forwarding reactor.
130    ///
131    /// You *must* call this function once a reactor is created.
132    pub async fn run(mut self) {
133        while let Some(event) = self.receiver.next().await {
134            match self.broadcast.try_broadcast(event) {
135                Ok(_) => {}
136                Err(TrySendError::Closed(_)) => break,
137                Err(TrySendError::Full(event)) => {
138                    // If the channel is full, do a blocking broadcast to wait for it to be
139                    // not full, and log a warning about receivers lagging behind.
140                    warn!("TorEventReceivers aren't receiving events fast enough!");
141                    if self.broadcast.broadcast(event).await.is_err() {
142                        break;
143                    }
144                }
145                Err(TrySendError::Inactive(_)) => {
146                    // no active receivers, so just drop the event on the floor.
147                }
148            }
149        }
150        // It shouldn't be possible to get here, since we have globals keeping the channels
151        // open. Still, if we somehow do, log an error about it.
152        error!("event reactor shutting down; this shouldn't ever happen");
153    }
154}
155
156/// An error encountered when trying to receive a `TorEvent`.
157#[derive(Clone, Debug, Error)]
158#[non_exhaustive]
159pub enum ReceiverError {
160    /// The receiver isn't subscribed to anything, so wouldn't ever return any events.
161    #[error("No event subscriptions")]
162    NoSubscriptions,
163    /// The internal broadcast channel was closed, which shouldn't ever happen.
164    #[error("Internal event broadcast channel closed")]
165    ChannelClosed,
166}
167
168/// A receiver for `TorEvent`s emitted by other users of this crate.
169///
170/// To use this type, first subscribe to some kinds of event by calling
171/// `TorEventReceiver::subscribe`. Then, consume events using the implementation of
172/// `futures::stream::Stream`.
173///
174/// # Warning
175///
176/// Once interest in events has been signalled with `subscribe`, events must be continuously
177/// read from the receiver in order to avoid excessive memory consumption.
178#[derive(Clone, Debug)]
179pub struct TorEventReceiver {
180    /// If no events have been subscribed to yet, this is an `InactiveReceiver`; otherwise,
181    /// it's a `Receiver`.
182    inner: Either<Receiver<TorEvent>, InactiveReceiver<TorEvent>>,
183    /// Whether we're subscribed to each event kind (if `subscribed[kind]` is true, we're
184    /// subscribed to `kind`).
185    subscribed: [bool; EVENT_KIND_COUNT],
186}
187
188impl futures::stream::Stream for TorEventReceiver {
189    type Item = TorEvent;
190
191    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
192        let this = self.get_mut();
193        match this.inner {
194            Either::Left(ref mut active) => loop {
195                match Pin::new(&mut *active).poll_next(cx) {
196                    Poll::Ready(Some(e)) => {
197                        if this.subscribed[e.kind() as usize] {
198                            return Poll::Ready(Some(e));
199                        }
200                        // loop, since we weren't subscribed to that event
201                    }
202                    x => return x,
203                }
204            },
205            Either::Right(_) => {
206                warn!("TorEventReceiver::poll_next() called without subscriptions!");
207                Poll::Ready(None)
208            }
209        }
210    }
211}
212
213impl TorEventReceiver {
214    /// Create a `TorEventReceiver` from an `InactiveReceiver` handle.
215    pub(crate) fn wrap(rx: InactiveReceiver<TorEvent>) -> Self {
216        Self {
217            inner: Either::Right(rx),
218            subscribed: [false; EVENT_KIND_COUNT],
219        }
220    }
221    /// Subscribe to a given kind of `TorEvent`.
222    ///
223    /// After calling this function, `TorEventReceiver::recv` will emit events of that kind.
224    /// This function is idempotent (subscribing twice has the same effect as doing so once).
225    pub fn subscribe(&mut self, kind: TorEventKind) {
226        if !self.subscribed[kind as usize] {
227            EVENT_SUBSCRIBERS[kind as usize].fetch_add(1, Ordering::SeqCst);
228            self.subscribed[kind as usize] = true;
229        }
230        // FIXME(eta): cloning is ungood, but hard to avoid
231        if let Either::Right(inactive) = self.inner.clone() {
232            self.inner = Either::Left(inactive.activate());
233        }
234    }
235    /// Unsubscribe from a given kind of `TorEvent`.
236    ///
237    /// After calling this function, `TorEventReceiver::recv` will no longer emit events of that
238    /// kind.
239    /// This function is idempotent (unsubscribing twice has the same effect as doing so once).
240    pub fn unsubscribe(&mut self, kind: TorEventKind) {
241        if self.subscribed[kind as usize] {
242            EVENT_SUBSCRIBERS[kind as usize].fetch_sub(1, Ordering::SeqCst);
243            self.subscribed[kind as usize] = false;
244        }
245        // If we're now not subscribed to anything, deactivate our channel.
246        if self.subscribed.iter().all(|x| !*x) {
247            // FIXME(eta): cloning is ungood, but hard to avoid
248            if let Either::Left(active) = self.inner.clone() {
249                self.inner = Either::Right(active.deactivate());
250            }
251        }
252    }
253}
254
255impl Drop for TorEventReceiver {
256    fn drop(&mut self) {
257        for (i, subscribed) in self.subscribed.iter().enumerate() {
258            // FIXME(eta): duplicates logic from Self::unsubscribe, because it's not possible
259            //             to go from a `usize` to a `TorEventKind`
260            if *subscribed {
261                EVENT_SUBSCRIBERS[i].fetch_sub(1, Ordering::SeqCst);
262            }
263        }
264    }
265}
266
267/// Returns a boolean indicating whether the event `kind` has any subscribers (as in,
268/// whether `TorEventReceiver::subscribe` has been called with that event kind).
269///
270/// This is useful to avoid doing work to generate events that might be computationally expensive
271/// to generate.
272pub fn event_has_subscribers(kind: TorEventKind) -> bool {
273    EVENT_SUBSCRIBERS[kind as usize].load(Ordering::SeqCst) > 0
274}
275
276/// Broadcast the given `TorEvent` to any interested subscribers.
277///
278/// As an optimization, does nothing if the event has no subscribers (`event_has_subscribers`
279/// returns false). (also does nothing if the event subsystem hasn't been initialized yet)
280///
281/// This function isn't intended for use outside Arti crates (as in, library consumers of Arti
282/// shouldn't broadcast events!).
283pub fn broadcast(event: TorEvent) {
284    if !event_has_subscribers(event.kind()) {
285        return;
286    }
287    if let Some(sender) = EVENT_SENDER.get() {
288        // If this fails, there isn't much we can really do about it!
289        let _ = sender.unbounded_send(event);
290    }
291}
292
293#[cfg(test)]
294mod test {
295    // @@ begin test lint list maintained by maint/add_warning @@
296    #![allow(clippy::bool_assert_comparison)]
297    #![allow(clippy::clone_on_copy)]
298    #![allow(clippy::dbg_macro)]
299    #![allow(clippy::mixed_attributes_style)]
300    #![allow(clippy::print_stderr)]
301    #![allow(clippy::print_stdout)]
302    #![allow(clippy::single_char_pattern)]
303    #![allow(clippy::unwrap_used)]
304    #![allow(clippy::unchecked_duration_subtraction)]
305    #![allow(clippy::useless_vec)]
306    #![allow(clippy::needless_pass_by_value)]
307    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
308    use crate::{
309        broadcast, event_has_subscribers, EventReactor, StreamExt, TorEvent, TorEventKind,
310    };
311    use std::sync::{Mutex, MutexGuard, OnceLock};
312    use std::time::Duration;
313    use tokio::runtime::Runtime;
314
315    // HACK(eta): these tests need to run effectively singlethreaded, since they mutate global
316    //            state. They *also* need to share the same tokio runtime, which the
317    //            #[tokio::test] thing doesn't do (it makes a new runtime per test), because of
318    //            the need to have a background singleton EventReactor.
319    //
320    //            To hack around this, we just have a global runtime protected by a mutex!
321    static TEST_MUTEX: OnceLock<Mutex<Runtime>> = OnceLock::new();
322
323    /// Locks the mutex, and makes sure the event reactor is initialized.
324    fn test_setup() -> MutexGuard<'static, Runtime> {
325        let mutex = TEST_MUTEX.get_or_init(|| Mutex::new(Runtime::new().unwrap()));
326        let runtime = mutex
327            .lock()
328            .expect("mutex poisoned, probably by other failing tests");
329        if let Some(reactor) = EventReactor::new() {
330            runtime.handle().spawn(reactor.run());
331        }
332        runtime
333    }
334
335    #[test]
336    fn subscriptions() {
337        let rt = test_setup();
338
339        rt.block_on(async move {
340            // shouldn't have any subscribers at the start
341            assert!(!event_has_subscribers(TorEventKind::Empty));
342
343            let mut rx = EventReactor::receiver().unwrap();
344            // creating a receiver shouldn't result in any subscriptions
345            assert!(!event_has_subscribers(TorEventKind::Empty));
346
347            rx.subscribe(TorEventKind::Empty);
348            // subscription should work
349            assert!(event_has_subscribers(TorEventKind::Empty));
350
351            rx.unsubscribe(TorEventKind::Empty);
352            // unsubscribing should work
353            assert!(!event_has_subscribers(TorEventKind::Empty));
354
355            // subscription should be idempotent
356            rx.subscribe(TorEventKind::Empty);
357            rx.subscribe(TorEventKind::Empty);
358            rx.subscribe(TorEventKind::Empty);
359            assert!(event_has_subscribers(TorEventKind::Empty));
360
361            rx.unsubscribe(TorEventKind::Empty);
362            assert!(!event_has_subscribers(TorEventKind::Empty));
363
364            rx.subscribe(TorEventKind::Empty);
365            assert!(event_has_subscribers(TorEventKind::Empty));
366
367            std::mem::drop(rx);
368            // dropping the receiver should auto-unsubscribe
369            assert!(!event_has_subscribers(TorEventKind::Empty));
370        });
371    }
372
373    #[test]
374    fn empty_recv() {
375        let rt = test_setup();
376
377        rt.block_on(async move {
378            let mut rx = EventReactor::receiver().unwrap();
379            // attempting to read from a receiver with no subscriptions should return None
380            let result = rx.next().await;
381            assert!(result.is_none());
382        });
383    }
384
385    #[test]
386    fn receives_events() {
387        let rt = test_setup();
388
389        rt.block_on(async move {
390            let mut rx = EventReactor::receiver().unwrap();
391            rx.subscribe(TorEventKind::Empty);
392            // HACK(eta): give the event reactor time to run
393            tokio::time::sleep(Duration::from_millis(100)).await;
394            broadcast(TorEvent::Empty);
395
396            let result = rx.next().await;
397            assert_eq!(result, Some(TorEvent::Empty));
398        });
399    }
400
401    #[test]
402    fn does_not_send_to_no_subscribers() {
403        let rt = test_setup();
404
405        rt.block_on(async move {
406            // this event should just get dropped on the floor, because no subscribers exist
407            broadcast(TorEvent::Empty);
408
409            let mut rx = EventReactor::receiver().unwrap();
410            rx.subscribe(TorEventKind::Empty);
411
412            // this shouldn't have an event to receive now
413            let result = tokio::time::timeout(Duration::from_millis(100), rx.next()).await;
414            assert!(result.is_err());
415        });
416    }
417}
tor_events/lib.rs

tor_events/
lib.rs
