WIP: separate gossip broadcasts into their own queue in PeerManager

This allows us to better prioritize channel messages over gossip broadcasts and
lays groundwork for rate limiting onion messages more simply.

Author: valentinewallace

lightning/src/ln/peer_handler.rs

@@ -337,6 +337,8 @@ struct Peer {
 
 	pending_outbound_buffer: LinkedList<Vec<u8>>,
 	pending_outbound_buffer_first_msg_offset: usize,
+	gossip_broadcast_buffer: LinkedList<Vec<u8>>,
+	gossip_broadcast_buffer_first_msg_offset: usize,
 	awaiting_write_event: bool,
 
 	pending_read_buffer: Vec<u8>,
@@ -393,15 +395,18 @@ impl Peer {
 	/// 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
+		self.pending_outbound_buffer.is_empty() && !self.buffer_full_drop_gossip_broadcast() // TODO: are broadcasts higher priority than background sync? seems like maybe they should be
+			&& self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK
 	}
 
-	/// Returns whether this peer's buffer is full and we should drop gossip messages.
+	/// Returns whether this peer's outbound buffers are full and we should drop gossip broadcasts.
 	fn buffer_full_drop_gossip_broadcast(&self) -> bool {
-		if self.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP
-			|| self.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO {
-				return false
+		let total_outbound_buffered =
+			self.gossip_broadcast_buffer.len() + self.pending_outbound_buffer.len();
+
+		if total_outbound_buffered > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP ||
+			self.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO
+		{
 				return true
 		}
 		false
@@ -672,6 +677,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 
 			pending_outbound_buffer: LinkedList::new(),
 			pending_outbound_buffer_first_msg_offset: 0,
+			gossip_broadcast_buffer: LinkedList::new(),
+			gossip_broadcast_buffer_first_msg_offset: 0,
 			awaiting_write_event: false,
 
 			pending_read_buffer,
@@ -718,6 +725,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 
 			pending_outbound_buffer: LinkedList::new(),
 			pending_outbound_buffer_first_msg_offset: 0,
+			gossip_broadcast_buffer: LinkedList::new(),
+			gossip_broadcast_buffer_first_msg_offset: 0,
 			awaiting_write_event: false,
 
 			pending_read_buffer,
@@ -780,20 +789,29 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 				self.maybe_send_extra_ping(peer);
 			}
 
-			let next_buff = match peer.pending_outbound_buffer.front() {
-				None => return,
-				Some(buff) => buff,
-			};
-
-			let pending = &next_buff[peer.pending_outbound_buffer_first_msg_offset..];
-			let data_sent = descriptor.send_data(pending, peer.should_read());
-			peer.pending_outbound_buffer_first_msg_offset += data_sent;
-			if peer.pending_outbound_buffer_first_msg_offset == next_buff.len() {
-				peer.pending_outbound_buffer_first_msg_offset = 0;
-				peer.pending_outbound_buffer.pop_front();
-			} else {
-				peer.awaiting_write_event = true;
-			}
+			// TODO: break into methods in `peer` or somehow DRY
+			// We prioritize channel messages over gossip broadcasts
+			if let Some(next_buff) = peer.pending_outbound_buffer.front() {
+				let pending = &next_buff[peer.pending_outbound_buffer_first_msg_offset..];
+				let data_sent = descriptor.send_data(pending, peer.should_read());
+				peer.pending_outbound_buffer_first_msg_offset += data_sent;
+				if peer.pending_outbound_buffer_first_msg_offset == next_buff.len() {
+					peer.pending_outbound_buffer_first_msg_offset = 0;
+					peer.pending_outbound_buffer.pop_front();
+				} else {
+					peer.awaiting_write_event = true;
+				}
+			} else if let Some(next_buff) = peer.gossip_broadcast_buffer.front() {
+				let pending = &next_buff[peer.gossip_broadcast_buffer_first_msg_offset..];
+				let data_sent = descriptor.send_data(pending, peer.should_read());
+				peer.gossip_broadcast_buffer_first_msg_offset += data_sent;
+				if peer.gossip_broadcast_buffer_first_msg_offset == next_buff.len() {
+					peer.gossip_broadcast_buffer_first_msg_offset = 0;
+					peer.gossip_broadcast_buffer.pop_front();
+				} else {
+					peer.awaiting_write_event = true;
+				}
+			} else { return }
 		}
 	}
 
@@ -858,6 +876,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 		peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_message[..]));
 	}
 
+	/// Append a message to a peer's pending outbound/write gossip broadcast buffer
+	fn enqueue_encoded_gossip_broadcast(&self, peer: &mut Peer, encoded_message: &Vec<u8>) {
+		peer.msgs_sent_since_pong += 1;
+		peer.gossip_broadcast_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_message[..]));
+	}
+
 	/// Append a message to a peer's pending outbound/write buffer
 	fn enqueue_message<M: wire::Type>(&self, peer: &mut Peer, message: &M) {
 		let mut buffer = VecWriter(Vec::with_capacity(2048));
@@ -1337,7 +1361,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 					if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
 						continue;
 					}
-					self.enqueue_encoded_message(&mut *peer, &encoded_msg);
+					self.enqueue_encoded_gossip_broadcast(&mut *peer, &encoded_msg);
 				}
 			},
 			wire::Message::NodeAnnouncement(ref msg) => {
@@ -1360,7 +1384,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 					if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
 						continue;
 					}
-					self.enqueue_encoded_message(&mut *peer, &encoded_msg);
+					self.enqueue_encoded_gossip_broadcast(&mut *peer, &encoded_msg);
 				}
 			},
 			wire::Message::ChannelUpdate(ref msg) => {
@@ -1380,7 +1404,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 					if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
 						continue;
 					}
-					self.enqueue_encoded_message(&mut *peer, &encoded_msg);
+					self.enqueue_encoded_gossip_broadcast(&mut *peer, &encoded_msg);
 				}
 			},
 			_ => debug_assert!(false, "We shouldn't attempt to forward anything but gossip messages"),