quinn-proto: unify API for ordered and unordered reads

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<Option<(Bytes, u64)>, 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<Option<Bytes>, 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<Option<(Bytes, u64)>, ReadError> {
+        let result = self.streams.read(id, max_length, ordered);
         self.post_read(id, &result);
         Ok(result?.map(|x| x.result))
     }

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<Bytes> {
-        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

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<Bytes> {
-        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

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)
+        );
     }
 }
 

quinn/src/streams.rs

@@ -354,8 +354,8 @@ where
     ) -> Poll<Result<(), ReadError>> {
         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<Result<Option<(Bytes, u64)>, 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<Result<Option<Bytes>, 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