From ca692c233dd64560ea10a11f9fc9d1f8b4e93dc2 Mon Sep 17 00:00:00 2001 From: Dirkjan Ochtman Date: Tue, 26 Jan 2021 10:16:23 +0100 Subject: [PATCH] quinn-proto: unify API for ordered and unordered reads --- quinn-proto/src/connection/mod.rs | 30 ++++---- quinn-proto/src/connection/streams.rs | 22 +++--- quinn-proto/src/connection/streams/recv.rs | 25 ++---- quinn-proto/src/tests/mod.rs | 89 ++++++++++++++-------- quinn/src/streams.rs | 11 ++- 5 files changed, 99 insertions(+), 78 deletions(-) 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