From a9e129c3dcd0ce5b036fe6dd7e170450627be705 Mon Sep 17 00:00:00 2001 From: Arik Sosman Date: Wed, 20 Mar 2024 17:45:18 -0700 Subject: [PATCH 1/3] Fix CI compilation in beta environment. --- src/config.rs | 1 - src/tracking.rs | 1 - src/verifier.rs | 1 - 3 files changed, 3 deletions(-) diff --git a/src/config.rs b/src/config.rs index dd07890..4318a72 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,6 +1,5 @@ use crate::hex_utils; -use std::convert::TryInto; use std::env; use std::io::Cursor; use std::net::{SocketAddr, ToSocketAddrs}; diff --git a/src/tracking.rs b/src/tracking.rs index 2f1e3b3..c8d6bac 100644 --- a/src/tracking.rs +++ b/src/tracking.rs @@ -7,7 +7,6 @@ use std::time::{Duration, Instant}; use bitcoin::hashes::hex::ToHex; use bitcoin::secp256k1::PublicKey; -use lightning; use lightning::ln::peer_handler::{ ErroringMessageHandler, IgnoringMessageHandler, MessageHandler, PeerManager, }; diff --git a/src/verifier.rs b/src/verifier.rs index 3d58ee0..c44cb66 100644 --- a/src/verifier.rs +++ b/src/verifier.rs @@ -1,4 +1,3 @@ -use std::convert::TryInto; use std::io::ErrorKind; use std::ops::Deref; use std::sync::Arc; From 3f77a1de51c39809b45b6065d1d533a2e7080a7f Mon Sep 17 00:00:00 2001 From: Arik Sosman Date: Mon, 11 Mar 2024 23:41:27 -0700 Subject: [PATCH 2/3] 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 From 418d7765b5f0f7b443a2e53ccc78f159f3d4b91d Mon Sep 17 00:00:00 2001 From: Arik Sosman Date: Wed, 20 Mar 2024 17:41:34 -0700 Subject: [PATCH 3/3] Simplify client creation. --- src/persistence.rs | 37 +++++++++++++------------------------ 1 file changed, 13 insertions(+), 24 deletions(-) diff --git a/src/persistence.rs b/src/persistence.rs index 8e299ee..857cbc3 100644 --- a/src/persistence.rs +++ b/src/persistence.rs @@ -1,7 +1,6 @@ use std::fs::OpenOptions; use std::io::{BufWriter, Write}; use std::ops::Deref; -use std::mem; use std::sync::Arc; use std::time::{Duration, Instant}; use lightning::log_info; @@ -38,10 +37,10 @@ impl GossipPersister where L::Target: Logger { } pub(crate) async fn persist_gossip(&mut self) { - let mut client = crate::connect_to_db().await; + { // initialize the database + // this client instance is only used once + let mut client = crate::connect_to_db().await; - { - // initialize the database let initialization = client .execute(config::db_config_table_creation_query(), &[]) .await; @@ -122,6 +121,16 @@ impl GossipPersister where L::Target: Logger { insert_limiter.acquire().await.unwrap().forget(); let limiter_ref = Arc::clone(&insert_limiter); + let client = { + let mut connections_set = connections_cache.lock().await; + let client = if connections_set.is_empty() { + crate::connect_to_db().await + } else { + connections_set.pop().unwrap() + }; + client + }; + let connections_cache_ref = Arc::clone(&connections_cache); match gossip_message { GossipMessage::ChannelAnnouncement(announcement, seen_override) => { @@ -132,16 +141,6 @@ impl GossipPersister where L::Target: Logger { announcement.write(&mut announcement_signed).unwrap(); let _task = self.tokio_runtime.spawn(async move { - let client; - { - let mut connections_set = connections_cache_ref.lock().await; - if connections_set.is_empty() { - mem::drop(connections_set); - client = crate::connect_to_db().await; - } else { - client = connections_set.pop().unwrap(); - } - } if cfg!(test) && seen_override.is_some() { tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client .execute("INSERT INTO channel_announcements (\ @@ -224,16 +223,6 @@ impl GossipPersister where L::Target: Logger { let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64; let _task = self.tokio_runtime.spawn(async move { - let client; - { - let mut connections_set = connections_cache_ref.lock().await; - if connections_set.is_empty() { - mem::drop(connections_set); - client = crate::connect_to_db().await; - } else { - client = connections_set.pop().unwrap(); - } - } tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client .execute(insertion_statement, &[ &scid,