1 // This file is Copyright its original authors, visible in version control
4 // This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE
5 // or http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
6 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
7 // You may not use this file except in accordance with one or both of these
10 //! Utilities which allow users to block on some future notification from LDK. These are
11 //! specifically used by [`ChannelManager`] to allow waiting until the [`ChannelManager`] needs to
14 //! [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
18 use crate::sync::{Condvar, Mutex, MutexGuard};
20 use crate::prelude::*;
22 #[cfg(any(test, feature = "std"))]
23 use std::time::{Duration, Instant};
25 use core::future::Future as StdFuture;
26 use core::task::{Context, Poll};
30 /// Used to signal to one of many waiters that the condition they're waiting on has happened.
31 pub(crate) struct Notifier {
32 notify_pending: Mutex<(bool, Option<Arc<Mutex<FutureState>>>)>,
36 macro_rules! check_woken {
37 ($guard: expr, $retval: expr) => { {
40 if $guard.1.as_ref().map(|l| l.lock().unwrap().complete).unwrap_or(false) {
41 // If we're about to return as woken, and the future state is marked complete, wipe
42 // the future state and let the next future wait until we get a new notify.
51 pub(crate) fn new() -> Self {
53 notify_pending: Mutex::new((false, None)),
54 condvar: Condvar::new(),
58 fn propagate_future_state_to_notify_flag(&self) -> MutexGuard<(bool, Option<Arc<Mutex<FutureState>>>)> {
59 let mut lock = self.notify_pending.lock().unwrap();
60 if let Some(existing_state) = &lock.1 {
61 if existing_state.lock().unwrap().callbacks_made {
62 // If the existing `FutureState` has completed and actually made callbacks,
63 // consider the notification flag to have been cleared and reset the future state.
71 pub(crate) fn wait(&self) {
73 let mut guard = self.propagate_future_state_to_notify_flag();
74 check_woken!(guard, ());
75 guard = self.condvar.wait(guard).unwrap();
76 check_woken!(guard, ());
80 #[cfg(any(test, feature = "std"))]
81 pub(crate) fn wait_timeout(&self, max_wait: Duration) -> bool {
82 let current_time = Instant::now();
84 let mut guard = self.propagate_future_state_to_notify_flag();
85 check_woken!(guard, true);
86 guard = self.condvar.wait_timeout(guard, max_wait).unwrap().0;
87 check_woken!(guard, true);
88 // Due to spurious wakeups that can happen on `wait_timeout`, here we need to check if the
89 // desired wait time has actually passed, and if not then restart the loop with a reduced wait
90 // time. Note that this logic can be highly simplified through the use of
91 // `Condvar::wait_while` and `Condvar::wait_timeout_while`, if and when our MSRV is raised to
93 let elapsed = current_time.elapsed();
94 if elapsed >= max_wait {
97 match max_wait.checked_sub(elapsed) {
104 /// Wake waiters, tracking that wake needs to occur even if there are currently no waiters.
105 pub(crate) fn notify(&self) {
106 let mut lock = self.notify_pending.lock().unwrap();
107 if let Some(future_state) = &lock.1 {
108 future_state.lock().unwrap().complete();
112 self.condvar.notify_all();
115 /// Gets a [`Future`] that will get woken up with any waiters
116 pub(crate) fn get_future(&self) -> Future {
117 let mut lock = self.propagate_future_state_to_notify_flag();
118 if let Some(existing_state) = &lock.1 {
119 Future { state: Arc::clone(&existing_state) }
121 let state = Arc::new(Mutex::new(FutureState {
122 callbacks: Vec::new(),
124 callbacks_made: false,
126 lock.1 = Some(Arc::clone(&state));
131 #[cfg(any(test, feature = "_test_utils"))]
132 pub fn notify_pending(&self) -> bool {
133 self.notify_pending.lock().unwrap().0
137 /// A callback which is called when a [`Future`] completes.
139 /// Note that this MUST NOT call back into LDK directly, it must instead schedule actions to be
140 /// taken later. Rust users should use the [`std::future::Future`] implementation for [`Future`]
143 /// Note that the [`std::future::Future`] implementation may only work for runtimes which schedule
144 /// futures when they receive a wake, rather than immediately executing them.
145 pub trait FutureCallback : Send {
146 /// The method which is called.
150 impl<F: Fn() + Send> FutureCallback for F {
151 fn call(&self) { (self)(); }
154 pub(crate) struct FutureState {
155 callbacks: Vec<Box<dyn FutureCallback>>,
157 callbacks_made: bool,
161 fn complete(&mut self) {
162 for callback in self.callbacks.drain(..) {
164 self.callbacks_made = true;
166 self.complete = true;
170 /// A simple future which can complete once, and calls some callback(s) when it does so.
172 state: Arc<Mutex<FutureState>>,
176 /// Registers a callback to be called upon completion of this future. If the future has already
177 /// completed, the callback will be called immediately.
179 /// (C-not exported) use the bindings-only `register_callback_fn` instead
180 pub fn register_callback(&self, callback: Box<dyn FutureCallback>) {
181 let mut state = self.state.lock().unwrap();
183 state.callbacks_made = true;
187 state.callbacks.push(callback);
191 // C bindings don't (currently) know how to map `Box<dyn Trait>`, and while it could add the
192 // following wrapper, doing it in the bindings is currently much more work than simply doing it
194 /// Registers a callback to be called upon completion of this future. If the future has already
195 /// completed, the callback will be called immediately.
197 pub fn register_callback_fn<F: 'static + FutureCallback>(&self, callback: F) {
198 self.register_callback(Box::new(callback));
202 use core::task::Waker;
203 struct StdWaker(pub Waker);
204 impl FutureCallback for StdWaker {
205 fn call(&self) { self.0.wake_by_ref() }
208 /// (C-not exported) as Rust Futures aren't usable in language bindings.
209 impl<'a> StdFuture for Future {
212 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
213 let mut state = self.state.lock().unwrap();
217 let waker = cx.waker().clone();
218 state.callbacks.push(Box::new(StdWaker(waker)));
227 use core::sync::atomic::{AtomicBool, Ordering};
228 use core::future::Future as FutureTrait;
229 use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
232 fn notifier_pre_notified_future() {
233 // Previously, if we generated a future after a `Notifier` had been notified, the future
234 // would never complete. This tests this behavior, ensuring the future instead completes
236 let notifier = Notifier::new();
239 let callback = Arc::new(AtomicBool::new(false));
240 let callback_ref = Arc::clone(&callback);
241 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
242 assert!(callback.load(Ordering::SeqCst));
246 fn notifier_future_completes_wake() {
247 // Previously, if we were only using the `Future` interface to learn when a `Notifier` has
248 // been notified, we'd never mark the notifier as not-awaiting-notify. This caused the
249 // `lightning-background-processor` to persist in a tight loop.
250 let notifier = Notifier::new();
252 // First check the simple case, ensuring if we get notified a new future isn't woken until
253 // a second `notify`.
254 let callback = Arc::new(AtomicBool::new(false));
255 let callback_ref = Arc::clone(&callback);
256 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
257 assert!(!callback.load(Ordering::SeqCst));
260 assert!(callback.load(Ordering::SeqCst));
262 let callback = Arc::new(AtomicBool::new(false));
263 let callback_ref = Arc::clone(&callback);
264 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
265 assert!(!callback.load(Ordering::SeqCst));
268 assert!(callback.load(Ordering::SeqCst));
270 // Then check the case where the future is fetched before the notification, but a callback
271 // is only registered after the `notify`, ensuring that it is still sufficient to ensure we
272 // don't get an instant-wake when we get a new future.
273 let future = notifier.get_future();
276 let callback = Arc::new(AtomicBool::new(false));
277 let callback_ref = Arc::clone(&callback);
278 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
279 assert!(callback.load(Ordering::SeqCst));
281 let callback = Arc::new(AtomicBool::new(false));
282 let callback_ref = Arc::clone(&callback);
283 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
284 assert!(!callback.load(Ordering::SeqCst));
288 fn new_future_wipes_notify_bit() {
289 // Previously, if we were only using the `Future` interface to learn when a `Notifier` has
290 // been notified, we'd never mark the notifier as not-awaiting-notify if a `Future` is
291 // fetched after the notify bit has been set.
292 let notifier = Notifier::new();
295 let callback = Arc::new(AtomicBool::new(false));
296 let callback_ref = Arc::clone(&callback);
297 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
298 assert!(callback.load(Ordering::SeqCst));
300 let callback = Arc::new(AtomicBool::new(false));
301 let callback_ref = Arc::clone(&callback);
302 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
303 assert!(!callback.load(Ordering::SeqCst));
306 assert!(callback.load(Ordering::SeqCst));
309 #[cfg(feature = "std")]
311 fn test_wait_timeout() {
312 use crate::sync::Arc;
315 let persistence_notifier = Arc::new(Notifier::new());
316 let thread_notifier = Arc::clone(&persistence_notifier);
318 let exit_thread = Arc::new(AtomicBool::new(false));
319 let exit_thread_clone = exit_thread.clone();
320 thread::spawn(move || {
322 let mut lock = thread_notifier.notify_pending.lock().unwrap();
324 thread_notifier.condvar.notify_all();
326 if exit_thread_clone.load(Ordering::SeqCst) {
332 // Check that we can block indefinitely until updates are available.
333 let _ = persistence_notifier.wait();
335 // Check that the Notifier will return after the given duration if updates are
338 if persistence_notifier.wait_timeout(Duration::from_millis(100)) {
343 exit_thread.store(true, Ordering::SeqCst);
345 // Check that the Notifier will return after the given duration even if no updates
348 if !persistence_notifier.wait_timeout(Duration::from_millis(100)) {
355 fn test_future_callbacks() {
356 let future = Future {
357 state: Arc::new(Mutex::new(FutureState {
358 callbacks: Vec::new(),
360 callbacks_made: false,
363 let callback = Arc::new(AtomicBool::new(false));
364 let callback_ref = Arc::clone(&callback);
365 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
367 assert!(!callback.load(Ordering::SeqCst));
368 future.state.lock().unwrap().complete();
369 assert!(callback.load(Ordering::SeqCst));
370 future.state.lock().unwrap().complete();
374 fn test_pre_completed_future_callbacks() {
375 let future = Future {
376 state: Arc::new(Mutex::new(FutureState {
377 callbacks: Vec::new(),
379 callbacks_made: false,
382 future.state.lock().unwrap().complete();
384 let callback = Arc::new(AtomicBool::new(false));
385 let callback_ref = Arc::clone(&callback);
386 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
388 assert!(callback.load(Ordering::SeqCst));
389 assert!(future.state.lock().unwrap().callbacks.is_empty());
392 // Rather annoyingly, there's no safe way in Rust std to construct a Waker despite it being
393 // totally possible to construct from a trait implementation (though somewhat less effecient
394 // compared to a raw VTable). Instead, we have to write out a lot of boilerplate to build a
395 // waker, which we do here with a trivial Arc<AtomicBool> data element to track woke-ness.
396 const WAKER_V_TABLE: RawWakerVTable = RawWakerVTable::new(waker_clone, wake, wake_by_ref, drop);
397 unsafe fn wake_by_ref(ptr: *const ()) { let p = ptr as *const Arc<AtomicBool>; assert!(!(*p).fetch_or(true, Ordering::SeqCst)); }
398 unsafe fn drop(ptr: *const ()) { let p = ptr as *mut Arc<AtomicBool>; let _freed = Box::from_raw(p); }
399 unsafe fn wake(ptr: *const ()) { wake_by_ref(ptr); drop(ptr); }
400 unsafe fn waker_clone(ptr: *const ()) -> RawWaker {
401 let p = ptr as *const Arc<AtomicBool>;
402 RawWaker::new(Box::into_raw(Box::new(Arc::clone(&*p))) as *const (), &WAKER_V_TABLE)
405 fn create_waker() -> (Arc<AtomicBool>, Waker) {
406 let a = Arc::new(AtomicBool::new(false));
407 let waker = unsafe { Waker::from_raw(waker_clone((&a as *const Arc<AtomicBool>) as *const ())) };
413 let mut future = Future {
414 state: Arc::new(Mutex::new(FutureState {
415 callbacks: Vec::new(),
417 callbacks_made: false,
420 let mut second_future = Future { state: Arc::clone(&future.state) };
422 let (woken, waker) = create_waker();
423 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
424 assert!(!woken.load(Ordering::SeqCst));
426 let (second_woken, second_waker) = create_waker();
427 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Pending);
428 assert!(!second_woken.load(Ordering::SeqCst));
430 future.state.lock().unwrap().complete();
431 assert!(woken.load(Ordering::SeqCst));
432 assert!(second_woken.load(Ordering::SeqCst));
433 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
434 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Ready(()));