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 if future_state.lock().unwrap().complete() {
115 self.condvar.notify_all();
118 /// Gets a [`Future`] that will get woken up with any waiters
119 pub(crate) fn get_future(&self) -> Future {
120 let mut lock = self.propagate_future_state_to_notify_flag();
121 if let Some(existing_state) = &lock.1 {
122 Future { state: Arc::clone(&existing_state) }
124 let state = Arc::new(Mutex::new(FutureState {
125 callbacks: Vec::new(),
127 callbacks_made: false,
129 lock.1 = Some(Arc::clone(&state));
134 #[cfg(any(test, feature = "_test_utils"))]
135 pub fn notify_pending(&self) -> bool {
136 self.notify_pending.lock().unwrap().0
140 /// A callback which is called when a [`Future`] completes.
142 /// Note that this MUST NOT call back into LDK directly, it must instead schedule actions to be
143 /// taken later. Rust users should use the [`std::future::Future`] implementation for [`Future`]
146 /// Note that the [`std::future::Future`] implementation may only work for runtimes which schedule
147 /// futures when they receive a wake, rather than immediately executing them.
148 pub trait FutureCallback : Send {
149 /// The method which is called.
153 impl<F: Fn() + Send> FutureCallback for F {
154 fn call(&self) { (self)(); }
157 pub(crate) struct FutureState {
158 // When we're tracking whether a callback counts as having woken the user's code, we check the
159 // first bool - set to false if we're just calling a Waker, and true if we're calling an actual
160 // user-provided function.
161 callbacks: Vec<(bool, Box<dyn FutureCallback>)>,
163 callbacks_made: bool,
167 fn complete(&mut self) -> bool {
168 for (counts_as_call, callback) in self.callbacks.drain(..) {
170 self.callbacks_made |= counts_as_call;
172 self.complete = true;
177 /// A simple future which can complete once, and calls some callback(s) when it does so.
179 state: Arc<Mutex<FutureState>>,
183 /// Registers a callback to be called upon completion of this future. If the future has already
184 /// completed, the callback will be called immediately.
186 /// (C-not exported) use the bindings-only `register_callback_fn` instead
187 pub fn register_callback(&self, callback: Box<dyn FutureCallback>) {
188 let mut state = self.state.lock().unwrap();
190 state.callbacks_made = true;
194 state.callbacks.push((true, callback));
198 // C bindings don't (currently) know how to map `Box<dyn Trait>`, and while it could add the
199 // following wrapper, doing it in the bindings is currently much more work than simply doing it
201 /// Registers a callback to be called upon completion of this future. If the future has already
202 /// completed, the callback will be called immediately.
204 pub fn register_callback_fn<F: 'static + FutureCallback>(&self, callback: F) {
205 self.register_callback(Box::new(callback));
209 use core::task::Waker;
210 struct StdWaker(pub Waker);
211 impl FutureCallback for StdWaker {
212 fn call(&self) { self.0.wake_by_ref() }
215 /// (C-not exported) as Rust Futures aren't usable in language bindings.
216 impl<'a> StdFuture for Future {
219 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
220 let mut state = self.state.lock().unwrap();
222 state.callbacks_made = true;
225 let waker = cx.waker().clone();
226 state.callbacks.push((false, Box::new(StdWaker(waker))));
235 use core::sync::atomic::{AtomicBool, Ordering};
236 use core::future::Future as FutureTrait;
237 use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
240 fn notifier_pre_notified_future() {
241 // Previously, if we generated a future after a `Notifier` had been notified, the future
242 // would never complete. This tests this behavior, ensuring the future instead completes
244 let notifier = Notifier::new();
247 let callback = Arc::new(AtomicBool::new(false));
248 let callback_ref = Arc::clone(&callback);
249 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
250 assert!(callback.load(Ordering::SeqCst));
254 fn notifier_future_completes_wake() {
255 // Previously, if we were only using the `Future` interface to learn when a `Notifier` has
256 // been notified, we'd never mark the notifier as not-awaiting-notify. This caused the
257 // `lightning-background-processor` to persist in a tight loop.
258 let notifier = Notifier::new();
260 // First check the simple case, ensuring if we get notified a new future isn't woken until
261 // a second `notify`.
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 let callback = Arc::new(AtomicBool::new(false));
271 let callback_ref = Arc::clone(&callback);
272 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
273 assert!(!callback.load(Ordering::SeqCst));
276 assert!(callback.load(Ordering::SeqCst));
278 // Then check the case where the future is fetched before the notification, but a callback
279 // is only registered after the `notify`, ensuring that it is still sufficient to ensure we
280 // don't get an instant-wake when we get a new future.
281 let future = notifier.get_future();
284 let callback = Arc::new(AtomicBool::new(false));
285 let callback_ref = Arc::clone(&callback);
286 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
287 assert!(callback.load(Ordering::SeqCst));
289 let callback = Arc::new(AtomicBool::new(false));
290 let callback_ref = Arc::clone(&callback);
291 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
292 assert!(!callback.load(Ordering::SeqCst));
296 fn new_future_wipes_notify_bit() {
297 // Previously, if we were only using the `Future` interface to learn when a `Notifier` has
298 // been notified, we'd never mark the notifier as not-awaiting-notify if a `Future` is
299 // fetched after the notify bit has been set.
300 let notifier = Notifier::new();
303 let callback = Arc::new(AtomicBool::new(false));
304 let callback_ref = Arc::clone(&callback);
305 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
306 assert!(callback.load(Ordering::SeqCst));
308 let callback = Arc::new(AtomicBool::new(false));
309 let callback_ref = Arc::clone(&callback);
310 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
311 assert!(!callback.load(Ordering::SeqCst));
314 assert!(callback.load(Ordering::SeqCst));
317 #[cfg(feature = "std")]
319 fn test_wait_timeout() {
320 use crate::sync::Arc;
323 let persistence_notifier = Arc::new(Notifier::new());
324 let thread_notifier = Arc::clone(&persistence_notifier);
326 let exit_thread = Arc::new(AtomicBool::new(false));
327 let exit_thread_clone = exit_thread.clone();
328 thread::spawn(move || {
330 let mut lock = thread_notifier.notify_pending.lock().unwrap();
332 thread_notifier.condvar.notify_all();
334 if exit_thread_clone.load(Ordering::SeqCst) {
340 // Check that we can block indefinitely until updates are available.
341 let _ = persistence_notifier.wait();
343 // Check that the Notifier will return after the given duration if updates are
346 if persistence_notifier.wait_timeout(Duration::from_millis(100)) {
351 exit_thread.store(true, Ordering::SeqCst);
353 // Check that the Notifier will return after the given duration even if no updates
356 if !persistence_notifier.wait_timeout(Duration::from_millis(100)) {
363 fn test_future_callbacks() {
364 let future = Future {
365 state: Arc::new(Mutex::new(FutureState {
366 callbacks: Vec::new(),
368 callbacks_made: false,
371 let callback = Arc::new(AtomicBool::new(false));
372 let callback_ref = Arc::clone(&callback);
373 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
375 assert!(!callback.load(Ordering::SeqCst));
376 future.state.lock().unwrap().complete();
377 assert!(callback.load(Ordering::SeqCst));
378 future.state.lock().unwrap().complete();
382 fn test_pre_completed_future_callbacks() {
383 let future = Future {
384 state: Arc::new(Mutex::new(FutureState {
385 callbacks: Vec::new(),
387 callbacks_made: false,
390 future.state.lock().unwrap().complete();
392 let callback = Arc::new(AtomicBool::new(false));
393 let callback_ref = Arc::clone(&callback);
394 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
396 assert!(callback.load(Ordering::SeqCst));
397 assert!(future.state.lock().unwrap().callbacks.is_empty());
400 // Rather annoyingly, there's no safe way in Rust std to construct a Waker despite it being
401 // totally possible to construct from a trait implementation (though somewhat less effecient
402 // compared to a raw VTable). Instead, we have to write out a lot of boilerplate to build a
403 // waker, which we do here with a trivial Arc<AtomicBool> data element to track woke-ness.
404 const WAKER_V_TABLE: RawWakerVTable = RawWakerVTable::new(waker_clone, wake, wake_by_ref, drop);
405 unsafe fn wake_by_ref(ptr: *const ()) { let p = ptr as *const Arc<AtomicBool>; assert!(!(*p).fetch_or(true, Ordering::SeqCst)); }
406 unsafe fn drop(ptr: *const ()) { let p = ptr as *mut Arc<AtomicBool>; let _freed = Box::from_raw(p); }
407 unsafe fn wake(ptr: *const ()) { wake_by_ref(ptr); drop(ptr); }
408 unsafe fn waker_clone(ptr: *const ()) -> RawWaker {
409 let p = ptr as *const Arc<AtomicBool>;
410 RawWaker::new(Box::into_raw(Box::new(Arc::clone(&*p))) as *const (), &WAKER_V_TABLE)
413 fn create_waker() -> (Arc<AtomicBool>, Waker) {
414 let a = Arc::new(AtomicBool::new(false));
415 let waker = unsafe { Waker::from_raw(waker_clone((&a as *const Arc<AtomicBool>) as *const ())) };
421 let mut future = Future {
422 state: Arc::new(Mutex::new(FutureState {
423 callbacks: Vec::new(),
425 callbacks_made: false,
428 let mut second_future = Future { state: Arc::clone(&future.state) };
430 let (woken, waker) = create_waker();
431 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
432 assert!(!woken.load(Ordering::SeqCst));
434 let (second_woken, second_waker) = create_waker();
435 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Pending);
436 assert!(!second_woken.load(Ordering::SeqCst));
438 future.state.lock().unwrap().complete();
439 assert!(woken.load(Ordering::SeqCst));
440 assert!(second_woken.load(Ordering::SeqCst));
441 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
442 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Ready(()));
446 fn test_dropped_future_doesnt_count() {
447 // Tests that if a Future gets drop'd before it is poll()ed `Ready` it doesn't count as
448 // having been woken, leaving the notify-required flag set.
449 let notifier = Notifier::new();
452 // If we get a future and don't touch it we're definitely still notify-required.
453 notifier.get_future();
454 assert!(notifier.wait_timeout(Duration::from_millis(1)));
455 assert!(!notifier.wait_timeout(Duration::from_millis(1)));
457 // Even if we poll'd once but didn't observe a `Ready`, we should be notify-required.
458 let mut future = notifier.get_future();
459 let (woken, waker) = create_waker();
460 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
463 assert!(woken.load(Ordering::SeqCst));
464 assert!(notifier.wait_timeout(Duration::from_millis(1)));
466 // However, once we do poll `Ready` it should wipe the notify-required flag.
467 let mut future = notifier.get_future();
468 let (woken, waker) = create_waker();
469 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
472 assert!(woken.load(Ordering::SeqCst));
473 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
474 assert!(!notifier.wait_timeout(Duration::from_millis(1)));
478 fn test_poll_post_notify_completes() {
479 // Tests that if we have a future state that has completed, and we haven't yet requested a
480 // new future, if we get a notify prior to requesting that second future it is generated
482 let notifier = Notifier::new();
485 let mut future = notifier.get_future();
486 let (woken, waker) = create_waker();
487 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
488 assert!(!woken.load(Ordering::SeqCst));
491 let mut future = notifier.get_future();
492 let (woken, waker) = create_waker();
493 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
494 assert!(!woken.load(Ordering::SeqCst));
496 let mut future = notifier.get_future();
497 let (woken, waker) = create_waker();
498 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
499 assert!(!woken.load(Ordering::SeqCst));
502 assert!(woken.load(Ordering::SeqCst));
503 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
507 fn test_poll_post_notify_completes_initial_notified() {
508 // Identical to the previous test, but the first future completes via a wake rather than an
509 // immediate `Poll::Ready`.
510 let notifier = Notifier::new();
512 let mut future = notifier.get_future();
513 let (woken, waker) = create_waker();
514 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
517 assert!(woken.load(Ordering::SeqCst));
518 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
521 let mut future = notifier.get_future();
522 let (woken, waker) = create_waker();
523 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
524 assert!(!woken.load(Ordering::SeqCst));
526 let mut future = notifier.get_future();
527 let (woken, waker) = create_waker();
528 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
529 assert!(!woken.load(Ordering::SeqCst));
532 assert!(woken.load(Ordering::SeqCst));
533 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));