X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fpersistence.rs;h=04c6b9a0c4076cab725b90b5a4ec769718d183a0;hb=49858cac9e83132572ea59b786922ab4a135f48a;hp=4ede02f5258cac98cc977f71853f120d17e7aadf;hpb=f1a3c914dfccbdf1ed38d2b51075015b5d766498;p=rapid-gossip-sync-server diff --git a/src/persistence.rs b/src/persistence.rs index 4ede02f..04c6b9a 100644 --- a/src/persistence.rs +++ b/src/persistence.rs @@ -7,16 +7,19 @@ use lightning::log_info; use lightning::routing::gossip::NetworkGraph; use lightning::util::logger::Logger; use lightning::util::ser::Writeable; -use tokio::sync::mpsc; +use tokio::runtime::Runtime; +use tokio::sync::{mpsc, Mutex, Semaphore}; use crate::config; use crate::types::GossipMessage; const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15); +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 } @@ -24,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; @@ -64,21 +69,20 @@ impl GossipPersister where L::Target: Logger { panic!("db init error: {}", initialization_error); } - let initialization = client - .execute(config::db_announcement_table_creation_query(), &[]) - .await; - if let Err(initialization_error) = initialization { - panic!("db init error: {}", initialization_error); - } + let table_creation_queries = [ + config::db_announcement_table_creation_query(), + config::db_channel_update_table_creation_query(), + config::db_channel_update_table_creation_query(), + config::db_node_announcement_table_creation_query() + ]; - let initialization = client - .execute( - config::db_channel_update_table_creation_query(), - &[], - ) - .await; - if let Err(initialization_error) = initialization { - panic!("db init error: {}", initialization_error); + for current_table_creation_query in table_creation_queries { + let initialization = client + .execute(current_table_creation_query, &[]) + .await; + if let Err(initialization_error) = initialization { + panic!("db init error: {}", initialization_error); + } } let initialization = client @@ -93,10 +97,14 @@ impl GossipPersister where L::Target: Logger { let mut latest_persistence_log = Instant::now() - Duration::from_secs(60); let mut i = 0u32; let mut latest_graph_cache_time = Instant::now(); + let insert_limiter = Arc::new(Semaphore::new(INSERT_PARALELLISM)); + let connections_cache = Arc::new(Mutex::new(Vec::with_capacity(INSERT_PARALELLISM))); + #[cfg(test)] + let mut tasks_spawned = Vec::new(); // TODO: it would be nice to have some sort of timeout here so after 10 seconds of // inactivity, some sort of message could be broadcast signaling the activation of request // processing - while let Some(gossip_message) = &self.gossip_persistence_receiver.recv().await { + while let Some(gossip_message) = self.gossip_persistence_receiver.recv().await { i += 1; // count the persisted gossip messages if latest_persistence_log.elapsed().as_secs() >= 60 { @@ -109,8 +117,74 @@ impl GossipPersister where L::Target: Logger { self.persist_network_graph(); latest_graph_cache_time = Instant::now(); } + 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::NodeAnnouncement(announcement, seen_override) => { + let public_key_hex = announcement.contents.node_id.to_string(); + + let mut announcement_signed = Vec::new(); + announcement.write(&mut announcement_signed).unwrap(); - match &gossip_message { + let features = announcement.contents.features.encode(); + let timestamp = announcement.contents.timestamp as i64; + + let mut serialized_addresses = Vec::new(); + announcement.contents.addresses.write(&mut serialized_addresses).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 node_announcements (\ + public_key, \ + features, \ + socket_addresses, \ + timestamp, \ + announcement_signed, \ + seen \ + ) VALUES ($1, $2, $3, $4, $5, TO_TIMESTAMP($6))", &[ + &public_key_hex, + &features, + &serialized_addresses, + ×tamp, + &announcement_signed, + &(seen_override.unwrap() as f64) + ])).await.unwrap().unwrap(); + } else { + tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client + .execute("INSERT INTO node_announcements (\ + public_key, \ + features, \ + socket_addresses, \ + timestamp, \ + announcement_signed \ + ) VALUES ($1, $2, $3, $4, $5)", &[ + &public_key_hex, + &features, + &serialized_addresses, + ×tamp, + &announcement_signed, + ])).await.unwrap().unwrap(); + } + let mut connections_set = connections_cache_ref.lock().await; + connections_set.push(client); + limiter_ref.add_permits(1); + }); + #[cfg(test)] + tasks_spawned.push(_task); + }, GossipMessage::ChannelAnnouncement(announcement, seen_override) => { let scid = announcement.contents.short_channel_id as i64; @@ -118,27 +192,34 @@ impl GossipPersister where L::Target: Logger { let mut announcement_signed = Vec::new(); announcement.write(&mut announcement_signed).unwrap(); - if cfg!(test) && seen_override.is_some() { - tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client - .execute("INSERT INTO channel_announcements (\ - short_channel_id, \ - announcement_signed, \ - seen \ - ) VALUES ($1, $2, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[ - &scid, - &announcement_signed, - &(seen_override.unwrap() as f64) - ])).await.unwrap().unwrap(); - } else { - tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client - .execute("INSERT INTO channel_announcements (\ - short_channel_id, \ - announcement_signed \ - ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[ - &scid, - &announcement_signed - ])).await.unwrap().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 (\ + short_channel_id, \ + announcement_signed, \ + seen \ + ) VALUES ($1, $2, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[ + &scid, + &announcement_signed, + &(seen_override.unwrap() as f64) + ])).await.unwrap().unwrap(); + } else { + tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client + .execute("INSERT INTO channel_announcements (\ + short_channel_id, \ + announcement_signed \ + ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[ + &scid, + &announcement_signed + ])).await.unwrap().unwrap(); + } + let mut connections_set = connections_cache_ref.lock().await; + connections_set.push(client); + limiter_ref.add_permits(1); + }); + #[cfg(test)] + tasks_spawned.push(_task); } GossipMessage::ChannelUpdate(update, seen_override) => { let scid = update.contents.short_channel_id as i64; @@ -193,25 +274,36 @@ 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; - tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client - .execute(insertion_statement, &[ - &scid, - ×tamp, - #[cfg(test)] - &_seen_timestamp, - &(update.contents.flags as i16), - &direction, - &disable, - &cltv_expiry_delta, - &htlc_minimum_msat, - &fee_base_msat, - &fee_proportional_millionths, - &htlc_maximum_msat, - &update_signed - ])).await.unwrap().unwrap(); + let _task = self.tokio_runtime.spawn(async move { + tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client + .execute(insertion_statement, &[ + &scid, + ×tamp, + #[cfg(test)] + &_seen_timestamp, + &(update.contents.flags as i16), + &direction, + &disable, + &cltv_expiry_delta, + &htlc_minimum_msat, + &fee_base_msat, + &fee_proportional_millionths, + &htlc_maximum_msat, + &update_signed + ])).await.unwrap().unwrap(); + let mut connections_set = connections_cache_ref.lock().await; + connections_set.push(client); + limiter_ref.add_permits(1); + }); + #[cfg(test)] + tasks_spawned.push(_task); } } } + #[cfg(test)] + for task in tasks_spawned { + task.await.unwrap(); + } } fn persist_network_graph(&self) {