Batch discard operations

fuzz/fuzzers/deframer.rs

@@ -20,7 +20,10 @@ fuzz_target!(|data: &[u8]| {
     buf.has_pending();
 
     let mut rl = RecordLayer::new();
-    while let Ok(Some(decrypted)) = dfm.pop(&mut rl, None, &mut buf) {
+    let mut borrowed_buf = buf.borrow();
+    while let Ok(Some(decrypted)) = dfm.pop(&mut rl, None, &mut borrowed_buf) {
         Message::try_from(decrypted.message).ok();
     }
+    let discard = borrowed_buf.pending_discard();
+    buf.discard(discard);
 });

rustls/src/conn.rs

@@ -3,7 +3,7 @@ use crate::enums::{AlertDescription, ContentType};
 use crate::error::{Error, PeerMisbehaved};
 #[cfg(feature = "logging")]
 use crate::log::trace;
-use crate::msgs::deframer::{Deframed, DeframerVecBuffer, MessageDeframer};
+use crate::msgs::deframer::{Deframed, DeframerSliceBuffer, DeframerVecBuffer, MessageDeframer};
 use crate::msgs::handshake::Random;
 use crate::msgs::message::{Message, MessagePayload, PlainMessage};
 use crate::suites::{ExtractedSecrets, PartiallyExtractedSecrets};
@@ -452,11 +452,14 @@ impl<Data> ConnectionCommon<Data> {
     /// This is a shortcut to the `process_new_packets()` -> `process_msg()` ->
     /// `process_handshake_messages()` path, specialized for the first handshake message.
     pub(crate) fn first_handshake_message(&mut self) -> Result<Option<Message>, Error> {
-        match self
+        let mut deframer_buffer = self.deframer_buffer.borrow();
+        let res = self
             .core
-            .deframe(None, &mut self.deframer_buffer)?
-            .map(Message::try_from)
-        {
+            .deframe(None, &mut deframer_buffer);
+        let discard = deframer_buffer.pending_discard();
+        self.deframer_buffer.discard(discard);
+
+        match res?.map(Message::try_from) {
             Some(Ok(msg)) => Ok(Some(msg)),
             Some(Err(err)) => Err(self.send_fatal_alert(AlertDescription::DecodeError, err)),
             None => Ok(None),
@@ -646,16 +649,21 @@ impl<Data> ConnectionCore<Data> {
             }
         };
 
-        while let Some(msg) = self.deframe(Some(&*state), deframer_buffer)? {
+        let mut borrowed_buffer = deframer_buffer.borrow();
+        while let Some(msg) = self.deframe(Some(&*state), &mut borrowed_buffer)? {
             match self.process_msg(msg, state) {
                 Ok(new) => state = new,
                 Err(e) => {
                     self.state = Err(e.clone());
+                    let discard = borrowed_buffer.pending_discard();
+                    deframer_buffer.discard(discard);
                     return Err(e);
                 }
             }
         }
 
+        let discard = borrowed_buffer.pending_discard();
+        deframer_buffer.discard(discard);
         self.state = Ok(state);
         Ok(self.common_state.current_io_state())
     }
@@ -664,7 +672,7 @@ impl<Data> ConnectionCore<Data> {
     fn deframe(
         &mut self,
         state: Option<&dyn State<Data>>,
-        deframer_buffer: &mut DeframerVecBuffer,
+        deframer_buffer: &mut DeframerSliceBuffer,
     ) -> Result<Option<PlainMessage>, Error> {
         match self.message_deframer.pop(
             &mut self.common_state.record_layer,

rustls/src/msgs/deframer.rs

@@ -37,7 +37,7 @@ impl MessageDeframer {
         &mut self,
         record_layer: &mut RecordLayer,
         negotiated_version: Option<ProtocolVersion>,
-        buffer: &mut DeframerVecBuffer,
+        buffer: &mut DeframerSliceBuffer,
     ) -> Result<Option<Deframed>, Error> {
         if let Some(last_err) = self.last_error.clone() {
             return Err(last_err);
@@ -110,7 +110,7 @@ impl MessageDeframer {
             };
             if self.joining_hs.is_none() && allowed_plaintext {
                 // This is unencrypted. We check the contents later.
-                buffer.discard(end);
+                buffer.queue_discard(end);
                 return Ok(Some(Deframed {
                     want_close_before_decrypt: false,
                     aligned: true,
@@ -137,7 +137,7 @@ impl MessageDeframer {
                     ));
                 }
                 Ok(None) => {
-                    buffer.discard(end);
+                    buffer.queue_discard(end);
                     continue;
                 }
                 Err(e) => return Err(e),
@@ -154,7 +154,7 @@ impl MessageDeframer {
             // If it's not a handshake message, just return it -- no joining necessary.
             if msg.typ != ContentType::Handshake {
                 let end = start + rd.used();
-                buffer.discard(end);
+                buffer.queue_discard(end);
                 return Ok(Some(Deframed {
                     want_close_before_decrypt: false,
                     aligned: true,
@@ -196,7 +196,7 @@ impl MessageDeframer {
             // discard all of the bytes that we're previously buffered as handshake data.
             let end = meta.message.end;
             self.joining_hs = None;
-            buffer.discard(end);
+            buffer.queue_discard(end);
         }
 
         Ok(Some(Deframed {
@@ -373,7 +373,7 @@ impl DeframerVecBuffer {
     }
 
     /// Discard `taken` bytes from the start of our buffer.
-    fn discard(&mut self, taken: usize) {
+    pub fn discard(&mut self, taken: usize) {
         #[allow(clippy::comparison_chain)]
         if taken < self.used {
             /* Before:
@@ -855,8 +855,13 @@ mod tests {
             record_layer: &mut RecordLayer,
             negotiated_version: Option<ProtocolVersion>,
         ) -> Result<Option<Deframed>, Error> {
-            self.inner
-                .pop(record_layer, negotiated_version, &mut self.buffer)
+            let mut deframer_buffer = self.buffer.borrow();
+            let res = self
+                .inner
+                .pop(record_layer, negotiated_version, &mut deframer_buffer);
+            let discard = deframer_buffer.pending_discard();
+            self.buffer.discard(discard);
+            res
         }
 
         fn read(&mut self, rd: &mut dyn io::Read) -> io::Result<usize> {