diff options
| -rw-r--r-- | crates/tor-chanmgr/src/lib.rs | 1 | ||||
| -rw-r--r-- | crates/tor-chanmgr/src/mgr.rs | 73 | ||||
| -rw-r--r-- | crates/tor-chanmgr/src/util.rs | 3 | ||||
| -rw-r--r-- | crates/tor-chanmgr/src/util/defer.rs | 75 |
4 files changed, 114 insertions, 38 deletions
diff --git a/crates/tor-chanmgr/src/lib.rs b/crates/tor-chanmgr/src/lib.rs index d9b3e0d04..03289918a 100644 --- a/crates/tor-chanmgr/src/lib.rs +++ b/crates/tor-chanmgr/src/lib.rs @@ -50,6 +50,7 @@ mod mgr; #[cfg(test)] mod testing; pub mod transport; +pub(crate) mod util; use futures::select_biased; use futures::task::SpawnExt; diff --git a/crates/tor-chanmgr/src/mgr.rs b/crates/tor-chanmgr/src/mgr.rs index 7daa30e4a..91bc1fbe0 100644 --- a/crates/tor-chanmgr/src/mgr.rs +++ b/crates/tor-chanmgr/src/mgr.rs @@ -1,6 +1,7 @@ //! Abstract implementation of a channel manager use crate::mgr::state::{ChannelForTarget, PendingChannelHandle}; +use crate::util::defer::Defer; use crate::{ChanProvenance, ChannelConfig, ChannelUsage, Dormancy, Error, Result}; use crate::factory::BootstrapReporter; @@ -257,50 +258,46 @@ impl<CF: AbstractChannelFactory + Clone> AbstractChanMgr<CF> { } // We need to launch a channel. Some(Action::Launch((handle, send))) => { - // WARNING: do not drop the handle without calling - // `upgrade_pending_channel_to_open` or `remove_pending_channel`. If you don't, - // then the pending entry will remain in the channel map forever, and arti will - // be unable to build new channels to the target relay of that pending channel. + // If the remainder of this code returns early or is cancelled, we still want to + // clean up our pending entry in the channel map. The following closure will be + // run when dropped to ensure that it's cleaned up properly. // - // This code is ugly since it tries to use IIFE-inspired immediately awaited - // async block expressions to prevent the code from returning early (for example - // due to a `?` operator). When modifying this code, be careful to not add a - // code path that returns early before properly passing `handle` back to - // `MgrState` (see `PendingChannelHandle` for details). - let outcome = async { - let chan_result = async { - let connector = self.channels.builder(); - let memquota = ChannelAccount::new(&self.memquota)?; - - connector - .build_channel(&target, self.reporter.clone(), memquota) - .await + // The `remove_pending_channel` will acquire the lock within `MgrState`, but + // this won't lead to deadlocks since the lock is only ever acquired within + // methods of `MgrState`. When this `Defer` is being dropped, no other + // `MgrState` methods will be running on this thread, so the lock will not have + // already been acquired. + let defer_remove_pending = Defer::new(handle, |handle| { + if let Err(e) = self.channels.remove_pending_channel(handle) { + // Just log an error if we're unable to remove it, since there's + // nothing else we can do here, and returning the error would + // hide the actual error that we care about (the channel build + // failure). + #[allow(clippy::missing_docs_in_private_items)] + const MSG: &str = "Unable to remove the pending channel"; + error_report!(internal!("{e}"), "{}", MSG); } + }); + + let connector = self.channels.builder(); + let memquota = ChannelAccount::new(&self.memquota)?; + + let outcome = connector + .build_channel(&target, self.reporter.clone(), memquota) .await; - match chan_result { - Ok(ref chan) => { - // Replace the pending channel with the newly built channel. - self.channels - .upgrade_pending_channel_to_open(handle, Arc::clone(chan))?; - } - Err(_) => { - // Remove the pending channel. - if let Err(e) = self.channels.remove_pending_channel(handle) { - // Just log an error if we're unable to remove it, since there's - // nothing else we can do here, and returning the error would - // hide the actual error that we care about (the channel build - // failure). - #[allow(clippy::missing_docs_in_private_items)] - const MSG: &str = "Unable to remove the pending channel"; - error_report!(internal!("{e}"), "{}", MSG); - } - } + match outcome { + Ok(ref chan) => { + // Replace the pending channel with the newly built channel. + let handle = defer_remove_pending.cancel(); + self.channels + .upgrade_pending_channel_to_open(handle, Arc::clone(chan))?; + } + Err(_) => { + // Remove the pending channel. + drop(defer_remove_pending); } - - chan_result } - .await; // It's okay if all the receivers went away: // that means that nobody was waiting for this channel. diff --git a/crates/tor-chanmgr/src/util.rs b/crates/tor-chanmgr/src/util.rs new file mode 100644 index 000000000..7e72e90b0 --- /dev/null +++ b/crates/tor-chanmgr/src/util.rs @@ -0,0 +1,3 @@ +//! Utilities used for the channel manager. + +pub(crate) mod defer; diff --git a/crates/tor-chanmgr/src/util/defer.rs b/crates/tor-chanmgr/src/util/defer.rs new file mode 100644 index 000000000..6b33c7447 --- /dev/null +++ b/crates/tor-chanmgr/src/util/defer.rs @@ -0,0 +1,75 @@ +//! Defer a closure until later. + +/// Runs a closure when dropped. +pub(crate) struct Defer<T, F: FnOnce(T)>(Option<DeferInner<T, F>>); + +/// Everything contained by a [`Defer`]. +struct DeferInner<T, F: FnOnce(T)> { + /// The argument `f` should be called with when [`Defer`] is dropped. + arg: T, + /// The function to call. + f: F, +} + +impl<T, F: FnOnce(T)> Defer<T, F> { + /// Defer running the provided closure `f` with `arg` until the returned [`Defer`] is dropped. + #[must_use] + pub(crate) fn new(arg: T, f: F) -> Self { + Self(Some(DeferInner { arg, f })) + } + + /// Return the provided `T` and drop the provided closure without running it. + pub(crate) fn cancel(mut self) -> T { + // other than the drop handler, there are no other places that mutate the `Option`, so it + // should always be `Some` here + self.0.take().expect("`Defer` is missing a value").arg + } +} + +impl<T, F: FnOnce(T)> std::ops::Drop for Defer<T, F> { + fn drop(&mut self) { + if let Some(DeferInner { arg, f }) = self.0.take() { + f(arg); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use std::sync::atomic::{AtomicU32, Ordering}; + + #[test] + fn test_drop() { + let x = AtomicU32::new(0); + { + let _defer = Defer::new(5, |inc| { + x.fetch_add(inc, Ordering::Relaxed); + }); + assert_eq!(x.load(Ordering::Relaxed), 0); + } + assert_eq!(x.load(Ordering::Relaxed), 5); + } + + #[test] + fn test_cancel() { + let x = AtomicU32::new(0); + { + let defer = Defer::new(5, |inc| { + x.fetch_add(inc, Ordering::Relaxed); + }); + assert_eq!(defer.cancel(), 5); + assert_eq!(x.load(Ordering::Relaxed), 0); + } + assert_eq!(x.load(Ordering::Relaxed), 0); + } + + #[test] + #[should_panic] + fn test_panic() { + let _ = Defer::new((), |()| { + panic!("intentional panic"); + }); + } +} |