feat: don't split header and body across TCP packets

yamux/Cargo.toml

@@ -17,6 +17,7 @@ parking_lot = "0.12"
 rand = "0.8.3"
 static_assertions = "1"
 pin-project = "1.1.0"
+zerocopy = { version = "0.7.0", features = ["derive"] }
 
 [dev-dependencies]
 quickcheck = "1.0"

yamux/src/chunks.rs

@@ -36,12 +36,15 @@ impl Chunks {
     }
 
     /// Add another chunk of bytes to the end.
-    pub(crate) fn push(&mut self, x: Vec<u8>) {
-        self.len += x.len();
-        if !x.is_empty() {
-            self.seq.push_back(Chunk {
-                cursor: io::Cursor::new(x),
-            })
+    pub(crate) fn push(&mut self, x: Vec<u8>, offset: usize) {
+        let x_len = x.len();
+        let cursor = io::Cursor::new(x);
+        let mut chunk = Chunk { cursor };
+        chunk.advance(offset);
+        if !chunk.is_empty() {
+            assert_eq!(chunk.len(), x_len - offset);
+            self.len += chunk.len() + offset;
+            self.seq.push_back(chunk);
         }
     }
 

yamux/src/connection/cleanup.rs

@@ -33,7 +33,7 @@ impl Future for Cleanup {
     type Output = ConnectionError;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.get_mut();
+        let this = self.get_mut();
 
         loop {
             match this.state {

yamux/src/connection/closing.rs

@@ -45,7 +45,7 @@ where
     type Output = Result<()>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.get_mut();
+        let this = self.get_mut();
 
         loop {
             match this.state {

yamux/src/connection/stream.rs

@@ -8,12 +8,12 @@
 // at https://www.apache.org/licenses/LICENSE-2.0 and a copy of the MIT license
 // at https://opensource.org/licenses/MIT.
 
-use crate::frame::header::ACK;
+use crate::frame::header::{Flags, ACK, HEADER_SIZE};
 use crate::{
     chunks::Chunks,
     connection::{self, StreamCommand},
     frame::{
-        header::{Data, Header, StreamId, WindowUpdate},
+        header::{Data, StreamId, WindowUpdate},
         Frame,
     },
     Config, WindowUpdateMode, DEFAULT_CREDIT,
@@ -183,18 +183,21 @@ impl Stream {
     }
 
     /// Set ACK or SYN flag if necessary.
-    fn add_flag(&mut self, header: &mut Header<Either<Data, WindowUpdate>>) {
+    fn add_flag(&mut self, frame: &mut Frame<Either<Data, WindowUpdate>>) -> Flags {
+        let mut parsed_frame = frame.parse_mut().expect("valid frame");
+        let header = parsed_frame.header_mut();
         match self.flag {
-            Flag::None => (),
+            Flag::None => {}
             Flag::Syn => {
                 header.syn();
-                self.flag = Flag::None
+                self.flag = Flag::None;
             }
             Flag::Ack => {
                 header.ack();
-                self.flag = Flag::None
+                self.flag = Flag::None;
             }
         }
+        header.flags()
     }
 
     /// Send new credit to the sending side via a window update message if
@@ -218,7 +221,7 @@ impl Stream {
             drop(shared);
 
             let mut frame = Frame::window_update(self.id, credit).right();
-            self.add_flag(frame.header_mut());
+            self.add_flag(&mut frame);
             let cmd = StreamCommand::SendFrame(frame);
             self.sender
                 .start_send(cmd)
@@ -257,7 +260,8 @@ impl futures::stream::Stream for Stream {
         let mut shared = self.shared();
 
         if let Some(bytes) = shared.buffer.pop() {
-            let off = bytes.offset();
+            // Every chunk starts with a frame header, so we add HEADER_SIZE to offset.
+            let off = bytes.offset() + HEADER_SIZE;
             let mut vec = bytes.into_vec();
             if off != 0 {
                 // This should generally not happen when the stream is used only as
@@ -269,7 +273,7 @@ impl futures::stream::Stream for Stream {
                     self.conn,
                     self.id
                 );
-                vec = vec.split_off(off)
+                vec = vec.split_off(off);
             }
             return Poll::Ready(Some(Ok(Packet(vec))));
         }
@@ -367,18 +371,18 @@ impl AsyncWrite for Stream {
             let k = std::cmp::min(shared.credit as usize, buf.len());
             let k = std::cmp::min(k, self.config.split_send_size);
             shared.credit = shared.credit.saturating_sub(k as u32);
-            Vec::from(&buf[..k])
+            &buf[..k]
         };
         let n = body.len();
         let mut frame = Frame::data(self.id, body).expect("body <= u32::MAX").left();
-        self.add_flag(frame.header_mut());
+        let flags = self.add_flag(&mut frame);
         log::trace!("{}/{}: write {} bytes", self.conn, self.id, n);
 
         // technically, the frame hasn't been sent yet on the wire but from the perspective of this data structure, we've queued the frame for sending
         // We are tracking this information:
         // a) to be consistent with outbound streams
         // b) to correctly test our behaviour around timing of when ACKs are sent. See `ack_timing.rs` test.
-        if frame.header().flags().contains(ACK) {
+        if flags.contains(ACK) {
             self.shared()
                 .update_state(self.conn, self.id, State::Open { acknowledged: true });
         }