Introduce a queue that allows to respect peer window size #817
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,5 @@ | ||
| use std::collections::VecDeque; | ||
|
|
||
| use openmina_core::{bug_condition, fuzz_maybe, fuzzed_maybe, Substate, SubstateAccess}; | ||
|
|
||
| use crate::P2pLimits; | ||
|
|
@@ -155,14 +157,14 @@ impl P2pNetworkYamuxState { | |
| data, | ||
| mut flags, | ||
| } => { | ||
| let stream_state = yamux_state | ||
| .streams | ||
| .get(&stream_id) | ||
| .ok_or_else(|| format!("Stream with id {stream_id} not found for `P2pNetworkYamuxAction::OutgoingData`"))?; | ||
|
|
||
| if !stream_state.incoming && !stream_state.established && !stream_state.syn_sent { | ||
| flags.insert(YamuxFlags::SYN); | ||
| } else if stream_state.incoming && !stream_state.established { | ||
| flags.insert(YamuxFlags::ACK); | ||
| } | ||
|
|
||
|
|
@@ -180,6 +182,7 @@ impl P2pNetworkYamuxState { | |
| Ok(()) | ||
| } | ||
| P2pNetworkYamuxAction::IncomingFrame { addr, frame } => { | ||
| let mut pending_outgoing = VecDeque::default(); | ||
| if let Some(frame) = yamux_state.incoming.pop_front() { | ||
| if frame.flags.contains(YamuxFlags::SYN) { | ||
| yamux_state | ||
|
|
@@ -211,11 +214,32 @@ impl P2pNetworkYamuxState { | |
| } | ||
| } | ||
| YamuxFrameInner::WindowUpdate { difference } => { | ||
| let stream = yamux_state | ||
| .streams | ||
| .entry(frame.stream_id) | ||
| .or_insert_with(YamuxStreamState::incoming); | ||
| stream.update_window(false, difference); | ||
| if difference > 0 { | ||
| // have some fresh space in the window | ||
| // try send as many frames as can | ||
| let mut window = stream.window_theirs; | ||
| while let Some(mut frame) = stream.pending.pop_front() { | ||
| let len = frame.len() as u32; | ||
| if let Some(new_window) = window.checked_sub(len) { | ||
| pending_outgoing.push_back(frame); | ||
| window = new_window; | ||
| } else { | ||
| if let Some(remaining) = | ||
| frame.split_at((len - window) as usize) | ||
| { | ||
| stream.pending.push_front(remaining); | ||
| } | ||
| pending_outgoing.push_back(frame); | ||
|
|
||
| break; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| YamuxFrameInner::Ping { .. } => {} | ||
| YamuxFrameInner::GoAway(res) => yamux_state.set_res(res), | ||
|
|
@@ -282,15 +306,17 @@ impl P2pNetworkYamuxState { | |
| } | ||
| match &frame.inner { | ||
| YamuxFrameInner::Data(data) => { | ||
| // here we are very permissive | ||
| // always when our window is smaller 64 kb, just increase it by 256 kb | ||
| // if we need fine grained back pressure, it should be implemented here | ||
| if stream.window_ours < 64 * 1024 { | ||
| let difference = 256 * 1024; | ||
| dispatcher.push(P2pNetworkYamuxAction::OutgoingFrame { | ||
| addr, | ||
| frame: YamuxFrame { | ||
| stream_id: frame.stream_id, | ||
| flags: YamuxFlags::empty(), | ||
| inner: YamuxFrameInner::WindowUpdate { difference }, | ||
| }, | ||
| }); | ||
| } | ||
|
|
@@ -318,21 +344,61 @@ impl P2pNetworkYamuxState { | |
| }); | ||
| } | ||
| } | ||
| YamuxFrameInner::WindowUpdate { difference } => { | ||
| if *difference < 0 { | ||
|
Collaborator
There was a problem hiding this comment. @vlad9486 what about
Contributor
Author
There was a problem hiding this comment. @tizoc Yes, initially I wrote |
||
| let error = | ||
| P2pNetworkConnectionError::YamuxBadWindowUpdate(frame.stream_id); | ||
| dispatcher.push(P2pNetworkSchedulerAction::Error { addr, error }); | ||
| } else { | ||
| while let Some(frame) = pending_outgoing.pop_front() { | ||
| dispatcher | ||
| .push(P2pNetworkYamuxAction::OutgoingFrame { addr, frame }); | ||
| } | ||
| } | ||
| } | ||
| _ => {} | ||
| } | ||
|
|
||
| Ok(()) | ||
| } | ||
| P2pNetworkYamuxAction::OutgoingFrame { mut frame, addr } => { | ||
| let stream_id = frame.stream_id; | ||
| let Some(stream) = yamux_state.streams.get_mut(&stream_id) else { | ||
| return Ok(()); | ||
| }; | ||
| match &mut frame.inner { | ||
| YamuxFrameInner::Data(data) => { | ||
| if let Some(new_window) = | ||
| stream.window_theirs.checked_sub(data.len() as u32) | ||
| { | ||
| // their window is big enough, decrease the size | ||
| // and send the whole frame | ||
| stream.window_theirs = new_window; | ||
| } else if stream.window_theirs != 0 && stream.pending.is_empty() { | ||
| // their window is not big enough, but has some space, | ||
| // and the queue is empty, | ||
| // do not send the whole frame, | ||
| // split it and put remaining in the queue, | ||
| if let Some(remaining) = frame.split_at(stream.window_theirs as usize) { | ||
| stream.pending.push_back(remaining); | ||
| } | ||
| // the window will be zero after sending | ||
| stream.window_theirs = 0; | ||
| } else { | ||
| // either the window cannot accept any byte, | ||
| // or the queue is already not empty | ||
| // in both cases the whole frame goes in the queue and nothing to send | ||
| stream.pending.push_back(frame); | ||
|
Contributor
Author
There was a problem hiding this comment. Need to add a limit here. Malicious peer can keep its window small and send an RPC request in a loop. |
||
| if stream.pending.iter().map(YamuxFrame::len).sum::<usize>() | ||
| > yamux_state.pending_outgoing_limit | ||
| { | ||
| let dispatcher = state_context.into_dispatcher(); | ||
| let error = P2pNetworkConnectionError::YamuxOverflow(stream_id); | ||
| dispatcher.push(P2pNetworkSchedulerAction::Error { addr, error }); | ||
| } | ||
|
|
||
| return Ok(()); | ||
| } | ||
| } | ||
| YamuxFrameInner::WindowUpdate { difference } => { | ||
| stream.update_window(true, *difference); | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this can be 0 or negative? in which cases does that happen?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
In my implementation it is signed, but in go implementation it is unsigned, so the window could only grow. So, this is just sanity check, maybe redundant. Peer cannot harm us by making its own window too big.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ok, please add a
bug_condition!call in the else branch so that if this ever happens for some reason we will know about it. That is something we want to do in general, to avoid such unexpected (unenforced) cases happening silently without us noticingThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I add a runtime error, not bug_condition, because peer can send us this and we should not crash.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You are right, this is based on external input, so it is not exactly a
bug_condition!. Btwbug_conditiononly panics if a flag is enabled in the environment, otherwise it just logs an error (but again this specific case is not for it)