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 dd078904916fc7787e59f87c1b2f6b65d5a8d72a..4318a72ca8b65a41e5207a083dbff8491dbcf408 100644 (file)
--- 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 9dc537d835530f6c0bd63870785a11b21b4c031d..857cbc34da21f71525dd4020a947f8ea80985a0f 100644 (file)
--- 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<L: Deref> where L::Target: Logger {
        gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
        network_graph: Arc<NetworkGraph<L>>,
+       tokio_runtime: Runtime,
        logger: L
 }
 
@@ -26,18 +27,20 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger {
        pub fn new(network_graph: Arc<NetworkGraph<L>>, logger: L) -> (Self, mpsc::Sender<GossipMessage>) {
                let (gossip_persistence_sender, gossip_persistence_receiver) =
                        mpsc::channel::<GossipMessage>(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<L: Deref> GossipPersister<L> 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<L: Deref> GossipPersister<L> 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<L: Deref> GossipPersister<L> 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 42de18a4fb63f7aea2ec4a3a19f34b393a26dc07..ef25ef0782d947c1e591a1cd348c6b39c1e80fb4 100644 (file)
--- 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 2f1e3b32d7bef336b58db16f9756b27af554b1dd..c8d6bac888bf5f292ad624d6e521cc3623dfd561 100644 (file)
--- 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 3d58ee0b7bab4c5349b7170692133e321383869b..c44cb663158e19d7e1c850b22bd57f6f952665c1 100644 (file)
--- 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;