Handle events asynchronously in the BackgroundProcessor's async variant

Author: wpaulino

lightning-background-processor/src/lib.rs

@@ -23,6 +23,8 @@ use lightning::ln::msgs::{ChannelMessageHandler, OnionMessageHandler, RoutingMes
 use lightning::ln::peer_handler::{CustomMessageHandler, PeerManager, SocketDescriptor};
 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
 use lightning::routing::scoring::WriteableScore;
+#[cfg(feature = "futures")]
+use lightning::util::events::AsyncEventHandler;
 use lightning::util::events::{Event, EventHandler, EventsProvider};
 use lightning::util::logger::Logger;
 use lightning::util::persist::Persister;
@@ -34,6 +36,11 @@ use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 use std::ops::Deref;
 
+#[cfg(feature = "futures")]
+use core::future::Future as StdFuture;
+#[cfg(feature = "futures")]
+use core::pin::Pin;
+
 #[cfg(feature = "futures")]
 use futures_util::{select_biased, future::FutureExt};
 
@@ -225,16 +232,49 @@ where A::Target: chain::Access, L::Target: Logger {
 	}
 }
 
+#[cfg(feature = "futures")]
+/// Decorates an [`EventHandler`] with common functionality provided by standard [`EventHandler`]s.
+struct DecoratingAsyncEventHandler<
+	'a,
+	E: AsyncEventHandler,
+	PGS: Deref<Target = P2PGossipSync<G, A, L>>,
+	RGS: Deref<Target = RapidGossipSync<G, L>>,
+	G: Deref<Target = NetworkGraph<L>>,
+	A: Deref,
+	L: Deref,
+>
+where A::Target: chain::Access, L::Target: Logger {
+	async_event_handler: E,
+	gossip_sync: &'a GossipSync<PGS, RGS, G, A, L>,
+}
+
+#[cfg(feature = "futures")]
+impl<
+	'a,
+	E: AsyncEventHandler,
+	PGS: Deref<Target = P2PGossipSync<G, A, L>>,
+	RGS: Deref<Target = RapidGossipSync<G, L>>,
+	G: Deref<Target = NetworkGraph<L>>,
+	A: Deref,
+	L: Deref,
+> AsyncEventHandler for DecoratingAsyncEventHandler<'a, E, PGS, RGS, G, A, L>
+where A::Target: chain::Access, L::Target: Logger {
+	fn handle_event_async<'b>(&'b self, event: &'b Event) -> Pin<Box<dyn StdFuture<Output = ()> + '_>> {
+		Box::pin(async move {
+			if let Some(network_graph) = self.gossip_sync.network_graph() {
+				network_graph.handle_event(event); // TODO: Also make async?
+			}
+			self.async_event_handler.handle_event_async(event).await;
+		})
+	}
+}
+
 macro_rules! define_run_body {
-	($persister: ident, $event_handler: ident, $chain_monitor: ident, $channel_manager: ident,
+	($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
+	 $channel_manager: ident, $process_channel_manager_events: expr,
 	 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
 	 $loop_exit_check: expr, $await: expr)
 	=> { {
-		let event_handler = DecoratingEventHandler {
-			event_handler: $event_handler,
-			gossip_sync: &$gossip_sync,
-		};
-
 		log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
 		$channel_manager.timer_tick_occurred();
 
@@ -245,8 +285,8 @@ macro_rules! define_run_body {
 		let mut have_pruned = false;
 
 		loop {
-			$channel_manager.process_pending_events(&event_handler);
-			$chain_monitor.process_pending_events(&event_handler);
+			$process_channel_manager_events;
+			$process_chain_monitor_events;
 
 			// Note that the PeerManager::process_events may block on ChannelManager's locks,
 			// hence it comes last here. When the ChannelManager finishes whatever it's doing,
@@ -379,7 +419,7 @@ pub async fn process_events_async<
 	CMH: 'static + Deref + Send + Sync,
 	RMH: 'static + Deref + Send + Sync,
 	OMH: 'static + Deref + Send + Sync,
-	EH: 'static + EventHandler + Send,
+	EH: 'static + AsyncEventHandler + Send,
 	PS: 'static + Deref + Send,
 	M: 'static + Deref<Target = ChainMonitor<Signer, CF, T, F, L, P>> + Send + Sync,
 	CM: 'static + Deref<Target = ChannelManager<Signer, CW, T, K, F, L>> + Send + Sync,
@@ -392,7 +432,7 @@ pub async fn process_events_async<
 	SleepFuture: core::future::Future<Output = bool>,
 	Sleeper: Fn(Duration) -> SleepFuture
 >(
-	persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
+	persister: PS, async_event_handler: EH, chain_monitor: M, channel_manager: CM,
 	gossip_sync: GossipSync<PGS, RGS, G, CA, L>, peer_manager: PM, logger: L, scorer: Option<S>,
 	sleeper: Sleeper,
 ) -> Result<(), std::io::Error>
@@ -412,7 +452,12 @@ where
 	PS::Target: 'static + Persister<'a, Signer, CW, T, K, F, L, SC>,
 {
 	let mut should_continue = true;
-	define_run_body!(persister, event_handler, chain_monitor, channel_manager,
+	let async_event_handler = DecoratingAsyncEventHandler {
+		async_event_handler,
+		gossip_sync: &gossip_sync,
+	};
+	define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events_async(&async_event_handler).await,
+		channel_manager, channel_manager.process_pending_events_async(&async_event_handler).await,
 		gossip_sync, peer_manager, logger, scorer, should_continue, {
 			select_biased! {
 				_ = channel_manager.get_persistable_update_future().fuse() => true,
@@ -517,7 +562,12 @@ impl BackgroundProcessor {
 		let stop_thread = Arc::new(AtomicBool::new(false));
 		let stop_thread_clone = stop_thread.clone();
 		let handle = thread::spawn(move || -> Result<(), std::io::Error> {
-			define_run_body!(persister, event_handler, chain_monitor, channel_manager,
+			let event_handler = DecoratingEventHandler {
+				event_handler,
+				gossip_sync: &gossip_sync,
+			};
+			define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
+				channel_manager, channel_manager.process_pending_events(&event_handler),
 				gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
 				channel_manager.await_persistable_update_timeout(Duration::from_millis(100)))
 		});