Pass gossip_queries messages to handler via ownership
authorbmancini55 <bmancini@gmail.com>
Thu, 3 Dec 2020 16:52:54 +0000 (11:52 -0500)
committerbmancini55 <bmancini@gmail.com>
Wed, 9 Dec 2020 20:02:32 +0000 (15:02 -0500)
This change modifies gossip_queries methods in RoutingMessageHandler to
move the message instead of passing a reference. This allows the message
handler to be more efficient by not requiring a full copy of SCIDs
passed in messages.

lightning-net-tokio/src/lib.rs
lightning/src/ln/msgs.rs
lightning/src/ln/peer_handler.rs
lightning/src/routing/network_graph.rs
lightning/src/util/test_utils.rs

index df0d6cb1f0ba78a08a87079738b25a0d60adfcd1..a90d673b6fd817dd65128db30715d445ab14695b 100644 (file)
@@ -537,10 +537,10 @@ mod tests {
                fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<NodeAnnouncement> { Vec::new() }
                fn should_request_full_sync(&self, _node_id: &PublicKey) -> bool { false }
                fn sync_routing_table(&self, _their_node_id: &PublicKey) { }
-               fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: &ReplyChannelRange) -> Result<(), LightningError> { Ok(()) }
-               fn handle_reply_short_channel_ids_end(&self, _their_node_id: &PublicKey, _msg: &ReplyShortChannelIdsEnd) -> Result<(), LightningError> { Ok(()) }
-               fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: &QueryChannelRange) -> Result<(), LightningError> { Ok(()) }
-               fn handle_query_short_channel_ids(&self, _their_node_id: &PublicKey, _msg: &QueryShortChannelIds) -> Result<(), LightningError> { Ok(()) }
+               fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: ReplyChannelRange) -> Result<(), LightningError> { Ok(()) }
+               fn handle_reply_short_channel_ids_end(&self, _their_node_id: &PublicKey, _msg: ReplyShortChannelIdsEnd) -> Result<(), LightningError> { Ok(()) }
+               fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: QueryChannelRange) -> Result<(), LightningError> { Ok(()) }
+               fn handle_query_short_channel_ids(&self, _their_node_id: &PublicKey, _msg: QueryShortChannelIds) -> Result<(), LightningError> { Ok(()) }
        }
        impl ChannelMessageHandler for MsgHandler {
                fn handle_open_channel(&self, _their_node_id: &PublicKey, _their_features: InitFeatures, _msg: &OpenChannel) {}
index 671f4211ccc6de7c27b02f5ab741226c3d82c53c..8cd3f8acc7c694209cf5e6ccdf941c7bcd18d3e1 100644 (file)
@@ -834,18 +834,22 @@ pub trait RoutingMessageHandler : Send + Sync + events::MessageSendEventsProvide
        /// Handles the reply of a query we initiated to learn about channels
        /// for a given range of blocks. We can expect to receive one or more
        /// replies to a single query.
-       fn handle_reply_channel_range(&self, their_node_id: &PublicKey, msg: &ReplyChannelRange) -> Result<(), LightningError>;
+       fn handle_reply_channel_range(&self, their_node_id: &PublicKey, msg: ReplyChannelRange) -> Result<(), LightningError>;
        /// Handles the reply of a query we initiated asking for routing gossip
        /// messages for a list of channels. We should receive this message when
        /// a node has completed its best effort to send us the pertaining routing
        /// gossip messages.
-       fn handle_reply_short_channel_ids_end(&self, their_node_id: &PublicKey, msg: &ReplyShortChannelIdsEnd) -> Result<(), LightningError>;
+       fn handle_reply_short_channel_ids_end(&self, their_node_id: &PublicKey, msg: ReplyShortChannelIdsEnd) -> Result<(), LightningError>;
        /// Handles when a peer asks us to send a list of short_channel_ids
-       /// for the requested range of blocks.
-       fn handle_query_channel_range(&self, their_node_id: &PublicKey, msg: &QueryChannelRange) -> Result<(), LightningError>;
+       /// for the requested range of blocks. There are potential DoS vectors when
+       /// handling inbound queries. Handling requests with first_blocknum very far
+       /// away may trigger repeated disk I/O if the NetworkGraph is not fully in-memory.
+       fn handle_query_channel_range(&self, their_node_id: &PublicKey, msg: QueryChannelRange) -> Result<(), LightningError>;
        /// Handles when a peer asks us to send routing gossip messages for a
-       /// list of short_channel_ids.
-       fn handle_query_short_channel_ids(&self, their_node_id: &PublicKey, msg: &QueryShortChannelIds) -> Result<(), LightningError>;
+       /// list of short_channel_ids. There are potential DoS vectors when handling
+       /// inbound queries. Handling requests with first_blocknum very far away may
+       /// trigger repeated disk I/O if the NetworkGraph is not fully in-memory.
+       fn handle_query_short_channel_ids(&self, their_node_id: &PublicKey, msg: QueryShortChannelIds) -> Result<(), LightningError>;
 }
 
 mod fuzzy_internal_msgs {
index beac542abddd56f53281bd5895ed359e542fa07a..daa74ad5b0090e19fc1b15f1eb193acf0927e21e 100644 (file)
@@ -842,16 +842,16 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
                                }
                        },
                        wire::Message::QueryShortChannelIds(msg) => {
-                               self.message_handler.route_handler.handle_query_short_channel_ids(&peer.their_node_id.unwrap(), &msg)?;
+                               self.message_handler.route_handler.handle_query_short_channel_ids(&peer.their_node_id.unwrap(), msg)?;
                        },
                        wire::Message::ReplyShortChannelIdsEnd(msg) => {
-                               self.message_handler.route_handler.handle_reply_short_channel_ids_end(&peer.their_node_id.unwrap(), &msg)?;
+                               self.message_handler.route_handler.handle_reply_short_channel_ids_end(&peer.their_node_id.unwrap(), msg)?;
                        },
                        wire::Message::QueryChannelRange(msg) => {
-                               self.message_handler.route_handler.handle_query_channel_range(&peer.their_node_id.unwrap(), &msg)?;
+                               self.message_handler.route_handler.handle_query_channel_range(&peer.their_node_id.unwrap(), msg)?;
                        },
                        wire::Message::ReplyChannelRange(msg) => {
-                               self.message_handler.route_handler.handle_reply_channel_range(&peer.their_node_id.unwrap(), &msg)?;
+                               self.message_handler.route_handler.handle_reply_channel_range(&peer.their_node_id.unwrap(), msg)?;
                        },
                        wire::Message::GossipTimestampFilter(_msg) => {
                                // TODO: handle message
index c1258ce37a2b1a6406dc53b5dea00b55401aff23..48900ecc446ffa7ef456c3a1d78f9782cceded92 100644 (file)
@@ -246,10 +246,8 @@ impl<C: Deref + Sync + Send, L: Deref + Sync + Send> RoutingMessageHandler for N
        /// stateless, it does not validate the sequencing of replies for multi-
        /// reply ranges. It does not validate whether the reply(ies) cover the
        /// queried range. It also does not filter SCIDs to only those in the
-       /// original query range. In the event of a failure, we may have received
-       /// some channel information. Before trying with another peer, the
-       /// caller should update its set of SCIDs that need to be queried.
-       fn handle_reply_channel_range(&self, their_node_id: &PublicKey, msg: &ReplyChannelRange) -> Result<(), LightningError> {
+       /// original query range.
+       fn handle_reply_channel_range(&self, their_node_id: &PublicKey, msg: ReplyChannelRange) -> Result<(), LightningError> {
                log_debug!(self.logger, "Handling reply_channel_range peer={}, first_blocknum={}, number_of_blocks={}, full_information={}, scids={}", log_pubkey!(their_node_id), msg.first_blocknum, msg.number_of_blocks, msg.full_information, msg.short_channel_ids.len(),);
 
                // Validate that the remote node maintains up-to-date channel
@@ -263,20 +261,13 @@ impl<C: Deref + Sync + Send, L: Deref + Sync + Send> RoutingMessageHandler for N
                        });
                }
 
-               // Copy the SCIDs into a new vector to be sent in the SCID query
-               let scid_size = msg.short_channel_ids.len();
-               let mut short_channel_ids: Vec<u64> = Vec::with_capacity(scid_size);
-               for scid in msg.short_channel_ids.iter() {
-                       short_channel_ids.push(scid.clone());
-               }
-
-               log_debug!(self.logger, "Sending query_short_channel_ids peer={}, batch_size={}", log_pubkey!(their_node_id), scid_size);
+               log_debug!(self.logger, "Sending query_short_channel_ids peer={}, batch_size={}", log_pubkey!(their_node_id), msg.short_channel_ids.len());
                let mut pending_events = self.pending_events.lock().unwrap();
                pending_events.push(events::MessageSendEvent::SendShortIdsQuery {
                        node_id: their_node_id.clone(),
                        msg: QueryShortChannelIds {
-                               chain_hash: msg.chain_hash.clone(),
-                               short_channel_ids,
+                               chain_hash: msg.chain_hash,
+                               short_channel_ids: msg.short_channel_ids,
                        }
                });
 
@@ -287,7 +278,7 @@ impl<C: Deref + Sync + Send, L: Deref + Sync + Send> RoutingMessageHandler for N
        /// gossip messages. In the event of a failure, we may have received
        /// some channel information. Before trying with another peer, the
        /// caller should update its set of SCIDs that need to be queried.
-       fn handle_reply_short_channel_ids_end(&self, their_node_id: &PublicKey, msg: &ReplyShortChannelIdsEnd) -> Result<(), LightningError> {
+       fn handle_reply_short_channel_ids_end(&self, their_node_id: &PublicKey, msg: ReplyShortChannelIdsEnd) -> Result<(), LightningError> {
                log_debug!(self.logger, "Handling reply_short_channel_ids_end peer={}, full_information={}", log_pubkey!(their_node_id), msg.full_information);
 
                // If the remote node does not have up-to-date information for the
@@ -303,10 +294,7 @@ impl<C: Deref + Sync + Send, L: Deref + Sync + Send> RoutingMessageHandler for N
                Ok(())
        }
 
-       /// There are potential DoS vectors when handling inbound queries.
-       /// Handling requests with first_blocknum very far away may trigger repeated
-       /// disk I/O if the NetworkGraph is not fully in-memory.
-       fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: &QueryChannelRange) -> Result<(), LightningError> {
+       fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: QueryChannelRange) -> Result<(), LightningError> {
                // TODO
                Err(LightningError {
                        err: String::from("Not implemented"),
@@ -314,10 +302,7 @@ impl<C: Deref + Sync + Send, L: Deref + Sync + Send> RoutingMessageHandler for N
                })
        }
 
-       /// There are potential DoS vectors when handling inbound queries.
-       /// Handling requests with first_blocknum very far away may trigger repeated
-       /// disk I/O if the NetworkGraph is not fully in-memory.
-       fn handle_query_short_channel_ids(&self, _their_node_id: &PublicKey, _msg: &QueryShortChannelIds) -> Result<(), LightningError> {
+       fn handle_query_short_channel_ids(&self, _their_node_id: &PublicKey, _msg: QueryShortChannelIds) -> Result<(), LightningError> {
                // TODO
                Err(LightningError {
                        err: String::from("Not implemented"),
@@ -1982,7 +1967,7 @@ mod tests {
                // matching the SCIDs in the reply
                {
                        // Handle a single successful reply that encompasses the queried channel range
-                       let result = net_graph_msg_handler.handle_reply_channel_range(&node_id_1, &ReplyChannelRange {
+                       let result = net_graph_msg_handler.handle_reply_channel_range(&node_id_1, ReplyChannelRange {
                                chain_hash,
                                full_information: true,
                                first_blocknum: 0,
@@ -2023,7 +2008,7 @@ mod tests {
                // full_information=false and short_channel_ids=[] as the signal.
                {
                        // Handle the reply indicating the peer was unable to fulfill our request.
-                       let result = net_graph_msg_handler.handle_reply_channel_range(&node_id_1, &ReplyChannelRange {
+                       let result = net_graph_msg_handler.handle_reply_channel_range(&node_id_1, ReplyChannelRange {
                                chain_hash,
                                full_information: false,
                                first_blocknum: 1000,
@@ -2045,7 +2030,7 @@ mod tests {
 
                // Test receipt of a successful reply
                {
-                       let result = net_graph_msg_handler.handle_reply_short_channel_ids_end(&node_id, &ReplyShortChannelIdsEnd {
+                       let result = net_graph_msg_handler.handle_reply_short_channel_ids_end(&node_id, ReplyShortChannelIdsEnd {
                                chain_hash,
                                full_information: true,
                        });
@@ -2055,7 +2040,7 @@ mod tests {
                // Test receipt of a reply that indicates the peer does not maintain up-to-date information
                // for the chain_hash requested in the query.
                {
-                       let result = net_graph_msg_handler.handle_reply_short_channel_ids_end(&node_id, &ReplyShortChannelIdsEnd {
+                       let result = net_graph_msg_handler.handle_reply_short_channel_ids_end(&node_id, ReplyShortChannelIdsEnd {
                                chain_hash,
                                full_information: false,
                        });
@@ -2072,7 +2057,7 @@ mod tests {
 
                let chain_hash = genesis_block(Network::Testnet).header.block_hash();
 
-               let result = net_graph_msg_handler.handle_query_channel_range(&node_id, &QueryChannelRange {
+               let result = net_graph_msg_handler.handle_query_channel_range(&node_id, QueryChannelRange {
                        chain_hash,
                        first_blocknum: 0,
                        number_of_blocks: 0xffff_ffff,
@@ -2088,7 +2073,7 @@ mod tests {
 
                let chain_hash = genesis_block(Network::Testnet).header.block_hash();
 
-               let result = net_graph_msg_handler.handle_query_short_channel_ids(&node_id, &QueryShortChannelIds {
+               let result = net_graph_msg_handler.handle_query_short_channel_ids(&node_id, QueryShortChannelIds {
                        chain_hash,
                        short_channel_ids: vec![0x0003e8_000000_0000],
                });
index dfbb283b1a4fac7811ddde09c82087e517e63df0..62bfda0948768f6835e5dca7c5e4d8fb11825b5b 100644 (file)
@@ -322,19 +322,19 @@ impl msgs::RoutingMessageHandler for TestRoutingMessageHandler {
 
        fn sync_routing_table(&self, _their_node_id: &PublicKey) {}
 
-       fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: &msgs::ReplyChannelRange) -> Result<(), msgs::LightningError> {
+       fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: msgs::ReplyChannelRange) -> Result<(), msgs::LightningError> {
                Ok(())
        }
 
-       fn handle_reply_short_channel_ids_end(&self, _their_node_id: &PublicKey, _msg: &msgs::ReplyShortChannelIdsEnd) -> Result<(), msgs::LightningError> {
+       fn handle_reply_short_channel_ids_end(&self, _their_node_id: &PublicKey, _msg: msgs::ReplyShortChannelIdsEnd) -> Result<(), msgs::LightningError> {
                Ok(())
        }
 
-       fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: &msgs::QueryChannelRange) -> Result<(), msgs::LightningError> {
+       fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: msgs::QueryChannelRange) -> Result<(), msgs::LightningError> {
                Ok(())
        }
 
-       fn handle_query_short_channel_ids(&self, _their_node_id: &PublicKey, _msg: &msgs::QueryShortChannelIds) -> Result<(), msgs::LightningError> {
+       fn handle_query_short_channel_ids(&self, _their_node_id: &PublicKey, _msg: msgs::QueryShortChannelIds) -> Result<(), msgs::LightningError> {
                Ok(())
        }
 }