From c6890cfc3317c0b1ba4d1e6cd71b3bb75386571e Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Tue, 9 Aug 2022 04:15:21 +0000 Subject: [PATCH] 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 is added) include the future in no-std bindings as well. Fixes #1595 --- lightning/src/ln/channelmanager.rs | 9 +- lightning/src/util/wakers.rs | 245 +++++++++++++++++++++++++---- 2 files changed, 222 insertions(+), 32 deletions(-) diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index 911db0db0..467db740f 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/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::Notifier; +use util::wakers::{Future, Notifier}; 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`] that 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.notify_pending() diff --git a/lightning/src/util/wakers.rs b/lightning/src/util/wakers.rs index 9aeb5370b..9636466aa 100644 --- a/lightning/src/util/wakers.rs +++ b/lightning/src/util/wakers.rs @@ -13,6 +13,7 @@ //! //! [`ChannelManager`]: crate::ln::channelmanager::ChannelManager +use alloc::sync::Arc; use core::mem; use core::time::Duration; use sync::{Condvar, Mutex}; @@ -20,32 +21,37 @@ use sync::{Condvar, Mutex}; #[cfg(any(test, feature = "std"))] use std::time::Instant; +use core::future::Future as StdFuture; +use core::task::{Context, Poll}; +use core::pin::Pin; + +use prelude::*; + /// Used to signal to one of many waiters that the condition they're waiting on has happened. pub(crate) struct Notifier { - /// Users won't access the lock directly, but rather wait on its bool using - /// `wait_timeout` and `wait`. - lock: (Mutex, Condvar), + notify_pending: Mutex<(bool, Option>>)>, + condvar: Condvar, } impl Notifier { pub(crate) fn new() -> Self { Self { - lock: (Mutex::new(false), Condvar::new()), + notify_pending: Mutex::new((false, None)), + condvar: Condvar::new(), } } pub(crate) fn wait(&self) { loop { - let &(ref mtx, ref cvar) = &self.lock; - let mut guard = mtx.lock().unwrap(); - if *guard { - *guard = false; + let mut guard = self.notify_pending.lock().unwrap(); + if guard.0 { + guard.0 = false; return; } - guard = cvar.wait(guard).unwrap(); - let result = *guard; + guard = self.condvar.wait(guard).unwrap(); + let result = guard.0; if result { - *guard = false; + guard.0 = false; return } } @@ -55,22 +61,21 @@ impl Notifier { pub(crate) fn wait_timeout(&self, max_wait: Duration) -> bool { let current_time = Instant::now(); loop { - let &(ref mtx, ref cvar) = &self.lock; - let mut guard = mtx.lock().unwrap(); - if *guard { - *guard = false; + let mut guard = self.notify_pending.lock().unwrap(); + if guard.0 { + guard.0 = false; return true; } - guard = cvar.wait_timeout(guard, max_wait).unwrap().0; + guard = self.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) { @@ -82,29 +87,128 @@ impl Notifier { /// Wake waiters, tracking that wake needs to occur even if there are currently no waiters. pub(crate) fn notify(&self) { - let &(ref persist_mtx, ref cnd) = &self.lock; - let mut lock = persist_mtx.lock().unwrap(); - *lock = true; + let mut lock = self.notify_pending.lock().unwrap(); + lock.0 = true; + if let Some(future_state) = lock.1.take() { + future_state.lock().unwrap().complete(); + } mem::drop(lock); - cnd.notify_all(); + 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.notify_pending.lock().unwrap(); + if lock.0 { + Future { + state: Arc::new(Mutex::new(FutureState { + callbacks: Vec::new(), + complete: false, + })) + } + } else if let Some(existing_state) = &lock.1 { + Future { state: Arc::clone(&existing_state) } + } else { + let state = Arc::new(Mutex::new(FutureState { + callbacks: Vec::new(), + complete: false, + })); + lock.1 = Some(Arc::clone(&state)); + Future { state } + } } #[cfg(any(test, feature = "_test_utils"))] pub fn notify_pending(&self) -> bool { - let &(ref mtx, _) = &self.lock; - let guard = mtx.lock().unwrap(); - *guard + self.notify_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); +} + +impl FutureCallback for F { + fn call(&self) { (self)(); } +} + +pub(crate) struct FutureState { + callbacks: Vec>, + 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>, +} + +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) { + let mut state = self.state.lock().unwrap(); + if state.complete { + mem::drop(state); + callback.call(); + } else { + state.callbacks.push(callback); + } + } +} + +mod std_future { + use core::task::Waker; + pub struct StdWaker(pub Waker); + impl super::FutureCallback for StdWaker { + fn call(&self) { self.0.wake_by_ref() } + } +} + +/// (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 { + 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 + } } } #[cfg(test)] mod tests { + use super::*; + use core::sync::atomic::{AtomicBool, Ordering}; + use core::future::Future as FutureTrait; + use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker}; + #[cfg(feature = "std")] #[test] fn test_wait_timeout() { - use super::*; use sync::Arc; - use core::sync::atomic::{AtomicBool, Ordering}; use std::thread; let persistence_notifier = Arc::new(Notifier::new()); @@ -114,10 +218,9 @@ mod tests { let exit_thread_clone = exit_thread.clone(); thread::spawn(move || { loop { - let &(ref persist_mtx, ref cnd) = &thread_notifier.lock; - let mut lock = persist_mtx.lock().unwrap(); - *lock = true; - cnd.notify_all(); + let mut lock = thread_notifier.notify_pending.lock().unwrap(); + lock.0 = true; + thread_notifier.condvar.notify_all(); if exit_thread_clone.load(Ordering::SeqCst) { break @@ -146,4 +249,84 @@ mod tests { } } } + + #[test] + fn test_future_callbacks() { + let future = Future { + state: Arc::new(Mutex::new(FutureState { + callbacks: Vec::new(), + complete: false, + })) + }; + let callback = Arc::new(AtomicBool::new(false)); + let callback_ref = Arc::clone(&callback); + future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst)))); + + assert!(!callback.load(Ordering::SeqCst)); + future.state.lock().unwrap().complete(); + assert!(callback.load(Ordering::SeqCst)); + future.state.lock().unwrap().complete(); + } + + #[test] + fn test_pre_completed_future_callbacks() { + let future = Future { + state: Arc::new(Mutex::new(FutureState { + callbacks: Vec::new(), + complete: false, + })) + }; + future.state.lock().unwrap().complete(); + + let callback = Arc::new(AtomicBool::new(false)); + let callback_ref = Arc::clone(&callback); + future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst)))); + + assert!(callback.load(Ordering::SeqCst)); + assert!(future.state.lock().unwrap().callbacks.is_empty()); + } + + // Rather annoyingly, there's no safe way in Rust std to construct a Waker despite it being + // totally possible to construct from a trait implementation (though somewhat less effecient + // compared to a raw VTable). Instead, we have to write out a lot of boilerplate to build a + // waker, which we do here with a trivial Arc data element to track woke-ness. + const WAKER_V_TABLE: RawWakerVTable = RawWakerVTable::new(waker_clone, wake, wake_by_ref, drop); + unsafe fn wake_by_ref(ptr: *const ()) { let p = ptr as *const Arc; assert!(!(*p).fetch_or(true, Ordering::SeqCst)); } + unsafe fn drop(ptr: *const ()) { let p = ptr as *mut Arc; Box::from_raw(p); } + unsafe fn wake(ptr: *const ()) { wake_by_ref(ptr); drop(ptr); } + unsafe fn waker_clone(ptr: *const ()) -> RawWaker { + let p = ptr as *const Arc; + RawWaker::new(Box::into_raw(Box::new(Arc::clone(&*p))) as *const (), &WAKER_V_TABLE) + } + + fn create_waker() -> (Arc, Waker) { + let a = Arc::new(AtomicBool::new(false)); + let waker = unsafe { Waker::from_raw(waker_clone((&a as *const Arc) as *const ())) }; + (a, waker) + } + + #[test] + fn test_future() { + let mut future = Future { + state: Arc::new(Mutex::new(FutureState { + callbacks: Vec::new(), + complete: false, + })) + }; + let mut second_future = Future { state: Arc::clone(&future.state) }; + + let (woken, waker) = create_waker(); + assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending); + assert!(!woken.load(Ordering::SeqCst)); + + let (second_woken, second_waker) = create_waker(); + assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Pending); + assert!(!second_woken.load(Ordering::SeqCst)); + + future.state.lock().unwrap().complete(); + assert!(woken.load(Ordering::SeqCst)); + assert!(second_woken.load(Ordering::SeqCst)); + assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(())); + assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Ready(())); + } } -- 2.39.5