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 core::time::Duration;
19 use sync::{Condvar, Mutex};
23 #[cfg(any(test, feature = "std"))]
24 use std::time::Instant;
26 use core::future::Future as StdFuture;
27 use core::task::{Context, Poll};
31 /// Used to signal to one of many waiters that the condition they're waiting on has happened.
32 pub(crate) struct Notifier {
33 notify_pending: Mutex<(bool, Option<Arc<Mutex<FutureState>>>)>,
38 pub(crate) fn new() -> Self {
40 notify_pending: Mutex::new((false, None)),
41 condvar: Condvar::new(),
45 pub(crate) fn wait(&self) {
47 let mut guard = self.notify_pending.lock().unwrap();
52 guard = self.condvar.wait(guard).unwrap();
61 #[cfg(any(test, feature = "std"))]
62 pub(crate) fn wait_timeout(&self, max_wait: Duration) -> bool {
63 let current_time = Instant::now();
65 let mut guard = self.notify_pending.lock().unwrap();
70 guard = self.condvar.wait_timeout(guard, max_wait).unwrap().0;
71 // Due to spurious wakeups that can happen on `wait_timeout`, here we need to check if the
72 // desired wait time has actually passed, and if not then restart the loop with a reduced wait
73 // time. Note that this logic can be highly simplified through the use of
74 // `Condvar::wait_while` and `Condvar::wait_timeout_while`, if and when our MSRV is raised to
76 let elapsed = current_time.elapsed();
78 if result || elapsed >= max_wait {
82 match max_wait.checked_sub(elapsed) {
83 None => return result,
89 /// Wake waiters, tracking that wake needs to occur even if there are currently no waiters.
90 pub(crate) fn notify(&self) {
91 let mut lock = self.notify_pending.lock().unwrap();
93 if let Some(future_state) = lock.1.take() {
94 future_state.lock().unwrap().complete();
97 self.condvar.notify_all();
100 /// Gets a [`Future`] that will get woken up with any waiters
101 pub(crate) fn get_future(&self) -> Future {
102 let mut lock = self.notify_pending.lock().unwrap();
105 state: Arc::new(Mutex::new(FutureState {
106 callbacks: Vec::new(),
110 } else if let Some(existing_state) = &lock.1 {
111 Future { state: Arc::clone(&existing_state) }
113 let state = Arc::new(Mutex::new(FutureState {
114 callbacks: Vec::new(),
117 lock.1 = Some(Arc::clone(&state));
122 #[cfg(any(test, feature = "_test_utils"))]
123 pub fn notify_pending(&self) -> bool {
124 self.notify_pending.lock().unwrap().0
128 /// A callback which is called when a [`Future`] completes.
130 /// Note that this MUST NOT call back into LDK directly, it must instead schedule actions to be
131 /// taken later. Rust users should use the [`std::future::Future`] implementation for [`Future`]
134 /// Note that the [`std::future::Future`] implementation may only work for runtimes which schedule
135 /// futures when they receive a wake, rather than immediately executing them.
136 pub trait FutureCallback : Send {
137 /// The method which is called.
141 impl<F: Fn() + Send> FutureCallback for F {
142 fn call(&self) { (self)(); }
145 pub(crate) struct FutureState {
146 callbacks: Vec<Box<dyn FutureCallback>>,
151 fn complete(&mut self) {
152 for callback in self.callbacks.drain(..) {
155 self.complete = true;
159 /// A simple future which can complete once, and calls some callback(s) when it does so.
161 state: Arc<Mutex<FutureState>>,
165 /// Registers a callback to be called upon completion of this future. If the future has already
166 /// completed, the callback will be called immediately.
167 pub fn register_callback(&self, callback: Box<dyn FutureCallback>) {
168 let mut state = self.state.lock().unwrap();
173 state.callbacks.push(callback);
179 use core::task::Waker;
180 pub struct StdWaker(pub Waker);
181 impl super::FutureCallback for StdWaker {
182 fn call(&self) { self.0.wake_by_ref() }
186 /// (C-not exported) as Rust Futures aren't usable in language bindings.
187 impl<'a> StdFuture for Future {
190 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
191 let mut state = self.state.lock().unwrap();
195 let waker = cx.waker().clone();
196 state.callbacks.push(Box::new(std_future::StdWaker(waker)));
205 use core::sync::atomic::{AtomicBool, Ordering};
206 use core::future::Future as FutureTrait;
207 use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
209 #[cfg(feature = "std")]
211 fn test_wait_timeout() {
215 let persistence_notifier = Arc::new(Notifier::new());
216 let thread_notifier = Arc::clone(&persistence_notifier);
218 let exit_thread = Arc::new(AtomicBool::new(false));
219 let exit_thread_clone = exit_thread.clone();
220 thread::spawn(move || {
222 let mut lock = thread_notifier.notify_pending.lock().unwrap();
224 thread_notifier.condvar.notify_all();
226 if exit_thread_clone.load(Ordering::SeqCst) {
232 // Check that we can block indefinitely until updates are available.
233 let _ = persistence_notifier.wait();
235 // Check that the Notifier will return after the given duration if updates are
238 if persistence_notifier.wait_timeout(Duration::from_millis(100)) {
243 exit_thread.store(true, Ordering::SeqCst);
245 // Check that the Notifier will return after the given duration even if no updates
248 if !persistence_notifier.wait_timeout(Duration::from_millis(100)) {
255 fn test_future_callbacks() {
256 let future = Future {
257 state: Arc::new(Mutex::new(FutureState {
258 callbacks: Vec::new(),
262 let callback = Arc::new(AtomicBool::new(false));
263 let callback_ref = Arc::clone(&callback);
264 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
266 assert!(!callback.load(Ordering::SeqCst));
267 future.state.lock().unwrap().complete();
268 assert!(callback.load(Ordering::SeqCst));
269 future.state.lock().unwrap().complete();
273 fn test_pre_completed_future_callbacks() {
274 let future = Future {
275 state: Arc::new(Mutex::new(FutureState {
276 callbacks: Vec::new(),
280 future.state.lock().unwrap().complete();
282 let callback = Arc::new(AtomicBool::new(false));
283 let callback_ref = Arc::clone(&callback);
284 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
286 assert!(callback.load(Ordering::SeqCst));
287 assert!(future.state.lock().unwrap().callbacks.is_empty());
290 // Rather annoyingly, there's no safe way in Rust std to construct a Waker despite it being
291 // totally possible to construct from a trait implementation (though somewhat less effecient
292 // compared to a raw VTable). Instead, we have to write out a lot of boilerplate to build a
293 // waker, which we do here with a trivial Arc<AtomicBool> data element to track woke-ness.
294 const WAKER_V_TABLE: RawWakerVTable = RawWakerVTable::new(waker_clone, wake, wake_by_ref, drop);
295 unsafe fn wake_by_ref(ptr: *const ()) { let p = ptr as *const Arc<AtomicBool>; assert!(!(*p).fetch_or(true, Ordering::SeqCst)); }
296 unsafe fn drop(ptr: *const ()) { let p = ptr as *mut Arc<AtomicBool>; let _freed = Box::from_raw(p); }
297 unsafe fn wake(ptr: *const ()) { wake_by_ref(ptr); drop(ptr); }
298 unsafe fn waker_clone(ptr: *const ()) -> RawWaker {
299 let p = ptr as *const Arc<AtomicBool>;
300 RawWaker::new(Box::into_raw(Box::new(Arc::clone(&*p))) as *const (), &WAKER_V_TABLE)
303 fn create_waker() -> (Arc<AtomicBool>, Waker) {
304 let a = Arc::new(AtomicBool::new(false));
305 let waker = unsafe { Waker::from_raw(waker_clone((&a as *const Arc<AtomicBool>) as *const ())) };
311 let mut future = Future {
312 state: Arc::new(Mutex::new(FutureState {
313 callbacks: Vec::new(),
317 let mut second_future = Future { state: Arc::clone(&future.state) };
319 let (woken, waker) = create_waker();
320 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
321 assert!(!woken.load(Ordering::SeqCst));
323 let (second_woken, second_waker) = create_waker();
324 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Pending);
325 assert!(!second_woken.load(Ordering::SeqCst));
327 future.state.lock().unwrap().complete();
328 assert!(woken.load(Ordering::SeqCst));
329 assert!(second_woken.load(Ordering::SeqCst));
330 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
331 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Ready(()));