Parallelize ChannelMonitor loading from async KVStores

Reading `ChannelMonitor`s on startup is one of the slowest parts of
LDK initialization. Now that we have an async `KVStore`, there's no
need for that, we can simply paralellize their loading, which we do
here.

Sadly, because Rust futures are pretty unergonomic, we have to add
some `unsafe {}` here, but arguing its fine is relatively
straightforward.

Author: TheBlueMatt

lightning/src/chain/channelmonitor.rs

@@ -6684,8 +6684,8 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 				// return Ok(None) to allow it to be skipped and not loaded.
 				return Ok(None);
 			} else {
-				panic!("Found monitor for channel {channel_id} with no updates since v0.0.118.\
-					These monitors are no longer supported.\
+				panic!("Found monitor for channel {channel_id} with no updates since v0.0.118. \
+					These monitors are no longer supported. \
 					To continue, run a v0.1 release, send/route a payment over the channel or close it.");
 			}
 		}

lightning/src/util/async_poll.rs

@@ -16,26 +16,31 @@ use core::marker::Unpin;
 use core::pin::Pin;
 use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
 
-pub(crate) enum ResultFuture<F: Future<Output = Result<(), E>>, E: Unpin> {
+pub(crate) enum ResultFuture<F: Future<Output = Result<O, E>> + Unpin, O, E> {
 	Pending(F),
-	Ready(Result<(), E>),
+	Ready(Result<O, E>),
 }
 
-pub(crate) struct MultiResultFuturePoller<F: Future<Output = Result<(), E>> + Unpin, E: Unpin> {
-	futures_state: Vec<ResultFuture<F, E>>,
+pub(crate) struct MultiResultFuturePoller<F: Future<Output = Result<O, E>> + Unpin, O, E> {
+	futures_state: Vec<ResultFuture<F, O, E>>,
 }
 
-impl<F: Future<Output = Result<(), E>> + Unpin, E: Unpin> MultiResultFuturePoller<F, E> {
-	pub fn new(futures_state: Vec<ResultFuture<F, E>>) -> Self {
+impl<F: Future<Output = Result<O, E>> + Unpin, O, E> MultiResultFuturePoller<F, O, E> {
+	pub fn new(futures_state: Vec<ResultFuture<F, O, E>>) -> Self {
 		Self { futures_state }
 	}
 }
 
-impl<F: Future<Output = Result<(), E>> + Unpin, E: Unpin> Future for MultiResultFuturePoller<F, E> {
-	type Output = Vec<Result<(), E>>;
-	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Vec<Result<(), E>>> {
+impl<F: Future<Output = Result<O, E>> + Unpin, O, E> Future for MultiResultFuturePoller<F, O, E> {
+	type Output = Vec<Result<O, E>>;
+	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Vec<Result<O, E>>> {
 		let mut have_pending_futures = false;
-		let futures_state = &mut self.get_mut().futures_state;
+		// SAFETY: While we are pinned, we can't get direct access to `futures_state` because we
+		// aren't `Unpin`. However, we don't actually need the `Pin` - we only use it below on the
+		// `Future` in the `ResultFuture::Pending` case, and the `Future` is bound by `Unpin`.
+		// Thus, the `Pin` is not actually used, and its safe to bypass it and access the inner
+		// reference directly.
+		let futures_state = unsafe { &mut self.get_unchecked_mut().futures_state };
 		for state in futures_state.iter_mut() {
 			match state {
 				ResultFuture::Pending(ref mut fut) => match Pin::new(fut).poll(cx) {

lightning/src/util/persist.rs

@@ -34,7 +34,9 @@ use crate::chain::transaction::OutPoint;
 use crate::ln::types::ChannelId;
 use crate::sign::{ecdsa::EcdsaChannelSigner, EntropySource, SignerProvider};
 use crate::sync::Mutex;
-use crate::util::async_poll::{dummy_waker, MaybeSend, MaybeSync};
+use crate::util::async_poll::{
+	dummy_waker, MaybeSend, MaybeSync, MultiResultFuturePoller, ResultFuture,
+};
 use crate::util::logger::Logger;
 use crate::util::native_async::FutureSpawner;
 use crate::util::ser::{Readable, ReadableArgs, Writeable};
@@ -783,11 +785,16 @@ where
 		let primary = CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE;
 		let secondary = CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE;
 		let monitor_list = self.0.kv_store.list(primary, secondary).await?;
-		let mut res = Vec::with_capacity(monitor_list.len());
+		let mut futures = Vec::with_capacity(monitor_list.len());
 		for monitor_key in monitor_list {
-			let result =
-				self.0.maybe_read_channel_monitor_with_updates(monitor_key.as_str()).await?;
-			if let Some(read_res) = result {
+			futures.push(ResultFuture::Pending(Box::pin(async move {
+				self.0.maybe_read_channel_monitor_with_updates(monitor_key.as_str()).await
+			})));
+		}
+		let future_results = MultiResultFuturePoller::new(futures).await;
+		let mut res = Vec::with_capacity(future_results.len());
+		for result in future_results {
+			if let Some(read_res) = result? {
 				res.push(read_res);
 			}
 		}