1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
//! Implement background tasks used by guard managers.
//!
//! These background tasks keep a weak reference to the [`GuardMgrInner`]
//! and use that to notice when they should shut down.
use crate::pending::{GuardStatus, RequestId};
use crate::GuardMgrInner;
use futures::{
channel::{mpsc, oneshot},
stream::{self, StreamExt},
};
use std::sync::{Mutex, Weak};
/// A message sent by to the [`report_status_events()`] task.
#[derive(Debug)]
pub(crate) enum Msg {
/// Tells the task to add another [`oneshot::Receiver`] to the list
/// of receivers it's listening to.
///
/// This message is sent by guard manager whenever it hands out a
/// guard; the receiver will be notified when the requester's circuit
/// succeeds, fails, or is abandoned. The receiver corresponds
/// to the sender in some [`GuardMonitor`](crate::GuardMonitor).
Observe(oneshot::Receiver<Msg>),
/// A message sent by a [`GuardMonitor`](crate::GuardMonitor) to
/// report the status of an attempt to use a guard.
Status(RequestId, GuardStatus),
/// Tells the task to reply on the provided oneshot::Sender once
/// it has seen this message. Used to indicate that the message
/// queue is flushed.
#[cfg(test)]
Ping(oneshot::Sender<()>),
}
/// Wrapper type to unify returns from mpsc and oneshots
pub(crate) type MsgResult = Result<Msg, futures::channel::oneshot::Canceled>;
/// Background task: wait for messages about guard statuses, and
/// tell a guard manager about them. Runs indefinitely.
///
/// Takes the [`GuardMgrInner`] by weak reference; if the guard
/// manager goes away, then this task exits.
///
/// Requires a `mpsc::Receiver` that is used to tell the task about
/// new status events to wait for.
pub(crate) async fn report_status_events(
runtime: impl tor_rtcompat::SleepProvider,
inner: Weak<Mutex<GuardMgrInner>>,
ctrl: mpsc::UnboundedReceiver<MsgResult>,
) {
// Multiplexes a bunch of one-shot receivers to tell us about guard
// status outcomes.
let notifications = stream::FuturesUnordered::new();
// If I don't put this dummy receiver into notifications, then
// notifications will be finished prematurely and not get polled any more.
// TODO: Is there a better way to do this?
let (_dummy_snd, rcv) = oneshot::channel();
notifications.push(rcv);
// Multiplexes `notifications` with events from `ctrl`.
let mut events = stream::select(notifications, ctrl);
loop {
match events.next().await {
Some(Ok(Msg::Observe(rcv))) => {
// We've been told to wait for a new event; add it to
// `notifications`.
events.get_ref().0.push(rcv);
}
Some(Ok(Msg::Status(id, status))) => {
// We've got a report about a guard status.
if let Some(inner) = inner.upgrade() {
let mut inner = inner.lock().expect("Poisoned lock");
inner.handle_msg(id, status, &runtime);
} else {
// The guard manager has gone away.
return;
}
}
Some(Err(_)) => {
// TODO: Unfortunately, we don't know which future was cancelled.
// It shouldn't be possible for this to occur, though, since
// GuardMonitor always sends a message, even on drop.
tracing::warn!("bug: Somehow a guard success event was dropped.");
}
#[cfg(test)]
Some(Ok(Msg::Ping(sender))) => {
let _ignore = sender.send(());
}
// The streams have all closed. (I think this is impossible?)
None => return,
}
// TODO: Is this task guaranteed to exit?
}
}
/// Background task to run periodic events on the guard manager.
///
/// The only role of this task is to invoke
/// [`GuardMgrInner::run_periodic_events`] from time to time, so that
/// it can perform housekeeping tasks.
///
/// Takes the [`GuardMgrInner`] by weak reference; if the guard
/// manager goes away, then this task exits.
pub(crate) async fn run_periodic<R: tor_rtcompat::SleepProvider>(
runtime: R,
inner: Weak<Mutex<GuardMgrInner>>,
) {
loop {
let delay = if let Some(inner) = inner.upgrade() {
let mut inner = inner.lock().expect("Poisoned lock");
let wallclock = runtime.wallclock();
let now = runtime.now();
inner.run_periodic_events(wallclock, now)
} else {
// The guard manager has gone away.
return;
};
runtime.sleep(delay).await;
}
}
|
