f - remove peers through timer_tick_occurred

Author: ViktorTigerstrom

lightning/src/ln/channelmanager.rs

@@ -595,8 +595,6 @@ pub type SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, M, T, F, L> = C
 //  |               |
 //  |               |__`best_block`
 //  |               |
-//  |               |__`pending_peers_awaiting_removal`
-//  |               |
 //  |               |__`pending_events`
 //  |                   |
 //  |                   |__`pending_background_events`
@@ -764,16 +762,6 @@ where
 
 	/// See `ChannelManager` struct-level documentation for lock order requirements.
 	pending_events: Mutex<Vec<events::Event>>,
-	/// When a peer disconnects but still has channels, the peer's `peer_state` entry in the
-	/// `per_peer_state` is not removed by the `peer_disconnected` function. If the channels of
-	/// to that peer is later closed while still being disconnected (i.e. force closed), we
-	/// therefore need to remove the peer from `peer_state` separately.
-	/// To avoid having to take the `per_peer_state` `write` lock once the channels are closed, we
-	/// instead store such peers awaiting removal in this field, and remove them on a timer to
-	/// limit the negative effects on parallelism as much as possible.
-	///
-	/// See `ChannelManager` struct-level documentation for lock order requirements.
-	pending_peers_awaiting_removal: Mutex<HashSet<PublicKey>>,
 	/// See `ChannelManager` struct-level documentation for lock order requirements.
 	pending_background_events: Mutex<Vec<BackgroundEvent>>,
 	/// Used when we have to take a BIG lock to make sure everything is self-consistent.
@@ -1304,11 +1292,10 @@ macro_rules! try_chan_entry {
 }
 
 macro_rules! remove_channel {
-	($self: expr, $entry: expr, $peer_state: expr) => {
+	($self: expr, $entry: expr) => {
 		{
 			let channel = $entry.remove_entry().1;
 			update_maps_on_chan_removal!($self, channel);
-			$self.add_pending_peer_to_be_removed(channel.get_counterparty_node_id(), $peer_state);
 			channel
 		}
 	}
@@ -1481,7 +1468,6 @@ where
 			per_peer_state: FairRwLock::new(HashMap::new()),
 
 			pending_events: Mutex::new(Vec::new()),
-			pending_peers_awaiting_removal: Mutex::new(HashSet::new()),
 			pending_background_events: Mutex::new(Vec::new()),
 			total_consistency_lock: RwLock::new(()),
 			persistence_notifier: Notifier::new(),
@@ -1720,7 +1706,7 @@ where
 						let (result, is_permanent) =
 							handle_monitor_update_res!(self, update_res, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, chan_entry.key(), NO_UPDATE);
 						if is_permanent {
-							remove_channel!(self, chan_entry, peer_state);
+							remove_channel!(self, chan_entry);
 							break result;
 						}
 					}
@@ -1731,7 +1717,7 @@ where
 					});
 
 					if chan_entry.get().is_shutdown() {
-						let channel = remove_channel!(self, chan_entry, peer_state);
+						let channel = remove_channel!(self, chan_entry);
 						if let Ok(channel_update) = self.get_channel_update_for_broadcast(&channel) {
 							peer_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
 								msg: channel_update
@@ -1834,7 +1820,7 @@ where
 				} else {
 					self.issue_channel_close_events(chan.get(),ClosureReason::HolderForceClosed);
 				}
-				remove_channel!(self, chan, peer_state)
+				remove_channel!(self, chan)
 			} else {
 				return Err(APIError::ChannelUnavailable{ err: format!("Channel with id {} not found for the passed counterparty node_id {}", log_bytes!(*channel_id), peer_node_id) });
 			}
@@ -1873,13 +1859,6 @@ where
 		}
 	}
 
-	fn add_pending_peer_to_be_removed(&self, counterparty_node_id: PublicKey, peer_state: &mut PeerState<<SP::Target as SignerProvider>::Signer>) {
-		let peer_should_be_removed = !peer_state.connected && peer_state.channel_by_id.len() == 0;
-		if peer_should_be_removed {
-			self.pending_peers_awaiting_removal.lock().unwrap().insert(counterparty_node_id);
-		}
-	}
-
 	/// Force closes a channel, immediately broadcasting the latest local transaction(s) and
 	/// rejecting new HTLCs on the given channel. Fails if `channel_id` is unknown to
 	/// the manager, or if the `counterparty_node_id` isn't the counterparty of the corresponding
@@ -3304,16 +3283,20 @@ where
 		true
 	}
 
-	/// Removes peers which have been been added to `pending_peers_awaiting_removal` which are
-	/// still disconnected and we have no channels to.
+	/// When a peer disconnects but still has channels, the peer's `peer_state` entry in the
+	/// `per_peer_state` is not removed by the `peer_disconnected` function. If the channels of
+	/// to that peer is later closed while still being disconnected (i.e. force closed), we
+	/// therefore need to remove the peer from `peer_state` separately.
+	/// To avoid having to take the `per_peer_state` `write` lock once the channels are closed, we
+	/// instead remove such peers awaiting removal through this function, which is called on a
+	/// timer through `timer_tick_occurred`, passing the peers disconnected peers with no channels,
+	/// to limit the negative effects on parallelism as much as possible.
 	///
 	/// Must be called without the `per_peer_state` lock acquired.
-	fn remove_peers_awaiting_removal(&self) {
-		let mut pending_peers_awaiting_removal = HashSet::new();
-		mem::swap(&mut *self.pending_peers_awaiting_removal.lock().unwrap(), &mut pending_peers_awaiting_removal);
+	fn remove_peers_awaiting_removal(&self, pending_peers_awaiting_removal: HashSet<PublicKey>) {
 		if pending_peers_awaiting_removal.len() > 0 {
 			let mut per_peer_state = self.per_peer_state.write().unwrap();
-			for counterparty_node_id in pending_peers_awaiting_removal.drain() {
+			for counterparty_node_id in pending_peers_awaiting_removal {
 				match per_peer_state.entry(counterparty_node_id) {
 					hash_map::Entry::Occupied(entry) => {
 						// Remove the entry if the peer is still disconnected and we still
@@ -3392,6 +3375,7 @@ where
 	///    the channel.
 	///  * Expiring a channel's previous `ChannelConfig` if necessary to only allow forwarding HTLCs
 	///    with the current `ChannelConfig`.
+	///  * Removing peers which have disconnected but and no longer have any channels.
 	///
 	/// Note that this may cause reentrancy through `chain::Watch::update_channel` calls or feerate
 	/// estimate fetches.
@@ -3404,6 +3388,7 @@ where
 
 			let mut handle_errors: Vec<(Result<(), _>, _)> = Vec::new();
 			let mut timed_out_mpp_htlcs = Vec::new();
+			let mut pending_peers_awaiting_removal = HashSet::new();
 			{
 				let per_peer_state = self.per_peer_state.read().unwrap();
 				for (counterparty_node_id, peer_state_mutex) in per_peer_state.iter() {
@@ -3451,10 +3436,13 @@ where
 
 						true
 					});
-					self.add_pending_peer_to_be_removed(counterparty_node_id, peer_state);
+					let peer_should_be_removed = !peer_state.connected && peer_state.channel_by_id.len() == 0;
+					if peer_should_be_removed {
+						pending_peers_awaiting_removal.insert(counterparty_node_id);
+					}
 				}
 			}
-			self.remove_peers_awaiting_removal();
+			self.remove_peers_awaiting_removal(pending_peers_awaiting_removal);
 
 			self.claimable_payments.lock().unwrap().claimable_htlcs.retain(|payment_hash, (_, htlcs)| {
 				if htlcs.is_empty() {
@@ -4190,7 +4178,7 @@ where
 						}
 					};
 					peer_state.pending_msg_events.push(send_msg_err_event);
-					let _ = remove_channel!(self, channel, peer_state);
+					let _ = remove_channel!(self, channel);
 					return Err(APIError::APIMisuseError { err: "Please use accept_inbound_channel_from_trusted_peer_0conf to accept channels with zero confirmations.".to_owned() });
 				}
 
@@ -4476,7 +4464,7 @@ where
 						let (result, is_permanent) =
 							handle_monitor_update_res!(self, update_res, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, chan_entry.key(), NO_UPDATE);
 						if is_permanent {
-							remove_channel!(self, chan_entry, peer_state);
+							remove_channel!(self, chan_entry);
 							break result;
 						}
 					}
@@ -4525,7 +4513,7 @@ where
 						// also implies there are no pending HTLCs left on the channel, so we can
 						// fully delete it from tracking (the channel monitor is still around to
 						// watch for old state broadcasts)!
-						(tx, Some(remove_channel!(self, chan_entry, peer_state)))
+						(tx, Some(remove_channel!(self, chan_entry)))
 					} else { (tx, None) }
 				},
 				hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close(format!("Got a message for a channel from the wrong node! No such channel for the passed counterparty_node_id {}", counterparty_node_id), msg.channel_id))
@@ -5028,11 +5016,12 @@ where
 							if let Some(peer_state_mutex) = per_peer_state.get(&counterparty_node_id) {
 								let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 								let peer_state = &mut *peer_state_lock;
+								let pending_msg_events = &mut peer_state.pending_msg_events;
 								if let hash_map::Entry::Occupied(chan_entry) = peer_state.channel_by_id.entry(funding_outpoint.to_channel_id()) {
-									let mut chan = remove_channel!(self, chan_entry, peer_state);
+									let mut chan = remove_channel!(self, chan_entry);
 									failed_channels.push(chan.force_shutdown(false));
 									if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-										peer_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+										pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
 											msg: update
 										});
 									}
@@ -5042,7 +5031,7 @@ where
 										ClosureReason::CommitmentTxConfirmed
 									};
 									self.issue_channel_close_events(&chan, reason);
-									peer_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
+									pending_msg_events.push(events::MessageSendEvent::HandleError {
 										node_id: chan.get_counterparty_node_id(),
 										action: msgs::ErrorAction::SendErrorMessage {
 											msg: msgs::ErrorMessage { channel_id: chan.channel_id(), data: "Channel force-closed".to_owned() }
@@ -5084,7 +5073,7 @@ where
 		{
 			let per_peer_state = self.per_peer_state.read().unwrap();
 
-			for (cp_id, peer_state_mutex) in per_peer_state.iter() {
+			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
 				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 				let peer_state = &mut *peer_state_lock;
 				let pending_msg_events = &mut peer_state.pending_msg_events;
@@ -5124,7 +5113,6 @@ where
 						}
 					}
 				});
-				self.add_pending_peer_to_be_removed(*cp_id, peer_state);
 			}
 		}
 
@@ -5149,7 +5137,7 @@ where
 		{
 			let per_peer_state = self.per_peer_state.read().unwrap();
 
-			for (cp_id, peer_state_mutex) in per_peer_state.iter() {
+			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
 				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 				let peer_state = &mut *peer_state_lock;
 				let pending_msg_events = &mut peer_state.pending_msg_events;
@@ -5187,7 +5175,6 @@ where
 						}
 					}
 				});
-				self.add_pending_peer_to_be_removed(*cp_id, peer_state);
 			}
 		}
 
@@ -5751,7 +5738,7 @@ where
 		let mut timed_out_htlcs = Vec::new();
 		{
 			let per_peer_state = self.per_peer_state.read().unwrap();
-			for (cp_id, peer_state_mutex) in per_peer_state.iter() {
+			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
 				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 				let peer_state = &mut *peer_state_lock;
 				let pending_msg_events = &mut peer_state.pending_msg_events;
@@ -5835,7 +5822,6 @@ where
 					}
 					true
 				});
-				self.add_pending_peer_to_be_removed(*cp_id, peer_state);
 			}
 		}
 
@@ -6161,7 +6147,7 @@ where
 
 		let per_peer_state = self.per_peer_state.read().unwrap();
 
-		for (cp_id, peer_state_mutex) in per_peer_state.iter() {
+		for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
 			let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 			let peer_state = &mut *peer_state_lock;
 			let pending_msg_events = &mut peer_state.pending_msg_events;
@@ -6193,7 +6179,6 @@ where
 				}
 				retain
 			});
-			self.add_pending_peer_to_be_removed(*cp_id, peer_state);
 		}
 		//TODO: Also re-broadcast announcement_signatures
 		Ok(())
@@ -6707,8 +6692,6 @@ where
 
 		write_ver_prefix!(writer, SERIALIZATION_VERSION, MIN_SERIALIZATION_VERSION);
 
-		self.remove_peers_awaiting_removal();
-
 		self.genesis_hash.write(writer)?;
 		{
 			let best_block = self.best_block.read().unwrap();
@@ -7530,7 +7513,6 @@ where
 			per_peer_state: FairRwLock::new(per_peer_state),
 
 			pending_events: Mutex::new(pending_events_read),
-			pending_peers_awaiting_removal: Mutex::new(HashSet::new()),
 			pending_background_events: Mutex::new(pending_background_events_read),
 			total_consistency_lock: RwLock::new(()),
 			persistence_notifier: Notifier::new(),
@@ -8019,9 +8001,6 @@ mod tests {
 			// Assert that nodes[1] is awaiting removal for nodes[0] once nodes[1] has been
 			// disconnected and the channel between has been force closed.
 			let nodes_0_per_peer_state = nodes[0].node.per_peer_state.read().unwrap();
-			let nodes_0_pending_peers_awaiting_removal = nodes[0].node.pending_peers_awaiting_removal.lock().unwrap();
-			assert_eq!(nodes_0_pending_peers_awaiting_removal.len(), 1);
-			assert!(nodes_0_pending_peers_awaiting_removal.get(&nodes[1].node.get_our_node_id()).is_some());
 			// Assert that nodes[1] isn't removed before `timer_tick_occurred` has been executed.
 			assert_eq!(nodes_0_per_peer_state.len(), 1);
 			assert!(nodes_0_per_peer_state.get(&nodes[1].node.get_our_node_id()).is_some());
@@ -8032,7 +8011,6 @@ mod tests {
 		{
 			// Assert that nodes[1] has now been removed.
 			assert_eq!(nodes[0].node.per_peer_state.read().unwrap().len(), 0);
-			assert_eq!(nodes[0].node.pending_peers_awaiting_removal.lock().unwrap().len(), 0);
 		}
 	}