Backfill gossip without buffering directly in LDK

Instead of backfilling gossip by buffering (up to) ten messages at
a time, only buffer one message at a time, as the peers' outbound
socket buffer drains. This moves the outbound backfill messages out
of `PeerHandler` and into the operating system buffer, where it
arguably belongs.

Not buffering causes us to walk the gossip B-Trees somewhat more
often, but avoids allocating vecs for the responses. While its
probably (without having benchmarked it) a net performance loss, it
simplifies buffer tracking and leaves us with more room to play
with the buffer sizing constants as we add onion message forwarding
which is an important win.

Note that because we change how often we check if we're out of
messages to send before pinging, we slightly change how many
messages are exchanged at once, impacting the
`test_do_attempt_write_data` constants.

Author: TheBlueMatt

lightning-net-tokio/src/lib.rs

@@ -562,8 +562,8 @@ mod tests {
 		fn handle_node_announcement(&self, _msg: &NodeAnnouncement) -> Result<bool, LightningError> { Ok(false) }
 		fn handle_channel_announcement(&self, _msg: &ChannelAnnouncement) -> Result<bool, LightningError> { Ok(false) }
 		fn handle_channel_update(&self, _msg: &ChannelUpdate) -> Result<bool, LightningError> { Ok(false) }
-		fn get_next_channel_announcements(&self, _starting_point: u64, _batch_amount: u8) -> Vec<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> { Vec::new() }
-		fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<NodeAnnouncement> { Vec::new() }
+		fn get_next_channel_announcements(&self, _starting_point: u64) -> Option<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> { None }
+		fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>) -> Option<NodeAnnouncement> { None }
 		fn peer_connected(&self, _their_node_id: &PublicKey, _init_msg: &Init) { }
 		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(()) }

lightning/src/ln/functional_test_utils.rs

@@ -318,20 +318,20 @@ impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> {
 				);
 				let mut chan_progress = 0;
 				loop {
-					let orig_announcements = self.gossip_sync.get_next_channel_announcements(chan_progress, 255);
-					let deserialized_announcements = gossip_sync.get_next_channel_announcements(chan_progress, 255);
+					let orig_announcements = self.gossip_sync.get_next_channel_announcements(chan_progress);
+					let deserialized_announcements = gossip_sync.get_next_channel_announcements(chan_progress);
 					assert!(orig_announcements == deserialized_announcements);
-					chan_progress = match orig_announcements.last() {
+					chan_progress = match orig_announcements {
 						Some(announcement) => announcement.0.contents.short_channel_id + 1,
 						None => break,
 					};
 				}
 				let mut node_progress = None;
 				loop {
-					let orig_announcements = self.gossip_sync.get_next_node_announcements(node_progress.as_ref(), 255);
-					let deserialized_announcements = gossip_sync.get_next_node_announcements(node_progress.as_ref(), 255);
+					let orig_announcements = self.gossip_sync.get_next_node_announcements(node_progress.as_ref());
+					let deserialized_announcements = gossip_sync.get_next_node_announcements(node_progress.as_ref());
 					assert!(orig_announcements == deserialized_announcements);
-					node_progress = match orig_announcements.last() {
+					node_progress = match orig_announcements {
 						Some(announcement) => Some(announcement.contents.node_id),
 						None => break,
 					};

lightning/src/ln/msgs.rs

@@ -915,15 +915,15 @@ pub trait RoutingMessageHandler : MessageSendEventsProvider {
 	/// Handle an incoming channel_update message, returning true if it should be forwarded on,
 	/// false or returning an Err otherwise.
 	fn handle_channel_update(&self, msg: &ChannelUpdate) -> Result<bool, LightningError>;
-	/// Gets a subset of the channel announcements and updates required to dump our routing table
-	/// to a remote node, starting at the short_channel_id indicated by starting_point and
-	/// including the batch_amount entries immediately higher in numerical value than starting_point.
-	fn get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)>;
-	/// Gets a subset of the node announcements required to dump our routing table to a remote node,
-	/// starting at the node *after* the provided publickey and including batch_amount entries
-	/// immediately higher (as defined by <PublicKey as Ord>::cmp) than starting_point.
+	/// Gets channel announcements and updates required to dump our routing table to a remote node,
+	/// starting at the short_channel_id indicated by starting_point and including announcements
+	/// for a single channel.
+	fn get_next_channel_announcements(&self, starting_point: u64) -> Option<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)>;
+	/// Gets a node announcement required to dump our routing table to a remote node, starting at
+	/// the node *after* the provided publickey and including up to one announcement immediately
+	/// higher (as defined by <PublicKey as Ord>::cmp) than starting_point.
 	/// If None is provided for starting_point, we start at the first node.
-	fn get_next_node_announcements(&self, starting_point: Option<&PublicKey>, batch_amount: u8) -> Vec<NodeAnnouncement>;
+	fn get_next_node_announcements(&self, starting_point: Option<&PublicKey>) -> Option<NodeAnnouncement>;
 	/// Called when a connection is established with a peer. This can be used to
 	/// perform routing table synchronization using a strategy defined by the
 	/// implementor.

lightning/src/ln/peer_handler.rs

@@ -67,9 +67,9 @@ impl RoutingMessageHandler for IgnoringMessageHandler {
 	fn handle_node_announcement(&self, _msg: &msgs::NodeAnnouncement) -> Result<bool, LightningError> { Ok(false) }
 	fn handle_channel_announcement(&self, _msg: &msgs::ChannelAnnouncement) -> Result<bool, LightningError> { Ok(false) }
 	fn handle_channel_update(&self, _msg: &msgs::ChannelUpdate) -> Result<bool, LightningError> { Ok(false) }
-	fn get_next_channel_announcements(&self, _starting_point: u64, _batch_amount: u8) ->
-		Vec<(msgs::ChannelAnnouncement, Option<msgs::ChannelUpdate>, Option<msgs::ChannelUpdate>)> { Vec::new() }
-	fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<msgs::NodeAnnouncement> { Vec::new() }
+	fn get_next_channel_announcements(&self, _starting_point: u64) ->
+		Option<(msgs::ChannelAnnouncement, Option<msgs::ChannelUpdate>, Option<msgs::ChannelUpdate>)> { None }
+	fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>) -> Option<msgs::NodeAnnouncement> { None }
 	fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init) {}
 	fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: msgs::ReplyChannelRange) -> Result<(), LightningError> { Ok(()) }
 	fn handle_reply_short_channel_ids_end(&self, _their_node_id: &PublicKey, _msg: msgs::ReplyShortChannelIdsEnd) -> Result<(), LightningError> { Ok(()) }
@@ -383,19 +383,17 @@ impl Peer {
 		}
 	}
 
-	/// Returns the number of gossip messages we can fit in this peer's buffer.
-	fn gossip_buffer_slots_available(&self) -> usize {
-		OUTBOUND_BUFFER_LIMIT_READ_PAUSE.saturating_sub(self.pending_outbound_buffer.len())
-	}
-
 	/// Returns whether we should be reading bytes from this peer, based on whether its outbound
 	/// buffer still has space and we don't need to pause reads to get some writes out.
 	fn should_read(&self) -> bool {
 		self.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE
 	}
 
-	fn should_backfill_gossip(&self) -> bool {
-		self.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE &&
+	/// Determines if we should push additional gossip messages onto a peer's outbound buffer for
+	/// backfilling gossip data to the peer. This is checked every time the peer's buffer may have
+	/// been drained.
+	fn should_buffer_gossip_backfill(&self) -> bool {
+		self.pending_outbound_buffer.is_empty() &&
 			self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK
 	}
 
@@ -739,46 +737,39 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 
 	fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) {
 		while !peer.awaiting_write_event {
-			if peer.should_backfill_gossip() {
+			if peer.should_buffer_gossip_backfill() {
 				match peer.sync_status {
 					InitSyncTracker::NoSyncRequested => {},
 					InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => {
-						let steps = ((peer.gossip_buffer_slots_available() + 2) / 3) as u8;
-						let all_messages = self.message_handler.route_handler.get_next_channel_announcements(c, steps);
-						for &(ref announce, ref update_a_option, ref update_b_option) in all_messages.iter() {
-							self.enqueue_message(peer, announce);
-							if let &Some(ref update_a) = update_a_option {
-								self.enqueue_message(peer, update_a);
+						if let Some((announce, update_a_option, update_b_option)) =
+							self.message_handler.route_handler.get_next_channel_announcements(c)
+						{
+							self.enqueue_message(peer, &announce);
+							if let Some(update_a) = update_a_option {
+								self.enqueue_message(peer, &update_a);
 							}
-							if let &Some(ref update_b) = update_b_option {
-								self.enqueue_message(peer, update_b);
+							if let Some(update_b) = update_b_option {
+								self.enqueue_message(peer, &update_b);
 							}
 							peer.sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1);
-						}
-						if all_messages.is_empty() || all_messages.len() != steps as usize {
+						} else {
 							peer.sync_status = InitSyncTracker::ChannelsSyncing(0xffff_ffff_ffff_ffff);
 						}
 					},
 					InitSyncTracker::ChannelsSyncing(c) if c == 0xffff_ffff_ffff_ffff => {
-						let steps = peer.gossip_buffer_slots_available() as u8;
-						let all_messages = self.message_handler.route_handler.get_next_node_announcements(None, steps);
-						for msg in all_messages.iter() {
-							self.enqueue_message(peer, msg);
+						if let Some(msg) = self.message_handler.route_handler.get_next_node_announcements(None) {
+							self.enqueue_message(peer, &msg);
 							peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
-						}
-						if all_messages.is_empty() || all_messages.len() != steps as usize {
+						} else {
 							peer.sync_status = InitSyncTracker::NoSyncRequested;
 						}
 					},
 					InitSyncTracker::ChannelsSyncing(_) => unreachable!(),
 					InitSyncTracker::NodesSyncing(key) => {
-						let steps = peer.gossip_buffer_slots_available() as u8;
-						let all_messages = self.message_handler.route_handler.get_next_node_announcements(Some(&key), steps);
-						for msg in all_messages.iter() {
-							self.enqueue_message(peer, msg);
+						if let Some(msg) = self.message_handler.route_handler.get_next_node_announcements(Some(&key)) {
+							self.enqueue_message(peer, &msg);
 							peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
-						}
-						if all_messages.is_empty() || all_messages.len() != steps as usize {
+						} else {
 							peer.sync_status = InitSyncTracker::NoSyncRequested;
 						}
 					},
@@ -2082,10 +2073,10 @@ mod tests {
 
 		// Check that each peer has received the expected number of channel updates and channel
 		// announcements.
-		assert_eq!(cfgs[0].routing_handler.chan_upds_recvd.load(Ordering::Acquire), 100);
-		assert_eq!(cfgs[0].routing_handler.chan_anns_recvd.load(Ordering::Acquire), 50);
-		assert_eq!(cfgs[1].routing_handler.chan_upds_recvd.load(Ordering::Acquire), 100);
-		assert_eq!(cfgs[1].routing_handler.chan_anns_recvd.load(Ordering::Acquire), 50);
+		assert_eq!(cfgs[0].routing_handler.chan_upds_recvd.load(Ordering::Acquire), 108);
+		assert_eq!(cfgs[0].routing_handler.chan_anns_recvd.load(Ordering::Acquire), 54);
+		assert_eq!(cfgs[1].routing_handler.chan_upds_recvd.load(Ordering::Acquire), 108);
+		assert_eq!(cfgs[1].routing_handler.chan_anns_recvd.load(Ordering::Acquire), 54);
 	}
 
 	#[test]

lightning/src/routing/gossip.rs

@@ -318,11 +318,10 @@ where C::Target: chain::Access, L::Target: Logger
 		Ok(msg.contents.excess_data.len() <= MAX_EXCESS_BYTES_FOR_RELAY)
 	}
 
-	fn get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> {
-		let mut result = Vec::with_capacity(batch_amount as usize);
+	fn get_next_channel_announcements(&self, starting_point: u64) -> Option<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> {
 		let channels = self.network_graph.channels.read().unwrap();
 		let mut iter = channels.range(starting_point..);
-		while result.len() < batch_amount as usize {
+		loop {
 			if let Some((_, ref chan)) = iter.next() {
 				if chan.announcement_message.is_some() {
 					let chan_announcement = chan.announcement_message.clone().unwrap();
@@ -334,20 +333,18 @@ where C::Target: chain::Access, L::Target: Logger
 					if let Some(two_to_one) = chan.two_to_one.as_ref() {
 						two_to_one_announcement = two_to_one.last_update_message.clone();
 					}
-					result.push((chan_announcement, one_to_two_announcement, two_to_one_announcement));
+					return Some((chan_announcement, one_to_two_announcement, two_to_one_announcement));
 				} else {
 					// TODO: We may end up sending un-announced channel_updates if we are sending
 					// initial sync data while receiving announce/updates for this channel.
 				}
 			} else {
-				return result;
+				return None;
 			}
 		}
-		result
 	}
 
-	fn get_next_node_announcements(&self, starting_point: Option<&PublicKey>, batch_amount: u8) -> Vec<NodeAnnouncement> {
-		let mut result = Vec::with_capacity(batch_amount as usize);
+	fn get_next_node_announcements(&self, starting_point: Option<&PublicKey>) -> Option<NodeAnnouncement> {
 		let nodes = self.network_graph.nodes.read().unwrap();
 		let mut iter = if let Some(pubkey) = starting_point {
 				let mut iter = nodes.range(NodeId::from_pubkey(pubkey)..);
@@ -356,18 +353,17 @@ where C::Target: chain::Access, L::Target: Logger
 			} else {
 				nodes.range::<NodeId, _>(..)
 			};
-		while result.len() < batch_amount as usize {
+		loop {
 			if let Some((_, ref node)) = iter.next() {
 				if let Some(node_info) = node.announcement_info.as_ref() {
-					if node_info.announcement_message.is_some() {
-						result.push(node_info.announcement_message.clone().unwrap());
+					if let Some(msg) = node_info.announcement_message.clone() {
+						return Some(msg);
 					}
 				}
 			} else {
-				return result;
+				return None;
 			}
 		}
-		result
 	}
 
 	/// Initiates a stateless sync of routing gossip information with a peer
@@ -2412,8 +2408,8 @@ mod tests {
 		let node_2_privkey = &SecretKey::from_slice(&[41; 32]).unwrap();
 
 		// Channels were not announced yet.
-		let channels_with_announcements = gossip_sync.get_next_channel_announcements(0, 1);
-		assert_eq!(channels_with_announcements.len(), 0);
+		let channels_with_announcements = gossip_sync.get_next_channel_announcements(0);
+		assert!(channels_with_announcements.is_none());
 
 		let short_channel_id;
 		{
@@ -2427,17 +2423,15 @@ mod tests {
 		}
 
 		// Contains initial channel announcement now.
-		let channels_with_announcements = gossip_sync.get_next_channel_announcements(short_channel_id, 1);
-		assert_eq!(channels_with_announcements.len(), 1);
-		if let Some(channel_announcements) = channels_with_announcements.first() {
-			let &(_, ref update_1, ref update_2) = channel_announcements;
+		let channels_with_announcements = gossip_sync.get_next_channel_announcements(short_channel_id);
+		if let Some(channel_announcements) = channels_with_announcements {
+			let (_, ref update_1, ref update_2) = channel_announcements;
 			assert_eq!(update_1, &None);
 			assert_eq!(update_2, &None);
 		} else {
 			panic!();
 		}
 
-
 		{
 			// Valid channel update
 			let valid_channel_update = get_signed_channel_update(|unsigned_channel_update| {
@@ -2450,10 +2444,9 @@ mod tests {
 		}
 
 		// Now contains an initial announcement and an update.
-		let channels_with_announcements = gossip_sync.get_next_channel_announcements(short_channel_id, 1);
-		assert_eq!(channels_with_announcements.len(), 1);
-		if let Some(channel_announcements) = channels_with_announcements.first() {
-			let &(_, ref update_1, ref update_2) = channel_announcements;
+		let channels_with_announcements = gossip_sync.get_next_channel_announcements(short_channel_id);
+		if let Some(channel_announcements) = channels_with_announcements {
+			let (_, ref update_1, ref update_2) = channel_announcements;
 			assert_ne!(update_1, &None);
 			assert_eq!(update_2, &None);
 		} else {
@@ -2473,19 +2466,18 @@ mod tests {
 		}
 
 		// Test that announcements with excess data won't be returned
-		let channels_with_announcements = gossip_sync.get_next_channel_announcements(short_channel_id, 1);
-		assert_eq!(channels_with_announcements.len(), 1);
-		if let Some(channel_announcements) = channels_with_announcements.first() {
-			let &(_, ref update_1, ref update_2) = channel_announcements;
+		let channels_with_announcements = gossip_sync.get_next_channel_announcements(short_channel_id);
+		if let Some(channel_announcements) = channels_with_announcements {
+			let (_, ref update_1, ref update_2) = channel_announcements;
 			assert_eq!(update_1, &None);
 			assert_eq!(update_2, &None);
 		} else {
 			panic!();
 		}
 
 		// Further starting point have no channels after it
-		let channels_with_announcements = gossip_sync.get_next_channel_announcements(short_channel_id + 1000, 1);
-		assert_eq!(channels_with_announcements.len(), 0);
+		let channels_with_announcements = gossip_sync.get_next_channel_announcements(short_channel_id + 1000);
+		assert!(channels_with_announcements.is_none());
 	}
 
 	#[test]
@@ -2497,8 +2489,8 @@ mod tests {
 		let node_id_1 = PublicKey::from_secret_key(&secp_ctx, node_1_privkey);
 
 		// No nodes yet.
-		let next_announcements = gossip_sync.get_next_node_announcements(None, 10);
-		assert_eq!(next_announcements.len(), 0);
+		let next_announcements = gossip_sync.get_next_node_announcements(None);
+		assert!(next_announcements.is_none());
 
 		{
 			// Announce a channel to add 2 nodes
@@ -2509,10 +2501,9 @@ mod tests {
 			};
 		}
 
-
 		// Nodes were never announced
-		let next_announcements = gossip_sync.get_next_node_announcements(None, 3);
-		assert_eq!(next_announcements.len(), 0);
+		let next_announcements = gossip_sync.get_next_node_announcements(None);
+		assert!(next_announcements.is_none());
 
 		{
 			let valid_announcement = get_signed_node_announcement(|_| {}, node_1_privkey, &secp_ctx);
@@ -2528,12 +2519,12 @@ mod tests {
 			};
 		}
 
-		let next_announcements = gossip_sync.get_next_node_announcements(None, 3);
-		assert_eq!(next_announcements.len(), 2);
+		let next_announcements = gossip_sync.get_next_node_announcements(None);
+		assert!(next_announcements.is_some());
 
 		// Skip the first node.
-		let next_announcements = gossip_sync.get_next_node_announcements(Some(&node_id_1), 2);
-		assert_eq!(next_announcements.len(), 1);
+		let next_announcements = gossip_sync.get_next_node_announcements(Some(&node_id_1));
+		assert!(next_announcements.is_some());
 
 		{
 			// Later announcement which should not be relayed (excess data) prevent us from sharing a node
@@ -2547,8 +2538,8 @@ mod tests {
 			};
 		}
 
-		let next_announcements = gossip_sync.get_next_node_announcements(Some(&node_id_1), 2);
-		assert_eq!(next_announcements.len(), 0);
+		let next_announcements = gossip_sync.get_next_node_announcements(Some(&node_id_1));
+		assert!(next_announcements.is_none());
 	}
 
 	#[test]