Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add discv4 terminate #4879

Merged
merged 6 commits into from
Oct 3, 2023
Merged

feat: add discv4 terminate #4879

Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
aef57f6
Update lib.rs
DoTheBestToGetTheBest Oct 2, 2023
d77482a
Update lib.rs
DoTheBestToGetTheBest Oct 2, 2023
0bc07a3
Update lib.rs
DoTheBestToGetTheBest Oct 2, 2023
ce93096
Update lib.rs
DoTheBestToGetTheBest Oct 2, 2023
40eb946
Update lib.rs
DoTheBestToGetTheBest Oct 3, 2023
d39b4cf
Update crates/net/discv4/src/lib.rs
mattsse Oct 3, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 54 additions & 31 deletions crates/net/discv4/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -389,6 +389,12 @@ impl Discv4 {
self.to_service.send(cmd)?;
Ok(rx.await?)
}
/// Terminates the Discv4Service.

pub fn terminate(&self) {
let cmd = Discv4Command::Terminated;
self.send_to_service(cmd);
}
}

/// Manages discv4 peer discovery over UDP.
Expand Down Expand Up @@ -711,8 +717,8 @@ impl Discv4Service {
self.kbuckets
.closest_values(&target_key)
.filter(|node| {
node.value.has_endpoint_proof &&
!self.pending_find_nodes.contains_key(&node.key.preimage().0)
node.value.has_endpoint_proof
&& !self.pending_find_nodes.contains_key(&node.key.preimage().0)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please use nightly formatting cargo +nightly fmt

})
.take(MAX_NODES_PER_BUCKET)
.map(|n| (target_key.distance(&n.key), n.value.record)),
Expand All @@ -728,7 +734,7 @@ impl Discv4Service {
// (e.g. connectivity problems over a long period of time, or issues during initial
// bootstrapping) so we attempt to bootstrap again
self.bootstrap();
return
return;
}

trace!(target : "discv4", ?target, num = closest.len(), "Start lookup closest nodes");
Expand Down Expand Up @@ -806,7 +812,7 @@ impl Discv4Service {
fn has_bond(&self, remote_id: PeerId, remote_ip: IpAddr) -> bool {
if let Some(timestamp) = self.received_pongs.last_pong(remote_id, remote_ip) {
if timestamp.elapsed() < self.config.bond_expiration {
return true
return true;
}
}
false
Expand All @@ -818,7 +824,7 @@ impl Discv4Service {
/// followup request to retrieve the updated ENR
fn update_on_reping(&mut self, record: NodeRecord, mut last_enr_seq: Option<u64>) {
if record.id == self.local_node_record.id {
return
return;
}

// If EIP868 extension is disabled then we want to ignore this
Expand Down Expand Up @@ -852,7 +858,7 @@ impl Discv4Service {
/// Callback invoked when we receive a pong from the peer.
fn update_on_pong(&mut self, record: NodeRecord, mut last_enr_seq: Option<u64>) {
if record.id == *self.local_peer_id() {
return
return;
}

// If EIP868 extension is disabled then we want to ignore this
Expand Down Expand Up @@ -959,7 +965,7 @@ impl Discv4Service {
fn on_ping(&mut self, ping: Ping, remote_addr: SocketAddr, remote_id: PeerId, hash: B256) {
if self.is_expired(ping.expire) {
// ping's expiration timestamp is in the past
return
return;
}

// create the record
Expand Down Expand Up @@ -1057,17 +1063,17 @@ impl Discv4Service {
fn try_ping(&mut self, node: NodeRecord, reason: PingReason) {
if node.id == *self.local_peer_id() {
// don't ping ourselves
return
return;
}

if self.pending_pings.contains_key(&node.id) ||
self.pending_find_nodes.contains_key(&node.id)
if self.pending_pings.contains_key(&node.id)
|| self.pending_find_nodes.contains_key(&node.id)
{
return
return;
}

if self.queued_pings.iter().any(|(n, _)| n.id == node.id) {
return
return;
}

if self.pending_pings.len() < MAX_NODES_PING {
Expand Down Expand Up @@ -1102,7 +1108,7 @@ impl Discv4Service {
/// Returns the echo hash of the ping message.
pub(crate) fn send_enr_request(&mut self, node: NodeRecord) {
if !self.config.enable_eip868 {
return
return;
}
let remote_addr = node.udp_addr();
let enr_request = EnrRequest { expire: self.enr_request_expiration() };
Expand All @@ -1117,7 +1123,7 @@ impl Discv4Service {
/// Message handler for an incoming `Pong`.
fn on_pong(&mut self, pong: Pong, remote_addr: SocketAddr, remote_id: PeerId) {
if self.is_expired(pong.expire) {
return
return;
}

let PingRequest { node, reason, .. } = match self.pending_pings.entry(remote_id) {
Expand All @@ -1126,7 +1132,7 @@ impl Discv4Service {
let request = entry.get();
if request.echo_hash != pong.echo {
debug!( target : "discv4", from=?remote_addr, expected=?request.echo_hash, echo_hash=?pong.echo,"Got unexpected Pong");
return
return;
}
}
entry.remove()
Expand Down Expand Up @@ -1164,11 +1170,11 @@ impl Discv4Service {
fn on_find_node(&mut self, msg: FindNode, remote_addr: SocketAddr, node_id: PeerId) {
if self.is_expired(msg.expire) {
// ping's expiration timestamp is in the past
return
return;
}
if node_id == *self.local_peer_id() {
// ignore find node requests to ourselves
return
return;
}

if self.has_bond(node_id, remote_addr.ip()) {
Expand Down Expand Up @@ -1216,7 +1222,7 @@ impl Discv4Service {
request_hash: B256,
) {
if !self.config.enable_eip868 || self.is_expired(msg.expire) {
return
return;
}

if self.has_bond(id, remote_addr.ip()) {
Expand All @@ -1235,7 +1241,7 @@ impl Discv4Service {
fn on_neighbours(&mut self, msg: Neighbours, remote_addr: SocketAddr, node_id: PeerId) {
if self.is_expired(msg.expire) {
// response is expired
return
return;
}
// check if this request was expected
let ctx = match self.pending_find_nodes.entry(node_id) {
Expand All @@ -1251,7 +1257,7 @@ impl Discv4Service {
request.response_count = total;
} else {
debug!(target : "discv4", total, from=?remote_addr, "Received neighbors packet entries exceeds max nodes per bucket");
return
return;
}
};

Expand All @@ -1267,7 +1273,7 @@ impl Discv4Service {
Entry::Vacant(_) => {
// received neighbours response without requesting it
debug!( target : "discv4", from=?remote_addr, "Received unsolicited Neighbours");
return
return;
}
};

Expand All @@ -1277,7 +1283,7 @@ impl Discv4Service {
// prevent banned peers from being added to the context
if self.config.ban_list.is_banned(&node.id, &node.address) {
trace!(target: "discv4", peer_id=?node.id, ip=?node.address, "ignoring banned record");
continue
continue;
}

ctx.add_node(node);
Expand Down Expand Up @@ -1346,7 +1352,7 @@ impl Discv4Service {
self.pending_pings.retain(|node_id, ping_request| {
if now.duration_since(ping_request.sent_at) > self.config.ping_expiration {
failed_pings.push(*node_id);
return false
return false;
}
true
});
Expand All @@ -1371,7 +1377,7 @@ impl Discv4Service {
// treat this as an hard error since it responded.
failed_neighbours.push(*node_id);
}
return false
return false;
}
true
});
Expand Down Expand Up @@ -1399,7 +1405,7 @@ impl Discv4Service {
if let Some(bucket) = self.kbuckets.get_bucket(&key) {
if bucket.num_entries() < MAX_NODES_PER_BUCKET / 2 {
// skip half empty bucket
continue
continue;
}
}
self.remove_node(node_id);
Expand Down Expand Up @@ -1446,7 +1452,7 @@ impl Discv4Service {
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
if self.config.enforce_expiration_timestamps && timestamp < now {
debug!(target: "discv4", "Expired packet");
return Err(())
return Err(());
}
Ok(())
}
Expand Down Expand Up @@ -1490,7 +1496,7 @@ impl Discv4Service {
loop {
// drain buffered events first
if let Some(event) = self.queued_events.pop_front() {
return Poll::Ready(event)
return Poll::Ready(event);
}

// trigger self lookup
Expand Down Expand Up @@ -1554,6 +1560,12 @@ impl Discv4Service {
let _ = self.local_eip_868_enr.set_tcp6(port, &self.secret_key);
}
}

Discv4Command::Terminated => {
//self.terminate();
//todos!
self.queued_events.push_back(Discv4Event::Terminated);
}
}
}

Expand Down Expand Up @@ -1612,7 +1624,7 @@ impl Discv4Service {
}

if self.queued_events.is_empty() {
return Poll::Pending
return Poll::Pending;
}
}
}
Expand All @@ -1623,7 +1635,15 @@ impl Stream for Discv4Service {
type Item = Discv4Event;

fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Poll::Ready(Some(ready!(self.get_mut().poll(cx))))
// Poll the internal poll method
let event = ready!(self.get_mut().poll(cx));

match event {
// If the event is Terminated, return Poll::Ready(None)
Discv4Event::Terminated => Poll::Ready(None),
// For any other event, return Poll::Ready(Some(event))
_ => Poll::Ready(Some(event)),
}
}
}

Expand All @@ -1644,6 +1664,8 @@ pub enum Discv4Event {
EnrRequest,
/// A `EnrResponse` message was handled.
EnrResponse,
/// A
mattsse marked this conversation as resolved.
Show resolved Hide resolved
Terminated,
}

/// Continuously reads new messages from the channel and writes them to the socket
Expand Down Expand Up @@ -1688,7 +1710,7 @@ pub(crate) async fn receive_loop(udp: Arc<UdpSocket>, tx: IngressSender, local_i
if packet.node_id == local_id {
// received our own message
debug!(target : "discv4", ?remote_addr, "Received own packet.");
continue
continue;
}
send(IngressEvent::Packet(remote_addr, packet)).await;
}
Expand All @@ -1714,6 +1736,7 @@ enum Discv4Command {
Lookup { node_id: Option<PeerId>, tx: Option<NodeRecordSender> },
SetLookupInterval(Duration),
Updates(OneshotSender<ReceiverStream<DiscoveryUpdate>>),
Terminated,
}

/// Event type receiver produces
Expand Down Expand Up @@ -1772,7 +1795,7 @@ impl LookupTargetRotator {
self.counter += 1;
self.counter %= self.interval;
if self.counter == 0 {
return *local
return *local;
}
PeerId::random()
}
Expand Down