From: Matt Corallo <649246+TheBlueMatt@users.noreply.github.com> Date: Fri, 22 Mar 2024 22:44:41 +0000 (+0000) Subject: Merge pull request #73 from arik-so/arik/2024-03-tokio-runtime-split X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=commitdiff_plain;h=47cb2cab4456e966ac1ed453c2b6e11136f4b5ee;hp=eaebdf774e7f3f0f87b459f83a84c4be12313da1;p=rapid-gossip-sync-server Merge pull request #73 from arik-so/arik/2024-03-tokio-runtime-split Use separate Tokio runtime for gossip persistence. --- 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/persistence.rs b/src/persistence.rs index 9dc537d..857cbc3 100644 --- a/src/persistence.rs +++ b/src/persistence.rs @@ -1,13 +1,13 @@ 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; 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 +19,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,18 +27,20 @@ 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) } 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; @@ -118,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) => { @@ -127,17 +140,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 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(); - } - } + let _task = self.tokio_runtime.spawn(async move { if cfg!(test) && seen_override.is_some() { tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client .execute("INSERT INTO channel_announcements (\ @@ -219,17 +222,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 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(); - } - } + let _task = self.tokio_runtime.spawn(async move { tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client .execute(insertion_statement, &[ &scid, 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 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;