Give peers which are sending us messages longer to respond to ping

See comment for rationale.

Author: TheBlueMatt

lightning-background-processor/src/lib.rs

@@ -14,8 +14,8 @@ use lightning::chain::chainmonitor::{ChainMonitor, Persist};
 use lightning::chain::keysinterface::{Sign, KeysInterface};
 use lightning::ln::channelmanager::ChannelManager;
 use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
-use lightning::ln::peer_handler::{PeerManager, SocketDescriptor};
-use lightning::ln::peer_handler::CustomMessageHandler;
+use lightning::ln::peer_handler::{CustomMessageHandler, PeerManager, SocketDescriptor};
+use lightning::ln::peer_handler;
 use lightning::routing::network_graph::NetGraphMsgHandler;
 use lightning::util::events::{Event, EventHandler, EventsProvider};
 use lightning::util::logger::Logger;
@@ -236,8 +236,9 @@ impl BackgroundProcessor {
 					// timer, we should have disconnected all sockets by now (and they're probably
 					// dead anyway), so disconnect them by calling `timer_tick_occurred()` twice.
 					log_trace!(logger, "Awoke after more than double our ping timer, disconnecting peers.");
-					peer_manager.timer_tick_occurred();
-					peer_manager.timer_tick_occurred();
+					for _ in 0..peer_handler::MAX_BUFFER_DRAIN_TICK_INTERVALS + 1 {
+						peer_manager.timer_tick_occurred();
+					}
 					last_ping_call = Instant::now();
 				} else if last_ping_call.elapsed().as_secs() > PING_TIMER {
 					log_trace!(logger, "Calling PeerManager's timer_tick_occurred");

lightning/src/ln/peer_handler.rs

@@ -294,6 +294,24 @@ const OUTBOUND_BUFFER_LIMIT_READ_PAUSE: usize = 10;
 /// the peer.
 const OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP: usize = 20;
 
+/// If we've sent a ping, and are still awaiting a response, we (or our peer) may need to churn our
+/// (or their) way through the socket receive buffer before receiving the ping.
+///
+/// On a fairly old Arm64 board, with Linux defaults, this can take as long as 20 seconds, not
+/// including any network delays or outbound traffic.
+///
+/// Thus, to avoid needlessly disconnecting a peer, we allow a peer to take this many timer ticks
+/// to respond to a ping, as long as they send us at least one message during each tick or if we
+/// sent a lot of messages, ensuring we aren't actually just disconnected. With a timer tick
+/// interval of five seconds, this translates to about 30 seconds.
+pub const MAX_BUFFER_DRAIN_TICK_INTERVALS: i8 = 6;
+
+/// This is the minimum number of messages we expect a peer to be able to handle within one timer
+/// tick. Once we have sent this many messages since the last ping, we send a ping right away to
+/// ensures we don't just fill up our send buffer and leave the peer with too many messages to
+/// process before the next ping.
+pub const BUFFER_DRAIN_MSGS_PER_TICK: usize = 32;
+
 struct Peer {
 	channel_encryptor: PeerChannelEncryptor,
 	their_node_id: Option<PublicKey>,
@@ -309,7 +327,9 @@ struct Peer {
 
 	sync_status: InitSyncTracker,
 
-	awaiting_pong: bool,
+	msgs_sent_since_pong: usize,
+	awaiting_pong_tick_intervals: i8,
+	received_message_since_timer_tick: bool,
 }
 
 impl Peer {
@@ -551,7 +571,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 
 			sync_status: InitSyncTracker::NoSyncRequested,
 
-			awaiting_pong: false,
+			msgs_sent_since_pong: 0,
+			awaiting_pong_tick_intervals: 0,
+			received_message_since_timer_tick: false,
 		}).is_some() {
 			panic!("PeerManager driver duplicated descriptors!");
 		};
@@ -589,7 +611,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 
 			sync_status: InitSyncTracker::NoSyncRequested,
 
-			awaiting_pong: false,
+			msgs_sent_since_pong: 0,
+			awaiting_pong_tick_intervals: 0,
+			received_message_since_timer_tick: false,
 		}).is_some() {
 			panic!("PeerManager driver duplicated descriptors!");
 		};
@@ -598,7 +622,7 @@ 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.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE {
+			if peer.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE && peer.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK {
 				match peer.sync_status {
 					InitSyncTracker::NoSyncRequested => {},
 					InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => {
@@ -643,6 +667,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 					},
 				}
 			}
+			if peer.msgs_sent_since_pong >= BUFFER_DRAIN_MSGS_PER_TICK {
+				self.maybe_send_extra_ping(peer);
+			}
 
 			if {
 				let next_buff = match peer.pending_outbound_buffer.front() {
@@ -725,6 +752,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 		let encoded_message = buffer.0;
 
 		log_trace!(self.logger, "Enqueueing message {:?} to {}", message, log_pubkey!(peer.their_node_id.unwrap()));
+		peer.msgs_sent_since_pong += 1;
 		peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_message[..]));
 	}
 
@@ -919,6 +947,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 		message: wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>
 	) -> Result<Option<wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>>, MessageHandlingError> {
 		log_trace!(self.logger, "Received message {:?} from {}", message, log_pubkey!(peer.their_node_id.unwrap()));
+		peer.received_message_since_timer_tick = true;
 
 		// Need an Init as first message
 		if let wire::Message::Init(_) = message {
@@ -982,7 +1011,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 				}
 			},
 			wire::Message::Pong(_msg) => {
-				peer.awaiting_pong = false;
+				peer.awaiting_pong_tick_intervals = 0;
+				peer.msgs_sent_since_pong = 0;
 			},
 
 			// Channel messages:
@@ -1103,7 +1133,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 							!peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
 						continue
 					}
-					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP || peer.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * 2 {
 						log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
 						continue;
 					}
@@ -1115,6 +1145,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 						continue;
 					}
 					peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+					peer.msgs_sent_since_pong += 1;
 				}
 			},
 			wire::Message::NodeAnnouncement(ref msg) => {
@@ -1126,7 +1157,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 							!peer.should_forward_node_announcement(msg.contents.node_id) {
 						continue
 					}
-					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP || peer.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * 2 {
 						log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
 						continue;
 					}
@@ -1137,6 +1168,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 						continue;
 					}
 					peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+					peer.msgs_sent_since_pong += 1;
 				}
 			},
 			wire::Message::ChannelUpdate(ref msg) => {
@@ -1148,14 +1180,15 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 							!peer.should_forward_channel_announcement(msg.contents.short_channel_id)  {
 						continue
 					}
-					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP || peer.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * 2 {
 						log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
 						continue;
 					}
 					if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
 						continue;
 					}
 					peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+					peer.msgs_sent_since_pong += 1;
 				}
 			},
 			_ => debug_assert!(false, "We shouldn't attempt to forward anything but gossip messages"),
@@ -1431,6 +1464,20 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 		}
 	}
 
+	/// This is called when we're blocked on sending additional gossip messages until we receive a
+	/// pong. If we aren't waiting on a pong, we take this opportunity to send a ping (setting
+	/// `awaiting_pong_tick_intervals` to a special flag value to indicate this).
+	fn maybe_send_extra_ping(&self, peer: &mut Peer) {
+		if peer.awaiting_pong_tick_intervals == 0 {
+			peer.awaiting_pong_tick_intervals = -1;
+			let ping = msgs::Ping {
+				ponglen: 0,
+				byteslen: 64,
+			};
+			self.enqueue_message(peer, &ping);
+		}
+	}
+
 	/// Send pings to each peer and disconnect those which did not respond to the last round of
 	/// pings.
 	///
@@ -1451,7 +1498,14 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 			let mut descriptors_needing_disconnect = Vec::new();
 
 			peers.retain(|descriptor, peer| {
-				if peer.awaiting_pong {
+				if !peer.channel_encryptor.is_ready_for_encryption() {
+					// The peer needs to complete its handshake before we can exchange messages
+					return true;
+				}
+
+				if (peer.awaiting_pong_tick_intervals > 0 && !peer.received_message_since_timer_tick)
+					|| peer.awaiting_pong_tick_intervals > MAX_BUFFER_DRAIN_TICK_INTERVALS
+				{
 					descriptors_needing_disconnect.push(descriptor.clone());
 					match peer.their_node_id {
 						Some(node_id) => {
@@ -1468,21 +1522,26 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 					return false;
 				}
 
-				if !peer.channel_encryptor.is_ready_for_encryption() {
-					// The peer needs to complete its handshake before we can exchange messages
+				peer.received_message_since_timer_tick = false;
+				if peer.awaiting_pong_tick_intervals == -1 {
+					// Magic value set in `maybe_send_extra_ping`.
+					peer.awaiting_pong_tick_intervals = 1;
+					return true;
+				}
+
+				if peer.awaiting_pong_tick_intervals > 0 {
+					peer.awaiting_pong_tick_intervals += 1;
 					return true;
 				}
 
+				peer.awaiting_pong_tick_intervals = 1;
 				let ping = msgs::Ping {
 					ponglen: 0,
 					byteslen: 64,
 				};
 				self.enqueue_message(peer, &ping);
+				self.do_attempt_write_data(&mut (descriptor.clone()), &mut *peer);
 
-				let mut descriptor_clone = descriptor.clone();
-				self.do_attempt_write_data(&mut descriptor_clone, peer);
-
-				peer.awaiting_pong = true;
 				true
 			});
 
@@ -1641,11 +1700,14 @@ mod tests {
 		// than can fit into a peer's buffer).
 		let (mut fd_a, mut fd_b) = establish_connection(&peers[0], &peers[1]);
 
-		// Make each peer to read the messages that the other peer just wrote to them.
-		peers[0].process_events();
-		peers[1].read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap();
-		peers[1].process_events();
-		peers[0].read_event(&mut fd_a, &fd_b.outbound_data.lock().unwrap().split_off(0)).unwrap();
+		// Make each peer to read the messages that the other peer just wrote to them. Note that
+		// due to the max-messagse-before-bing limits this may take a few iterations to complete.
+		for _ in 0..10 {
+			peers[0].process_events();
+			peers[1].read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap();
+			peers[1].process_events();
+			peers[0].read_event(&mut fd_a, &fd_b.outbound_data.lock().unwrap().split_off(0)).unwrap();
+		}
 
 		// Check that each peer has received the expected number of channel updates and channel
 		// announcements.