From 3f77a1de51c39809b45b6065d1d533a2e7080a7f Mon Sep 17 00:00:00 2001 From: Arik Sosman Date: Mon, 11 Mar 2024 23:41:27 -0700 Subject: [PATCH] Use separate tokio runtime for gossip persistence. --- src/persistence.rs | 8 +++++-- src/tests/mod.rs | 56 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/src/persistence.rs b/src/persistence.rs index 9dc537d..8e299ee 100644 --- a/src/persistence.rs +++ b/src/persistence.rs @@ -8,6 +8,7 @@ use lightning::log_info; use lightning::routing::gossip::NetworkGraph; use lightning::util::logger::Logger; use lightning::util::ser::Writeable; +use tokio::runtime::Runtime; use tokio::sync::{mpsc, Mutex, Semaphore}; use crate::config; @@ -19,6 +20,7 @@ const INSERT_PARALELLISM: usize = 16; pub(crate) struct GossipPersister where L::Target: Logger { gossip_persistence_receiver: mpsc::Receiver, network_graph: Arc>, + tokio_runtime: Runtime, logger: L } @@ -26,9 +28,11 @@ impl GossipPersister where L::Target: Logger { pub fn new(network_graph: Arc>, logger: L) -> (Self, mpsc::Sender) { let (gossip_persistence_sender, gossip_persistence_receiver) = mpsc::channel::(100); + let runtime = Runtime::new().unwrap(); (GossipPersister { gossip_persistence_receiver, network_graph, + tokio_runtime: runtime, logger }, gossip_persistence_sender) } @@ -127,7 +131,7 @@ impl GossipPersister where L::Target: Logger { let mut announcement_signed = Vec::new(); announcement.write(&mut announcement_signed).unwrap(); - let _task = tokio::spawn(async move { + let _task = self.tokio_runtime.spawn(async move { let client; { let mut connections_set = connections_cache_ref.lock().await; @@ -219,7 +223,7 @@ impl GossipPersister where L::Target: Logger { // this may not be used outside test cfg let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64; - let _task = tokio::spawn(async move { + let _task = self.tokio_runtime.spawn(async move { let client; { let mut connections_set = connections_cache_ref.lock().await; diff --git a/src/tests/mod.rs b/src/tests/mod.rs index 42de18a..ef25ef0 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -173,6 +173,22 @@ async fn clean_test_db() { }); } +#[tokio::test] +async fn test_persistence_runtime() { + let _sanitizer = SchemaSanitizer::new(); + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + let (_persister, _receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + + tokio::task::spawn_blocking(move || { + drop(_persister); + }).await.unwrap(); + + clean_test_db().await; +} + + #[tokio::test] async fn test_trivial_setup() { let _sanitizer = SchemaSanitizer::new(); @@ -240,6 +256,10 @@ async fn test_trivial_setup() { println!("last update b: {}", last_update_seen_b); assert_eq!(last_update_seen_a, update_result - CLIENT_BACKDATE_INTERVAL); assert_eq!(last_update_seen_b, update_result - CLIENT_BACKDATE_INTERVAL); + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); } /// If a channel has only seen updates in one direction, it should not be announced @@ -303,6 +323,10 @@ async fn test_unidirectional_intermediate_update_consideration() { let client_channel_count = channels.len(); assert_eq!(client_channel_count, 1); + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + clean_test_db().await; } @@ -357,6 +381,10 @@ async fn test_bidirectional_intermediate_update_consideration() { assert_eq!(serialization.update_count_full, 0); assert_eq!(serialization.update_count_incremental, 1); + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + clean_test_db().await; } @@ -399,6 +427,10 @@ async fn test_full_snapshot_recency() { drop(receiver); persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); } let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); @@ -475,6 +507,10 @@ async fn test_full_snapshot_recency_with_wrong_seen_order() { drop(receiver); persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); } let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); @@ -550,6 +586,10 @@ async fn test_full_snapshot_recency_with_wrong_propagation_order() { drop(receiver); persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); } let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); @@ -679,6 +719,10 @@ async fn test_full_snapshot_mutiny_scenario() { drop(receiver); persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); } let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); @@ -788,6 +832,10 @@ async fn test_full_snapshot_interlaced_channel_timestamps() { drop(receiver); persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); } let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); @@ -859,6 +907,10 @@ async fn test_full_snapshot_persistence() { drop(receiver); persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); } let cache_path = cache_sanitizer.cache_path(); @@ -900,6 +952,10 @@ async fn test_full_snapshot_persistence() { drop(receiver); persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); } // regenerate snapshots -- 2.39.5