diff --git a/quinn-proto/src/connection/mod.rs b/quinn-proto/src/connection/mod.rs
index 10662a94d..a9bef6453 100644
--- a/quinn-proto/src/connection/mod.rs
+++ b/quinn-proto/src/connection/mod.rs
@@ -889,23 +889,23 @@ where
Some(id)
}
- /// Read from the given recv stream, in undefined order
+ /// Read from the given recv stream
///
- /// While stream data is typically processed by applications in-order, unordered reads improve
- /// performance when packet loss occurs and data cannot be retransmitted before the flow control
- /// window is filled. When in-order delivery is required, the sibling `read()` or `read_chunk[s]()`
- /// methods should be used.
+ /// If `Ok`, the return value contains the bytes and their offset in the stream. For ordered
+ /// reads, the offset could be ignored, since it provides no new information.
///
- /// The return value if `Ok` contains the bytes and their offset in the stream.
- pub fn read_unordered(&mut self, id: StreamId) -> Result, ReadError> {
- let result = self.streams.read_unordered(id);
- self.post_read(id, &result);
- Ok(result?.map(|x| x.result))
- }
-
- /// Read the next ordered chunk from the given recv stream
- pub fn read(&mut self, id: StreamId, max_length: usize) -> Result , ReadError> {
- let result = self.streams.read(id, max_length);
+ /// While most applications will prefer to consume stream data in order, unordered reads can
+ /// improve performance when packet loss occurs and data cannot be retransmitted before the flow
+ /// control window is filled. On any given stream, you can switch from ordered to unordered
+ /// reads, but ordered reads on streams that have seen previous unordered reads will return
+ /// `ReadError::IllegalOrderedRead`.
+ pub fn read(
+ &mut self,
+ id: StreamId,
+ max_length: usize,
+ ordered: bool,
+ ) -> Result , ReadError> {
+ let result = self.streams.read(id, max_length, ordered);
self.post_read(id, &result);
Ok(result?.map(|x| x.result))
}
diff --git a/quinn-proto/src/connection/streams.rs b/quinn-proto/src/connection/streams.rs
index d5d5ec3b9..6d3497c65 100644
--- a/quinn-proto/src/connection/streams.rs
+++ b/quinn-proto/src/connection/streams.rs
@@ -200,12 +200,13 @@ impl Streams {
self.connection_blocked.clear();
}
- pub(crate) fn read_unordered(&mut self, id: StreamId) -> ReadResult<(Bytes, u64)> {
- self.try_read(id, |rs| rs.read_unordered())
- }
-
- pub(crate) fn read(&mut self, id: StreamId, max_length: usize) -> ReadResult {
- self.try_read(id, |rs| rs.read(max_length))
+ pub(crate) fn read(
+ &mut self,
+ id: StreamId,
+ max_length: usize,
+ ordered: bool,
+ ) -> ReadResult<(Bytes, u64)> {
+ self.try_read(id, |rs| rs.read(max_length, ordered))
}
pub(crate) fn read_chunks(
@@ -1021,7 +1022,7 @@ mod tests {
);
assert_eq!(client.data_recvd, 2048);
assert_eq!(client.local_max_data - initial_max, 0);
- client.read(id, 1024).unwrap();
+ client.read(id, 1024, true).unwrap();
assert_eq!(client.local_max_data - initial_max, 1024);
assert_eq!(
client
@@ -1122,8 +1123,11 @@ mod tests {
}
);
assert!(client.stop(id).is_err());
- assert_eq!(client.read(id, 0), Err(ReadError::UnknownStream));
- assert_eq!(client.read_unordered(id), Err(ReadError::UnknownStream));
+ assert_eq!(client.read(id, 0, true), Err(ReadError::UnknownStream));
+ assert_eq!(
+ client.read(id, usize::MAX, false),
+ Err(ReadError::UnknownStream)
+ );
assert_eq!(client.local_max_data - initial_max, 32);
assert_eq!(
client
diff --git a/quinn-proto/src/connection/streams/recv.rs b/quinn-proto/src/connection/streams/recv.rs
index 59dd7fe4e..f400a7d4f 100644
--- a/quinn-proto/src/connection/streams/recv.rs
+++ b/quinn-proto/src/connection/streams/recv.rs
@@ -59,18 +59,13 @@ impl Recv {
Ok(new_bytes)
}
- pub(super) fn read_unordered(&mut self) -> StreamReadResult<(Bytes, u64)> {
- // Return data we already have buffered, regardless of state
- if let Some((offset, bytes)) = self.assembler.read(usize::MAX, false)? {
- Ok(Some((bytes, offset)))
- } else {
- self.read_blocked().map(|()| None)
- }
- }
-
- pub(super) fn read(&mut self, max_length: usize) -> StreamReadResult {
- match self.assembler.read(max_length, true)? {
- Some((_, bytes)) => Ok(Some(bytes)),
+ pub(super) fn read(
+ &mut self,
+ max_length: usize,
+ ordered: bool,
+ ) -> StreamReadResult<(Bytes, u64)> {
+ match self.assembler.read(max_length, ordered)? {
+ Some((offset, bytes)) => Ok(Some((bytes, offset))),
None => self.read_blocked().map(|()| None),
}
}
@@ -252,12 +247,6 @@ pub(crate) trait BytesRead {
fn bytes_read(&self) -> u64;
}
-impl BytesRead for Bytes {
- fn bytes_read(&self) -> u64 {
- self.len() as u64
- }
-}
-
impl BytesRead for (Bytes, u64) {
fn bytes_read(&self) -> u64 {
self.0.len() as u64
diff --git a/quinn-proto/src/tests/mod.rs b/quinn-proto/src/tests/mod.rs
index bb26763e8..89822a211 100644
--- a/quinn-proto/src/tests/mod.rs
+++ b/quinn-proto/src/tests/mod.rs
@@ -239,10 +239,13 @@ fn finish_stream_simple() {
assert_matches!(pair.server_conn_mut(server_ch).accept(Dir::Uni), Some(stream) if stream == s);
assert_matches!(pair.server_conn_mut(server_ch).poll(), None);
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 0))) if data == MSG
);
- assert_matches!(pair.server_conn_mut(server_ch).read_unordered(s), Ok(None));
+ assert_matches!(
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
+ Ok(None)
+ );
}
#[test]
@@ -268,7 +271,7 @@ fn reset_stream() {
);
assert_matches!(pair.server_conn_mut(server_ch).accept(Dir::Uni), Some(stream) if stream == s);
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Err(ReadError::Reset(ERROR))
);
assert_matches!(pair.client_conn_mut(client_ch).poll(), None);
@@ -430,7 +433,7 @@ fn zero_rtt_happypath() {
assert!(pair.client_conn_mut(client_ch).accepted_0rtt());
let server_ch = pair.server.assert_accept();
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 0))) if data == MSG
);
assert_eq!(pair.client_conn_mut(client_ch).lost_packets(), 0);
@@ -501,7 +504,8 @@ fn zero_rtt_rejection() {
let s2 = pair.client_conn_mut(client_ch).open(Dir::Uni).unwrap();
assert_eq!(s, s2);
assert_eq!(
- pair.server_conn_mut(server_conn).read_unordered(s2),
+ pair.server_conn_mut(server_conn)
+ .read(s2, usize::MAX, false),
Err(ReadError::Blocked)
);
assert_eq!(pair.client_conn_mut(client_ch).lost_packets(), 0);
@@ -639,8 +643,11 @@ fn stream_id_limit() {
Some(Event::Stream(StreamEvent::Opened { dir: Dir::Uni }))
);
assert_matches!(pair.server_conn_mut(server_ch).accept(Dir::Uni), Some(stream) if stream == s);
- assert_matches!(pair.server_conn_mut(server_ch).read_unordered(s), Ok(Some((msg, 0))) if msg == MSG);
- assert_eq!(pair.server_conn_mut(server_ch).read_unordered(s), Ok(None));
+ assert_matches!(pair.server_conn_mut(server_ch).read(s, usize::MAX, false), Ok(Some((msg, 0))) if msg == MSG);
+ assert_eq!(
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
+ Ok(None)
+ );
// Server will only send MAX_STREAM_ID now that the application's been notified
pair.drive();
assert_matches!(
@@ -666,7 +673,10 @@ fn stream_id_limit() {
);
assert_matches!(pair.server_conn_mut(server_ch).accept(Dir::Uni), Some(stream) if stream == s);
assert_matches!(pair.server_conn_mut(server_ch).poll(), None);
- assert_matches!(pair.server_conn_mut(server_ch).read_unordered(s), Ok(None));
+ assert_matches!(
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
+ Ok(None)
+ );
}
#[test]
@@ -693,7 +703,7 @@ fn key_update_simple() {
assert_matches!(pair.server_conn_mut(server_ch).accept(Dir::Bi), Some(stream) if stream == s);
assert_matches!(pair.server_conn_mut(server_ch).poll(), None);
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 0))) if data == MSG1
);
@@ -707,7 +717,7 @@ fn key_update_simple() {
assert_matches!(pair.server_conn_mut(server_ch).poll(), Some(Event::Stream(StreamEvent::Readable { id })) if id == s);
assert_matches!(pair.server_conn_mut(server_ch).poll(), None);
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 6))) if data == MSG2
);
@@ -752,16 +762,16 @@ fn key_update_reordered() {
let buf1 = pair
.server_conn_mut(server_ch)
- .read(s, usize::MAX)
+ .read(s, usize::MAX, true)
.unwrap()
.unwrap();
- assert_matches!(&*buf1, MSG1);
+ assert_matches!(&*buf1.0, MSG1);
let buf2 = pair
.server_conn_mut(server_ch)
- .read(s, usize::MAX)
+ .read(s, usize::MAX, true)
.unwrap()
.unwrap();
- assert_eq!(buf2, MSG2);
+ assert_eq!(buf2.0, MSG2);
assert_eq!(pair.client_conn_mut(client_ch).lost_packets(), 0);
assert_eq!(pair.server_conn_mut(server_ch).lost_packets(), 0);
@@ -978,7 +988,7 @@ fn test_flow_control(config: TransportConfig, window_size: usize) {
.unwrap();
pair.drive();
assert_eq!(
- pair.server_conn_mut(server_conn).read(s, usize::MAX),
+ pair.server_conn_mut(server_conn).read(s, usize::MAX, true),
Err(ReadError::Reset(VarInt(42)))
);
@@ -997,8 +1007,8 @@ fn test_flow_control(config: TransportConfig, window_size: usize) {
pair.drive();
let mut cursor = 0;
loop {
- match pair.server_conn_mut(server_conn).read(s, usize::MAX) {
- Ok(Some(buf)) => {
+ match pair.server_conn_mut(server_conn).read(s, usize::MAX, true) {
+ Ok(Some((buf, _))) => {
cursor += buf.len();
}
Ok(None) => {
@@ -1028,8 +1038,8 @@ fn test_flow_control(config: TransportConfig, window_size: usize) {
pair.drive();
let mut cursor = 0;
loop {
- match pair.server_conn_mut(server_conn).read(s, usize::MAX) {
- Ok(Some(buf)) => {
+ match pair.server_conn_mut(server_conn).read(s, usize::MAX, true) {
+ Ok(Some((buf, _))) => {
cursor += buf.len();
}
Ok(None) => {
@@ -1093,7 +1103,7 @@ fn stop_opens_bidi() {
assert_matches!(pair.server_conn_mut(server_conn).accept(Dir::Bi), Some(stream) if stream == s);
assert_eq!(pair.server_conn_mut(client_conn).send_streams(), 1);
assert_matches!(
- pair.server_conn_mut(server_conn).read_unordered(s),
+ pair.server_conn_mut(server_conn).read(s, usize::MAX, false),
Err(ReadError::Blocked)
);
assert_matches!(
@@ -1293,7 +1303,7 @@ fn finish_stream_flow_control_reordered() {
// Issue flow control credit
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 0))) if data == MSG
);
pair.server.drive(pair.time, pair.client.addr);
@@ -1315,7 +1325,10 @@ fn finish_stream_flow_control_reordered() {
Some(Event::Stream(StreamEvent::Opened { dir: Dir::Uni }))
);
assert_matches!(pair.server_conn_mut(server_ch).accept(Dir::Uni), Some(stream) if stream == s);
- assert_matches!(pair.server_conn_mut(server_ch).read_unordered(s), Ok(None));
+ assert_matches!(
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
+ Ok(None)
+ );
}
#[test]
@@ -1345,7 +1358,7 @@ fn handshake_1rtt_handling() {
assert!(pair.client_conn_mut(client_ch).lost_packets() != 0);
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 0))) if data == MSG
);
}
@@ -1591,11 +1604,11 @@ fn finish_acked() {
assert_matches!(pair.server_conn_mut(server_ch).accept(Dir::Uni), Some(stream) if stream == s);
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 0))) if data == MSG
);
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Err(ReadError::Blocked)
);
@@ -1613,7 +1626,10 @@ fn finish_acked() {
pair.client_conn_mut(client_ch).poll(),
Some(Event::Stream(StreamEvent::Finished { id })) if id == s
);
- assert_matches!(pair.server_conn_mut(server_ch).read_unordered(s), Ok(None));
+ assert_matches!(
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
+ Ok(None)
+ );
}
#[test]
@@ -1653,10 +1669,13 @@ fn finish_retransmit() {
assert_matches!(pair.server_conn_mut(server_ch).accept(Dir::Uni), Some(stream) if stream == s);
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 0))) if data == MSG
);
- assert_matches!(pair.server_conn_mut(server_ch).read_unordered(s), Ok(None));
+ assert_matches!(
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
+ Ok(None)
+ );
}
/// Ensures that exchanging data on a client-initiated bidirectional stream works past the initial
@@ -1685,20 +1704,26 @@ fn repeated_request_response() {
assert_eq!(pair.server_conn_mut(server_ch).accept(Dir::Bi), Some(s));
assert_matches!(
- pair.server_conn_mut(server_ch).read_unordered(s),
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 0))) if data == REQUEST
);
- assert_matches!(pair.server_conn_mut(server_ch).read_unordered(s), Ok(None));
+ assert_matches!(
+ pair.server_conn_mut(server_ch).read(s, usize::MAX, false),
+ Ok(None)
+ );
pair.server_conn_mut(server_ch).write(s, RESPONSE).unwrap();
pair.server_conn_mut(server_ch).finish(s).unwrap();
pair.drive();
assert_matches!(
- pair.client_conn_mut(client_ch).read_unordered(s),
+ pair.client_conn_mut(client_ch).read(s, usize::MAX, false),
Ok(Some((ref data, 0))) if data == RESPONSE
);
- assert_matches!(pair.client_conn_mut(client_ch).read_unordered(s), Ok(None));
+ assert_matches!(
+ pair.client_conn_mut(client_ch).read(s, usize::MAX, false),
+ Ok(None)
+ );
}
}
diff --git a/quinn/src/streams.rs b/quinn/src/streams.rs
index c896353f3..b3bab49dc 100644
--- a/quinn/src/streams.rs
+++ b/quinn/src/streams.rs
@@ -354,8 +354,8 @@ where
) -> Poll> {
self.poll_read_generic(cx, |conn, stream| {
conn.inner
- .read(stream, buf.remaining())
- .map(|val| val.map(|chunk| buf.put_slice(&chunk)))
+ .read(stream, buf.remaining(), true)
+ .map(|val| val.map(|(chunk, _)| buf.put_slice(&chunk)))
})
.map(|res| res.map(|_| ()))
}
@@ -375,7 +375,9 @@ where
&mut self,
cx: &mut Context,
) -> Poll, ReadError>> {
- self.poll_read_generic(cx, |conn, stream| conn.inner.read_unordered(stream))
+ self.poll_read_generic(cx, |conn, stream| {
+ conn.inner.read(stream, usize::MAX, false)
+ })
}
/// Read the next segment of data
@@ -398,7 +400,8 @@ where
cx: &mut Context,
max_length: usize,
) -> Poll, ReadError>> {
- self.poll_read_generic(cx, |conn, stream| conn.inner.read(stream, max_length))
+ self.poll_read_generic(cx, |conn, stream| conn.inner.read(stream, max_length, true))
+ .map(|ready| ready.map(|ok| ok.map(|(bytes, _)| bytes)))
}
/// Read the next segments of data