Skip to content

Commit 64c502c

Browse files
joostjagerjoostjager
committed
Remove unnecessary Copy bound from MultiResultFuturePoller
1 parent 4fcc291 commit 64c502c

File tree

3 files changed

+122
-60
lines changed

3 files changed

+122
-60
lines changed

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

Lines changed: 112 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -70,9 +70,13 @@ use lightning_rapid_gossip_sync::RapidGossipSync;
7070

7171
use lightning_liquidity::ALiquidityManager;
7272

73+
use core::future::Future;
7374
use core::ops::Deref;
75+
use core::pin::Pin;
7476
use core::time::Duration;
7577

78+
use lightning::util::async_poll::{MultiResultFuturePoller, ResultFuture};
79+
7680
#[cfg(feature = "std")]
7781
use core::sync::atomic::{AtomicBool, Ordering};
7882
#[cfg(feature = "std")]
@@ -627,11 +631,11 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
627631
pub async fn process_events_async<
628632
'a,
629633
UL: 'static + Deref,
630-
CF: 'static + Deref,
631-
T: 'static + Deref,
632-
F: 'static + Deref,
634+
CF: 'static + Deref + Sync,
635+
T: 'static + Deref + Sync,
636+
F: 'static + Deref + Sync,
633637
G: 'static + Deref<Target = NetworkGraph<L>>,
634-
L: 'static + Deref,
638+
L: 'static + Deref + Sync,
635639
P: 'static + Deref,
636640
EventHandlerFuture: core::future::Future<Output = Result<(), ReplayEvent>>,
637641
EventHandler: Fn(Event) -> EventHandlerFuture,
@@ -646,10 +650,10 @@ pub async fn process_events_async<
646650
RGS: 'static + Deref<Target = RapidGossipSync<G, L>>,
647651
PM: 'static + Deref,
648652
LM: 'static + Deref,
649-
D: 'static + Deref,
650-
O: 'static + Deref,
651-
K: 'static + Deref,
652-
OS: 'static + Deref<Target = OutputSweeper<T, D, F, CF, K, L, O>>,
653+
D: 'static + Deref + Sync,
654+
O: 'static + Deref + Sync,
655+
K: 'static + Deref + Sync,
656+
OS: 'static + Deref<Target = OutputSweeper<T, D, F, CF, K, L, O>> + Clone + Send,
653657
S: 'static + Deref<Target = SC> + Send + Sync,
654658
SC: for<'b> WriteableScore<'b>,
655659
SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
@@ -826,17 +830,24 @@ where
826830
None => {},
827831
}
828832

833+
let mut futures = Vec::new();
834+
829835
// Persist channel manager.
830836
if channel_manager.get_cm().get_and_clear_needs_persistence() {
831837
log_trace!(logger, "Persisting ChannelManager...");
832-
kv_store
833-
.write(
834-
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
835-
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
836-
CHANNEL_MANAGER_PERSISTENCE_KEY,
837-
&channel_manager.get_cm().encode(),
838-
)
839-
.await?;
838+
let res = kv_store.write(
839+
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
840+
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
841+
CHANNEL_MANAGER_PERSISTENCE_KEY,
842+
&channel_manager.get_cm().encode(),
843+
);
844+
845+
let fut: Pin<
846+
Box<dyn Future<Output = Result<(), (lightning::io::Error, bool)>> + Send + 'static>,
847+
> = Box::pin(async move { res.await.map_err(|e| (e, true)) });
848+
849+
futures.push(ResultFuture::Pending(fut));
850+
840851
log_trace!(logger, "Done persisting ChannelManager.");
841852
}
842853

@@ -864,17 +875,29 @@ where
864875
log_warn!(logger, "Not pruning network graph, consider implementing the fetch_time argument or calling remove_stale_channels_and_tracking_with_time manually.");
865876
log_trace!(logger, "Persisting network graph.");
866877
}
867-
if let Err(e) = kv_store
868-
.write(
869-
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
870-
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
871-
NETWORK_GRAPH_PERSISTENCE_KEY,
872-
&network_graph.encode(),
873-
)
874-
.await
875-
{
876-
log_error!(logger, "Error: Failed to persist network graph, check your disk and permissions {}",e);
877-
}
878+
let res = kv_store.write(
879+
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
880+
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
881+
NETWORK_GRAPH_PERSISTENCE_KEY,
882+
&network_graph.encode(),
883+
);
884+
let fut: Pin<
885+
Box<
886+
dyn Future<Output = Result<(), (lightning::io::Error, bool)>>
887+
+ Send
888+
+ 'static,
889+
>,
890+
> = Box::pin(async move {
891+
res.await.map_err(|e| {
892+
(lightning::io::Error::new(
893+
lightning::io::ErrorKind::Other,
894+
format!("failed to persist network graph, check your disk and permissions {}", e)),
895+
false)
896+
})
897+
});
898+
899+
futures.push(ResultFuture::Pending(fut));
900+
878901
have_pruned = true;
879902
}
880903
let prune_timer =
@@ -901,21 +924,28 @@ where
901924
} else {
902925
log_trace!(logger, "Persisting scorer");
903926
}
904-
if let Err(e) = kv_store
905-
.write(
906-
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
907-
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
908-
SCORER_PERSISTENCE_KEY,
909-
&scorer.encode(),
910-
)
911-
.await
912-
{
913-
log_error!(
914-
logger,
915-
"Error: Failed to persist scorer, check your disk and permissions {}",
916-
e
917-
);
918-
}
927+
let res = kv_store.write(
928+
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
929+
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
930+
SCORER_PERSISTENCE_KEY,
931+
&scorer.encode(),
932+
);
933+
let fut: Pin<
934+
Box<
935+
dyn Future<Output = Result<(), (lightning::io::Error, bool)>>
936+
+ Send
937+
+ 'static,
938+
>,
939+
> = Box::pin(async move {
940+
res.await.map_err(|e| {
941+
(lightning::io::Error::new(
942+
lightning::io::ErrorKind::Other,
943+
format!("failed to persist scorer, check your disk and permissions {}", e)),
944+
false)
945+
})
946+
});
947+
948+
futures.push(ResultFuture::Pending(fut));
919949
}
920950
last_scorer_persist_call = sleeper(SCORER_PERSIST_TIMER);
921951
},
@@ -928,14 +958,43 @@ where
928958
Some(false) => {
929959
log_trace!(logger, "Regenerating sweeper spends if necessary");
930960
if let Some(ref sweeper) = sweeper {
931-
let _ = sweeper.regenerate_and_broadcast_spend_if_necessary().await;
961+
let sweeper = sweeper.clone();
962+
let fut: Pin<
963+
Box<
964+
dyn Future<Output = Result<(), (lightning::io::Error, bool)>>
965+
+ Send
966+
+ 'static,
967+
>,
968+
> = Box::pin(async move {
969+
sweeper.regenerate_and_broadcast_spend_if_necessary().await.map_err(|_| {
970+
(lightning::io::Error::new(
971+
lightning::io::ErrorKind::Other,
972+
format!("failed to persist sweeper, check your disk and permissions")),
973+
false)
974+
})
975+
});
976+
977+
futures.push(ResultFuture::Pending(fut));
932978
}
933979
last_sweeper_call = sleeper(SWEEPER_TIMER);
934980
},
935981
Some(true) => break,
936982
None => {},
937983
}
938984

985+
// Run persistence tasks in parallel.
986+
let multi_res = MultiResultFuturePoller::new(futures).await;
987+
for res in multi_res {
988+
if let Err((e, exit)) = res {
989+
log_error!(logger, "Error: {}", e);
990+
991+
if exit {
992+
log_error!(logger, "Exiting background processor");
993+
return Err(e);
994+
}
995+
}
996+
}
997+
939998
// Onion messenger timer tick.
940999
match check_sleeper(&mut last_onion_message_handler_call) {
9411000
Some(false) => {
@@ -1025,9 +1084,9 @@ fn check_sleeper<SleepFuture: core::future::Future<Output = bool> + core::marker
10251084
/// synchronous background persistence.
10261085
pub async fn process_events_async_with_kv_store_sync<
10271086
UL: 'static + Deref,
1028-
CF: 'static + Deref,
1029-
T: 'static + Deref,
1030-
F: 'static + Deref,
1087+
CF: 'static + Deref + Sync,
1088+
T: 'static + Deref + Sync,
1089+
F: 'static + Deref + Sync,
10311090
G: 'static + Deref<Target = NetworkGraph<L>>,
10321091
L: 'static + Deref + Send + Sync,
10331092
P: 'static + Deref,
@@ -1044,10 +1103,13 @@ pub async fn process_events_async_with_kv_store_sync<
10441103
RGS: 'static + Deref<Target = RapidGossipSync<G, L>>,
10451104
PM: 'static + Deref,
10461105
LM: 'static + Deref,
1047-
D: 'static + Deref,
1048-
O: 'static + Deref,
1049-
K: 'static + Deref,
1050-
OS: 'static + Deref<Target = OutputSweeper<T, D, F, CF, KVStoreSyncWrapper<K>, L, O>>,
1106+
D: 'static + Deref + Sync,
1107+
O: 'static + Deref + Sync,
1108+
K: 'static + Deref + Sync,
1109+
OS: 'static
1110+
+ Deref<Target = OutputSweeper<T, D, F, CF, KVStoreSyncWrapper<K>, L, O>>
1111+
+ Clone
1112+
+ Send,
10511113
S: 'static + Deref<Target = SC> + Send + Sync,
10521114
SC: for<'b> WriteableScore<'b>,
10531115
SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,

‎lightning/src/util/async_poll.rs

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -15,27 +15,27 @@ use core::marker::Unpin;
1515
use core::pin::Pin;
1616
use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
1717

18-
pub(crate) enum ResultFuture<F: Future<Output = Result<(), E>>, E: Copy + Unpin> {
18+
/// A future that can be in a pending or ready state, where the ready state contains a `Result`.
19+
pub enum ResultFuture<F: Future<Output = Result<(), E>>, E: Unpin> {
20+
/// The future is still pending and needs to be polled again.
1921
Pending(F),
22+
/// The future has completed and contains a result.
2023
Ready(Result<(), E>),
2124
}
2225

23-
pub(crate) struct MultiResultFuturePoller<
24-
F: Future<Output = Result<(), E>> + Unpin,
25-
E: Copy + Unpin,
26-
> {
26+
/// A utility to poll multiple futures that return results, collecting their results into a vector.
27+
pub struct MultiResultFuturePoller<F: Future<Output = Result<(), E>> + Unpin, E: Unpin> {
2728
futures_state: Vec<ResultFuture<F, E>>,
2829
}
2930

30-
impl<F: Future<Output = Result<(), E>> + Unpin, E: Copy + Unpin> MultiResultFuturePoller<F, E> {
31+
impl<F: Future<Output = Result<(), E>> + Unpin, E: Unpin> MultiResultFuturePoller<F, E> {
32+
/// Creates a new `MultiResultFuturePoller` with the given futures.
3133
pub fn new(futures_state: Vec<ResultFuture<F, E>>) -> Self {
3234
Self { futures_state }
3335
}
3436
}
3537

36-
impl<F: Future<Output = Result<(), E>> + Unpin, E: Copy + Unpin> Future
37-
for MultiResultFuturePoller<F, E>
38-
{
38+
impl<F: Future<Output = Result<(), E>> + Unpin, E: Unpin> Future for MultiResultFuturePoller<F, E> {
3939
type Output = Vec<Result<(), E>>;
4040
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Vec<Result<(), E>>> {
4141
let mut have_pending_futures = false;

‎lightning/src/util/mod.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ pub mod ser;
3232
pub mod sweep;
3333
pub mod wakers;
3434

35-
pub(crate) mod async_poll;
35+
pub mod async_poll;
3636
pub(crate) mod atomic_counter;
3737
pub(crate) mod byte_utils;
3838
pub mod hash_tables;

0 commit comments

Comments
 (0)