Add a Future which can receive manager persistence events

This allows users who don't wish to block a full thread to receive
persistence events.

The `Future` added here is really just a trivial list of callbacks,
but from that we can build a (somewhat ineffecient)
std::future::Future implementation and can (at least once a mapping
for Box<dyn Trait> is added) include the future in no-std bindings
as well.

Fixes #1595

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -55,7 +55,7 @@ use util::config::{UserConfig, ChannelConfig};
 use util::events::{EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider, ClosureReason, HTLCDestination};
 use util::{byte_utils, events};
 use util::crypto::sign;
-use util::wakers::PersistenceNotifier;
+use util::wakers::{Future, PersistenceNotifier};
 use util::scid_utils::fake_scid;
 use util::ser::{BigSize, FixedLengthReader, Readable, ReadableArgs, MaybeReadable, Writeable, Writer, VecWriter};
 use util::logger::{Level, Logger};
@@ -5988,6 +5988,13 @@ where
 		self.persistence_notifier.wait()
 	}
 
+	/// Gets a [`Future`] which completes when a persistable update is available. Note that
+	/// callbacks registered on the [`Future`] MUST NOT call back into this [`ChannelManager`] and
+	/// should instead register actions to be taken later.
+	pub fn get_persistable_update_future(&self) -> Future {
+		self.persistence_notifier.get_future()
+	}
+
 	#[cfg(any(test, feature = "_test_utils"))]
 	pub fn get_persistence_condvar_value(&self) -> bool {
 		self.persistence_notifier.needs_persist()

lightning/src/util/wakers.rs

@@ -13,40 +13,47 @@
 //!
 //! [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
 
+use alloc::sync::Arc;
 use core::mem;
 use core::time::Duration;
 use sync::{Condvar, Mutex};
 
 #[cfg(any(test, feature = "std"))]
 use std::time::Instant;
 
+#[cfg(feature = "std")]
+use std::future::Future as StdFuture;
+#[cfg(feature = "std")]
+use std::task::{Context, Poll};
+#[cfg(feature = "std")]
+use std::pin::Pin;
+
 /// Used to signal to the ChannelManager persister that the manager needs to be re-persisted to
 /// disk/backups, through `await_persistable_update_timeout` and `await_persistable_update`.
 pub(crate) struct PersistenceNotifier {
-	/// Users won't access the persistence_lock directly, but rather wait on its bool using
-	/// `wait_timeout` and `wait`.
-	persistence_lock: (Mutex<bool>, Condvar),
+	persist_pending: Mutex<(bool, Option<Arc<Mutex<FutureState>>>)>,
+	persistence_condvar: Condvar,
 }
 
 impl PersistenceNotifier {
 	pub(crate) fn new() -> Self {
 		Self {
-			persistence_lock: (Mutex::new(false), Condvar::new()),
+			persist_pending: Mutex::new((false, None)),
+			persistence_condvar: Condvar::new(),
 		}
 	}
 
 	pub(crate) fn wait(&self) {
 		loop {
-			let &(ref mtx, ref cvar) = &self.persistence_lock;
-			let mut guard = mtx.lock().unwrap();
-			if *guard {
-				*guard = false;
+			let mut guard = self.persist_pending.lock().unwrap();
+			if guard.0 {
+				guard.0 = false;
 				return;
 			}
-			guard = cvar.wait(guard).unwrap();
-			let result = *guard;
+			guard = self.persistence_condvar.wait(guard).unwrap();
+			let result = guard.0;
 			if result {
-				*guard = false;
+				guard.0 = false;
 				return
 			}
 		}
@@ -56,22 +63,21 @@ impl PersistenceNotifier {
 	pub(crate) fn wait_timeout(&self, max_wait: Duration) -> bool {
 		let current_time = Instant::now();
 		loop {
-			let &(ref mtx, ref cvar) = &self.persistence_lock;
-			let mut guard = mtx.lock().unwrap();
-			if *guard {
-				*guard = false;
+			let mut guard = self.persist_pending.lock().unwrap();
+			if guard.0 {
+				guard.0 = false;
 				return true;
 			}
-			guard = cvar.wait_timeout(guard, max_wait).unwrap().0;
+			guard = self.persistence_condvar.wait_timeout(guard, max_wait).unwrap().0;
 			// 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();
-			let result = *guard;
+			let result = guard.0;
 			if result || elapsed >= max_wait {
-				*guard = false;
+				guard.0 = false;
 				return result;
 			}
 			match max_wait.checked_sub(elapsed) {
@@ -83,18 +89,112 @@ impl PersistenceNotifier {
 
 	/// Wake waiters, tracking that persistence needs to occur.
 	pub(crate) fn notify(&self) {
-		let &(ref persist_mtx, ref cnd) = &self.persistence_lock;
-		let mut persistence_lock = persist_mtx.lock().unwrap();
-		*persistence_lock = true;
+		let mut persistence_lock = self.persist_pending.lock().unwrap();
+		persistence_lock.0 = true;
+		if let Some(future_state) = persistence_lock.1.take() {
+			future_state.lock().unwrap().complete();
+		}
 		mem::drop(persistence_lock);
-		cnd.notify_all();
+		self.persistence_condvar.notify_all();
+	}
+
+	/// Gets a [`Future`] that will get woken up with any waiters
+	pub(crate) fn get_future(&self) -> Future {
+		let mut persistence_lock = self.persist_pending.lock().unwrap();
+		if persistence_lock.0 {
+			Future {
+				state: Arc::new(Mutex::new(FutureState {
+					callbacks: Vec::new(),
+					complete: false,
+				}))
+			}
+		} else if let Some(existing_state) = &persistence_lock.1 {
+			Future { state: Arc::clone(&existing_state) }
+		} else {
+			let state = Arc::new(Mutex::new(FutureState {
+				callbacks: Vec::new(),
+				complete: false,
+			}));
+			persistence_lock.1 = Some(Arc::clone(&state));
+			Future { state }
+		}
 	}
 
 	#[cfg(any(test, feature = "_test_utils"))]
 	pub fn needs_persist(&self) -> bool {
-		let &(ref mtx, _) = &self.persistence_lock;
-		let guard = mtx.lock().unwrap();
-		*guard
+		self.persist_pending.lock().unwrap().0
+	}
+}
+
+/// A callback which is called when a [`Future`] completes.
+///
+/// Note that this MUST NOT call back into LDK directly, it must instead schedule actions to be
+/// taken later. Rust users should use the [`std::future::Future`] implementation for [`Future`]
+/// instead.
+///
+/// Note that the [`std::future::Future`] implementation may only work for runtimes which schedule
+/// futures when they receive a wake, rather than immediately executing them.
+pub trait FutureCallback : Send {
+	/// The method which is called.
+	fn call(&self);
+}
+
+pub struct FutureState {
+	callbacks: Vec<Box<dyn FutureCallback>>,
+	complete: bool,
+}
+
+impl FutureState {
+	fn complete(&mut self) {
+		for callback in self.callbacks.drain(..) {
+			callback.call();
+		}
+		self.complete = true;
+	}
+}
+
+/// A simple future which can complete once, and calls some callback(s) when it does so.
+pub struct Future {
+	state: Arc<Mutex<FutureState>>,
+}
+
+impl Future {
+	/// Registers a callback to be called upon completion of this future. If the future has already
+	/// completed, the callback will be called immediately.
+	pub fn register_callback(&self, callback: Box<dyn FutureCallback>) {
+		let mut state = self.state.lock().unwrap();
+		if state.complete {
+			mem::drop(state);
+			callback.call();
+		} else {
+			state.callbacks.push(callback);
+		}
+	}
+}
+
+#[cfg(feature = "std")]
+mod std_future {
+	use std::task::Waker;
+	pub struct StdWaker(pub Waker);
+	impl super::FutureCallback for StdWaker {
+		fn call(&self) { self.0.wake_by_ref() }
+	}
+}
+
+#[cfg(feature = "std")]
+/// (C-not exported) as Rust Futures aren't usable in language bindings.
+impl<'a> StdFuture for Future {
+	type Output = ();
+
+	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+		let mut state = self.state.lock().unwrap();
+		if state.complete {
+			Poll::Ready(())
+		} else {
+			let waker = cx.waker().clone();
+			state.callbacks.push(Box::new(std_future::StdWaker(waker)));
+			Poll::Pending
+		}
 	}
 }
 
@@ -115,10 +215,9 @@ mod tests {
 		let exit_thread_clone = exit_thread.clone();
 		thread::spawn(move || {
 			loop {
-				let &(ref persist_mtx, ref cnd) = &thread_notifier.persistence_lock;
-				let mut persistence_lock = persist_mtx.lock().unwrap();
-				*persistence_lock = true;
-				cnd.notify_all();
+				let mut persistence_lock = thread_notifier.persist_pending.lock().unwrap();
+				persistence_lock.0 = true;
+				thread_notifier.persistence_condvar.notify_all();
 
 				if exit_thread_clone.load(Ordering::SeqCst) {
 					break