accelerate local syncing mechanism to first take all announcements, and then all updates

Author: arik-so

lightning-graph-sync/src/lib.rs

@@ -54,13 +54,14 @@ where
 
 #[cfg(test)]
 mod tests {
-    use bitcoin::blockdata::constants::genesis_block;
+	use std::time::Instant;
+	use bitcoin::blockdata::constants::genesis_block;
     use bitcoin::Network;
 
-    use lightning::routing::network_graph::NetworkGraph;
+	use lightning::routing::network_graph::NetworkGraph;
 
     use crate::error::GraphSyncError;
-    use crate::sync_network_graph_with_url;
+    use crate::{sync_network_graph_with_file_path, sync_network_graph_with_url};
 
 
 
@@ -92,4 +93,29 @@ mod tests {
 
         assert!(sync_result.is_ok())
     }
+
+	#[test]
+	fn test_file_load() {
+		let block_hash = genesis_block(Network::Bitcoin).block_hash();
+		let network_graph = NetworkGraph::new(block_hash);
+
+		let chain_source: Option<Box<dyn lightning::chain::Access>> = None;
+
+		let start = Instant::now();
+
+		let path = "/Users/arik/Developer/rust-lightning/lightning-graph-sync/res/full_compressed.lngossip";
+		// let path = "/Users/arik/Developer/rust-lightning/lightning-graph-sync/res/one_day.lngossip";
+		// let path = "/Users/arik/Developer/rust-lightning/lightning-graph-sync/res/one_week.lngossip";
+
+		let sync_result = sync_network_graph_with_file_path(
+			&network_graph,
+			path,
+			&chain_source,
+		);
+
+		let duration = start.elapsed();
+		let elapsed_time = format!("{:?}", duration);
+
+		println!("sync result: {}, time: {}", sync_result.is_ok(), elapsed_time);
+	}
 }

lightning-graph-sync/src/processing.rs

@@ -50,48 +50,28 @@ where
 
     // shadow variable for ownership release after length-restricted reads
     let mut read_cursor = read_cursor;
-    loop {
-        let message_length_result = Readable::read(read_cursor);
-        let message_length: BigSize = if let Ok(message_length) = message_length_result {
-            message_length
-        } else if let Err(msgs::DecodeError::ShortRead) = message_length_result {
-            break;
-        } else {
-            return Err(message_length_result.err().unwrap().into());
-        };
-        // let mut restricted_cursor = FixedLengthReader::new(&read_cursor, message_length.0);
-        let mut restricted_cursor = read_cursor.take(message_length.0);
-
-        let message_type: u16 = Readable::read(&mut restricted_cursor)?;
-
-        let update_result = match message_type {
-            msgs::ChannelAnnouncement::TYPE => {
-                let channel_announcement: msgs::UnsignedChannelAnnouncement =
-                    Readable::read(&mut restricted_cursor).unwrap();
-                network_graph
-                    .update_channel_from_unsigned_announcement(&channel_announcement, &chain_source)
-            }
-            msgs::NodeAnnouncement::TYPE => {
-                let node_announcement: msgs::UnsignedNodeAnnouncement =
-                    Readable::read(&mut restricted_cursor)?;
-                network_graph.update_node_from_unsigned_announcement(&node_announcement)
-            }
-            msgs::ChannelUpdate::TYPE => {
-                let channel_update: msgs::UnsignedChannelUpdate =
-                    Readable::read(&mut restricted_cursor)?;
-                network_graph.update_channel_unsigned(&channel_update)
-            }
-            _ => {
-                let error_string = format!("Unexpected graph sync message type: {}", message_type);
-                return Err(GraphSyncError::ProcessingError(error_string));
-            }
-        };
 
-        // propagate the error
-        let _update = update_result?;
-
-        read_cursor = restricted_cursor.into_inner();
-    }
+	let announcement_count: BigSize = Readable::read(read_cursor)?;
+	for _ in 0..announcement_count.0 {
+		let message_length: BigSize = Readable::read(read_cursor)?;
+		let mut restricted_cursor = read_cursor.take(message_length.0);
+		let channel_announcement: msgs::UnsignedChannelAnnouncement =
+			Readable::read(&mut restricted_cursor).unwrap();
+		network_graph.update_channel_from_unsigned_announcement(&channel_announcement, &chain_source)?;
+
+		read_cursor = restricted_cursor.into_inner();
+	}
+
+	let update_count: BigSize = Readable::read(read_cursor)?;
+	for _ in 0..update_count.0 {
+		let message_length: BigSize = Readable::read(read_cursor)?;
+		let mut restricted_cursor = read_cursor.take(message_length.0);
+		let channel_update: msgs::UnsignedChannelUpdate =
+			Readable::read(&mut restricted_cursor).unwrap();
+		network_graph.update_channel_unsigned(&channel_update)?;
+
+		read_cursor = restricted_cursor.into_inner();
+	}
 
     Ok(())
 }

lightning/src/ln/wire.rs

@@ -242,7 +242,8 @@ pub(crate) fn write<M: Type + Writeable, W: Writer>(message: &M, buffer: &mut W)
 	message.write(buffer)
 }
 
-mod encode {
+/// Encode module (should really be private, but exposure necessary for graph sync)
+pub mod encode {
 	/// Defines a constant type identifier for reading messages from the wire.
 	pub trait Encode {
 		/// The type identifying the message payload.