Merge pull request #4147 from TheBlueMatt/2025-10-parallel-reads

Parallelize `ChannelMonitor` loading from async `KVStore`s

Author: TheBlueMatt

lightning-block-sync/src/gossip.rs

@@ -47,8 +47,12 @@ pub trait UtxoSource: BlockSource + 'static {
 pub struct TokioSpawner;
 #[cfg(feature = "tokio")]
 impl FutureSpawner for TokioSpawner {
-	fn spawn<T: Future<Output = ()> + Send + 'static>(&self, future: T) {
-		tokio::spawn(future);
+	type E = tokio::task::JoinError;
+	type SpawnedFutureResult<O> = tokio::task::JoinHandle<O>;
+	fn spawn<O: Send + 'static, F: Future<Output = O> + Send + 'static>(
+		&self, future: F,
+	) -> Self::SpawnedFutureResult<O> {
+		tokio::spawn(future)
 	}
 }
 
@@ -254,7 +258,7 @@ where
 		let fut = res.clone();
 		let source = self.source.clone();
 		let block_cache = Arc::clone(&self.block_cache);
-		self.spawn.spawn(async move {
+		let _not_polled = self.spawn.spawn(async move {
 			let res = Self::retrieve_utxo(source, block_cache, scid).await;
 			fut.resolve(res);
 		});

lightning/src/util/async_poll.rs

@@ -15,26 +15,100 @@ 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 = O> + Unpin, O> {
 	Pending(F),
-	Ready(Result<(), E>),
+	Ready(O),
 }
 
-pub(crate) struct MultiResultFuturePoller<F: Future<Output = Result<(), E>> + Unpin, E: Unpin> {
-	futures_state: Vec<ResultFuture<F, E>>,
+pub(crate) struct TwoFutureJoiner<
+	AO,
+	BO,
+	AF: Future<Output = AO> + Unpin,
+	BF: Future<Output = BO> + Unpin,
+> {
+	a: Option<ResultFuture<AF, AO>>,
+	b: Option<ResultFuture<BF, BO>>,
 }
 
-impl<F: Future<Output = Result<(), E>> + Unpin, E: Unpin> MultiResultFuturePoller<F, E> {
-	pub fn new(futures_state: Vec<ResultFuture<F, E>>) -> Self {
+impl<AO, BO, AF: Future<Output = AO> + Unpin, BF: Future<Output = BO> + Unpin>
+	TwoFutureJoiner<AO, BO, AF, BF>
+{
+	pub fn new(future_a: AF, future_b: BF) -> Self {
+		Self { a: Some(ResultFuture::Pending(future_a)), b: Some(ResultFuture::Pending(future_b)) }
+	}
+}
+
+impl<AO, BO, AF: Future<Output = AO> + Unpin, BF: Future<Output = BO> + Unpin> Future
+	for TwoFutureJoiner<AO, BO, AF, BF>
+{
+	type Output = (AO, BO);
+	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<(AO, BO)> {
+		let mut have_pending_futures = false;
+		// SAFETY: While we are pinned, we can't get direct access to our internal 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 state = unsafe { &mut self.get_unchecked_mut() };
+		macro_rules! poll_future {
+			($future: ident) => {
+				match state.$future {
+					Some(ResultFuture::Pending(ref mut fut)) => match Pin::new(fut).poll(cx) {
+						Poll::Ready(res) => {
+							state.$future = Some(ResultFuture::Ready(res));
+						},
+						Poll::Pending => {
+							have_pending_futures = true;
+						},
+					},
+					Some(ResultFuture::Ready(_)) => {},
+					None => {
+						debug_assert!(false, "Future polled after Ready");
+						return Poll::Pending;
+					},
+				}
+			};
+		}
+		poll_future!(a);
+		poll_future!(b);
+
+		if have_pending_futures {
+			Poll::Pending
+		} else {
+			Poll::Ready((
+				match state.a.take() {
+					Some(ResultFuture::Ready(a)) => a,
+					_ => unreachable!(),
+				},
+				match state.b.take() {
+					Some(ResultFuture::Ready(b)) => b,
+					_ => unreachable!(),
+				},
+			))
+		}
+	}
+}
+
+pub(crate) struct MultiResultFuturePoller<F: Future<Output = O> + Unpin, O> {
+	futures_state: Vec<ResultFuture<F, O>>,
+}
+
+impl<F: Future<Output = O> + Unpin, O> MultiResultFuturePoller<F, O> {
+	pub fn new(futures_state: Vec<ResultFuture<F, O>>) -> 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 = O> + Unpin, O> Future for MultiResultFuturePoller<F, O> {
+	type Output = Vec<O>;
+	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Vec<O>> {
 		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/native_async.rs

@@ -8,23 +8,44 @@
 //! environment.
 
 #[cfg(all(test, feature = "std"))]
-use crate::sync::Mutex;
+use crate::sync::{Arc, Mutex};
 use crate::util::async_poll::{MaybeSend, MaybeSync};
 
+#[cfg(all(test, not(feature = "std")))]
+use alloc::rc::Rc;
+
 #[cfg(all(test, not(feature = "std")))]
 use core::cell::RefCell;
+#[cfg(test)]
+use core::convert::Infallible;
 use core::future::Future;
 #[cfg(test)]
 use core::pin::Pin;
+#[cfg(test)]
+use core::task::{Context, Poll};
 
-/// A generic trait which is able to spawn futures in the background.
+/// A generic trait which is able to spawn futures to be polled in the background.
+///
+/// When the spawned future completes, the returned [`Self::SpawnedFutureResult`] should resolve
+/// with the output of the spawned future.
+///
+/// Spawned futures must be polled independently in the background even if the returned
+/// [`Self::SpawnedFutureResult`] is dropped without being polled. This matches the semantics of
+/// `tokio::spawn`.
 ///
 /// This is not exported to bindings users as async is only supported in Rust.
 pub trait FutureSpawner: MaybeSend + MaybeSync + 'static {
+	/// The error type of [`Self::SpawnedFutureResult`]. This can be used to indicate that the
+	/// spawned future was cancelled or panicked.
+	type E;
+	/// The result of [`Self::spawn`], a future which completes when the spawned future completes.
+	type SpawnedFutureResult<O>: Future<Output = Result<O, Self::E>> + Unpin;
 	/// Spawns the given future as a background task.
 	///
 	/// This method MUST NOT block on the given future immediately.
-	fn spawn<T: Future<Output = ()> + MaybeSend + 'static>(&self, future: T);
+	fn spawn<O: MaybeSend + 'static, T: Future<Output = O> + MaybeSend + 'static>(
+		&self, future: T,
+	) -> Self::SpawnedFutureResult<O>;
 }
 
 #[cfg(test)]
@@ -39,6 +60,77 @@ pub(crate) struct FutureQueue(Mutex<Vec<Pin<Box<dyn MaybeSendableFuture>>>>);
 #[cfg(all(test, not(feature = "std")))]
 pub(crate) struct FutureQueue(RefCell<Vec<Pin<Box<dyn MaybeSendableFuture>>>>);
 
+/// A simple future which can be completed later. Used to implement [`FutureQueue`].
+#[cfg(all(test, feature = "std"))]
+pub struct FutureQueueCompletion<O>(Arc<Mutex<Option<O>>>);
+#[cfg(all(test, not(feature = "std")))]
+pub struct FutureQueueCompletion<O>(Rc<RefCell<Option<O>>>);
+
+#[cfg(all(test, feature = "std"))]
+impl<O> FutureQueueCompletion<O> {
+	fn new() -> Self {
+		Self(Arc::new(Mutex::new(None)))
+	}
+
+	fn complete(&self, o: O) {
+		*self.0.lock().unwrap() = Some(o);
+	}
+}
+
+#[cfg(all(test, feature = "std"))]
+impl<O> Clone for FutureQueueCompletion<O> {
+	fn clone(&self) -> Self {
+		#[cfg(all(test, feature = "std"))]
+		{
+			Self(Arc::clone(&self.0))
+		}
+		#[cfg(all(test, not(feature = "std")))]
+		{
+			Self(Rc::clone(&self.0))
+		}
+	}
+}
+
+#[cfg(all(test, not(feature = "std")))]
+impl<O> FutureQueueCompletion<O> {
+	fn new() -> Self {
+		Self(Rc::new(RefCell::new(None)))
+	}
+
+	fn complete(&self, o: O) {
+		*self.0.borrow_mut() = Some(o);
+	}
+}
+
+#[cfg(all(test, not(feature = "std")))]
+impl<O> Clone for FutureQueueCompletion<O> {
+	fn clone(&self) -> Self {
+		Self(self.0.clone())
+	}
+}
+
+#[cfg(all(test, feature = "std"))]
+impl<O> Future for FutureQueueCompletion<O> {
+	type Output = Result<O, Infallible>;
+	fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<O, Infallible>> {
+		match Pin::into_inner(self).0.lock().unwrap().take() {
+			None => Poll::Pending,
+			Some(o) => Poll::Ready(Ok(o)),
+		}
+	}
+}
+
+#[cfg(all(test, not(feature = "std")))]
+impl<O> Future for FutureQueueCompletion<O> {
+	type Output = Result<O, Infallible>;
+	fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<O, Infallible>> {
+		match Pin::into_inner(self).0.borrow_mut().take() {
+			None => Poll::Pending,
+			Some(o) => Poll::Ready(Ok(o)),
+		}
+	}
+}
+
 #[cfg(test)]
 impl FutureQueue {
 	pub(crate) fn new() -> Self {
@@ -74,7 +166,6 @@ impl FutureQueue {
 			futures = self.0.borrow_mut();
 		}
 		futures.retain_mut(|fut| {
-			use core::task::{Context, Poll};
 			let waker = crate::util::async_poll::dummy_waker();
 			match fut.as_mut().poll(&mut Context::from_waker(&waker)) {
 				Poll::Ready(()) => false,
@@ -86,7 +177,16 @@ impl FutureQueue {
 
 #[cfg(test)]
 impl FutureSpawner for FutureQueue {
-	fn spawn<T: Future<Output = ()> + MaybeSend + 'static>(&self, future: T) {
+	type E = Infallible;
+	type SpawnedFutureResult<O> = FutureQueueCompletion<O>;
+	fn spawn<O: MaybeSend + 'static, F: Future<Output = O> + MaybeSend + 'static>(
+		&self, f: F,
+	) -> FutureQueueCompletion<O> {
+		let completion = FutureQueueCompletion::new();
+		let compl_ref = completion.clone();
+		let future = async move {
+			compl_ref.complete(f.await);
+		};
 		#[cfg(feature = "std")]
 		{
 			self.0.lock().unwrap().push(Box::pin(future));
@@ -95,14 +195,24 @@ impl FutureSpawner for FutureQueue {
 		{
 			self.0.borrow_mut().push(Box::pin(future));
 		}
+		completion
 	}
 }
 
 #[cfg(test)]
 impl<D: core::ops::Deref<Target = FutureQueue> + MaybeSend + MaybeSync + 'static> FutureSpawner
 	for D
 {
-	fn spawn<T: Future<Output = ()> + MaybeSend + 'static>(&self, future: T) {
+	type E = Infallible;
+	type SpawnedFutureResult<O> = FutureQueueCompletion<O>;
+	fn spawn<O: MaybeSend + 'static, F: Future<Output = O> + MaybeSend + 'static>(
+		&self, f: F,
+	) -> FutureQueueCompletion<O> {
+		let completion = FutureQueueCompletion::new();
+		let compl_ref = completion.clone();
+		let future = async move {
+			compl_ref.complete(f.await);
+		};
 		#[cfg(feature = "std")]
 		{
 			self.0.lock().unwrap().push(Box::pin(future));
@@ -111,5 +221,6 @@ impl<D: core::ops::Deref<Target = FutureQueue> + MaybeSend + MaybeSync + 'static
 		{
 			self.0.borrow_mut().push(Box::pin(future));
 		}
+		completion
 	}
 }