Implement the ability to block on multiple futures at once

In the next commits we'll be adding a second notify pipeline - from
the `ChainMonitor` back to the background processor. This will
cause the `background-processor` to need to await multiple wakers
at once, which we cannot do in the current scheme without taking on
a full async runtime.

Building a multi-future waiter isn't so bad, and notably will allow
us to remove the existing sleep pipeline entirely, reducing the
complexity of our wakers implementation by only having one notify
path to consider.

Author: TheBlueMatt

lightning/src/util/wakers.rs

@@ -30,75 +30,22 @@ use core::pin::Pin;
 /// Used to signal to one of many waiters that the condition they're waiting on has happened.
 pub(crate) struct Notifier {
 	notify_pending: Mutex<(bool, Option<Arc<Mutex<FutureState>>>)>,
-	condvar: Condvar,
-}
-
-macro_rules! check_woken {
-	($guard: expr, $retval: expr) => { {
-		if $guard.0 {
-			$guard.0 = false;
-			if $guard.1.as_ref().map(|l| l.lock().unwrap().complete).unwrap_or(false) {
-				// If we're about to return as woken, and the future state is marked complete, wipe
-				// the future state and let the next future wait until we get a new notify.
-				$guard.1.take();
-			}
-			return $retval;
-		}
-	} }
 }
 
 impl Notifier {
 	pub(crate) fn new() -> Self {
 		Self {
 			notify_pending: Mutex::new((false, None)),
-			condvar: Condvar::new(),
-		}
-	}
-
-	fn propagate_future_state_to_notify_flag(&self) -> MutexGuard<(bool, Option<Arc<Mutex<FutureState>>>)> {
-		let mut lock = self.notify_pending.lock().unwrap();
-		if let Some(existing_state) = &lock.1 {
-			if existing_state.lock().unwrap().callbacks_made {
-				// If the existing `FutureState` has completed and actually made callbacks,
-				// consider the notification flag to have been cleared and reset the future state.
-				lock.1.take();
-				lock.0 = false;
-			}
 		}
-		lock
 	}
 
 	pub(crate) fn wait(&self) {
-		loop {
-			let mut guard = self.propagate_future_state_to_notify_flag();
-			check_woken!(guard, ());
-			guard = self.condvar.wait(guard).unwrap();
-			check_woken!(guard, ());
-		}
+		Sleeper::from_single_future(self.get_future()).wait();
 	}
 
 	#[cfg(any(test, feature = "std"))]
 	pub(crate) fn wait_timeout(&self, max_wait: Duration) -> bool {
-		let current_time = Instant::now();
-		loop {
-			let mut guard = self.propagate_future_state_to_notify_flag();
-			check_woken!(guard, true);
-			guard = self.condvar.wait_timeout(guard, max_wait).unwrap().0;
-			check_woken!(guard, true);
-			// Due to spurious wakeups that can happen on `wait_timeout`, here we need to check if the
-			// desired wait time has actually passed, and if not then restart the loop with a reduced wait
-			// time. Note that this logic can be highly simplified through the use of
-			// `Condvar::wait_while` and `Condvar::wait_timeout_while`, if and when our MSRV is raised to
-			// 1.42.0.
-			let elapsed = current_time.elapsed();
-			if elapsed >= max_wait {
-				return false;
-			}
-			match max_wait.checked_sub(elapsed) {
-				None => return false,
-				Some(_) => continue
-			}
-		}
+		Sleeper::from_single_future(self.get_future()).wait_timeout(max_wait)
 	}
 
 	/// Wake waiters, tracking that wake needs to occur even if there are currently no waiters.
@@ -111,13 +58,19 @@ impl Notifier {
 			}
 		}
 		lock.0 = true;
-		mem::drop(lock);
-		self.condvar.notify_all();
 	}
 
 	/// Gets a [`Future`] that will get woken up with any waiters
 	pub(crate) fn get_future(&self) -> Future {
-		let mut lock = self.propagate_future_state_to_notify_flag();
+		let mut lock = self.notify_pending.lock().unwrap();
+		if let Some(existing_state) = &lock.1 {
+			if existing_state.lock().unwrap().callbacks_made {
+				// If the existing `FutureState` has completed and actually made callbacks,
+				// consider the notification flag to have been cleared and reset the future state.
+				lock.1.take();
+				lock.0 = false;
+			}
+		}
 		if let Some(existing_state) = &lock.1 {
 			Future { state: Arc::clone(&existing_state) }
 		} else {
@@ -182,6 +135,9 @@ impl FutureState {
 }
 
 /// A simple future which can complete once, and calls some callback(s) when it does so.
+///
+/// Clones can be made and all futures cloned from the same source will complete at the same time.
+#[derive(Clone)]
 pub struct Future {
 	state: Arc<Mutex<FutureState>>,
 }
@@ -236,6 +192,86 @@ impl<'a> StdFuture for Future {
 	}
 }
 
+/// A struct which can be used to await multiple [`Future`]s at once without relying on a full
+/// async context.
+pub struct Sleeper {
+	notifiers: Vec<Arc<Mutex<FutureState>>>,
+}
+
+impl Sleeper {
+	/// Constructs a new sleeper from one future, allowing blocking on it.
+	pub fn from_single_future(future: Future) -> Self {
+		Self { notifiers: vec![future.state] }
+	}
+	/// Constructs a new sleeper from two futures, allowing blocking on both at once.
+	// Note that this is the common case - a ChannelManager and ChainMonitor.
+	pub fn from_two_futures(fut_a: Future, fut_b: Future) -> Self {
+		Self { notifiers: vec![fut_a.state, fut_b.state] }
+	}
+	/// Constructs a new sleeper on many futures, allowing blocking on all at once.
+	pub fn new(futures: Vec<Future>) -> Self {
+		Self { notifiers: futures.into_iter().map(|f| f.state).collect() }
+	}
+	fn setup_wait(&self) -> (Arc<Condvar>, Arc<Mutex<Option<Arc<Mutex<FutureState>>>>>) {
+		let cv = Arc::new(Condvar::new());
+		let notified_fut_mtx = Arc::new(Mutex::new(None));
+		{
+			for notifier_mtx in self.notifiers.iter() {
+				let cv_ref = Arc::clone(&cv);
+				let notified_ref = Arc::clone(&notified_fut_mtx);
+				let notifier_ref = Arc::clone(&notifier_mtx);
+				let mut notifier = notifier_mtx.lock().unwrap();
+				if notifier.complete {
+					*notified_fut_mtx.lock().unwrap() = Some(notifier_ref);
+					break;
+				}
+				notifier.callbacks.push((false, Box::new(move || {
+					*notified_ref.lock().unwrap() = Some(Arc::clone(&notifier_ref));
+					cv_ref.notify_all();
+				})));
+			}
+		}
+		(cv, notified_fut_mtx)
+	}
+
+	/// Wait until one of the [`Future`]s registered with this [`Sleeper`] has completed.
+	pub fn wait(&self) {
+		let (cv, notified_fut_mtx) = self.setup_wait();
+		let notified_fut = {
+			let mut notified_fut_lck = notified_fut_mtx.lock().unwrap();
+			loop {
+				if let Some(notified_fut) = notified_fut_lck.take() {
+					break notified_fut;
+				}
+				notified_fut_lck = cv.wait(notified_fut_lck).unwrap();
+			}
+		};
+		notified_fut.lock().unwrap().callbacks_made = true;
+	}
+
+	/// Wait until one of the [`Future`]s registered with this [`Sleeper`] has completed or the
+	/// given amount of time has elapsed. Returns true if a [`Future`] completed, false if the time
+	/// elapsed.
+	#[cfg(any(test, feature = "std"))]
+	pub fn wait_timeout(&self, max_wait: Duration) -> bool {
+		let start_time = Instant::now();
+		let (cv, notified_fut_mtx) = self.setup_wait();
+		let notified_fut = {
+			let mut notified_fut_lck = notified_fut_mtx.lock().unwrap();
+			loop {
+				if let Some(notified_fut) = notified_fut_lck.take() {
+					break notified_fut;
+				}
+				let sleep_time = max_wait.saturating_sub(start_time.elapsed());
+				if sleep_time == Duration::from_secs(0) { return false; }
+				notified_fut_lck = cv.wait_timeout(notified_fut_lck, max_wait).unwrap().0;
+			}
+		};
+		notified_fut.lock().unwrap().callbacks_made = true;
+		true
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;
@@ -334,10 +370,7 @@ mod tests {
 		let exit_thread_clone = exit_thread.clone();
 		thread::spawn(move || {
 			loop {
-				let mut lock = thread_notifier.notify_pending.lock().unwrap();
-				lock.0 = true;
-				thread_notifier.condvar.notify_all();
-
+				thread_notifier.notify();
 				if exit_thread_clone.load(Ordering::SeqCst) {
 					break
 				}