pr feedback

Author: camshaft

quic/s2n-quic-core/src/buffer/duplex.rs

@@ -4,16 +4,17 @@
 use super::{Error, Reader, Writer};
 use crate::varint::VarInt;
 
-mod split;
+mod interposer;
 
-pub use split::Split;
+pub use interposer::Interposer;
 
 /// A buffer that is capable of both reading and writing
 pub trait Duplex: Reader + Writer {}
 
 impl<T: Reader + Writer> Duplex for T {}
 
-/// A buffer that is capable of both skipping a write and read with a given amount.
+/// A buffer which can be advanced forward without reading or writing payloads. This
+/// is essentially a forward-only [`std::io::Seek`].
 ///
 /// This can be used for scenarios where the buffer was written somewhere else but still needed to
 /// be tracked.

quic/s2n-quic-core/src/buffer/duplex/split.rs quic/s2n-quic-core/src/buffer/duplex/interposer.rs

@@ -12,9 +12,10 @@ use crate::{
 };
 use core::convert::Infallible;
 
-/// A split duplex that tries to write as much as possible to `storage`, while falling back to
-/// `duplex`.
-pub struct Split<'a, S, D>
+/// A wrapper around an underlying buffer (`duplex`) which will prefer to read/write from a
+/// user-provided temporary buffer (`storage`). The underlying buffer (`duplex`)'s current
+/// position and total length are updated if needed.
+pub struct Interposer<'a, S, D>
 where
     S: writer::Storage + ?Sized,
     D: duplex::Skip<Error = Infallible> + ?Sized,
@@ -23,19 +24,24 @@ where
     duplex: &'a mut D,
 }
 
-impl<'a, S, D> Split<'a, S, D>
+impl<'a, S, D> Interposer<'a, S, D>
 where
     S: writer::Storage + ?Sized,
     D: duplex::Skip<Error = Infallible> + ?Sized,
 {
     #[inline]
     pub fn new(storage: &'a mut S, duplex: &'a mut D) -> Self {
+        debug_assert!(
+            !storage.has_remaining_capacity() || duplex.buffer_is_empty(),
+            "`duplex` should be drained into `storage` before constructing an Interposer"
+        );
+
         Self { storage, duplex }
     }
 }
 
 /// Delegates to the inner Duplex
-impl<'a, S, D> reader::Storage for Split<'a, S, D>
+impl<'a, S, D> reader::Storage for Interposer<'a, S, D>
 where
     S: writer::Storage + ?Sized,
     D: duplex::Skip<Error = Infallible> + ?Sized,
@@ -78,7 +84,7 @@ where
 }
 
 /// Delegates to the inner Duplex
-impl<'a, C, D> Reader for Split<'a, C, D>
+impl<'a, C, D> Reader for Interposer<'a, C, D>
 where
     C: writer::Storage + ?Sized,
     D: duplex::Skip<Error = Infallible> + ?Sized,
@@ -104,7 +110,7 @@ where
     }
 }
 
-impl<'a, C, D> Writer for Split<'a, C, D>
+impl<'a, C, D> Writer for Interposer<'a, C, D>
 where
     C: writer::Storage + ?Sized,
     D: duplex::Skip<Error = Infallible> + ?Sized,
@@ -121,7 +127,7 @@ where
             // receive buffer, since that's what it stores
             let mut should_delegate = C::SPECIALIZES_BYTES || C::SPECIALIZES_BYTES_MUT;
 
-            // if the storage is empty then write into the duplex
+            // if the storage has no space left then write into the duplex
             should_delegate |= !self.storage.has_remaining_capacity();
 
             // if this packet is non-contiguous, then delegate to the wrapped writer
@@ -192,9 +198,9 @@ mod tests {
             // limit the storage capacity so we force writing into the duplex
             let mut storage = storage.with_write_limit(1);
 
-            let mut split = Split::new(&mut storage, &mut duplex);
+            let mut interposer = Interposer::new(&mut storage, &mut duplex);
 
-            split.read_from(&mut reader).unwrap();
+            interposer.read_from(&mut reader).unwrap();
         }
 
         // the storage was too small so we delegated to duplex
@@ -220,15 +226,15 @@ mod tests {
 
             let mut storage: Vec<u8> = vec![];
 
-            let mut split = Split::new(&mut storage, &mut duplex);
+            let mut interposer = Interposer::new(&mut storage, &mut duplex);
 
-            split.read_from(&mut reader).unwrap();
+            interposer.read_from(&mut reader).unwrap();
 
             // make sure we consumed the reader
             assert_eq!(reader.current_offset().as_u64(), 10);
 
-            assert_eq!(split.current_offset().as_u64(), 0);
-            assert_eq!(split.buffered_len(), 0);
+            assert_eq!(interposer.current_offset().as_u64(), 0);
+            assert_eq!(interposer.buffered_len(), 0);
 
             // make sure we didn't write to the storage, even if we had capacity, since the
             // current_offset doesn't match
@@ -241,15 +247,15 @@ mod tests {
 
             let mut storage: Vec<u8> = vec![];
 
-            let mut split = Split::new(&mut storage, &mut duplex);
+            let mut interposer = Interposer::new(&mut storage, &mut duplex);
 
-            split.read_from(&mut reader).unwrap();
+            interposer.read_from(&mut reader).unwrap();
 
             // make sure we consumed the reader
             assert_eq!(reader.current_offset().as_u64(), 10);
 
-            assert_eq!(split.current_offset().as_u64(), 10);
-            assert_eq!(split.buffered_len(), 0);
+            assert_eq!(interposer.current_offset().as_u64(), 10);
+            assert_eq!(interposer.buffered_len(), 0);
 
             // make sure we copied the entire reader
             assert_eq!(storage.len(), 10);
@@ -265,9 +271,9 @@ mod tests {
 
         let mut storage: Vec<u8> = vec![];
 
-        let mut split = Split::new(&mut storage, &mut duplex);
+        let mut interposer = Interposer::new(&mut storage, &mut duplex);
 
-        split.read_from(&mut reader).unwrap();
+        interposer.read_from(&mut reader).unwrap();
 
         assert_eq!(storage.len(), 10);
         assert_eq!(duplex.current_offset().as_u64(), 10);
@@ -283,19 +289,19 @@ mod tests {
 
         let mut storage = writer::storage::Empty;
 
-        let mut split = Split::new(&mut storage, &mut duplex);
+        let mut interposer = Interposer::new(&mut storage, &mut duplex);
 
-        split.read_from(&mut reader).unwrap();
+        interposer.read_from(&mut reader).unwrap();
 
-        assert_eq!(split.current_offset().as_u64(), 0);
-        assert_eq!(split.buffered_len(), 10);
+        assert_eq!(interposer.current_offset().as_u64(), 0);
+        assert_eq!(interposer.buffered_len(), 10);
 
-        checker.read_from(&mut split).unwrap();
+        checker.read_from(&mut interposer).unwrap();
 
-        assert_eq!(split.current_offset().as_u64(), 10);
-        assert!(split.buffer_is_empty());
-        assert_eq!(split.buffered_len(), 0);
-        assert!(split.is_consumed());
+        assert_eq!(interposer.current_offset().as_u64(), 10);
+        assert!(interposer.buffer_is_empty());
+        assert_eq!(interposer.buffered_len(), 0);
+        assert!(interposer.is_consumed());
     }
 
     #[test]
@@ -308,9 +314,9 @@ mod tests {
         {
             let mut storage = storage.with_write_limit(9);
 
-            let mut split = Split::new(&mut storage, &mut duplex);
+            let mut interposer = Interposer::new(&mut storage, &mut duplex);
 
-            split.read_from(&mut reader).unwrap();
+            interposer.read_from(&mut reader).unwrap();
         }
 
         // the storage was at least half the reader

quic/s2n-quic-core/src/buffer/reader/limit.rs

@@ -13,19 +13,16 @@ use crate::{
 ///
 /// This can be used for applying back pressure to the reader with flow control.
 pub struct Limit<'a, R: Reader + ?Sized> {
-    buffered_len: usize,
+    len: usize,
     reader: &'a mut R,
 }
 
 impl<'a, R: Reader + ?Sized> Limit<'a, R> {
     #[inline]
     pub fn new(reader: &'a mut R, max_buffered_len: usize) -> Self {
-        let buffered_len = max_buffered_len.min(reader.buffered_len());
+        let len = max_buffered_len.min(reader.buffered_len());
 
-        Self {
-            buffered_len,
-            reader,
-        }
+        Self { len, reader }
     }
 }
 
@@ -34,17 +31,17 @@ impl<'a, R: Reader + ?Sized> Storage for Limit<'a, R> {
 
     #[inline]
     fn buffered_len(&self) -> usize {
-        self.buffered_len
+        self.len
     }
 
     #[inline]
     fn read_chunk(&mut self, watermark: usize) -> Result<Chunk, Self::Error> {
-        let watermark = self.buffered_len.min(watermark);
+        let watermark = self.len.min(watermark);
         let chunk = self.reader.read_chunk(watermark)?;
         unsafe {
-            assume!(chunk.len() <= self.buffered_len);
+            assume!(chunk.len() <= self.len);
         }
-        self.buffered_len -= chunk.len();
+        self.len -= chunk.len();
         Ok(chunk)
     }
 
@@ -53,14 +50,14 @@ impl<'a, R: Reader + ?Sized> Storage for Limit<'a, R> {
     where
         Dest: writer::Storage + ?Sized,
     {
-        let mut dest = dest.with_write_limit(self.buffered_len);
+        let mut dest = dest.with_write_limit(self.len);
         let mut dest = dest.track_write();
         let chunk = self.reader.partial_copy_into(&mut dest)?;
         let len = dest.written_len() + chunk.len();
         unsafe {
-            assume!(len <= self.buffered_len);
+            assume!(len <= self.len);
         }
-        self.buffered_len -= len;
+        self.len -= len;
         Ok(chunk)
     }
 
@@ -69,14 +66,14 @@ impl<'a, R: Reader + ?Sized> Storage for Limit<'a, R> {
     where
         Dest: writer::Storage + ?Sized,
     {
-        let mut dest = dest.with_write_limit(self.buffered_len);
+        let mut dest = dest.with_write_limit(self.len);
         let mut dest = dest.track_write();
         self.reader.copy_into(&mut dest)?;
         let len = dest.written_len();
         unsafe {
-            assume!(len <= self.buffered_len);
+            assume!(len <= self.len);
         }
-        self.buffered_len -= len;
+        self.len -= len;
         Ok(())
     }
 }

quic/s2n-quic-core/src/buffer/reader/storage/io_slice.rs

@@ -45,12 +45,10 @@ where
             };
         }
 
-        // skip over any empty slices
-        let buf = &buf[first_non_empty..=last_non_empty];
-
-        unsafe {
-            assume!(!buf.is_empty());
-        }
+        let buf = unsafe {
+            // Safety: we checked above that this range is at least 1 element and is in-bounds
+            buf.get_unchecked(first_non_empty..=last_non_empty)
+        };
 
         let mut slice = Self {
             len,
@@ -115,25 +113,22 @@ where
             // head can be consumed and the watermark still has capacity
             Ordering::Less => {
                 let head = core::mem::take(&mut self.head);
-                unsafe {
-                    assume!(!self.buf.is_empty());
-                }
                 self.advance_buf();
                 self.sub_len(head.len());
                 ControlFlow::Continue(head.into())
             }
             // head can be consumed and the watermark is filled
             Ordering::Equal => {
                 let head = core::mem::take(&mut self.head);
-                unsafe {
-                    assume!(!self.buf.is_empty());
-                }
                 self.advance_buf();
                 self.sub_len(head.len());
                 ControlFlow::Break(head.into())
             }
             // head is partially consumed and the watermark is filled
             Ordering::Greater => {
+                unsafe {
+                    assume!(self.head.len() >= watermark);
+                }
                 let (head, tail) = self.head.split_at(watermark);
                 self.head = tail;
                 self.sub_len(head.len());