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

            
47
pub mod events;
48

            
49
use crate::events::{TorEvent, TorEventKind};
50
use async_broadcast::{InactiveReceiver, Receiver, Sender, TrySendError};
51
use futures::channel::mpsc;
52
use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
53
use futures::future::Either;
54
use futures::StreamExt;
55
use std::pin::Pin;
56
use std::sync::atomic::{AtomicUsize, Ordering};
57
use std::sync::OnceLock;
58
use std::task::{Context, Poll};
59
use thiserror::Error;
60
use tracing::{error, warn};
61

            
62
/// Pointer to an `UnboundedSender`, used to send events into the `EventReactor`.
63
static EVENT_SENDER: OnceLock<UnboundedSender<TorEvent>> = OnceLock::new();
64
/// An inactive receiver for the currently active broadcast channel, if there is one.
65
static CURRENT_RECEIVER: OnceLock<InactiveReceiver<TorEvent>> = OnceLock::new();
66
/// The number of `TorEventKind`s there are.
67
const 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.
74
static 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.
77
pub 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.
85
pub 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

            
96
impl 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
8
    pub fn new() -> Option<Self> {
106
8
        let (tx, rx) = mpsc::unbounded();
107
8
        if EVENT_SENDER.set(tx).is_ok() {
108
2
            let (btx, brx) = async_broadcast::broadcast(BROADCAST_CAPACITY);
109
2
            CURRENT_RECEIVER
110
2
                .set(brx.deactivate())
111
2
                .expect("CURRENT_RECEIVER can't be set if EVENT_SENDER is unset!");
112
2
            Some(Self {
113
2
                receiver: rx,
114
2
                broadcast: btx,
115
2
            })
116
        } else {
117
6
            None
118
        }
119
8
    }
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
8
    pub fn receiver() -> Option<TorEventReceiver> {
125
8
        CURRENT_RECEIVER
126
8
            .get()
127
12
            .map(|rx| TorEventReceiver::wrap(rx.clone()))
128
8
    }
129
    /// Run the event forwarding reactor.
130
    ///
131
    /// You *must* call this function once a reactor is created.
132
3
    pub async fn run(mut self) {
133
4
        while let Some(event) = self.receiver.next().await {
134
2
            match self.broadcast.try_broadcast(event) {
135
2
                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]
159
pub 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)]
179
pub 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

            
188
impl futures::stream::Stream for TorEventReceiver {
189
    type Item = TorEvent;
190

            
191
10
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
192
10
        let this = self.get_mut();
193
10
        match this.inner {
194
8
            Either::Left(ref mut active) => loop {
195
8
                match Pin::new(&mut *active).poll_next(cx) {
196
2
                    Poll::Ready(Some(e)) => {
197
2
                        if this.subscribed[e.kind() as usize] {
198
2
                            return Poll::Ready(Some(e));
199
                        }
200
                        // loop, since we weren't subscribed to that event
201
                    }
202
6
                    x => return x,
203
                }
204
            },
205
            Either::Right(_) => {
206
2
                warn!("TorEventReceiver::poll_next() called without subscriptions!");
207
2
                Poll::Ready(None)
208
            }
209
        }
210
10
    }
211
}
212

            
213
impl TorEventReceiver {
214
    /// Create a `TorEventReceiver` from an `InactiveReceiver` handle.
215
8
    pub(crate) fn wrap(rx: InactiveReceiver<TorEvent>) -> Self {
216
8
        Self {
217
8
            inner: Either::Right(rx),
218
8
            subscribed: [false; EVENT_KIND_COUNT],
219
8
        }
220
8
    }
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
14
    pub fn subscribe(&mut self, kind: TorEventKind) {
226
14
        if !self.subscribed[kind as usize] {
227
10
            EVENT_SUBSCRIBERS[kind as usize].fetch_add(1, Ordering::SeqCst);
228
10
            self.subscribed[kind as usize] = true;
229
10
        }
230
        // FIXME(eta): cloning is ungood, but hard to avoid
231
14
        if let Either::Right(inactive) = self.inner.clone() {
232
10
            self.inner = Either::Left(inactive.activate());
233
10
        }
234
14
    }
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
4
    pub fn unsubscribe(&mut self, kind: TorEventKind) {
241
4
        if self.subscribed[kind as usize] {
242
4
            EVENT_SUBSCRIBERS[kind as usize].fetch_sub(1, Ordering::SeqCst);
243
4
            self.subscribed[kind as usize] = false;
244
4
        }
245
        // If we're now not subscribed to anything, deactivate our channel.
246
6
        if self.subscribed.iter().all(|x| !*x) {
247
            // FIXME(eta): cloning is ungood, but hard to avoid
248
4
            if let Either::Left(active) = self.inner.clone() {
249
4
                self.inner = Either::Right(active.deactivate());
250
4
            }
251
        }
252
4
    }
253
}
254

            
255
impl Drop for TorEventReceiver {
256
8
    fn drop(&mut self) {
257
8
        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
8
            if *subscribed {
261
6
                EVENT_SUBSCRIBERS[i].fetch_sub(1, Ordering::SeqCst);
262
6
            }
263
        }
264
8
    }
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.
272
20
pub fn event_has_subscribers(kind: TorEventKind) -> bool {
273
20
    EVENT_SUBSCRIBERS[kind as usize].load(Ordering::SeqCst) > 0
274
20
}
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!).
283
4
pub fn broadcast(event: TorEvent) {
284
4
    if !event_has_subscribers(event.kind()) {
285
2
        return;
286
2
    }
287
2
    if let Some(sender) = EVENT_SENDER.get() {
288
2
        // If this fails, there isn't much we can really do about it!
289
2
        let _ = sender.unbounded_send(event);
290
2
    }
291
4
}
292

            
293
#[cfg(test)]
294
mod 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
}