feat(gossipsub): introduce backpressure #5595
Conversation
jxs
force-pushed
the
impl-gossipsub-backpressure-2
branch
from
6d406e1
to
5e349f5
Compare
jxs
force-pushed
the
impl-gossipsub-backpressure-2
branch
from
5e349f5
to
00cde64
Compare
protocols/gossipsub/src/behaviour.rs
Outdated
| // before we add all the gossip from this heartbeat in order to gain a true measure of | ||
| // steady-state size of the queues. | ||
| if let Some(m) = &mut self.metrics { | ||
| for sender_queue in self.connected_peers.values_mut().map(|v| &v.sender) { |
There was a problem hiding this comment.
| for sender_queue in self.connected_peers.values_mut().map(|v| &v.sender) { | |
| for sender_queue in self.connected_peers.values().map(|v| &v.sender) { |
| // Slow peer weighting | ||
| if peer_stats.slow_peer_penalty > self.params.slow_peer_threshold { | ||
| let excess = peer_stats.slow_peer_penalty - self.params.slow_peer_threshold; | ||
| score += excess * self.params.slow_peer_weight; | ||
| } |
There was a problem hiding this comment.
We should move this nested if statement (Slow peer weighting) outside of P7: behavioural pattern penalty?
There was a problem hiding this comment.
right, thanks Akihito. Updated!
| slow_peer_weight: -0.2, | ||
| slow_peer_threshold: 0.0, | ||
| slow_peer_decay: 0.2, |
There was a problem hiding this comment.
where do these numbers come from?
There was a problem hiding this comment.
They are made up as suggestions.
These are rough estimates from what we expect on real networks. I.e They are minor penalties relative to the behaviour penalty.
We didn't want to penalize peers too much, so having 50 failures to 1 behaviour penalty seemed reasonable.
Note that these are not on by default and it is generally recommended that users set their own scoring based on their own networks.
There was a problem hiding this comment.
We didn't want to penalize peers too much, so having 50 failures to 1 behaviour penalty seemed reasonable.
OK, makes sense to me! Since this is defining the default for PeerScoreParams, it should be a safe and documented value (even if not enabled by default). IMO a comment saying that the slow_peer_weight is 50x lower than behaviour_penalty_weight would help.
protocols/gossipsub/src/types.rs
Outdated
| pub(crate) fn new(cap: usize) -> RpcSender { | ||
| let (priority_sender, priority_receiver) = async_channel::unbounded(); | ||
| let (non_priority_sender, non_priority_receiver) = async_channel::bounded(cap / 2); | ||
| let len = Arc::new(AtomicUsize::new(0)); | ||
| RpcSender { | ||
| cap: cap / 2, |
There was a problem hiding this comment.
what is the reasoning of using cap / 2 instead of cap?
There was a problem hiding this comment.
cause we have two channels priority and non_priority left a comment elaborating on that, thanks Gui
| if self.len.load(Ordering::Relaxed) >= self.cap { | ||
| return Err(()); |
There was a problem hiding this comment.
self.cap is only used in publish. why?
There was a problem hiding this comment.
because publish calls to self.priority_sender which is unbounded, and we need it unbounded to send control messages such as GRAFT, PRUNE, SUBSCRIBE, and UNSUBSCRIBE. Left a comment explaining this, thanks Gui!
protocols/gossipsub/src/behaviour.rs
Outdated
| // before we add all the gossip from this heartbeat in order to gain a true measure of | ||
| // steady-state size of the queues. | ||
| if let Some(m) = &mut self.metrics { | ||
| for sender_queue in self.connected_peers.values_mut().map(|v| &v.sender) { |
| // Slow peer weighting | ||
| if peer_stats.slow_peer_penalty > self.params.slow_peer_threshold { | ||
| let excess = peer_stats.slow_peer_penalty - self.params.slow_peer_threshold; | ||
| score += excess * self.params.slow_peer_weight; | ||
| } |
There was a problem hiding this comment.
right, thanks Akihito. Updated!
protocols/gossipsub/src/types.rs
Outdated
| pub(crate) fn new(cap: usize) -> RpcSender { | ||
| let (priority_sender, priority_receiver) = async_channel::unbounded(); | ||
| let (non_priority_sender, non_priority_receiver) = async_channel::bounded(cap / 2); | ||
| let len = Arc::new(AtomicUsize::new(0)); | ||
| RpcSender { | ||
| cap: cap / 2, |
There was a problem hiding this comment.
cause we have two channels priority and non_priority left a comment elaborating on that, thanks Gui
| if self.len.load(Ordering::Relaxed) >= self.cap { | ||
| return Err(()); |
There was a problem hiding this comment.
because publish calls to self.priority_sender which is unbounded, and we need it unbounded to send control messages such as GRAFT, PRUNE, SUBSCRIBE, and UNSUBSCRIBE. Left a comment explaining this, thanks Gui!
Description
superseeds #4914 with some changes and improvements, namely:
DelayforForwardandPublishmessages, messages that take more than the configured delay to be sent are discarded