Latch the send_time at the beginning of the write loop

This will result in BBR sent packets having the correct send times
in cases where multiple packets are emitted in a burst, and result
in correct minRTT calculations.

The next release time is computed by QUIC connection path since each
path has it's own congestion controller.  Ideally sends should be
done in next release time order for all the paths to avoid generating
packets on slower paths first.

Author: antoniovicente

quiche/src/lib.rs

@@ -3833,6 +3833,91 @@ impl<F: BufFactory> Connection<F> {
     pub fn send_on_path(
         &mut self, out: &mut [u8], from: Option<SocketAddr>,
         to: Option<SocketAddr>,
+    ) -> Result<(usize, SendInfo)> {
+        let path_id = match (from, to) {
+            (Some(f), Some(t)) => self
+                .paths
+                .path_id_from_addrs(&(f, t))
+                .ok_or(Error::InvalidState)?,
+
+            _ => self.get_send_path_id(from, to)?,
+        };
+
+        let now = Instant::now();
+
+        self.send_with_path_id(out, PathId { path_id }, now)
+    }
+
+    /// Writes a single QUIC packet to be sent to the peer from the
+    /// specified path_id.  Call `paths_id_iter()` to get a list of
+    /// path_ids to call `send_with_path_id()` on.
+    ///
+    /// On success the number of bytes written to the output buffer is
+    /// returned, or [`Done`] if there was nothing to write.
+    ///
+    /// The application should call `send_with_path_id()` multiple times until
+    /// [`Done`] is returned, indicating that there are no more packets to
+    /// send. It is recommended that `send_with_path_id()` be called in the
+    /// following cases:
+    ///
+    ///  * When the application receives QUIC packets from the peer (that is,
+    ///    any time [`recv()`] is also called).
+    ///
+    ///  * When the connection timer expires (that is, any time [`on_timeout()`]
+    ///    is also called).
+    ///
+    ///  * When the application sends data to the peer (for examples, any time
+    ///    [`stream_send()`] or [`stream_shutdown()`] are called).
+    ///
+    ///  * When the application receives data from the peer (for example any
+    ///    time [`stream_recv()`] is called).
+    ///
+    /// Once [`is_draining()`] returns `true`, it is no longer necessary to call
+    /// `send_with_path_id()` and all calls will return [`Done`].
+    ///
+    /// [`Done`]: enum.Error.html#variant.Done
+    /// [`InvalidState`]: enum.Error.html#InvalidState
+    /// [`recv()`]: struct.Connection.html#method.recv
+    /// [`on_timeout()`]: struct.Connection.html#method.on_timeout
+    /// [`stream_send()`]: struct.Connection.html#method.stream_send
+    /// [`stream_shutdown()`]: struct.Connection.html#method.stream_shutdown
+    /// [`stream_recv()`]: struct.Connection.html#method.stream_recv
+    /// [`path_event_next()`]: struct.Connection.html#method.path_event_next
+    /// [`paths_iter()`]: struct.Connection.html#method.paths_iter
+    /// [`is_draining()`]: struct.Connection.html#method.is_draining
+    ///
+    /// ## Examples:
+    ///
+    /// ```no_run
+    /// # let mut out = [0; 512];
+    /// # let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+    /// # let mut config = quiche::Config::new(quiche::PROTOCOL_VERSION)?;
+    /// # let scid = quiche::ConnectionId::from_ref(&[0xba; 16]);
+    /// # let peer = "127.0.0.1:1234".parse().unwrap();
+    /// # let local = socket.local_addr().unwrap();
+    /// # let mut conn = quiche::accept(&scid, None, local, peer, &mut config)?;
+    /// loop {
+    ///     for path_id in self.paths_id_iter(local) {
+    ///     let (write, send_info) = match conn.send_with_path_id(&mut out, path_id, Instant::now()) {
+    ///         Ok(v) => v,
+    ///
+    ///         Err(quiche::Error::Done) => {
+    ///             // Done writing.
+    ///             break;
+    ///         },
+    ///
+    ///         Err(e) => {
+    ///             // An error occurred, handle it.
+    ///             break;
+    ///         },
+    ///     };
+    ///
+    ///     socket.send_to(&out[..write], &send_info.to).unwrap();
+    /// }
+    /// # Ok::<(), quiche::Error>(())
+    /// ```
+    pub fn send_with_path_id(
+        &mut self, out: &mut [u8], path_id: PathId, now: Instant,
     ) -> Result<(usize, SendInfo)> {
         if out.is_empty() {
             return Err(Error::BufferTooShort);
@@ -3842,8 +3927,6 @@ impl<F: BufFactory> Connection<F> {
             return Err(Error::Done);
         }
 
-        let now = Instant::now();
-
         if self.local_error.is_none() {
             self.do_handshake(now)?;
         }
@@ -3870,15 +3953,7 @@ impl<F: BufFactory> Connection<F> {
         // maximum UDP payload size limit.
         let mut left = cmp::min(out.len(), self.max_send_udp_payload_size());
 
-        let send_pid = match (from, to) {
-            (Some(f), Some(t)) => self
-                .paths
-                .path_id_from_addrs(&(f, t))
-                .ok_or(Error::InvalidState)?,
-
-            _ => self.get_send_path_id(from, to)?,
-        };
-
+        let send_pid = path_id.path_id;
         let send_path = self.paths.get_mut(send_pid)?;
 
         // Update max datagram size to allow path MTU discovery probe to be sent.
@@ -3938,13 +4013,6 @@ impl<F: BufFactory> Connection<F> {
                     break;
                 }
             }
-
-            // Don't coalesce packets that must go on different paths.
-            if !(from.is_some() && to.is_some()) &&
-                self.get_send_path_id(from, to)? != send_pid
-            {
-                break;
-            }
         }
 
         if done == 0 {
@@ -5226,18 +5294,45 @@ impl<F: BufFactory> Connection<F> {
         Ok(())
     }
 
-    /// Returns the desired send time for the next packet.
+    /// Returns the desired send time for the next packet on the desired path.
     #[inline]
-    pub fn get_next_release_time(&self) -> Option<ReleaseDecision> {
+    pub fn get_next_release_time(
+        &self, path_id: PathId,
+    ) -> Option<ReleaseDecision> {
         Some(
             self.paths
-                .get_active()
+                .get(path_id.path_id)
                 .ok()?
                 .recovery
                 .get_next_release_time(),
         )
     }
 
+    /// Latches the packet send time for the selected path.  The desired send
+    /// time must be >= now. Send time remains latched until cleared.
+    pub fn latch_packet_send_time(
+        &mut self, path_id: PathId, send_time: Instant,
+    ) -> Result<()> {
+        self.paths
+            .get_mut(path_id.path_id)?
+            .recovery
+            .latch_packet_send_time(send_time);
+
+        Ok(())
+    }
+
+    /// Clear the latched packet send time on the selected path.
+    pub fn clear_latched_packet_send_time(
+        &mut self, path_id: PathId,
+    ) -> Result<()> {
+        self.paths
+            .get_mut(path_id.path_id)?
+            .recovery
+            .clear_latched_packet_send_time();
+
+        Ok(())
+    }
+
     /// Returns whether gcongestion is enabled.
     #[inline]
     pub fn gcongestion_enabled(&self) -> Option<bool> {
@@ -6950,6 +7045,35 @@ impl<F: BufFactory> Connection<F> {
         }
     }
 
+    /// Returns an iterator over `PathId`s whose association with `from` forms
+    /// a known QUIC path on which packets can be sent to.
+    ///
+    /// This function is typically used in combination with [`send_with_path_id()`].
+    ///
+    /// Note that the iterator includes all the possible combination of
+    /// destination `PathId`s, even those whose sending is not required now.
+    /// In other words, this is another way for the application to recall from
+    /// past [`PathEvent::New`] events.
+    ///
+    /// [`PathEvent::New`]: enum.PathEvent.html#variant.New
+    /// [`send_with_path_id()`]: struct.Connection.html#method.send_with_path_id
+    pub fn paths_id_iter(&self, from: SocketAddr) -> PathIdIter {
+        // Instead of trying to identify whether packets will be sent on the
+        // given 4-tuple, simply filter paths that cannot be used.
+        PathIdIter {
+            paths: self
+                .paths
+                .iter()
+                .filter(|(_, p)| p.active_dcid_seq.is_some())
+                .filter(|(_, p)| p.usable() || p.probing_required())
+                .filter(|(_, p)| p.local_addr() == from)
+                .map(|(path_id, _p)| PathId { path_id })
+                .collect(),
+
+            index: 0,
+        }
+    }
+
     /// Closes the connection with the given error and reason.
     ///
     /// The `app` parameter specifies whether an application close should be
@@ -9417,6 +9541,8 @@ pub use crate::packet::Header;
 pub use crate::packet::Type;
 
 pub use crate::path::PathEvent;
+pub use crate::path::PathId;
+pub use crate::path::PathIdIter;
 pub use crate::path::PathStats;
 pub use crate::path::SocketAddrIter;
 

quiche/src/path.rs

@@ -569,6 +569,37 @@ impl Path {
     }
 }
 
+/// Opaque identifier for a Path on a QUIC connection.
+#[derive(Clone, Copy, Debug)]
+pub struct PathId {
+    pub(crate) path_id: usize,
+}
+
+/// An iterator over PathId.
+#[derive(Default)]
+pub struct PathIdIter {
+    pub(crate) paths: SmallVec<[PathId; 8]>,
+    pub(crate) index: usize,
+}
+
+impl Iterator for PathIdIter {
+    type Item = PathId;
+
+    #[inline]
+    fn next(&mut self) -> Option<Self::Item> {
+        let v = self.paths.get(self.index)?;
+        self.index += 1;
+        Some(*v)
+    }
+}
+
+impl ExactSizeIterator for PathIdIter {
+    #[inline]
+    fn len(&self) -> usize {
+        self.paths.len() - self.index
+    }
+}
+
 /// An iterator over SocketAddr.
 #[derive(Default)]
 pub struct SocketAddrIter {

quiche/src/recovery/congestion/recovery.rs

@@ -631,6 +631,13 @@ impl RecoveryOps for LegacyRecovery {
         now
     }
 
+    fn latch_packet_send_time(&mut self, _now: Instant) {
+        // Not needed because recovery algorithms in this directory do
+        // not implement pacing.
+    }
+
+    fn clear_latched_packet_send_time(&mut self) {}
+
     // `peer_sent_ack_ranges` should not be used without validation.
     fn on_ack_received(
         &mut self, peer_sent_ack_ranges: &RangeSet, ack_delay: u64, epoch: Epoch,

quiche/src/recovery/gcongestion/recovery.rs

@@ -465,6 +465,8 @@ pub struct GRecovery {
     lost_reuse: Vec<Lost>,
 
     pacer: Pacer,
+
+    latched_packet_send_time: Option<Instant>,
 }
 
 impl GRecovery {
@@ -518,6 +520,7 @@ impl GRecovery {
 
             newly_acked: Vec::new(),
             lost_reuse: Vec::new(),
+            latched_packet_send_time: None,
         })
     }
 
@@ -688,7 +691,7 @@ impl RecoveryOps for GRecovery {
         &mut self, pkt: Sent, epoch: packet::Epoch,
         handshake_status: HandshakeStatus, now: Instant, trace_id: &str,
     ) {
-        let time_sent = self.get_next_release_time().time(now).unwrap_or(now);
+        let time_sent = self.get_packet_send_time(now);
 
         let epoch = &mut self.epochs[epoch];
 
@@ -749,7 +752,18 @@ impl RecoveryOps for GRecovery {
     }
 
     fn get_packet_send_time(&self, now: Instant) -> Instant {
-        self.pacer.get_next_release_time().time(now).unwrap_or(now)
+        // TODO debug_assert that latched time >= now.
+        self.latched_packet_send_time
+            .filter(|v| *v > now)
+            .unwrap_or(now)
+    }
+
+    fn latch_packet_send_time(&mut self, send_time: Instant) {
+        self.latched_packet_send_time = Some(send_time);
+    }
+
+    fn clear_latched_packet_send_time(&mut self) {
+        self.latched_packet_send_time = None;
     }
 
     // `peer_sent_ack_ranges` should not be used without validation.
@@ -802,6 +816,7 @@ impl RecoveryOps for GRecovery {
             has_ack_eliciting;
         if update_rtt {
             let latest_rtt = now - largest_newly_acked.time_sent;
+
             self.rtt_stats.update_rtt(
                 latest_rtt,
                 Duration::from_micros(ack_delay),