Skip to content

Commit 1a78955

Browse files
joostjagerjoostjager
committed
Add async KVStore
1 parent d6b7ddd commit 1a78955

File tree

3 files changed

+398
-151
lines changed

3 files changed

+398
-151
lines changed

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

Lines changed: 143 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -42,11 +42,11 @@ use lightning::sign::EntropySource;
4242
use lightning::sign::OutputSpender;
4343
use lightning::util::logger::Logger;
4444
use lightning::util::persist::{
45-
KVStoreSync, CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
46-
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY,
47-
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
48-
SCORER_PERSISTENCE_KEY, SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
49-
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
45+
KVStore, KVStoreSync, KVStoreSyncWrapper, CHANNEL_MANAGER_PERSISTENCE_KEY,
46+
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
47+
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
48+
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, SCORER_PERSISTENCE_KEY,
49+
SCORER_PERSISTENCE_PRIMARY_NAMESPACE, SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
5050
};
5151
use lightning::util::sweep::OutputSweeper;
5252
#[cfg(feature = "std")]
@@ -318,6 +318,15 @@ fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + Wri
318318
true
319319
}
320320

321+
macro_rules! maybe_await {
322+
(true, $e:expr) => {
323+
$e.await
324+
};
325+
(false, $e:expr) => {
326+
$e
327+
};
328+
}
329+
321330
macro_rules! define_run_body {
322331
(
323332
$kv_store: ident,
@@ -327,7 +336,7 @@ macro_rules! define_run_body {
327336
$peer_manager: ident, $gossip_sync: ident,
328337
$process_sweeper: expr,
329338
$logger: ident, $scorer: ident, $loop_exit_check: expr, $await: expr, $get_timer: expr,
330-
$timer_elapsed: expr, $check_slow_await: expr, $time_fetch: expr,
339+
$timer_elapsed: expr, $check_slow_await: expr, $time_fetch: expr, $async_persist: tt,
331340
) => { {
332341
log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
333342
$channel_manager.get_cm().timer_tick_occurred();
@@ -383,12 +392,12 @@ macro_rules! define_run_body {
383392

384393
if $channel_manager.get_cm().get_and_clear_needs_persistence() {
385394
log_trace!($logger, "Persisting ChannelManager...");
386-
$kv_store.write(
395+
maybe_await!($async_persist, $kv_store.write(
387396
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
388397
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
389398
CHANNEL_MANAGER_PERSISTENCE_KEY,
390399
&$channel_manager.get_cm().encode(),
391-
)?;
400+
))?;
392401
log_trace!($logger, "Done persisting ChannelManager.");
393402
}
394403
if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
@@ -449,12 +458,12 @@ macro_rules! define_run_body {
449458
log_trace!($logger, "Persisting network graph.");
450459
}
451460

452-
if let Err(e) = $kv_store.write(
461+
if let Err(e) = maybe_await!($async_persist, $kv_store.write(
453462
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
454463
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
455464
NETWORK_GRAPH_PERSISTENCE_KEY,
456465
&network_graph.encode(),
457-
) {
466+
)) {
458467
log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
459468
}
460469

@@ -482,12 +491,12 @@ macro_rules! define_run_body {
482491
} else {
483492
log_trace!($logger, "Persisting scorer");
484493
}
485-
if let Err(e) = $kv_store.write(
494+
if let Err(e) = maybe_await!($async_persist, $kv_store.write(
486495
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
487496
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
488497
SCORER_PERSISTENCE_KEY,
489498
&scorer.encode(),
490-
) {
499+
)) {
491500
log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
492501
}
493502
}
@@ -510,31 +519,31 @@ macro_rules! define_run_body {
510519
// After we exit, ensure we persist the ChannelManager one final time - this avoids
511520
// some races where users quit while channel updates were in-flight, with
512521
// ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
513-
$kv_store.write(
522+
maybe_await!($async_persist, $kv_store.write(
514523
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
515524
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
516525
CHANNEL_MANAGER_PERSISTENCE_KEY,
517526
&$channel_manager.get_cm().encode(),
518-
)?;
527+
))?;
519528

520529
// Persist Scorer on exit
521530
if let Some(ref scorer) = $scorer {
522-
$kv_store.write(
531+
maybe_await!($async_persist, $kv_store.write(
523532
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
524533
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
525534
SCORER_PERSISTENCE_KEY,
526535
&scorer.encode(),
527-
)?;
536+
))?;
528537
}
529538

530539
// Persist NetworkGraph on exit
531540
if let Some(network_graph) = $gossip_sync.network_graph() {
532-
$kv_store.write(
541+
maybe_await!($async_persist, $kv_store.write(
533542
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
534543
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
535544
NETWORK_GRAPH_PERSISTENCE_KEY,
536545
&network_graph.encode(),
537-
)?;
546+
))?;
538547
}
539548

540549
Ok(())
@@ -681,11 +690,12 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
681690
/// ```
682691
/// # use lightning::io;
683692
/// # use lightning::events::ReplayEvent;
684-
/// # use lightning::util::sweep::OutputSweeper;
685693
/// # use std::sync::{Arc, RwLock};
686694
/// # use std::sync::atomic::{AtomicBool, Ordering};
687695
/// # use std::time::SystemTime;
688-
/// # use lightning_background_processor::{process_events_async, GossipSync};
696+
/// # use lightning_background_processor::{process_events_full_async, GossipSync};
697+
/// # use core::future::Future;
698+
/// # use core::pin::Pin;
689699
/// # struct Logger {}
690700
/// # impl lightning::util::logger::Logger for Logger {
691701
/// # fn log(&self, _record: lightning::util::logger::Record) {}
@@ -697,6 +707,13 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
697707
/// # fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> io::Result<()> { Ok(()) }
698708
/// # fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
699709
/// # }
710+
/// # struct Store {}
711+
/// # impl lightning::util::persist::KVStore for Store {
712+
/// # fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> Pin<Box<dyn Future<Output = Result<Vec<u8>, io::Error>> + 'static + Send>> { todo!() }
713+
/// # fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static + Send>> { todo!() }
714+
/// # fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static + Send>> { todo!() }
715+
/// # fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> Pin<Box<dyn Future<Output = Result<Vec<String>, io::Error>> + 'static + Send>> { todo!() }
716+
/// # }
700717
/// # struct EventHandler {}
701718
/// # impl EventHandler {
702719
/// # async fn handle_event(&self, _: lightning::events::Event) -> Result<(), ReplayEvent> { Ok(()) }
@@ -715,7 +732,8 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
715732
/// # type LiquidityManager<B, F, FE> = lightning_liquidity::LiquidityManager<Arc<lightning::sign::KeysManager>, Arc<ChannelManager<B, F, FE>>, Arc<F>>;
716733
/// # type Scorer = RwLock<lightning::routing::scoring::ProbabilisticScorer<Arc<NetworkGraph>, Arc<Logger>>>;
717734
/// # type PeerManager<B, F, FE, UL> = lightning::ln::peer_handler::SimpleArcPeerManager<SocketDescriptor, ChainMonitor<B, F, FE>, B, FE, Arc<UL>, Logger, F, StoreSync>;
718-
/// #
735+
/// # type OutputSweeper<B, D, FE, F, O> = lightning::util::sweep::OutputSweeper<Arc<B>, Arc<D>, Arc<FE>, Arc<F>, Arc<Store>, Arc<Logger>, Arc<O>>;
736+
///
719737
/// # struct Node<
720738
/// # B: lightning::chain::chaininterface::BroadcasterInterface + Send + Sync + 'static,
721739
/// # F: lightning::chain::Filter + Send + Sync + 'static,
@@ -731,10 +749,10 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
731749
/// # liquidity_manager: Arc<LiquidityManager<B, F, FE>>,
732750
/// # chain_monitor: Arc<ChainMonitor<B, F, FE>>,
733751
/// # gossip_sync: Arc<P2PGossipSync<UL>>,
734-
/// # persister: Arc<StoreSync>,
752+
/// # persister: Arc<Store>,
735753
/// # logger: Arc<Logger>,
736754
/// # scorer: Arc<Scorer>,
737-
/// # sweeper: Arc<OutputSweeper<Arc<B>, Arc<D>, Arc<FE>, Arc<F>, Arc<StoreSync>, Arc<Logger>, Arc<O>>>,
755+
/// # sweeper: Arc<OutputSweeper<B, D, FE, F, O>>,
738756
/// # }
739757
/// #
740758
/// # async fn setup_background_processing<
@@ -780,7 +798,7 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
780798
doc = " let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();"
781799
)]
782800
#[cfg_attr(not(feature = "std"), doc = " rt.block_on(async move {")]
783-
/// process_events_async(
801+
/// process_events_full_async(
784802
/// background_persister,
785803
/// |e| background_event_handler.handle_event(e),
786804
/// background_chain_mon,
@@ -805,7 +823,7 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
805823
#[cfg_attr(feature = "std", doc = " handle.await.unwrap()")]
806824
/// # }
807825
///```
808-
pub async fn process_events_async<
826+
pub async fn process_events_full_async<
809827
'a,
810828
UL: 'static + Deref,
811829
CF: 'static + Deref,
@@ -856,7 +874,7 @@ where
856874
LM::Target: ALiquidityManager,
857875
O::Target: 'static + OutputSpender,
858876
D::Target: 'static + ChangeDestinationSource,
859-
K::Target: 'static + KVStoreSync,
877+
K::Target: 'static + KVStore,
860878
{
861879
let mut should_break = false;
862880
let async_event_handler = |event| {
@@ -875,12 +893,15 @@ where
875893
if let Some(duration_since_epoch) = fetch_time() {
876894
if update_scorer(scorer, &event, duration_since_epoch) {
877895
log_trace!(logger, "Persisting scorer after update");
878-
if let Err(e) = kv_store.write(
879-
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
880-
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
881-
SCORER_PERSISTENCE_KEY,
882-
&scorer.encode(),
883-
) {
896+
if let Err(e) = kv_store
897+
.write(
898+
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
899+
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
900+
SCORER_PERSISTENCE_KEY,
901+
&scorer.encode(),
902+
)
903+
.await
904+
{
884905
log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e);
885906
// We opt not to abort early on persistence failure here as persisting
886907
// the scorer is non-critical and we still hope that it will have
@@ -958,7 +979,82 @@ where
958979
},
959980
mobile_interruptable_platform,
960981
fetch_time,
982+
true,
983+
)
984+
}
985+
986+
/// Async events processor that is based on [`process_events_async`] but allows for [`KVStoreSync`] to be used for
987+
/// synchronous background persistence.
988+
pub async fn process_events_async<
989+
UL: 'static + Deref,
990+
CF: 'static + Deref,
991+
T: 'static + Deref,
992+
F: 'static + Deref,
993+
G: 'static + Deref<Target = NetworkGraph<L>>,
994+
L: 'static + Deref + Send + Sync,
995+
P: 'static + Deref,
996+
EventHandlerFuture: core::future::Future<Output = Result<(), ReplayEvent>>,
997+
EventHandler: Fn(Event) -> EventHandlerFuture,
998+
ES: 'static + Deref + Send,
999+
M: 'static
1000+
+ Deref<Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P, ES>>
1001+
+ Send
1002+
+ Sync,
1003+
CM: 'static + Deref + Send + Sync,
1004+
OM: 'static + Deref,
1005+
PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>>,
1006+
RGS: 'static + Deref<Target = RapidGossipSync<G, L>>,
1007+
PM: 'static + Deref,
1008+
LM: 'static + Deref,
1009+
D: 'static + Deref,
1010+
O: 'static + Deref,
1011+
K: 'static + Deref,
1012+
OS: 'static + Deref<Target = OutputSweeper<T, D, F, CF, KVStoreSyncWrapper<K>, L, O>>,
1013+
S: 'static + Deref<Target = SC> + Send + Sync,
1014+
SC: for<'b> WriteableScore<'b>,
1015+
SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
1016+
Sleeper: Fn(Duration) -> SleepFuture,
1017+
FetchTime: Fn() -> Option<Duration>,
1018+
>(
1019+
kv_store: K, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
1020+
onion_messenger: Option<OM>, gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM,
1021+
liquidity_manager: Option<LM>, sweeper: Option<OS>, logger: L, scorer: Option<S>,
1022+
sleeper: Sleeper, mobile_interruptable_platform: bool, fetch_time: FetchTime,
1023+
) -> Result<(), lightning::io::Error>
1024+
where
1025+
UL::Target: 'static + UtxoLookup,
1026+
CF::Target: 'static + chain::Filter,
1027+
T::Target: 'static + BroadcasterInterface,
1028+
F::Target: 'static + FeeEstimator,
1029+
L::Target: 'static + Logger,
1030+
P::Target: 'static + Persist<<CM::Target as AChannelManager>::Signer>,
1031+
ES::Target: 'static + EntropySource,
1032+
CM::Target: AChannelManager,
1033+
OM::Target: AOnionMessenger,
1034+
PM::Target: APeerManager,
1035+
LM::Target: ALiquidityManager,
1036+
O::Target: 'static + OutputSpender,
1037+
D::Target: 'static + ChangeDestinationSource,
1038+
K::Target: 'static + KVStoreSync,
1039+
{
1040+
let kv_store = KVStoreSyncWrapper(kv_store);
1041+
process_events_full_async(
1042+
kv_store,
1043+
event_handler,
1044+
chain_monitor,
1045+
channel_manager,
1046+
onion_messenger,
1047+
gossip_sync,
1048+
peer_manager,
1049+
liquidity_manager,
1050+
sweeper,
1051+
logger,
1052+
scorer,
1053+
sleeper,
1054+
mobile_interruptable_platform,
1055+
fetch_time,
9611056
)
1057+
.await
9621058
}
9631059

9641060
#[cfg(feature = "std")]
@@ -1138,6 +1234,7 @@ impl BackgroundProcessor {
11381234
.expect("Time should be sometime after 1970"),
11391235
)
11401236
},
1237+
false,
11411238
)
11421239
});
11431240
Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
@@ -1226,7 +1323,7 @@ mod tests {
12261323
use lightning::types::payment::PaymentHash;
12271324
use lightning::util::config::UserConfig;
12281325
use lightning::util::persist::{
1229-
KVStoreSync, CHANNEL_MANAGER_PERSISTENCE_KEY,
1326+
KVStoreSync, KVStoreSyncWrapper, CHANNEL_MANAGER_PERSISTENCE_KEY,
12301327
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
12311328
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY,
12321329
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
@@ -2151,12 +2248,13 @@ mod tests {
21512248
open_channel!(nodes[0], nodes[1], 100000);
21522249

21532250
let data_dir = nodes[0].kv_store.get_data_dir();
2154-
let persister = Arc::new(
2251+
let kv_store_sync = Arc::new(
21552252
Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"),
21562253
);
2254+
let kv_store = Arc::new(KVStoreSyncWrapper(kv_store_sync));
21572255

2158-
let bp_future = super::process_events_async(
2159-
persister,
2256+
let bp_future = super::process_events_full_async(
2257+
kv_store,
21602258
|_: _| async { Ok(()) },
21612259
Arc::clone(&nodes[0].chain_monitor),
21622260
Arc::clone(&nodes[0].node),
@@ -2659,11 +2757,13 @@ mod tests {
26592757
let (_, nodes) =
26602758
create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async");
26612759
let data_dir = nodes[0].kv_store.get_data_dir();
2662-
let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
2760+
let kv_store_sync =
2761+
Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
2762+
let kv_store = Arc::new(KVStoreSyncWrapper(kv_store_sync));
26632763

26642764
let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
2665-
let bp_future = super::process_events_async(
2666-
persister,
2765+
let bp_future = super::process_events_full_async(
2766+
kv_store,
26672767
|_: _| async { Ok(()) },
26682768
Arc::clone(&nodes[0].chain_monitor),
26692769
Arc::clone(&nodes[0].node),
@@ -2875,12 +2975,13 @@ mod tests {
28752975

28762976
let (_, nodes) = create_nodes(1, "test_payment_path_scoring_async");
28772977
let data_dir = nodes[0].kv_store.get_data_dir();
2878-
let persister = Arc::new(Persister::new(data_dir));
2978+
let kv_store_sync = Arc::new(Persister::new(data_dir));
2979+
let kv_store = Arc::new(KVStoreSyncWrapper(kv_store_sync));
28792980

28802981
let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
28812982

2882-
let bp_future = super::process_events_async(
2883-
persister,
2983+
let bp_future = super::process_events_full_async(
2984+
kv_store,
28842985
event_handler,
28852986
Arc::clone(&nodes[0].chain_monitor),
28862987
Arc::clone(&nodes[0].node),

0 commit comments

Comments
 (0)