Skip to content

Commit 3b8bf93

Browse files
TheBlueMattTheBlueMatt
authored
Merge pull request #2090 from TheBlueMatt/2023-03-mon-wake-bp
Wake background-processor on async monitor update completion
2 parents 7ca3709 + 94a11f7 commit 3b8bf93

File tree

7 files changed

+292
-212
lines changed

7 files changed

+292
-212
lines changed

‎lightning-background-processor/src/lib.rs

+43-13
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@ use lightning::routing::router::Router;
3838
use lightning::routing::scoring::{Score, WriteableScore};
3939
use lightning::util::logger::Logger;
4040
use lightning::util::persist::Persister;
41+
#[cfg(feature = "std")]
42+
use lightning::util::wakers::Sleeper;
4143
use lightning_rapid_gossip_sync::RapidGossipSync;
4244

4345
use core::ops::Deref;
@@ -114,6 +116,13 @@ const FIRST_NETWORK_PRUNE_TIMER: u64 = 60;
114116
#[cfg(test)]
115117
const FIRST_NETWORK_PRUNE_TIMER: u64 = 1;
116118

119+
#[cfg(feature = "futures")]
120+
/// core::cmp::min is not currently const, so we define a trivial (and equivalent) replacement
121+
const fn min_u64(a: u64, b: u64) -> u64 { if a < b { a } else { b } }
122+
#[cfg(feature = "futures")]
123+
const FASTEST_TIMER: u64 = min_u64(min_u64(FRESHNESS_TIMER, PING_TIMER),
124+
min_u64(SCORER_PERSIST_TIMER, FIRST_NETWORK_PRUNE_TIMER));
125+
117126
/// Either [`P2PGossipSync`] or [`RapidGossipSync`].
118127
pub enum GossipSync<
119128
P: Deref<Target = P2PGossipSync<G, U, L>>,
@@ -256,7 +265,8 @@ macro_rules! define_run_body {
256265
($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
257266
$channel_manager: ident, $process_channel_manager_events: expr,
258267
$gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
259-
$loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr)
268+
$loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
269+
$check_slow_await: expr)
260270
=> { {
261271
log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
262272
$channel_manager.timer_tick_occurred();
@@ -286,9 +296,10 @@ macro_rules! define_run_body {
286296

287297
// We wait up to 100ms, but track how long it takes to detect being put to sleep,
288298
// see `await_start`'s use below.
289-
let mut await_start = $get_timer(1);
299+
let mut await_start = None;
300+
if $check_slow_await { await_start = Some($get_timer(1)); }
290301
let updates_available = $await;
291-
let await_slow = $timer_elapsed(&mut await_start, 1);
302+
let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
292303

293304
if updates_available {
294305
log_trace!($logger, "Persisting ChannelManager...");
@@ -388,23 +399,32 @@ pub(crate) mod futures_util {
388399
use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
389400
use core::pin::Pin;
390401
use core::marker::Unpin;
391-
pub(crate) struct Selector<A: Future<Output=()> + Unpin, B: Future<Output=bool> + Unpin> {
402+
pub(crate) struct Selector<
403+
A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
404+
> {
392405
pub a: A,
393406
pub b: B,
407+
pub c: C,
394408
}
395409
pub(crate) enum SelectorOutput {
396-
A, B(bool),
410+
A, B, C(bool),
397411
}
398412

399-
impl<A: Future<Output=()> + Unpin, B: Future<Output=bool> + Unpin> Future for Selector<A, B> {
413+
impl<
414+
A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
415+
> Future for Selector<A, B, C> {
400416
type Output = SelectorOutput;
401417
fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
402418
match Pin::new(&mut self.a).poll(ctx) {
403419
Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
404420
Poll::Pending => {},
405421
}
406422
match Pin::new(&mut self.b).poll(ctx) {
407-
Poll::Ready(res) => { return Poll::Ready(SelectorOutput::B(res)); },
423+
Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
424+
Poll::Pending => {},
425+
}
426+
match Pin::new(&mut self.c).poll(ctx) {
427+
Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
408428
Poll::Pending => {},
409429
}
410430
Poll::Pending
@@ -438,6 +458,11 @@ use core::task;
438458
/// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
439459
/// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
440460
/// manually instead.
461+
///
462+
/// The `mobile_interruptable_platform` flag should be set if we're currently running on a
463+
/// mobile device, where we may need to check for interruption of the application regularly. If you
464+
/// are unsure, you should set the flag, as the performance impact of it is minimal unless there
465+
/// are hundreds or thousands of simultaneous process calls running.
441466
#[cfg(feature = "futures")]
442467
pub async fn process_events_async<
443468
'a,
@@ -473,7 +498,7 @@ pub async fn process_events_async<
473498
>(
474499
persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
475500
gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
476-
sleeper: Sleeper,
501+
sleeper: Sleeper, mobile_interruptable_platform: bool,
477502
) -> Result<(), lightning::io::Error>
478503
where
479504
UL::Target: 'static + UtxoLookup,
@@ -514,11 +539,13 @@ where
514539
gossip_sync, peer_manager, logger, scorer, should_break, {
515540
let fut = Selector {
516541
a: channel_manager.get_persistable_update_future(),
517-
b: sleeper(Duration::from_millis(100)),
542+
b: chain_monitor.get_update_future(),
543+
c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
518544
};
519545
match fut.await {
520546
SelectorOutput::A => true,
521-
SelectorOutput::B(exit) => {
547+
SelectorOutput::B => false,
548+
SelectorOutput::C(exit) => {
522549
should_break = exit;
523550
false
524551
}
@@ -528,7 +555,7 @@ where
528555
let mut waker = dummy_waker();
529556
let mut ctx = task::Context::from_waker(&mut waker);
530557
core::pin::Pin::new(fut).poll(&mut ctx).is_ready()
531-
})
558+
}, mobile_interruptable_platform)
532559
}
533560

534561
#[cfg(feature = "std")]
@@ -643,8 +670,11 @@ impl BackgroundProcessor {
643670
define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
644671
channel_manager, channel_manager.process_pending_events(&event_handler),
645672
gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
646-
channel_manager.await_persistable_update_timeout(Duration::from_millis(100)),
647-
|_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur)
673+
Sleeper::from_two_futures(
674+
channel_manager.get_persistable_update_future(),
675+
chain_monitor.get_update_future()
676+
).wait_timeout(Duration::from_millis(100)),
677+
|_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
648678
});
649679
Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
650680
}

‎lightning-net-tokio/src/lib.rs

+8-53
Original file line numberDiff line numberDiff line change
@@ -8,64 +8,19 @@
88
// licenses.
99

1010
//! A socket handling library for those running in Tokio environments who wish to use
11-
//! rust-lightning with native TcpStreams.
11+
//! rust-lightning with native [`TcpStream`]s.
1212
//!
1313
//! Designed to be as simple as possible, the high-level usage is almost as simple as "hand over a
14-
//! TcpStream and a reference to a PeerManager and the rest is handled", except for the
15-
//! [Event](../lightning/util/events/enum.Event.html) handling mechanism; see example below.
14+
//! [`TcpStream`] and a reference to a [`PeerManager`] and the rest is handled".
1615
//!
17-
//! The PeerHandler, due to the fire-and-forget nature of this logic, must be an Arc, and must use
18-
//! the SocketDescriptor provided here as the PeerHandler's SocketDescriptor.
16+
//! The [`PeerManager`], due to the fire-and-forget nature of this logic, must be a reference,
17+
//! (e.g. an [`Arc`]) and must use the [`SocketDescriptor`] provided here as the [`PeerManager`]'s
18+
//! `SocketDescriptor` implementation.
1919
//!
20-
//! Three methods are exposed to register a new connection for handling in tokio::spawn calls; see
21-
//! their individual docs for details.
20+
//! Three methods are exposed to register a new connection for handling in [`tokio::spawn`] calls;
21+
//! see their individual docs for details.
2222
//!
23-
//! # Example
24-
//! ```
25-
//! use std::net::TcpStream;
26-
//! use bitcoin::secp256k1::PublicKey;
27-
//! use lightning::events::{Event, EventHandler, EventsProvider};
28-
//! use std::net::SocketAddr;
29-
//! use std::sync::Arc;
30-
//!
31-
//! // Define concrete types for our high-level objects:
32-
//! type TxBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync;
33-
//! type FeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync;
34-
//! type Logger = dyn lightning::util::logger::Logger + Send + Sync;
35-
//! type NodeSigner = dyn lightning::chain::keysinterface::NodeSigner + Send + Sync;
36-
//! type UtxoLookup = dyn lightning::routing::utxo::UtxoLookup + Send + Sync;
37-
//! type ChainFilter = dyn lightning::chain::Filter + Send + Sync;
38-
//! type DataPersister = dyn lightning::chain::chainmonitor::Persist<lightning::chain::keysinterface::InMemorySigner> + Send + Sync;
39-
//! type ChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::chain::keysinterface::InMemorySigner, Arc<ChainFilter>, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>, Arc<DataPersister>>;
40-
//! type ChannelManager = Arc<lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor, TxBroadcaster, FeeEstimator, Logger>>;
41-
//! type PeerManager = Arc<lightning::ln::peer_handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChainMonitor, TxBroadcaster, FeeEstimator, UtxoLookup, Logger>>;
42-
//!
43-
//! // Connect to node with pubkey their_node_id at addr:
44-
//! async fn connect_to_node(peer_manager: PeerManager, chain_monitor: Arc<ChainMonitor>, channel_manager: ChannelManager, their_node_id: PublicKey, addr: SocketAddr) {
45-
//! lightning_net_tokio::connect_outbound(peer_manager, their_node_id, addr).await;
46-
//! loop {
47-
//! let event_handler = |event: Event| {
48-
//! // Handle the event!
49-
//! };
50-
//! channel_manager.await_persistable_update();
51-
//! channel_manager.process_pending_events(&event_handler);
52-
//! chain_monitor.process_pending_events(&event_handler);
53-
//! }
54-
//! }
55-
//!
56-
//! // Begin reading from a newly accepted socket and talk to the peer:
57-
//! async fn accept_socket(peer_manager: PeerManager, chain_monitor: Arc<ChainMonitor>, channel_manager: ChannelManager, socket: TcpStream) {
58-
//! lightning_net_tokio::setup_inbound(peer_manager, socket);
59-
//! loop {
60-
//! let event_handler = |event: Event| {
61-
//! // Handle the event!
62-
//! };
63-
//! channel_manager.await_persistable_update();
64-
//! channel_manager.process_pending_events(&event_handler);
65-
//! chain_monitor.process_pending_events(&event_handler);
66-
//! }
67-
//! }
68-
//! ```
23+
//! [`PeerManager`]: lightning::ln::peer_handler::PeerManager
6924
7025
// Prefix these with `rustdoc::` when we update our MSRV to be >= 1.52 to remove warnings.
7126
#![deny(broken_intra_doc_links)]

‎lightning/src/chain/chainmonitor.rs

+19
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ use crate::events::{Event, EventHandler};
3737
use crate::util::atomic_counter::AtomicCounter;
3838
use crate::util::logger::Logger;
3939
use crate::util::errors::APIError;
40+
use crate::util::wakers::{Future, Notifier};
4041
use crate::ln::channelmanager::ChannelDetails;
4142

4243
use crate::prelude::*;
@@ -240,6 +241,8 @@ pub struct ChainMonitor<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T:
240241
pending_monitor_events: Mutex<Vec<(OutPoint, Vec<MonitorEvent>, Option<PublicKey>)>>,
241242
/// The best block height seen, used as a proxy for the passage of time.
242243
highest_chain_height: AtomicUsize,
244+
245+
event_notifier: Notifier,
243246
}
244247

245248
impl<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> ChainMonitor<ChannelSigner, C, T, F, L, P>
@@ -300,6 +303,7 @@ where C::Target: chain::Filter,
300303
ChannelMonitorUpdateStatus::PermanentFailure => {
301304
monitor_state.channel_perm_failed.store(true, Ordering::Release);
302305
self.pending_monitor_events.lock().unwrap().push((*funding_outpoint, vec![MonitorEvent::UpdateFailed(*funding_outpoint)], monitor.get_counterparty_node_id()));
306+
self.event_notifier.notify();
303307
},
304308
ChannelMonitorUpdateStatus::InProgress => {
305309
log_debug!(self.logger, "Channel Monitor sync for channel {} in progress, holding events until completion!", log_funding_info!(monitor));
@@ -345,6 +349,7 @@ where C::Target: chain::Filter,
345349
persister,
346350
pending_monitor_events: Mutex::new(Vec::new()),
347351
highest_chain_height: AtomicUsize::new(0),
352+
event_notifier: Notifier::new(),
348353
}
349354
}
350355

@@ -472,6 +477,7 @@ where C::Target: chain::Filter,
472477
}
473478
},
474479
}
480+
self.event_notifier.notify();
475481
Ok(())
476482
}
477483

@@ -486,6 +492,7 @@ where C::Target: chain::Filter,
486492
funding_txo,
487493
monitor_update_id,
488494
}], counterparty_node_id));
495+
self.event_notifier.notify();
489496
}
490497

491498
#[cfg(any(test, fuzzing, feature = "_test_utils"))]
@@ -514,6 +521,18 @@ where C::Target: chain::Filter,
514521
handler(event).await;
515522
}
516523
}
524+
525+
/// Gets a [`Future`] that completes when an event is available either via
526+
/// [`chain::Watch::release_pending_monitor_events`] or
527+
/// [`EventsProvider::process_pending_events`].
528+
///
529+
/// Note that callbacks registered on the [`Future`] MUST NOT call back into this
530+
/// [`ChainMonitor`] and should instead register actions to be taken later.
531+
///
532+
/// [`EventsProvider::process_pending_events`]: crate::events::EventsProvider::process_pending_events
533+
pub fn get_update_future(&self) -> Future {
534+
self.event_notifier.get_future()
535+
}
517536
}
518537

519538
impl<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>

0 commit comments

Comments
 (0)