Skip to content

Commit 137b77c

Browse files
TheBlueMattTheBlueMatt
authored
Merge pull request #2022 from alecchendev/2023-02-cache-peer-nodeid
Store a cached `NodeId` for each `Peer`
2 parents 56146e7 + 4c1055d commit 137b77c

File tree

2 files changed

+29
-25
lines changed

2 files changed

+29
-25
lines changed

‎lightning/src/ln/chan_utils.rs

-1
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@ use bitcoin::hash_types::{Txid, PubkeyHash};
2323

2424
use crate::ln::{PaymentHash, PaymentPreimage};
2525
use crate::ln::msgs::DecodeError;
26-
use crate::routing::gossip::NodeId;
2726
use crate::util::ser::{Readable, Writeable, Writer};
2827
use crate::util::transaction_utils;
2928

‎lightning/src/ln/peer_handler.rs

+29-24
Original file line numberDiff line numberDiff line change
@@ -389,7 +389,9 @@ const BUFFER_DRAIN_MSGS_PER_TICK: usize = 32;
389389

390390
struct Peer {
391391
channel_encryptor: PeerChannelEncryptor,
392-
their_node_id: Option<PublicKey>,
392+
/// We cache a `NodeId` here to avoid serializing peers' keys every time we forward gossip
393+
/// messages in `PeerManager`. Use `Peer::set_their_node_id` to modify this field.
394+
their_node_id: Option<(PublicKey, NodeId)>,
393395
their_features: Option<InitFeatures>,
394396
their_net_address: Option<NetAddress>,
395397

@@ -481,6 +483,10 @@ impl Peer {
481483
total_outbound_buffered > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP ||
482484
self.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO
483485
}
486+
487+
fn set_their_node_id(&mut self, node_id: PublicKey) {
488+
self.their_node_id = Some((node_id, NodeId::from_pubkey(&node_id)));
489+
}
484490
}
485491

486492
/// SimpleArcPeerManager is useful when you need a PeerManager with a static lifetime, e.g.
@@ -651,10 +657,10 @@ impl<Descriptor: SocketDescriptor, RM: Deref, L: Deref, NS: Deref> PeerManager<D
651657
/// This works around `format!()` taking a reference to each argument, preventing
652658
/// `if let Some(node_id) = peer.their_node_id { format!(.., node_id) } else { .. }` from compiling
653659
/// due to lifetime errors.
654-
struct OptionalFromDebugger<'a>(&'a Option<PublicKey>);
660+
struct OptionalFromDebugger<'a>(&'a Option<(PublicKey, NodeId)>);
655661
impl core::fmt::Display for OptionalFromDebugger<'_> {
656662
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> Result<(), core::fmt::Error> {
657-
if let Some(node_id) = self.0 { write!(f, " from {}", log_pubkey!(node_id)) } else { Ok(()) }
663+
if let Some((node_id, _)) = self.0 { write!(f, " from {}", log_pubkey!(node_id)) } else { Ok(()) }
658664
}
659665
}
660666

@@ -741,7 +747,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
741747
return None;
742748
}
743749
p.their_node_id
744-
}).collect()
750+
}).map(|(node_id, _)| node_id).collect()
745751
}
746752

747753
fn get_ephemeral_key(&self) -> SecretKey {
@@ -849,7 +855,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
849855
fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) {
850856
while !peer.awaiting_write_event {
851857
if peer.should_buffer_onion_message() {
852-
if let Some(peer_node_id) = peer.their_node_id {
858+
if let Some((peer_node_id, _)) = peer.their_node_id {
853859
if let Some(next_onion_message) =
854860
self.message_handler.onion_message_handler.next_onion_message_for_peer(peer_node_id) {
855861
self.enqueue_message(peer, &next_onion_message);
@@ -978,9 +984,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
978984
/// Append a message to a peer's pending outbound/write buffer
979985
fn enqueue_message<M: wire::Type>(&self, peer: &mut Peer, message: &M) {
980986
if is_gossip_msg(message.type_id()) {
981-
log_gossip!(self.logger, "Enqueueing message {:?} to {}", message, log_pubkey!(peer.their_node_id.unwrap()));
987+
log_gossip!(self.logger, "Enqueueing message {:?} to {}", message, log_pubkey!(peer.their_node_id.unwrap().0));
982988
} else {
983-
log_trace!(self.logger, "Enqueueing message {:?} to {}", message, log_pubkey!(peer.their_node_id.unwrap()))
989+
log_trace!(self.logger, "Enqueueing message {:?} to {}", message, log_pubkey!(peer.their_node_id.unwrap().0))
984990
}
985991
peer.msgs_sent_since_pong += 1;
986992
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(message));
@@ -1065,14 +1071,14 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
10651071

10661072
macro_rules! insert_node_id {
10671073
() => {
1068-
match self.node_id_to_descriptor.lock().unwrap().entry(peer.their_node_id.unwrap()) {
1074+
match self.node_id_to_descriptor.lock().unwrap().entry(peer.their_node_id.unwrap().0) {
10691075
hash_map::Entry::Occupied(_) => {
1070-
log_trace!(self.logger, "Got second connection with {}, closing", log_pubkey!(peer.their_node_id.unwrap()));
1076+
log_trace!(self.logger, "Got second connection with {}, closing", log_pubkey!(peer.their_node_id.unwrap().0));
10711077
peer.their_node_id = None; // Unset so that we don't generate a peer_disconnected event
10721078
return Err(PeerHandleError{ no_connection_possible: false })
10731079
},
10741080
hash_map::Entry::Vacant(entry) => {
1075-
log_debug!(self.logger, "Finished noise handshake for connection with {}", log_pubkey!(peer.their_node_id.unwrap()));
1081+
log_debug!(self.logger, "Finished noise handshake for connection with {}", log_pubkey!(peer.their_node_id.unwrap().0));
10761082
entry.insert(peer_descriptor.clone())
10771083
},
10781084
};
@@ -1096,7 +1102,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
10961102
peer.pending_read_buffer = [0; 18].to_vec(); // Message length header is 18 bytes
10971103
peer.pending_read_is_header = true;
10981104

1099-
peer.their_node_id = Some(their_node_id);
1105+
peer.set_their_node_id(their_node_id);
11001106
insert_node_id!();
11011107
let features = self.message_handler.chan_handler.provided_init_features(&their_node_id)
11021108
.or(self.message_handler.route_handler.provided_init_features(&their_node_id))
@@ -1110,7 +1116,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
11101116
peer.channel_encryptor.process_act_three(&peer.pending_read_buffer[..]));
11111117
peer.pending_read_buffer = [0; 18].to_vec(); // Message length header is 18 bytes
11121118
peer.pending_read_is_header = true;
1113-
peer.their_node_id = Some(their_node_id);
1119+
peer.set_their_node_id(their_node_id);
11141120
insert_node_id!();
11151121
let features = self.message_handler.chan_handler.provided_init_features(&their_node_id)
11161122
.or(self.message_handler.route_handler.provided_init_features(&their_node_id))
@@ -1212,7 +1218,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
12121218
}
12131219

12141220
for msg in msgs_to_forward.drain(..) {
1215-
self.forward_broadcast_msg(&*peers, &msg, peer_node_id.as_ref());
1221+
self.forward_broadcast_msg(&*peers, &msg, peer_node_id.as_ref().map(|(pk, _)| pk));
12161222
}
12171223

12181224
Ok(pause_read)
@@ -1226,7 +1232,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
12261232
mut peer_lock: MutexGuard<Peer>,
12271233
message: wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>
12281234
) -> Result<Option<wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>>, MessageHandlingError> {
1229-
let their_node_id = peer_lock.their_node_id.clone().expect("We know the peer's public key by the time we receive messages");
1235+
let their_node_id = peer_lock.their_node_id.clone().expect("We know the peer's public key by the time we receive messages").0;
12301236
peer_lock.received_message_since_timer_tick = true;
12311237

12321238
// Need an Init as first message
@@ -1467,13 +1473,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
14671473
log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
14681474
continue;
14691475
}
1470-
if let Some(their_node_id) = peer.their_node_id {
1471-
let their_node_id = NodeId::from_pubkey(&their_node_id);
1476+
if let Some((_, their_node_id)) = peer.their_node_id {
14721477
if their_node_id == msg.contents.node_id_1 || their_node_id == msg.contents.node_id_2 {
14731478
continue;
14741479
}
14751480
}
1476-
if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
1481+
if except_node.is_some() && peer.their_node_id.as_ref().map(|(pk, _)| pk) == except_node {
14771482
continue;
14781483
}
14791484
self.enqueue_encoded_gossip_broadcast(&mut *peer, encoded_msg.clone());
@@ -1493,12 +1498,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
14931498
log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
14941499
continue;
14951500
}
1496-
if let Some(their_node_id) = peer.their_node_id {
1497-
if NodeId::from_pubkey(&their_node_id) == msg.contents.node_id {
1501+
if let Some((_, their_node_id)) = peer.their_node_id {
1502+
if their_node_id == msg.contents.node_id {
14981503
continue;
14991504
}
15001505
}
1501-
if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
1506+
if except_node.is_some() && peer.their_node_id.as_ref().map(|(pk, _)| pk) == except_node {
15021507
continue;
15031508
}
15041509
self.enqueue_encoded_gossip_broadcast(&mut *peer, encoded_msg.clone());
@@ -1518,7 +1523,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
15181523
log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
15191524
continue;
15201525
}
1521-
if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
1526+
if except_node.is_some() && peer.their_node_id.as_ref().map(|(pk, _)| pk) == except_node {
15221527
continue;
15231528
}
15241529
self.enqueue_encoded_gossip_broadcast(&mut *peer, encoded_msg.clone());
@@ -1837,7 +1842,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
18371842
},
18381843
Some(peer_lock) => {
18391844
let peer = peer_lock.lock().unwrap();
1840-
if let Some(node_id) = peer.their_node_id {
1845+
if let Some((node_id, _)) = peer.their_node_id {
18411846
log_trace!(self.logger,
18421847
"Handling disconnection of peer {}, with {}future connection to the peer possible.",
18431848
log_pubkey!(node_id), if no_connection_possible { "no " } else { "" });
@@ -1877,7 +1882,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
18771882
self.node_id_to_descriptor.lock().unwrap().clear();
18781883
let peers = &mut *peers_lock;
18791884
for (mut descriptor, peer) in peers.drain() {
1880-
if let Some(node_id) = peer.lock().unwrap().their_node_id {
1885+
if let Some((node_id, _)) = peer.lock().unwrap().their_node_id {
18811886
log_trace!(self.logger, "Disconnecting peer with id {} due to client request to disconnect all peers", node_id);
18821887
self.message_handler.chan_handler.peer_disconnected(&node_id, false);
18831888
self.message_handler.onion_message_handler.peer_disconnected(&node_id, false);
@@ -1967,7 +1972,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
19671972
let mut peers_lock = self.peers.write().unwrap();
19681973
for descriptor in descriptors_needing_disconnect.iter() {
19691974
if let Some(peer) = peers_lock.remove(descriptor) {
1970-
if let Some(node_id) = peer.lock().unwrap().their_node_id {
1975+
if let Some((node_id, _)) = peer.lock().unwrap().their_node_id {
19711976
log_trace!(self.logger, "Disconnecting peer with id {} due to ping timeout", node_id);
19721977
self.node_id_to_descriptor.lock().unwrap().remove(&node_id);
19731978
self.message_handler.chan_handler.peer_disconnected(&node_id, false);

0 commit comments

Comments
 (0)