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 sync::{Condvar, Mutex};
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>>>)>,
37 pub(crate) fn new() -> Self {
39 notify_pending: Mutex::new((false, None)),
40 condvar: Condvar::new(),
44 pub(crate) fn wait(&self) {
46 let mut guard = self.notify_pending.lock().unwrap();
51 guard = self.condvar.wait(guard).unwrap();
60 #[cfg(any(test, feature = "std"))]
61 pub(crate) fn wait_timeout(&self, max_wait: Duration) -> bool {
62 let current_time = Instant::now();
64 let mut guard = self.notify_pending.lock().unwrap();
69 guard = self.condvar.wait_timeout(guard, max_wait).unwrap().0;
70 // Due to spurious wakeups that can happen on `wait_timeout`, here we need to check if the
71 // desired wait time has actually passed, and if not then restart the loop with a reduced wait
72 // time. Note that this logic can be highly simplified through the use of
73 // `Condvar::wait_while` and `Condvar::wait_timeout_while`, if and when our MSRV is raised to
75 let elapsed = current_time.elapsed();
77 if result || elapsed >= max_wait {
81 match max_wait.checked_sub(elapsed) {
82 None => return result,
88 /// Wake waiters, tracking that wake needs to occur even if there are currently no waiters.
89 pub(crate) fn notify(&self) {
90 let mut lock = self.notify_pending.lock().unwrap();
92 if let Some(future_state) = lock.1.take() {
93 future_state.lock().unwrap().complete();
96 self.condvar.notify_all();
99 /// Gets a [`Future`] that will get woken up with any waiters
100 pub(crate) fn get_future(&self) -> Future {
101 let mut lock = self.notify_pending.lock().unwrap();
104 state: Arc::new(Mutex::new(FutureState {
105 callbacks: Vec::new(),
109 } else if let Some(existing_state) = &lock.1 {
110 Future { state: Arc::clone(&existing_state) }
112 let state = Arc::new(Mutex::new(FutureState {
113 callbacks: Vec::new(),
116 lock.1 = Some(Arc::clone(&state));
121 #[cfg(any(test, feature = "_test_utils"))]
122 pub fn notify_pending(&self) -> bool {
123 self.notify_pending.lock().unwrap().0
127 /// A callback which is called when a [`Future`] completes.
129 /// Note that this MUST NOT call back into LDK directly, it must instead schedule actions to be
130 /// taken later. Rust users should use the [`std::future::Future`] implementation for [`Future`]
133 /// Note that the [`std::future::Future`] implementation may only work for runtimes which schedule
134 /// futures when they receive a wake, rather than immediately executing them.
135 pub trait FutureCallback : Send {
136 /// The method which is called.
140 impl<F: Fn() + Send> FutureCallback for F {
141 fn call(&self) { (self)(); }
144 pub(crate) struct FutureState {
145 callbacks: Vec<Box<dyn FutureCallback>>,
150 fn complete(&mut self) {
151 for callback in self.callbacks.drain(..) {
154 self.complete = true;
158 /// A simple future which can complete once, and calls some callback(s) when it does so.
160 state: Arc<Mutex<FutureState>>,
164 /// Registers a callback to be called upon completion of this future. If the future has already
165 /// completed, the callback will be called immediately.
166 pub fn register_callback(&self, callback: Box<dyn FutureCallback>) {
167 let mut state = self.state.lock().unwrap();
172 state.callbacks.push(callback);
178 use core::task::Waker;
179 pub struct StdWaker(pub Waker);
180 impl super::FutureCallback for StdWaker {
181 fn call(&self) { self.0.wake_by_ref() }
185 /// (C-not exported) as Rust Futures aren't usable in language bindings.
186 impl<'a> StdFuture for Future {
189 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
190 let mut state = self.state.lock().unwrap();
194 let waker = cx.waker().clone();
195 state.callbacks.push(Box::new(std_future::StdWaker(waker)));
204 use core::sync::atomic::{AtomicBool, Ordering};
205 use core::future::Future as FutureTrait;
206 use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
208 #[cfg(feature = "std")]
210 fn test_wait_timeout() {
214 let persistence_notifier = Arc::new(Notifier::new());
215 let thread_notifier = Arc::clone(&persistence_notifier);
217 let exit_thread = Arc::new(AtomicBool::new(false));
218 let exit_thread_clone = exit_thread.clone();
219 thread::spawn(move || {
221 let mut lock = thread_notifier.notify_pending.lock().unwrap();
223 thread_notifier.condvar.notify_all();
225 if exit_thread_clone.load(Ordering::SeqCst) {
231 // Check that we can block indefinitely until updates are available.
232 let _ = persistence_notifier.wait();
234 // Check that the Notifier will return after the given duration if updates are
237 if persistence_notifier.wait_timeout(Duration::from_millis(100)) {
242 exit_thread.store(true, Ordering::SeqCst);
244 // Check that the Notifier will return after the given duration even if no updates
247 if !persistence_notifier.wait_timeout(Duration::from_millis(100)) {
254 fn test_future_callbacks() {
255 let future = Future {
256 state: Arc::new(Mutex::new(FutureState {
257 callbacks: Vec::new(),
261 let callback = Arc::new(AtomicBool::new(false));
262 let callback_ref = Arc::clone(&callback);
263 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
265 assert!(!callback.load(Ordering::SeqCst));
266 future.state.lock().unwrap().complete();
267 assert!(callback.load(Ordering::SeqCst));
268 future.state.lock().unwrap().complete();
272 fn test_pre_completed_future_callbacks() {
273 let future = Future {
274 state: Arc::new(Mutex::new(FutureState {
275 callbacks: Vec::new(),
279 future.state.lock().unwrap().complete();
281 let callback = Arc::new(AtomicBool::new(false));
282 let callback_ref = Arc::clone(&callback);
283 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
285 assert!(callback.load(Ordering::SeqCst));
286 assert!(future.state.lock().unwrap().callbacks.is_empty());
289 // Rather annoyingly, there's no safe way in Rust std to construct a Waker despite it being
290 // totally possible to construct from a trait implementation (though somewhat less effecient
291 // compared to a raw VTable). Instead, we have to write out a lot of boilerplate to build a
292 // waker, which we do here with a trivial Arc<AtomicBool> data element to track woke-ness.
293 const WAKER_V_TABLE: RawWakerVTable = RawWakerVTable::new(waker_clone, wake, wake_by_ref, drop);
294 unsafe fn wake_by_ref(ptr: *const ()) { let p = ptr as *const Arc<AtomicBool>; assert!(!(*p).fetch_or(true, Ordering::SeqCst)); }
295 unsafe fn drop(ptr: *const ()) { let p = ptr as *mut Arc<AtomicBool>; let _freed = Box::from_raw(p); }
296 unsafe fn wake(ptr: *const ()) { wake_by_ref(ptr); drop(ptr); }
297 unsafe fn waker_clone(ptr: *const ()) -> RawWaker {
298 let p = ptr as *const Arc<AtomicBool>;
299 RawWaker::new(Box::into_raw(Box::new(Arc::clone(&*p))) as *const (), &WAKER_V_TABLE)
302 fn create_waker() -> (Arc<AtomicBool>, Waker) {
303 let a = Arc::new(AtomicBool::new(false));
304 let waker = unsafe { Waker::from_raw(waker_clone((&a as *const Arc<AtomicBool>) as *const ())) };
310 let mut future = Future {
311 state: Arc::new(Mutex::new(FutureState {
312 callbacks: Vec::new(),
316 let mut second_future = Future { state: Arc::clone(&future.state) };
318 let (woken, waker) = create_waker();
319 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
320 assert!(!woken.load(Ordering::SeqCst));
322 let (second_woken, second_waker) = create_waker();
323 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Pending);
324 assert!(!second_woken.load(Ordering::SeqCst));
326 future.state.lock().unwrap().complete();
327 assert!(woken.load(Ordering::SeqCst));
328 assert!(second_woken.load(Ordering::SeqCst));
329 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
330 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Ready(()));