Skip to content

Commit

Permalink
WIP: separate gossip broadcasts into their own queue in PeerManager
Browse files Browse the repository at this point in the history 
This allows us to better prioritize channel messages over gossip broadcasts and
lays groundwork for rate limiting onion messages more simply.
  • Loading branch information
@valentinewallace
valentinewallace committed Aug 24, 2022
1 parent 06d3daf commit 5f4e243
Showing 1 changed file with 47 additions and 23 deletions.
70 changes: 47 additions & 23 deletions lightning/src/ln/peer_handler.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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>,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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 }
}
}

Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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) => {
Expand All @@ -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) => {
Expand All @@ -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"),
Expand Down

0 comments on commit 5f4e243

Please sign in to comment.