]> git.bitcoin.ninja Git - rust-lightning/blob - lightning/src/util/wakers.rs
Merge pull request #1861 from TheBlueMatt/2022-11-tx-connection-idempotency
[rust-lightning] / lightning / src / util / wakers.rs
1 // This file is Copyright its original authors, visible in version control
2 // history.
3 //
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
8 // licenses.
9
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
12 //! be re-persisted.
13 //!
14 //! [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
15
16 use alloc::sync::Arc;
17 use core::mem;
18 use crate::sync::{Condvar, Mutex, MutexGuard};
19
20 use crate::prelude::*;
21
22 #[cfg(any(test, feature = "std"))]
23 use std::time::{Duration, Instant};
24
25 use core::future::Future as StdFuture;
26 use core::task::{Context, Poll};
27 use core::pin::Pin;
28
29
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>>>)>,
33         condvar: Condvar,
34 }
35
36 macro_rules! check_woken {
37         ($guard: expr, $retval: expr) => { {
38                 if $guard.0 {
39                         $guard.0 = false;
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.
43                                 $guard.1.take();
44                         }
45                         return $retval;
46                 }
47         } }
48 }
49
50 impl Notifier {
51         pub(crate) fn new() -> Self {
52                 Self {
53                         notify_pending: Mutex::new((false, None)),
54                         condvar: Condvar::new(),
55                 }
56         }
57
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.
64                                 lock.1.take();
65                                 lock.0 = false;
66                         }
67                 }
68                 lock
69         }
70
71         pub(crate) fn wait(&self) {
72                 loop {
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, ());
77                 }
78         }
79
80         #[cfg(any(test, feature = "std"))]
81         pub(crate) fn wait_timeout(&self, max_wait: Duration) -> bool {
82                 let current_time = Instant::now();
83                 loop {
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
92                         // 1.42.0.
93                         let elapsed = current_time.elapsed();
94                         if elapsed >= max_wait {
95                                 return false;
96                         }
97                         match max_wait.checked_sub(elapsed) {
98                                 None => return false,
99                                 Some(_) => continue
100                         }
101                 }
102         }
103
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();
109                 }
110                 lock.0 = true;
111                 mem::drop(lock);
112                 self.condvar.notify_all();
113         }
114
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) }
120                 } else {
121                         let state = Arc::new(Mutex::new(FutureState {
122                                 callbacks: Vec::new(),
123                                 complete: lock.0,
124                                 callbacks_made: false,
125                         }));
126                         lock.1 = Some(Arc::clone(&state));
127                         Future { state }
128                 }
129         }
130
131         #[cfg(any(test, feature = "_test_utils"))]
132         pub fn notify_pending(&self) -> bool {
133                 self.notify_pending.lock().unwrap().0
134         }
135 }
136
137 /// A callback which is called when a [`Future`] completes.
138 ///
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`]
141 /// instead.
142 ///
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.
147         fn call(&self);
148 }
149
150 impl<F: Fn() + Send> FutureCallback for F {
151         fn call(&self) { (self)(); }
152 }
153
154 pub(crate) struct FutureState {
155         // When we're tracking whether a callback counts as having woken the user's code, we check the
156         // first bool - set to false if we're just calling a Waker, and true if we're calling an actual
157         // user-provided function.
158         callbacks: Vec<(bool, Box<dyn FutureCallback>)>,
159         complete: bool,
160         callbacks_made: bool,
161 }
162
163 impl FutureState {
164         fn complete(&mut self) {
165                 for (counts_as_call, callback) in self.callbacks.drain(..) {
166                         callback.call();
167                         self.callbacks_made |= counts_as_call;
168                 }
169                 self.complete = true;
170         }
171 }
172
173 /// A simple future which can complete once, and calls some callback(s) when it does so.
174 pub struct Future {
175         state: Arc<Mutex<FutureState>>,
176 }
177
178 impl Future {
179         /// Registers a callback to be called upon completion of this future. If the future has already
180         /// completed, the callback will be called immediately.
181         ///
182         /// (C-not exported) use the bindings-only `register_callback_fn` instead
183         pub fn register_callback(&self, callback: Box<dyn FutureCallback>) {
184                 let mut state = self.state.lock().unwrap();
185                 if state.complete {
186                         state.callbacks_made = true;
187                         mem::drop(state);
188                         callback.call();
189                 } else {
190                         state.callbacks.push((true, callback));
191                 }
192         }
193
194         // C bindings don't (currently) know how to map `Box<dyn Trait>`, and while it could add the
195         // following wrapper, doing it in the bindings is currently much more work than simply doing it
196         // here.
197         /// Registers a callback to be called upon completion of this future. If the future has already
198         /// completed, the callback will be called immediately.
199         #[cfg(c_bindings)]
200         pub fn register_callback_fn<F: 'static + FutureCallback>(&self, callback: F) {
201                 self.register_callback(Box::new(callback));
202         }
203 }
204
205 use core::task::Waker;
206 struct StdWaker(pub Waker);
207 impl FutureCallback for StdWaker {
208         fn call(&self) { self.0.wake_by_ref() }
209 }
210
211 /// (C-not exported) as Rust Futures aren't usable in language bindings.
212 impl<'a> StdFuture for Future {
213         type Output = ();
214
215         fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
216                 let mut state = self.state.lock().unwrap();
217                 if state.complete {
218                         state.callbacks_made = true;
219                         Poll::Ready(())
220                 } else {
221                         let waker = cx.waker().clone();
222                         state.callbacks.push((false, Box::new(StdWaker(waker))));
223                         Poll::Pending
224                 }
225         }
226 }
227
228 #[cfg(test)]
229 mod tests {
230         use super::*;
231         use core::sync::atomic::{AtomicBool, Ordering};
232         use core::future::Future as FutureTrait;
233         use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
234
235         #[test]
236         fn notifier_pre_notified_future() {
237                 // Previously, if we generated a future after a `Notifier` had been notified, the future
238                 // would never complete. This tests this behavior, ensuring the future instead completes
239                 // immediately.
240                 let notifier = Notifier::new();
241                 notifier.notify();
242
243                 let callback = Arc::new(AtomicBool::new(false));
244                 let callback_ref = Arc::clone(&callback);
245                 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
246                 assert!(callback.load(Ordering::SeqCst));
247         }
248
249         #[test]
250         fn notifier_future_completes_wake() {
251                 // Previously, if we were only using the `Future` interface to learn when a `Notifier` has
252                 // been notified, we'd never mark the notifier as not-awaiting-notify. This caused the
253                 // `lightning-background-processor` to persist in a tight loop.
254                 let notifier = Notifier::new();
255
256                 // First check the simple case, ensuring if we get notified a new future isn't woken until
257                 // a second `notify`.
258                 let callback = Arc::new(AtomicBool::new(false));
259                 let callback_ref = Arc::clone(&callback);
260                 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
261                 assert!(!callback.load(Ordering::SeqCst));
262
263                 notifier.notify();
264                 assert!(callback.load(Ordering::SeqCst));
265
266                 let callback = Arc::new(AtomicBool::new(false));
267                 let callback_ref = Arc::clone(&callback);
268                 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
269                 assert!(!callback.load(Ordering::SeqCst));
270
271                 notifier.notify();
272                 assert!(callback.load(Ordering::SeqCst));
273
274                 // Then check the case where the future is fetched before the notification, but a callback
275                 // is only registered after the `notify`, ensuring that it is still sufficient to ensure we
276                 // don't get an instant-wake when we get a new future.
277                 let future = notifier.get_future();
278                 notifier.notify();
279
280                 let callback = Arc::new(AtomicBool::new(false));
281                 let callback_ref = Arc::clone(&callback);
282                 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
283                 assert!(callback.load(Ordering::SeqCst));
284
285                 let callback = Arc::new(AtomicBool::new(false));
286                 let callback_ref = Arc::clone(&callback);
287                 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
288                 assert!(!callback.load(Ordering::SeqCst));
289         }
290
291         #[test]
292         fn new_future_wipes_notify_bit() {
293                 // Previously, if we were only using the `Future` interface to learn when a `Notifier` has
294                 // been notified, we'd never mark the notifier as not-awaiting-notify if a `Future` is
295                 // fetched after the notify bit has been set.
296                 let notifier = Notifier::new();
297                 notifier.notify();
298
299                 let callback = Arc::new(AtomicBool::new(false));
300                 let callback_ref = Arc::clone(&callback);
301                 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
302                 assert!(callback.load(Ordering::SeqCst));
303
304                 let callback = Arc::new(AtomicBool::new(false));
305                 let callback_ref = Arc::clone(&callback);
306                 notifier.get_future().register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
307                 assert!(!callback.load(Ordering::SeqCst));
308
309                 notifier.notify();
310                 assert!(callback.load(Ordering::SeqCst));
311         }
312
313         #[cfg(feature = "std")]
314         #[test]
315         fn test_wait_timeout() {
316                 use crate::sync::Arc;
317                 use std::thread;
318
319                 let persistence_notifier = Arc::new(Notifier::new());
320                 let thread_notifier = Arc::clone(&persistence_notifier);
321
322                 let exit_thread = Arc::new(AtomicBool::new(false));
323                 let exit_thread_clone = exit_thread.clone();
324                 thread::spawn(move || {
325                         loop {
326                                 let mut lock = thread_notifier.notify_pending.lock().unwrap();
327                                 lock.0 = true;
328                                 thread_notifier.condvar.notify_all();
329
330                                 if exit_thread_clone.load(Ordering::SeqCst) {
331                                         break
332                                 }
333                         }
334                 });
335
336                 // Check that we can block indefinitely until updates are available.
337                 let _ = persistence_notifier.wait();
338
339                 // Check that the Notifier will return after the given duration if updates are
340                 // available.
341                 loop {
342                         if persistence_notifier.wait_timeout(Duration::from_millis(100)) {
343                                 break
344                         }
345                 }
346
347                 exit_thread.store(true, Ordering::SeqCst);
348
349                 // Check that the Notifier will return after the given duration even if no updates
350                 // are available.
351                 loop {
352                         if !persistence_notifier.wait_timeout(Duration::from_millis(100)) {
353                                 break
354                         }
355                 }
356         }
357
358         #[test]
359         fn test_future_callbacks() {
360                 let future = Future {
361                         state: Arc::new(Mutex::new(FutureState {
362                                 callbacks: Vec::new(),
363                                 complete: false,
364                                 callbacks_made: false,
365                         }))
366                 };
367                 let callback = Arc::new(AtomicBool::new(false));
368                 let callback_ref = Arc::clone(&callback);
369                 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
370
371                 assert!(!callback.load(Ordering::SeqCst));
372                 future.state.lock().unwrap().complete();
373                 assert!(callback.load(Ordering::SeqCst));
374                 future.state.lock().unwrap().complete();
375         }
376
377         #[test]
378         fn test_pre_completed_future_callbacks() {
379                 let future = Future {
380                         state: Arc::new(Mutex::new(FutureState {
381                                 callbacks: Vec::new(),
382                                 complete: false,
383                                 callbacks_made: false,
384                         }))
385                 };
386                 future.state.lock().unwrap().complete();
387
388                 let callback = Arc::new(AtomicBool::new(false));
389                 let callback_ref = Arc::clone(&callback);
390                 future.register_callback(Box::new(move || assert!(!callback_ref.fetch_or(true, Ordering::SeqCst))));
391
392                 assert!(callback.load(Ordering::SeqCst));
393                 assert!(future.state.lock().unwrap().callbacks.is_empty());
394         }
395
396         // Rather annoyingly, there's no safe way in Rust std to construct a Waker despite it being
397         // totally possible to construct from a trait implementation (though somewhat less effecient
398         // compared to a raw VTable). Instead, we have to write out a lot of boilerplate to build a
399         // waker, which we do here with a trivial Arc<AtomicBool> data element to track woke-ness.
400         const WAKER_V_TABLE: RawWakerVTable = RawWakerVTable::new(waker_clone, wake, wake_by_ref, drop);
401         unsafe fn wake_by_ref(ptr: *const ()) { let p = ptr as *const Arc<AtomicBool>; assert!(!(*p).fetch_or(true, Ordering::SeqCst)); }
402         unsafe fn drop(ptr: *const ()) { let p = ptr as *mut Arc<AtomicBool>; let _freed = Box::from_raw(p); }
403         unsafe fn wake(ptr: *const ()) { wake_by_ref(ptr); drop(ptr); }
404         unsafe fn waker_clone(ptr: *const ()) -> RawWaker {
405                 let p = ptr as *const Arc<AtomicBool>;
406                 RawWaker::new(Box::into_raw(Box::new(Arc::clone(&*p))) as *const (), &WAKER_V_TABLE)
407         }
408
409         fn create_waker() -> (Arc<AtomicBool>, Waker) {
410                 let a = Arc::new(AtomicBool::new(false));
411                 let waker = unsafe { Waker::from_raw(waker_clone((&a as *const Arc<AtomicBool>) as *const ())) };
412                 (a, waker)
413         }
414
415         #[test]
416         fn test_future() {
417                 let mut future = Future {
418                         state: Arc::new(Mutex::new(FutureState {
419                                 callbacks: Vec::new(),
420                                 complete: false,
421                                 callbacks_made: false,
422                         }))
423                 };
424                 let mut second_future = Future { state: Arc::clone(&future.state) };
425
426                 let (woken, waker) = create_waker();
427                 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
428                 assert!(!woken.load(Ordering::SeqCst));
429
430                 let (second_woken, second_waker) = create_waker();
431                 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Pending);
432                 assert!(!second_woken.load(Ordering::SeqCst));
433
434                 future.state.lock().unwrap().complete();
435                 assert!(woken.load(Ordering::SeqCst));
436                 assert!(second_woken.load(Ordering::SeqCst));
437                 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
438                 assert_eq!(Pin::new(&mut second_future).poll(&mut Context::from_waker(&second_waker)), Poll::Ready(()));
439         }
440
441         #[test]
442         fn test_dropped_future_doesnt_count() {
443                 // Tests that if a Future gets drop'd before it is poll()ed `Ready` it doesn't count as
444                 // having been woken, leaving the notify-required flag set.
445                 let notifier = Notifier::new();
446                 notifier.notify();
447
448                 // If we get a future and don't touch it we're definitely still notify-required.
449                 notifier.get_future();
450                 assert!(notifier.wait_timeout(Duration::from_millis(1)));
451                 assert!(!notifier.wait_timeout(Duration::from_millis(1)));
452
453                 // Even if we poll'd once but didn't observe a `Ready`, we should be notify-required.
454                 let mut future = notifier.get_future();
455                 let (woken, waker) = create_waker();
456                 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
457
458                 notifier.notify();
459                 assert!(woken.load(Ordering::SeqCst));
460                 assert!(notifier.wait_timeout(Duration::from_millis(1)));
461
462                 // However, once we do poll `Ready` it should wipe the notify-required flag.
463                 let mut future = notifier.get_future();
464                 let (woken, waker) = create_waker();
465                 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Pending);
466
467                 notifier.notify();
468                 assert!(woken.load(Ordering::SeqCst));
469                 assert_eq!(Pin::new(&mut future).poll(&mut Context::from_waker(&waker)), Poll::Ready(()));
470                 assert!(!notifier.wait_timeout(Duration::from_millis(1)));
471         }
472 }