Skip to content

Commit ca153fe

Browse files
TheBlueMattTheBlueMatt
committed
Drop peers_needing_send tracking and try to write for each peer
We do a lot of work to track which peers have buffers which need to be flushed, when we could instead just try to flush ever peer's buffer.
1 parent 57c286f commit ca153fe

File tree

1 file changed

+27
-59
lines changed

1 file changed

+27
-59
lines changed

‎lightning/src/ln/peer_handler.rs

Lines changed: 27 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -287,9 +287,6 @@ impl Peer {
287287

288288
struct PeerHolder<Descriptor: SocketDescriptor> {
289289
peers: HashMap<Descriptor, Peer>,
290-
/// Added to by do_read_event for cases where we pushed a message onto the send buffer but
291-
/// didn't call do_attempt_write_data to avoid reentrancy. Cleared in process_events()
292-
peers_needing_send: HashSet<Descriptor>,
293290
/// Only add to this set when noise completes:
294291
node_id_to_descriptor: HashMap<PublicKey, Descriptor>,
295292
}
@@ -420,7 +417,6 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
420417
message_handler,
421418
peers: Mutex::new(PeerHolder {
422419
peers: HashMap::new(),
423-
peers_needing_send: HashSet::new(),
424420
node_id_to_descriptor: HashMap::new()
425421
}),
426422
our_node_secret,
@@ -652,14 +648,13 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
652648
}
653649

654650
/// Append a message to a peer's pending outbound/write buffer, and update the map of peers needing sends accordingly.
655-
fn enqueue_message<M: Encode + Writeable>(&self, peers_needing_send: &mut HashSet<Descriptor>, peer: &mut Peer, descriptor: Descriptor, message: &M) {
651+
fn enqueue_message<M: Encode + Writeable>(&self, peer: &mut Peer, message: &M) {
656652
let mut buffer = VecWriter(Vec::new());
657653
wire::write(message, &mut buffer).unwrap(); // crash if the write failed
658654
let encoded_message = buffer.0;
659655

660656
log_trace!(self.logger, "Enqueueing message of type {} to {}", message.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
661657
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_message[..]));
662-
peers_needing_send.insert(descriptor);
663658
}
664659

665660
fn do_read_event(&self, peer_descriptor: &mut Descriptor, data: &[u8]) -> Result<bool, PeerHandleError> {
@@ -703,7 +698,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
703698
},
704699
msgs::ErrorAction::SendErrorMessage { msg } => {
705700
log_trace!(self.logger, "Got Err handling message, sending Error message because {}", e.err);
706-
self.enqueue_message(&mut peers.peers_needing_send, peer, peer_descriptor.clone(), &msg);
701+
self.enqueue_message(peer, &msg);
707702
continue;
708703
},
709704
}
@@ -745,7 +740,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
745740
insert_node_id!();
746741
let features = InitFeatures::known();
747742
let resp = msgs::Init { features };
748-
self.enqueue_message(&mut peers.peers_needing_send, peer, peer_descriptor.clone(), &resp);
743+
self.enqueue_message(peer, &resp);
749744
},
750745
NextNoiseStep::ActThree => {
751746
let their_node_id = try_potential_handleerror!(peer.channel_encryptor.process_act_three(&peer.pending_read_buffer[..]));
@@ -755,7 +750,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
755750
insert_node_id!();
756751
let features = InitFeatures::known();
757752
let resp = msgs::Init { features };
758-
self.enqueue_message(&mut peers.peers_needing_send, peer, peer_descriptor.clone(), &resp);
753+
self.enqueue_message(peer, &resp);
759754
},
760755
NextNoiseStep::NoiseComplete => {
761756
if peer.pending_read_is_header {
@@ -803,7 +798,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
803798
}
804799
};
805800

806-
match self.handle_message(&mut peers.peers_needing_send, peer, peer_descriptor.clone(), message) {
801+
match self.handle_message(peer, message) {
807802
Err(handling_error) => match handling_error {
808803
MessageHandlingError::PeerHandleError(e) => { return Err(e) },
809804
MessageHandlingError::LightningError(e) => {
@@ -838,7 +833,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
838833

839834
/// Process an incoming message and return a decision (ok, lightning error, peer handling error) regarding the next action with the peer
840835
/// Returns the message back if it needs to be broadcasted to all other peers.
841-
fn handle_message(&self, peers_needing_send: &mut HashSet<Descriptor>, peer: &mut Peer, peer_descriptor: Descriptor, message: wire::Message) -> Result<Option<wire::Message>, MessageHandlingError> {
836+
fn handle_message(&self, peer: &mut Peer, message: wire::Message) -> Result<Option<wire::Message>, MessageHandlingError> {
842837
log_trace!(self.logger, "Received message of type {} from {}", message.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
843838

844839
// Need an Init as first message
@@ -873,7 +868,6 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
873868

874869
if msg.features.initial_routing_sync() {
875870
peer.sync_status = InitSyncTracker::ChannelsSyncing(0);
876-
peers_needing_send.insert(peer_descriptor.clone());
877871
}
878872
if !msg.features.supports_static_remote_key() {
879873
log_debug!(self.logger, "Peer {} does not support static remote key, disconnecting with no_connection_possible", log_pubkey!(peer.their_node_id.unwrap()));
@@ -908,7 +902,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
908902
wire::Message::Ping(msg) => {
909903
if msg.ponglen < 65532 {
910904
let resp = msgs::Pong { byteslen: msg.ponglen };
911-
self.enqueue_message(peers_needing_send, peer, peer_descriptor.clone(), &resp);
905+
self.enqueue_message(peer, &resp);
912906
}
913907
},
914908
wire::Message::Pong(_msg) => {
@@ -1030,7 +1024,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
10301024
wire::Message::ChannelAnnouncement(ref msg) => {
10311025
let encoded_msg = encode_msg!(msg);
10321026

1033-
for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
1027+
for (_, ref mut peer) in peers.peers.iter_mut() {
10341028
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
10351029
!peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
10361030
continue
@@ -1046,13 +1040,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
10461040
continue;
10471041
}
10481042
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
1049-
peers.peers_needing_send.insert((*descriptor).clone());
10501043
}
10511044
},
10521045
wire::Message::NodeAnnouncement(ref msg) => {
10531046
let encoded_msg = encode_msg!(msg);
10541047

1055-
for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
1048+
for (_, ref mut peer) in peers.peers.iter_mut() {
10561049
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
10571050
!peer.should_forward_node_announcement(msg.contents.node_id) {
10581051
continue
@@ -1067,13 +1060,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
10671060
continue;
10681061
}
10691062
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
1070-
peers.peers_needing_send.insert((*descriptor).clone());
10711063
}
10721064
},
10731065
wire::Message::ChannelUpdate(ref msg) => {
10741066
let encoded_msg = encode_msg!(msg);
10751067

1076-
for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
1068+
for (_, ref mut peer) in peers.peers.iter_mut() {
10771069
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
10781070
!peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
10791071
continue
@@ -1085,7 +1077,6 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
10851077
continue;
10861078
}
10871079
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
1088-
peers.peers_needing_send.insert((*descriptor).clone());
10891080
}
10901081
},
10911082
_ => debug_assert!(false, "We shouldn't attempt to forward anything but gossip messages"),
@@ -1132,17 +1123,15 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
11321123
log_trace!(self.logger, "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
11331124
log_pubkey!(node_id),
11341125
log_bytes!(msg.temporary_channel_id));
1135-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1126+
let (_, peer) = get_peer_for_forwarding!(node_id);
11361127
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1137-
self.do_attempt_write_data(&mut descriptor, peer);
11381128
},
11391129
MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
11401130
log_trace!(self.logger, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
11411131
log_pubkey!(node_id),
11421132
log_bytes!(msg.temporary_channel_id));
1143-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1133+
let (_, peer) = get_peer_for_forwarding!(node_id);
11441134
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1145-
self.do_attempt_write_data(&mut descriptor, peer);
11461135
},
11471136
MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
11481137
log_trace!(self.logger, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
@@ -1151,33 +1140,29 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
11511140
log_funding_channel_id!(msg.funding_txid, msg.funding_output_index));
11521141
// TODO: If the peer is gone we should generate a DiscardFunding event
11531142
// indicating to the wallet that they should just throw away this funding transaction
1154-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1143+
let (_, peer) = get_peer_for_forwarding!(node_id);
11551144
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1156-
self.do_attempt_write_data(&mut descriptor, peer);
11571145
},
11581146
MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
11591147
log_trace!(self.logger, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
11601148
log_pubkey!(node_id),
11611149
log_bytes!(msg.channel_id));
1162-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1150+
let (_, peer) = get_peer_for_forwarding!(node_id);
11631151
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1164-
self.do_attempt_write_data(&mut descriptor, peer);
11651152
},
11661153
MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
11671154
log_trace!(self.logger, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
11681155
log_pubkey!(node_id),
11691156
log_bytes!(msg.channel_id));
1170-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1157+
let (_, peer) = get_peer_for_forwarding!(node_id);
11711158
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1172-
self.do_attempt_write_data(&mut descriptor, peer);
11731159
},
11741160
MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
11751161
log_trace!(self.logger, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
11761162
log_pubkey!(node_id),
11771163
log_bytes!(msg.channel_id));
1178-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1164+
let (_, peer) = get_peer_for_forwarding!(node_id);
11791165
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1180-
self.do_attempt_write_data(&mut descriptor, peer);
11811166
},
11821167
MessageSendEvent::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref update_fee, ref commitment_signed } } => {
11831168
log_trace!(self.logger, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
@@ -1186,7 +1171,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
11861171
update_fulfill_htlcs.len(),
11871172
update_fail_htlcs.len(),
11881173
log_bytes!(commitment_signed.channel_id));
1189-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1174+
let (_, peer) = get_peer_for_forwarding!(node_id);
11901175
for msg in update_add_htlcs {
11911176
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
11921177
}
@@ -1203,39 +1188,34 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
12031188
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
12041189
}
12051190
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed)));
1206-
self.do_attempt_write_data(&mut descriptor, peer);
12071191
},
12081192
MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
12091193
log_trace!(self.logger, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
12101194
log_pubkey!(node_id),
12111195
log_bytes!(msg.channel_id));
1212-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1196+
let (_, peer) = get_peer_for_forwarding!(node_id);
12131197
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1214-
self.do_attempt_write_data(&mut descriptor, peer);
12151198
},
12161199
MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => {
12171200
log_trace!(self.logger, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
12181201
log_pubkey!(node_id),
12191202
log_bytes!(msg.channel_id));
1220-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1203+
let (_, peer) = get_peer_for_forwarding!(node_id);
12211204
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1222-
self.do_attempt_write_data(&mut descriptor, peer);
12231205
},
12241206
MessageSendEvent::SendShutdown { ref node_id, ref msg } => {
12251207
log_trace!(self.logger, "Handling Shutdown event in peer_handler for node {} for channel {}",
12261208
log_pubkey!(node_id),
12271209
log_bytes!(msg.channel_id));
1228-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1210+
let (_, peer) = get_peer_for_forwarding!(node_id);
12291211
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1230-
self.do_attempt_write_data(&mut descriptor, peer);
12311212
},
12321213
MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
12331214
log_trace!(self.logger, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
12341215
log_pubkey!(node_id),
12351216
log_bytes!(msg.channel_id));
1236-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1217+
let (_, peer) = get_peer_for_forwarding!(node_id);
12371218
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1238-
self.do_attempt_write_data(&mut descriptor, peer);
12391219
},
12401220
MessageSendEvent::BroadcastChannelAnnouncement { msg, update_msg } => {
12411221
log_trace!(self.logger, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
@@ -1263,7 +1243,6 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
12631243
match *action {
12641244
msgs::ErrorAction::DisconnectPeer { ref msg } => {
12651245
if let Some(mut descriptor) = peers.node_id_to_descriptor.remove(node_id) {
1266-
peers.peers_needing_send.remove(&descriptor);
12671246
if let Some(mut peer) = peers.peers.remove(&descriptor) {
12681247
if let Some(ref msg) = *msg {
12691248
log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}",
@@ -1286,21 +1265,18 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
12861265
log_trace!(self.logger, "Handling SendErrorMessage HandleError event in peer_handler for node {} with message {}",
12871266
log_pubkey!(node_id),
12881267
msg.data);
1289-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1268+
let (_, peer) = get_peer_for_forwarding!(node_id);
12901269
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1291-
self.do_attempt_write_data(&mut descriptor, peer);
12921270
},
12931271
}
12941272
},
12951273
MessageSendEvent::SendChannelRangeQuery { ref node_id, ref msg } => {
1296-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1274+
let (_, peer) = get_peer_for_forwarding!(node_id);
12971275
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1298-
self.do_attempt_write_data(&mut descriptor, peer);
12991276
},
13001277
MessageSendEvent::SendShortIdsQuery { ref node_id, ref msg } => {
1301-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1278+
let (_, peer) = get_peer_for_forwarding!(node_id);
13021279
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1303-
self.do_attempt_write_data(&mut descriptor, peer);
13041280
}
13051281
MessageSendEvent::SendReplyChannelRange { ref node_id, ref msg } => {
13061282
log_trace!(self.logger, "Handling SendReplyChannelRange event in peer_handler for node {} with num_scids={} first_blocknum={} number_of_blocks={}, sync_complete={}",
@@ -1309,18 +1285,14 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
13091285
msg.first_blocknum,
13101286
msg.number_of_blocks,
13111287
msg.sync_complete);
1312-
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id);
1288+
let (_, peer) = get_peer_for_forwarding!(node_id);
13131289
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
1314-
self.do_attempt_write_data(&mut descriptor, peer);
13151290
}
13161291
}
13171292
}
13181293

1319-
for mut descriptor in peers.peers_needing_send.drain() {
1320-
match peers.peers.get_mut(&descriptor) {
1321-
Some(peer) => self.do_attempt_write_data(&mut descriptor, peer),
1322-
None => panic!("Inconsistent peers set state!"),
1323-
}
1294+
for (descriptor, ref mut peer) in peers.peers.iter_mut() {
1295+
self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
13241296
}
13251297
}
13261298
}
@@ -1339,7 +1311,6 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
13391311

13401312
fn disconnect_event_internal(&self, descriptor: &Descriptor, no_connection_possible: bool) {
13411313
let mut peers = self.peers.lock().unwrap();
1342-
peers.peers_needing_send.remove(descriptor);
13431314
let peer_option = peers.peers.remove(descriptor);
13441315
match peer_option {
13451316
None => panic!("Descriptor for disconnect_event is not already known to PeerManager"),
@@ -1367,7 +1338,6 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
13671338
if let Some(mut descriptor) = peers_lock.node_id_to_descriptor.remove(&node_id) {
13681339
log_trace!(self.logger, "Disconnecting peer with id {} due to client request", node_id);
13691340
peers_lock.peers.remove(&descriptor);
1370-
peers_lock.peers_needing_send.remove(&descriptor);
13711341
self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible);
13721342
descriptor.disconnect_socket();
13731343
}
@@ -1381,14 +1351,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
13811351
let mut peers_lock = self.peers.lock().unwrap();
13821352
{
13831353
let peers = &mut *peers_lock;
1384-
let peers_needing_send = &mut peers.peers_needing_send;
13851354
let node_id_to_descriptor = &mut peers.node_id_to_descriptor;
13861355
let peers = &mut peers.peers;
13871356
let mut descriptors_needing_disconnect = Vec::new();
13881357

13891358
peers.retain(|descriptor, peer| {
13901359
if peer.awaiting_pong {
1391-
peers_needing_send.remove(descriptor);
13921360
descriptors_needing_disconnect.push(descriptor.clone());
13931361
match peer.their_node_id {
13941362
Some(node_id) => {

0 commit comments

Comments
 (0)