perf: use GSO

Author: mxinden

neqo-bin/src/client/mod.rs

@@ -413,7 +413,8 @@ impl<'a, H: Handler> Runner<'a, H> {
             args,
             timeout: None,
             recv_buf: vec![0; neqo_udp::RECV_BUF_SIZE],
-            send_buf: Vec::new(),
+            // TODO
+            send_buf: Vec::with_capacity(neqo_udp::RECV_BUF_SIZE),
         }
     }
 

neqo-bin/src/lib.rs

@@ -146,7 +146,7 @@ impl Default for QuicParameters {
             max_streams_uni: 16,
             idle_timeout: 30,
             congestion_control: CongestionControlAlgorithm::NewReno,
-            no_pacing: false,
+            no_pacing: true,
             no_pmtud: false,
             preferred_address_v4: None,
             preferred_address_v6: None,

neqo-bin/src/server/mod.rs

@@ -227,7 +227,8 @@ impl ServerRunner {
             timeout: None,
             sockets,
             recv_buf: vec![0; neqo_udp::RECV_BUF_SIZE],
-            send_buf: vec![],
+            // TODO
+            send_buf: Vec::with_capacity(neqo_udp::RECV_BUF_SIZE),
         }
     }
 

neqo-common/src/codec.rs

@@ -196,12 +196,16 @@ impl<'a, 'b> PartialEq<Decoder<'b>> for Decoder<'a> {
 /// Encoder is good for building data structures.
 #[derive(PartialEq, Eq)]
 pub struct Encoder<'a> {
+    start: usize,
     buf: &'a mut Vec<u8>,
 }
 
 impl<'a> Encoder<'a> {
     pub fn new(buf: &'a mut Vec<u8>) -> Self {
-        Self { buf }
+        Self {
+            start: buf.len(),
+            buf,
+        }
     }
 
     /// Static helper function for previewing the results of encoding without doing it.
@@ -234,20 +238,20 @@ impl<'a> Encoder<'a> {
     /// been written to the buffer.
     #[must_use]
     pub fn len(&self) -> usize {
-        self.buf.len()
+        self.buf.len() - self.start
     }
 
     /// Returns true if the encoder buffer contains no elements.
     #[must_use]
     pub fn is_empty(&self) -> bool {
-        self.buf.is_empty()
+        self.start == self.buf.len()
     }
 
     /// Create a view of the current contents of the buffer.
     /// Note: for a view of a slice, use `Decoder::new(&enc[s..e])`
     #[must_use]
     pub fn as_decoder(&self) -> Decoder {
-        Decoder::new(self.buf)
+        Decoder::new(&self.buf[self.start..])
     }
 
     /// Don't use this except in testing.
@@ -272,7 +276,7 @@ impl<'a> Encoder<'a> {
 
     #[cfg(test)]
     fn to_hex(&self) -> String {
-        crate::hex(&self.buf)
+        crate::hex(&self.buf[self.start..])
     }
 
     /// Generic encode routine for arbitrary data.
@@ -402,19 +406,19 @@ impl<'a> Encoder<'a> {
 
     /// Truncate the encoder to the given size.
     pub fn truncate(&mut self, len: usize) {
-        self.buf.truncate(len);
+        self.buf.truncate(len + self.start);
     }
 
     /// Pad the buffer to `len` with bytes set to `v`.
     pub fn pad_to(&mut self, len: usize, v: u8) {
-        if len > self.buf.len() {
-            self.buf.resize(len, v);
+        if len > self.buf.len() - self.start {
+            self.buf.resize(len + self.start, v);
         }
     }
 
     #[must_use]
     pub fn to_vec(&self) -> Vec<u8> {
-        self.buf.clone()
+        self.buf[self.start..].to_vec()
     }
 }
 
@@ -426,29 +430,31 @@ impl<'a> Debug for Encoder<'a> {
 
 impl<'a> AsRef<[u8]> for Encoder<'a> {
     fn as_ref(&self) -> &[u8] {
-        self.buf
+        &self.buf[self.start..]
     }
 }
 
 impl<'a> AsMut<[u8]> for Encoder<'a> {
     fn as_mut(&mut self) -> &mut [u8] {
-        self.buf
-    }
-}
-
-impl<'a> From<Encoder<'a>> for &'a [u8] {
-    #[must_use]
-    fn from(encoder: Encoder<'a>) -> &'a [u8] {
-        encoder.buf
+        &mut self.buf[self.start..]
     }
 }
 
-impl<'a> From<Encoder<'a>> for &'a mut Vec<u8> {
-    #[must_use]
-    fn from(buf: Encoder<'a>) -> &'a mut Vec<u8> {
-        buf.buf
-    }
-}
+// TODO: Needed? Non-intuiv at least.
+// impl<'a> From<Encoder<'a>> for &'a [u8] {
+//     #[must_use]
+//     fn from(encoder: Encoder<'a>) -> &'a [u8] {
+//         encoder.buf
+//     }
+// }
+
+// TODO: Needed? Non-intuiv at least.
+// impl<'a> From<Encoder<'a>> for &'a mut Vec<u8> {
+//     #[must_use]
+//     fn from(buf: Encoder<'a>) -> &'a mut Vec<u8> {
+//         buf.buf
+//     }
+// }
 
 #[cfg(test)]
 mod tests {

neqo-common/src/datagram.rs

@@ -46,6 +46,10 @@ impl<D> Datagram<D> {
     pub const fn segment_size(&self) -> usize {
         self.segment_size
     }
+
+    pub fn set_segment_size(&mut self, segment_size: usize) {
+        self.segment_size = segment_size;
+    }
 }
 
 impl<D: AsRef<[u8]>> Datagram<D> {

neqo-transport/src/connection/mod.rs

@@ -2343,28 +2343,104 @@ impl Connection {
         self.stats.borrow_mut().frame_tx.connection_close += 1;
     }
 
+    // TODO: There is a limit to the number of segments supported. E.g. you
+    // can't pass 200 segments to the Linux Kernel.
+    fn output_path<'a>(
+        &mut self,
+        path: &PathRef,
+        now: Instant,
+        closing_frame: &Option<ClosingFrame>,
+        out: &'a mut Vec<u8>,
+    ) -> Res<SendOption<'a>> {
+        assert_eq!(out.len(), 0);
+
+        qinfo!("\n===== output_path");
+        let mut initial_capacity = out.capacity();
+        // TODO
+        if initial_capacity == 0 {
+            initial_capacity = usize::MAX;
+        }
+        qinfo!("initial_capacity: {initial_capacity}");
+        let mut segment_size = None;
+
+        loop {
+            let limit = segment_size.map(|s| min(s, initial_capacity - out.len()));
+            if limit < segment_size {
+                qinfo!("no more space for another segment");
+                break;
+            }
+            qinfo!("limit: {limit:?}, out len: {}", out.len());
+            // Determine how we are sending packets (PTO, etc..).
+            let profile = self
+                .loss_recovery
+                .send_profile_2(limit, &path.borrow(), now);
+            qinfo!("output_path send_profile {:?}", profile);
+            qdebug!([self], "output_path send_profile {:?}", profile);
+
+            let start = out.len();
+            if let Some(pace) = self.output_path_segment(profile, path, now, closing_frame, out)? {
+                qinfo!("SendOption::No");
+                if segment_size.is_none() {
+                    return Ok(SendOption::No(pace));
+                }
+                break;
+            }
+            qinfo!("SendOption::Yes");
+
+            if segment_size.is_none() {
+                segment_size = Some(out.len());
+                if out.len() < path.borrow().plpmtu() / 2 {
+                    qinfo!("Segment is smaller than half of the mtu: {}", out.len());
+                    break;
+                }
+            }
+
+            if out.len() % segment_size.expect("TODO") != 0 {
+                qinfo!("Segment is smaller than previous");
+                break;
+            }
+
+            if path.borrow().pmtud().needs_probe() {
+                qinfo!("about to probe, thus larger than previous segment");
+                break;
+            }
+
+            assert!(out.len() - start <= segment_size.unwrap());
+        }
+
+        let mut dgram = path
+            .borrow_mut()
+            .datagram(out, &mut self.stats.borrow_mut());
+        dgram.set_segment_size(segment_size.expect("TODO"));
+
+        qinfo!(
+            "Sending datagram: {} size {} segments",
+            dgram.len(),
+            dgram.num_segments()
+        );
+
+        return Ok(SendOption::Yes(dgram));
+    }
+
     /// Build a datagram, possibly from multiple packets (for different PN
     /// spaces) and each containing 1+ frames.
     #[allow(clippy::too_many_lines)] // Yeah, that's just the way it is.
-    fn output_path<'a>(
+    fn output_path_segment<'a>(
         &mut self,
+        profile: SendProfile,
         path: &PathRef,
         now: Instant,
         closing_frame: &Option<ClosingFrame>,
         out: &'a mut Vec<u8>,
-    ) -> Res<SendOption<'a>> {
+        // TODO: Option is misleading here.
+    ) -> Res<Option<bool>> {
         let mut initial_sent = None;
         let mut needs_padding = false;
         let grease_quic_bit = self.can_grease_quic_bit();
         let version = self.version();
 
-        // Determine how we are sending packets (PTO, etc..).
-        let profile = self.loss_recovery.send_profile(&path.borrow(), now);
-        qdebug!([self], "output_path send_profile {:?}", profile);
-
         // Frames for different epochs must go in different packets, but then these
         // packets can go in a single datagram
-        assert_eq!(out.len(), 0);
         let mut encoder = Encoder::new(out);
         for space in PacketNumberSpace::iter() {
             // Ensure we have tx crypto state for this epoch, or skip it.
@@ -2492,30 +2568,26 @@ impl Connection {
 
         if encoder.is_empty() {
             qdebug!("TX blocked, profile={:?}", profile);
-            Ok(SendOption::No(profile.paced()))
+            Ok(Some(profile.paced()))
         } else {
             // Perform additional padding for Initial packets as necessary.
-            let packets: &mut Vec<u8> = encoder.into();
             if let Some(mut initial) = initial_sent.take() {
                 if needs_padding {
                     qdebug!(
                         [self],
                         "pad Initial from {} to PLPMTU {}",
-                        packets.len(),
+                        encoder.len(),
                         profile.limit()
                     );
-                    initial.track_padding(profile.limit() - packets.len());
+                    initial.track_padding(profile.limit() - encoder.len());
                     // These zeros aren't padding frames, they are an invalid all-zero coalesced
                     // packet, which is why we don't increase `frame_tx.padding` count here.
-                    packets.resize(profile.limit(), 0);
+                    encoder.pad_to(profile.limit(), 0);
                 }
                 self.loss_recovery.on_packet_sent(path, initial);
             }
-            path.borrow_mut().add_sent(packets.len());
-            Ok(SendOption::Yes(
-                path.borrow_mut()
-                    .datagram(packets, &mut self.stats.borrow_mut()),
-            ))
+            path.borrow_mut().add_sent(encoder.len());
+            Ok(None)
         }
     }
 

neqo-transport/src/packet/mod.rs

@@ -499,8 +499,10 @@ impl<'a> PacketBuilder<'a> {
             Ok(aead.encrypt(0, encoder.as_ref(), &[], &mut buf)?.to_vec())
         })?;
         encoder.encode(&tag);
-        let complete: &[u8] = encoder.into();
-        Ok(&complete[start..])
+
+        // TODO: Tripple check that this is correct. Ideally improve Encoder instead of this hack here.
+        let encoder_len = encoder.len();
+        Ok(&out[out.len() - encoder_len + start..])
     }
 
     /// Make a Version Negotiation packet.
@@ -534,7 +536,9 @@ impl<'a> PacketBuilder<'a> {
         grease[3] = (client_version.wrapping_add(0x10) & 0xf0) as u8 | 0x0a;
         encoder.encode(&grease[..4]);
 
-        encoder.into()
+        // TODO
+        let encoder_len = encoder.len();
+        &out[(out.len() - encoder_len)..]
     }
 }
 

neqo-transport/src/recovery/mod.rs

@@ -885,13 +885,30 @@ impl LossRecovery {
         lost_packets
     }
 
+    #[cfg(test)]
+    pub fn send_profile(&mut self, path: &Path, now: Instant) -> SendProfile {
+        self.send_profile_2(None, path, now)
+    }
+
     /// Check how packets should be sent, based on whether there is a PTO,
     /// what the current congestion window is, and what the pacer says.
     #[allow(clippy::option_if_let_else)]
-    pub fn send_profile(&mut self, path: &Path, now: Instant) -> SendProfile {
+    pub fn send_profile_2(
+        &mut self,
+        // TODO: Maybe `segment_limit` is better? Because it might be the last
+        // in the vector, smaller than the overall segment size.
+        segment_size: Option<usize>,
+        path: &Path,
+        now: Instant,
+    ) -> SendProfile {
         qtrace!([self], "get send profile {:?}", now);
         let sender = path.sender();
-        let mtu = path.plpmtu();
+        let mut mtu = path.plpmtu();
+        // TODO: Cleanup?
+        if let Some(segment_size) = segment_size {
+            mtu = min(mtu, segment_size);
+        }
+
         if let Some(profile) = self
             .pto_state
             .as_mut()