Skip to content
This repository was archived by the owner on Nov 15, 2023. It is now read-only.

Further reduce the CPU overhead of networking metrics #10875

Merged
paritytech-processbot merged 5 commits into from
Feb 18, 2022
Merged

Further reduce the CPU overhead of networking metrics #10875

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
489fe01
Simplify `num_connected_peers`
koute Feb 17, 2022
9f1c870
Track requested peer counts
koute Feb 17, 2022
5d4eb05
Revert "Track requested peer counts"
koute Feb 18, 2022
73d769a
Remove `substrate_sub_libp2p_peerset_num_requested` metric
koute Feb 18, 2022
5ae994b
Remove two unused functions that I forgot to get rid of in previous c…
koute Feb 18, 2022
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
Prev Previous commit
Next Next commit
Revert "Track requested peer counts" 
This reverts commit 9f1c870.
  • Loading branch information
@koute
koute committed Feb 18, 2022
commit 5d4eb05129a273ead2cbab1d340c4d32c29c9ee5
6 changes: 3 additions & 3 deletions client/network/src/protocol.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -465,10 +465,10 @@ impl<B: BlockT> Protocol<B> {
self.behaviour.open_peers()
}

/// Returns the number of all the peers that the peerset currently requests us to be connected
/// Returns the list of all the peers that the peerset currently requests us to be connected
/// to on the default set.
pub fn requested_peers_count(&self) -> usize {
self.behaviour.requested_peers_count(HARDCODED_PEERSETS_SYNC)
pub fn requested_peers(&self) -> impl Iterator<Item = &PeerId> {
self.behaviour.requested_peers(HARDCODED_PEERSETS_SYNC)
}

/// Returns the number of discovered nodes that we keep in memory.
Expand Down
83 changes: 9 additions & 74 deletions client/network/src/protocol/notifications/behaviour.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,10 +110,6 @@ pub struct Notifications {
/// List of peers in our state.
peers: FnvHashMap<(PeerId, sc_peerset::SetId), PeerState>,

/// A map containing the number of all the peers that the peerset currently requests us to be
/// connected to.
requested_peer_counts: FnvHashMap<sc_peerset::SetId, usize>,

/// The elements in `peers` occasionally contain `Delay` objects that we would normally have
/// to be polled one by one. In order to avoid doing so, as an optimization, every `Delay` is
/// instead put inside of `delays` and reference by a [`DelayId`]. This stream
Expand Down Expand Up @@ -367,32 +363,6 @@ pub enum NotificationsOut {
},
}

fn increment_requested_peer_counts(
requested_peer_counts: &mut FnvHashMap<sc_peerset::SetId, usize>,
set_id: sc_peerset::SetId,
) {
*requested_peer_counts.entry(set_id).or_insert(0) += 1;
}

fn decrement_requested_peer_counts(
requested_peer_counts: &mut FnvHashMap<sc_peerset::SetId, usize>,
set_id: sc_peerset::SetId,
) {
match requested_peer_counts.entry(set_id) {
Entry::Occupied(mut entry) => {
let value = entry.get_mut();
if *value == 1 {
entry.remove();
} else {
*value -= 1;
}
},
Entry::Vacant(..) => {
warn!(target: "sub-libp2p", "Underflow of requested peer count for {:?}", set_id)
},
}
}

impl Notifications {
/// Creates a `CustomProtos`.
pub fn new(
Expand All @@ -414,7 +384,6 @@ impl Notifications {
notif_protocols,
peerset,
peers: FnvHashMap::default(),
requested_peer_counts: Default::default(),
delays: Default::default(),
next_delay_id: DelayId(0),
incoming: SmallVec::new(),
Expand Down Expand Up @@ -488,7 +457,6 @@ impl Notifications {
} else {
timer_deadline
});
decrement_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*entry.into_mut() = PeerState::Disabled { connections, backoff_until }
},

Expand Down Expand Up @@ -538,7 +506,6 @@ impl Notifications {
.any(|(_, s)| matches!(s, ConnectionState::Opening)));

let backoff_until = ban.map(|dur| Instant::now() + dur);
decrement_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*entry.into_mut() = PeerState::Disabled { connections, backoff_until }
},

Expand Down Expand Up @@ -593,18 +560,15 @@ impl Notifications {
}
}

/// Returns the number of all the peers that the peerset currently requests us to be connected
/// to.
pub fn requested_peers_count(&self, set_id: sc_peerset::SetId) -> usize {
let count = self.requested_peer_counts.get(&set_id).copied().unwrap_or(0);
debug_assert_eq!(
self.peers
.iter()
.filter(move |((_, set), state)| *set == set_id && state.is_requested())
.count(),
count
);
count
/// Returns the list of all the peers that the peerset currently requests us to be connected to.
pub fn requested_peers<'a>(
&'a self,
set_id: sc_peerset::SetId,
) -> impl Iterator<Item = &'a PeerId> + 'a {
self.peers
.iter()
.filter(move |((_, set), state)| *set == set_id && state.is_requested())
.map(|((id, _), _)| id)
}

/// Returns the list of reserved peers.
Expand Down Expand Up @@ -683,7 +647,6 @@ impl Notifications {
condition: DialPeerCondition::Disconnected,
handler,
});
increment_requested_peer_counts(&mut self.requested_peer_counts, set_id);
entry.insert(PeerState::Requested);
return
},
Expand All @@ -702,7 +665,6 @@ impl Notifications {
set_id,
timer_deadline,
);
increment_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*occ_entry.into_mut() =
PeerState::PendingRequest { timer: *timer, timer_deadline: *timer_deadline };
},
Expand All @@ -722,7 +684,6 @@ impl Notifications {
condition: DialPeerCondition::Disconnected,
handler,
});
increment_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*occ_entry.into_mut() = PeerState::Requested;
},

Expand Down Expand Up @@ -750,7 +711,6 @@ impl Notifications {
.boxed(),
);

increment_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*occ_entry.into_mut() = PeerState::DisabledPendingEnable {
connections,
timer: delay_id,
Expand All @@ -777,7 +737,6 @@ impl Notifications {
event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
});
*connec_state = ConnectionState::Opening;
increment_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*occ_entry.into_mut() = PeerState::Enabled { connections };
} else {
// If no connection is available, switch to `DisabledPendingEnable` in order
Expand Down Expand Up @@ -812,7 +771,6 @@ impl Notifications {
.boxed(),
);

increment_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*occ_entry.into_mut() = PeerState::DisabledPendingEnable {
connections,
timer: delay_id,
Expand Down Expand Up @@ -855,7 +813,6 @@ impl Notifications {
*connec_state = ConnectionState::Opening;
}

increment_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*occ_entry.into_mut() = PeerState::Enabled { connections };
},

Expand Down Expand Up @@ -913,7 +870,6 @@ impl Notifications {
trace!(target: "sub-libp2p",
"PSM => Drop({}, {:?}): Interrupting pending enabling.",
entry.key().0, set_id);
decrement_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*entry.into_mut() =
PeerState::Disabled { connections, backoff_until: Some(timer_deadline) };
},
Expand Down Expand Up @@ -963,7 +919,6 @@ impl Notifications {
*connec_state = ConnectionState::Closing;
}

decrement_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*entry.into_mut() = PeerState::Disabled { connections, backoff_until: None }
},

Expand All @@ -974,17 +929,13 @@ impl Notifications {
// well at the same time.
trace!(target: "sub-libp2p", "PSM => Drop({}, {:?}): Not yet connected.",
entry.key().0, set_id);

decrement_requested_peer_counts(&mut self.requested_peer_counts, set_id);
entry.remove();
},

// PendingRequest => Backoff
PeerState::PendingRequest { timer, timer_deadline } => {
trace!(target: "sub-libp2p", "PSM => Drop({}, {:?}): Not yet connected",
entry.key().0, set_id);

decrement_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*entry.into_mut() = PeerState::Backoff { timer, timer_deadline }
},

Expand Down Expand Up @@ -1059,7 +1010,6 @@ impl Notifications {
*connec_state = ConnectionState::Opening;
}

increment_requested_peer_counts(&mut self.requested_peer_counts, incoming.set_id);
*state = PeerState::Enabled { connections };
},

Expand Down Expand Up @@ -1280,7 +1230,6 @@ impl NetworkBehaviour for Notifications {
if connections.is_empty() {
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
self.peerset.dropped(set_id, *peer_id, DropReason::Unknown);
decrement_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*entry.get_mut() = PeerState::Backoff { timer, timer_deadline };
} else {
*entry.get_mut() =
Expand Down Expand Up @@ -1433,7 +1382,6 @@ impl NetworkBehaviour for Notifications {
.boxed(),
);

decrement_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*entry.get_mut() = PeerState::Backoff {
timer: delay_id,
timer_deadline: Instant::now() + Duration::from_secs(ban_dur),
Expand All @@ -1444,7 +1392,6 @@ impl NetworkBehaviour for Notifications {
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
self.peerset.dropped(set_id, *peer_id, DropReason::Unknown);

decrement_requested_peer_counts(&mut self.requested_peer_counts, set_id);
*entry.get_mut() = PeerState::Disabled { connections, backoff_until: None };
} else {
*entry.get_mut() = PeerState::Enabled { connections };
Expand Down Expand Up @@ -1518,10 +1465,6 @@ impl NetworkBehaviour for Notifications {
.boxed(),
);

decrement_requested_peer_counts(
&mut self.requested_peer_counts,
set_id,
);
*entry.into_mut() = PeerState::Backoff {
timer: delay_id,
timer_deadline: now + ban_duration,
Expand Down Expand Up @@ -1808,10 +1751,6 @@ impl NetworkBehaviour for Notifications {
{
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", source, set_id);
self.peerset.dropped(set_id, source, DropReason::Refused);
decrement_requested_peer_counts(
&mut self.requested_peer_counts,
set_id,
);
*entry.into_mut() =
PeerState::Disabled { connections, backoff_until: None };
} else {
Expand Down Expand Up @@ -1989,10 +1928,6 @@ impl NetworkBehaviour for Notifications {
self.peerset.dropped(set_id, source, DropReason::Refused);

let ban_dur = Uniform::new(5, 10).sample(&mut rand::thread_rng());
decrement_requested_peer_counts(
&mut self.requested_peer_counts,
set_id,
);
*entry.into_mut() = PeerState::Disabled {
connections,
backoff_until: Some(Instant::now() + Duration::from_secs(ban_dur)),
Expand Down
8 changes: 4 additions & 4 deletions client/network/src/service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2108,10 +2108,10 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
.peerset_num_discovered
.set(this.network_service.behaviour_mut().user_protocol().num_discovered_peers()
as u64);
metrics
.peerset_num_requested
.set(this.network_service.behaviour_mut().user_protocol().requested_peers_count()
as u64);
metrics.peerset_num_requested.set(
this.network_service.behaviour_mut().user_protocol().requested_peers().count()
as u64,
);
metrics.pending_connections.set(
Swarm::network_info(&this.network_service).connection_counters().num_pending()
as u64,
Expand Down