Parallelize `ChannelMonitor` loading from async `KVStore`s
What the developer wrote
Authored by Matt Corallo
Parallelize `ChannelMonitor` loading from async `KVStore`s
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.
What changed, and why it matters
This commit is a performance improvement, not a security fix. It changes how Lightning Dev Kit loads saved channel data on startup so that multiple pieces can be read at the same time rather than one after another. The change introduces a small amount of `unsafe` Rust code, but the included safety comment explains why the authors believe it is correct. There is no indication in the commit that this fixes a vulnerability or that it creates one.
No security action required. Reviewers may optionally verify the `unsafe` safety argument independently, but the commit presents it as a routine performance optimization.
Security signals we found
Use of `unsafe` Rust with an inline SAFETY justification
Async future polling helper generalized to arbitrary output types
Parallel I/O during startup for ChannelMonitor loading
Evidence from the diff
The patch generalizes MultiResultFuturePoller and ResultFuture so they can hold arbitrary future outputs, not just Result<(), E>. It then uses them in persist.rs to spawn parallel reads of ChannelMonitor records from an async KVStore during initialization. The only unsafe block is an Pin::get_unchecked_mut() call in the poller, justified by the bound that the inner future type is Unpin. No memory-safety bug, data race, or cryptographic issue is evident from the diff.
Changed components
lightning/src/util/async_poll.rslightning/src/util/persist.rsInspect captured patch +27 / −15
diff --git a/lightning/src/util/async_poll.rs b/lightning/src/util/async_poll.rs
index 9c2ca4c..931d281 100644
--- a/lightning/src/util/async_poll.rs
+++ b/lightning/src/util/async_poll.rs
@@ -15,26 +15,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 = 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 MultiResultFuturePoller<F: Future<Output = O> + Unpin, O> {
+ futures_state: Vec<ResultFuture<F, O>>,
}
-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 = 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) {
diff --git a/lightning/src/util/persist.rs b/lightning/src/util/persist.rs
index e152096..69b5c85 100644
--- a/lightning/src/util/persist.rs
+++ b/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};
@@ -875,11 +877,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);
}
}
Why this scored 11/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.