Skip to content

proto: Remove write_source (pull in IndexMap to avoid borrowing issues) #2260

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

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
+125 −378
Open

proto: Remove write_source (pull in IndexMap to avoid borrowing issues) #2260

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
729617b
proto: Internally use SendStream::new
gretchenfrage Jun 3, 2025
d676d4d
proto: Change StreamsState.send to IndexMap
gretchenfrage Jun 3, 2025
5d78e29
proto: Preemptive SendStream hashmap lookup
gretchenfrage Jun 3, 2025
df184d1
proto: Remove write_source
gretchenfrage Jun 3, 2025
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
7 changes: 4 additions & 3 deletions fuzz/fuzz_targets/streams.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,8 @@ fuzz_target!(|input: (StreamParams, Vec<Operation>)| {
Streams::new(&mut state, &conn_state).accept(dir);
}
Operation::Finish(id) => {
let _ = SendStream::new(id, &mut state, &mut pending, &conn_state).finish();
let _ =
SendStream::new_for_fuzzing(id, &mut state, &mut pending, &conn_state).finish();
}
Operation::ReceivedStopSending(sid, err_code) => {
Streams::new(&mut state, &conn_state)
Expand All @@ -63,8 +64,8 @@ fuzz_target!(|input: (StreamParams, Vec<Operation>)| {
.received_reset(rs);
}
Operation::Reset(id) => {
let _ =
SendStream::new(id, &mut state, &mut pending, &conn_state).reset(0u32.into());
let _ = SendStream::new_for_fuzzing(id, &mut state, &mut pending, &conn_state)
.reset(0u32.into());
}
}
}
Expand Down
10 changes: 5 additions & 5 deletions quinn-proto/src/connection/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -426,12 +426,12 @@ impl Connection {
#[must_use]
pub fn send_stream(&mut self, id: StreamId) -> SendStream<'_> {
assert!(id.dir() == Dir::Bi || id.initiator() == self.side.side());
SendStream {
SendStream::new(
id,
state: &mut self.streams,
pending: &mut self.spaces[SpaceId::Data].pending,
conn_state: &self.state,
}
&mut self.streams,
&mut self.spaces[SpaceId::Data].pending,
&self.state,
)
}

/// Returns packets to transmit
Expand Down
11 changes: 10 additions & 1 deletion quinn-proto/src/connection/streams/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,7 +203,16 @@ pub struct SendStream<'a> {
#[allow(clippy::needless_lifetimes)] // Needed for cfg(fuzzing)
impl<'a> SendStream<'a> {
#[cfg(fuzzing)]
pub fn new(
pub fn new_for_fuzzing(
Copy link
Collaborator

Choose a reason for hiding this comment

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

bikeshed: I'd prefer that this always be named new. We could achieve that by having two definitions, selected by #[cfg(fuzzing)] state and differing only in visibility, both of which delegate to a private new_inner or similar.

id: StreamId,
state: &'a mut StreamsState,
pending: &'a mut Retransmits,
conn_state: &'a super::State,
) -> Self {
Self::new(id, state, pending, conn_state)
}

pub(super) fn new(
id: StreamId,
state: &'a mut StreamsState,
pending: &'a mut Retransmits,
Expand Down
98 changes: 14 additions & 84 deletions quinn-proto/src/connection/streams/state.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1290,12 +1290,7 @@ mod tests {
.open(Dir::Uni)
.unwrap();

let mut stream = SendStream {
id,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut stream = SendStream::new(id, &mut server, &mut pending, &state);

let error_code = 0u32.into();
stream.state.received_stop_sending(id, error_code);
Expand Down Expand Up @@ -1353,29 +1348,14 @@ mod tests {
let id_mid = streams.open(Dir::Bi).unwrap();
let id_low = streams.open(Dir::Bi).unwrap();

let mut mid = SendStream {
id: id_mid,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut mid = SendStream::new(id_mid, &mut server, &mut pending, &state);
mid.write(b"mid").unwrap();

let mut low = SendStream {
id: id_low,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut low = SendStream::new(id_low, &mut server, &mut pending, &state);
low.set_priority(-1).unwrap();
low.write(b"low").unwrap();

let mut high = SendStream {
id: id_high,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut high = SendStream::new(id_high, &mut server, &mut pending, &state);
high.set_priority(1).unwrap();
high.write(b"high").unwrap();

Expand Down Expand Up @@ -1408,34 +1388,19 @@ mod tests {
let id_high = streams.open(Dir::Bi).unwrap();
let id_mid = streams.open(Dir::Bi).unwrap();

let mut mid = SendStream {
id: id_mid,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut mid = SendStream::new(id_mid, &mut server, &mut pending, &state);
assert_eq!(mid.write(b"mid").unwrap(), 3);
assert_eq!(server.pending.len(), 1);

let mut high = SendStream {
id: id_high,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut high = SendStream::new(id_high, &mut server, &mut pending, &state);
high.set_priority(1).unwrap();
assert_eq!(high.write(&[0; 200]).unwrap(), 200);
assert_eq!(server.pending.len(), 2);

// Requeue the high priority stream to lowest priority. The initial send
// still uses high priority since it's queued that way. After that it will
// switch to low priority
let mut high = SendStream {
id: id_high,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut high = SendStream::new(id_high, &mut server, &mut pending, &state);
high.set_priority(-1).unwrap();

let mut buf = Vec::with_capacity(1000);
Expand Down Expand Up @@ -1478,28 +1443,13 @@ mod tests {
let id_b = streams.open(Dir::Bi).unwrap();
let id_c = streams.open(Dir::Bi).unwrap();

let mut stream_a = SendStream {
id: id_a,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut stream_a = SendStream::new(id_a, &mut server, &mut pending, &state);
stream_a.write(&[b'a'; 100]).unwrap();

let mut stream_b = SendStream {
id: id_b,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut stream_b = SendStream::new(id_b, &mut server, &mut pending, &state);
stream_b.write(&[b'b'; 100]).unwrap();

let mut stream_c = SendStream {
id: id_c,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut stream_c = SendStream::new(id_c, &mut server, &mut pending, &state);
stream_c.write(&[b'c'; 100]).unwrap();

let mut metas = vec![];
Expand Down Expand Up @@ -1558,20 +1508,10 @@ mod tests {
let id_b = streams.open(Dir::Bi).unwrap();
let id_c = streams.open(Dir::Bi).unwrap();

let mut stream_a = SendStream {
id: id_a,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut stream_a = SendStream::new(id_a, &mut server, &mut pending, &state);
stream_a.write(&[b'a'; 100]).unwrap();

let mut stream_b = SendStream {
id: id_b,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut stream_b = SendStream::new(id_b, &mut server, &mut pending, &state);
stream_b.write(&[b'b'; 100]).unwrap();

let mut metas = vec![];
Expand All @@ -1584,12 +1524,7 @@ mod tests {
metas.extend(meta);

// Queue stream_c which has higher priority
let mut stream_c = SendStream {
id: id_c,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut stream_c = SendStream::new(id_c, &mut server, &mut pending, &state);
stream_c.set_priority(1).unwrap();
stream_c.write(&[b'b'; 100]).unwrap();

Expand Down Expand Up @@ -1658,12 +1593,7 @@ mod tests {
};

let id = streams.open(Dir::Uni).unwrap();
let mut stream = SendStream {
id,
state: &mut server,
pending: &mut pending,
conn_state: &state,
};
let mut stream = SendStream::new(id, &mut server, &mut pending, &state);
stream.write(b"hello").unwrap();
stream.reset(0u32.into()).unwrap();

Expand Down