Move network graph persistence to its own tokio task

The DB persistence loop is responsible for taking new gossip off of
the persistence queue and writing it to Postgres. If it gets
stalled, the logic pushing gossip onto the event queue might also
stall. To reduce the dependence of that on taking the
`NetworkGraph` mutexes and writing the full, large, graph to disk,
move it to its own task.

Author: TheBlueMatt

src/lib.rs

@@ -11,9 +11,10 @@ extern crate core;
 
 use std::collections::{HashMap, HashSet};
 use std::fs::File;
-use std::io::BufReader;
+use std::io::{BufReader, BufWriter, Write};
 use std::ops::Deref;
 use std::sync::Arc;
+use std::time::Duration;
 use bitcoin::blockdata::constants::ChainHash;
 use lightning::log_info;
 
@@ -104,7 +105,7 @@ impl<L: Deref + Clone + Send + Sync + 'static> RapidSyncProcessor<L> where L::Ta
 
 		if config::DOWNLOAD_NEW_GOSSIP {
 			let (mut persister, persistence_sender) =
-				GossipPersister::new(self.network_graph.clone(), self.logger.clone()).await;
+				GossipPersister::new(self.logger.clone()).await;
 			log_info!(self.logger, "Starting gossip db persistence listener");
 			tokio::spawn(async move { persister.persist_gossip().await; });
 
@@ -130,6 +131,16 @@ impl<L: Deref + Clone + Send + Sync + 'static> RapidSyncProcessor<L> where L::Ta
 			log_info!(self.logger, "Starting gossip download");
 			tokio::spawn(tracking::download_gossip(persistence_sender, sync_completion_sender,
 				Arc::clone(&self.network_graph), self.logger.clone()));
+
+			let graph = Arc::clone(&self.network_graph);
+			let logger = self.logger.clone();
+			tokio::spawn(async move {
+				let mut intvl = tokio::time::interval(Duration::from_secs(60 * 10));
+				loop {
+					intvl.tick().await;
+					persist_network_graph(&logger, &*graph);
+				}
+			});
 		} else {
 			sync_completion_sender.send(()).await.unwrap();
 		}
@@ -145,6 +156,22 @@ impl<L: Deref + Clone + Send + Sync + 'static> RapidSyncProcessor<L> where L::Ta
 	}
 }
 
+fn persist_network_graph<L: Deref>(logger: &L, graph: &NetworkGraph<L>) where L::Target: Logger {
+	log_info!(logger, "Caching network graph…");
+	let cache_path = config::network_graph_cache_path();
+	let file = std::fs::OpenOptions::new()
+		.create(true)
+		.write(true)
+		.truncate(true)
+		.open(&cache_path)
+		.unwrap();
+	graph.remove_stale_channels_and_tracking();
+	let mut writer = BufWriter::new(file);
+	graph.write(&mut writer).unwrap();
+	writer.flush().unwrap();
+	log_info!(logger, "Cached network graph!");
+}
+
 pub(crate) async fn connect_to_db() -> Client {
 	let connection_config = config::db_connection_config();
 	let (client, connection) = connection_config.connect(NoTls).await.unwrap();

src/persistence.rs

@@ -1,10 +1,7 @@
-use std::fs::OpenOptions;
-use std::io::{BufWriter, Write};
 use std::ops::Deref;
 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;
@@ -18,13 +15,12 @@ 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
 }
 
 impl<L: Deref + Clone + Send + Sync + 'static> GossipPersister<L> where L::Target: Logger {
-	pub async fn new(network_graph: Arc<NetworkGraph<L>>, logger: L) -> (Self, mpsc::Sender<GossipMessage>) {
+	pub async fn new(logger: L) -> (Self, mpsc::Sender<GossipMessage>) {
 		{ // initialize the database
 			// this client instance is only used once
 			let mut client = crate::connect_to_db().await;
@@ -86,7 +82,6 @@ impl<L: Deref + Clone + Send + Sync + 'static> GossipPersister<L> where L::Targe
 		let runtime = Runtime::new().unwrap();
 		(GossipPersister {
 			gossip_persistence_receiver,
-			network_graph,
 			tokio_runtime: runtime,
 			logger
 		}, gossip_persistence_sender)
@@ -96,7 +91,6 @@ impl<L: Deref + Clone + Send + Sync + 'static> GossipPersister<L> where L::Targe
 		// print log statement every minute
 		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)]
@@ -112,11 +106,6 @@ impl<L: Deref + Clone + Send + Sync + 'static> GossipPersister<L> where L::Targe
 				latest_persistence_log = Instant::now();
 			}
 
-			// has it been ten minutes? Just cache it
-			if latest_graph_cache_time.elapsed().as_secs() >= 600 {
-				self.persist_network_graph();
-				latest_graph_cache_time = Instant::now();
-			}
 			insert_limiter.acquire().await.unwrap().forget();
 
 			let limiter_ref = Arc::clone(&insert_limiter);
@@ -309,20 +298,4 @@ impl<L: Deref + Clone + Send + Sync + 'static> GossipPersister<L> where L::Targe
 			task.await.unwrap();
 		}
 	}
-
-	fn persist_network_graph(&self) {
-		log_info!(self.logger, "Caching network graph…");
-		let cache_path = config::network_graph_cache_path();
-		let file = OpenOptions::new()
-			.create(true)
-			.write(true)
-			.truncate(true)
-			.open(&cache_path)
-			.unwrap();
-		self.network_graph.remove_stale_channels_and_tracking();
-		let mut writer = BufWriter::new(file);
-		self.network_graph.write(&mut writer).unwrap();
-		writer.flush().unwrap();
-		log_info!(self.logger, "Cached network graph!");
-	}
 }