Eliminating an Instant::now() in BBR bandwidth estimation code

recatek · recatek · commit 575468ccd180 · 2025-02-19T13:42:24.000-05:00
This was the only case where `Instant::now()` is used in quinn-proto or quinn-udp for anything other than tests or rate-limiting logging. Making this change to keep these two crates more IO-agnostic.

Clean up Mutex/Instant::now() use for send message error logging
diff --git a/quinn-proto/src/congestion/bbr/bw_estimation.rs b/quinn-proto/src/congestion/bbr/bw_estimation.rs
@@ -3,15 +3,15 @@ use std::fmt::{Debug, Display, Formatter};
 use super::min_max::MinMax;
 use crate::{Duration, Instant};
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Default)]
 pub(crate) struct BandwidthEstimation {
     total_acked: u64,
     prev_total_acked: u64,
     acked_time: Option<Instant>,
     prev_acked_time: Option<Instant>,
     total_sent: u64,
     prev_total_sent: u64,
-    sent_time: Instant,
+    sent_time: Option<Instant>,
     prev_sent_time: Option<Instant>,
     max_filter: MinMax,
     acked_at_last_window: u64,
@@ -21,8 +21,8 @@ impl BandwidthEstimation {
     pub(crate) fn on_sent(&mut self, now: Instant, bytes: u64) {
         self.prev_total_sent = self.total_sent;
         self.total_sent += bytes;
-        self.prev_sent_time = Some(self.sent_time);
-        self.sent_time = now;
+        self.prev_sent_time = self.sent_time;
+        self.sent_time = Some(now);
     }
 
     pub(crate) fn on_ack(
@@ -43,14 +43,13 @@ impl BandwidthEstimation {
             None => return,
         };
 
-        let send_rate = if self.sent_time > prev_sent_time {
-            Self::bw_from_delta(
+        let send_rate = match self.sent_time {
+            Some(sent_time) if sent_time > prev_sent_time => Self::bw_from_delta(
                 self.total_sent - self.prev_total_sent,
-                self.sent_time - prev_sent_time,
+                sent_time - prev_sent_time,
             )
-            .unwrap_or(0)
-        } else {
-            u64::MAX // will take the min of send and ack, so this is just a skip
+            .unwrap_or(0),
+            _ => u64::MAX, // will take the min of send and ack, so this is just a skip
         };
 
         let ack_rate = match self.prev_acked_time {
@@ -91,25 +90,6 @@ impl BandwidthEstimation {
     }
 }
 
-impl Default for BandwidthEstimation {
-    fn default() -> Self {
-        Self {
-            total_acked: 0,
-            prev_total_acked: 0,
-            acked_time: None,
-            prev_acked_time: None,
-            total_sent: 0,
-            prev_total_sent: 0,
-            // The `sent_time` value set here is ignored; it is used in `on_ack()`, but will
-            // have been reset by `on_sent()` before that method is called.
-            sent_time: Instant::now(),
-            prev_sent_time: None,
-            max_filter: MinMax::default(),
-            acked_at_last_window: 0,
-        }
-    }
-}
-
 impl Display for BandwidthEstimation {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(
diff --git a/quinn-udp/src/fallback.rs b/quinn-udp/src/fallback.rs
@@ -1,26 +1,21 @@
-use std::{
-    io::{self, IoSliceMut},
-    sync::Mutex,
-    time::Instant,
-};
+use std::io::{self, IoSliceMut};
 
-use super::{log_sendmsg_error, RecvMeta, Transmit, UdpSockRef, IO_ERROR_LOG_INTERVAL};
+use super::{RecvMeta, SendErrorSink, Transmit, UdpSockRef};
 
 /// Fallback UDP socket interface that stubs out all special functionality
 ///
 /// Used when a better implementation is not available for a particular target, at the cost of
 /// reduced performance compared to that enabled by some target-specific interfaces.
 #[derive(Debug)]
 pub struct UdpSocketState {
-    last_send_error: Mutex<Instant>,
+    last_send_error: SendErrorSink,
 }
 
 impl UdpSocketState {
     pub fn new(socket: UdpSockRef<'_>) -> io::Result<Self> {
         socket.0.set_nonblocking(true)?;
-        let now = Instant::now();
         Ok(Self {
-            last_send_error: Mutex::new(now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now)),
+            last_send_error: SendErrorSink::new(),
         })
     }
 
@@ -40,7 +35,7 @@ impl UdpSocketState {
             Ok(()) => Ok(()),
             Err(e) if e.kind() == io::ErrorKind::WouldBlock => Err(e),
             Err(e) => {
-                log_sendmsg_error(&self.last_send_error, e, transmit);
+                self.last_send_error.log_sendmsg_error(e, transmit);
 
                 Ok(())
             }
diff --git a/quinn-udp/src/lib.rs b/quinn-udp/src/lib.rs
@@ -32,7 +32,7 @@ use std::net::{IpAddr, Ipv6Addr, SocketAddr};
 use std::os::unix::io::AsFd;
 #[cfg(windows)]
 use std::os::windows::io::AsSocket;
-#[cfg(not(wasm_browser))]
+#[cfg(all(not(wasm_browser), any(feature = "tracing", feature = "direct-log")))]
 use std::{
     sync::Mutex,
     time::{Duration, Instant},
@@ -146,33 +146,47 @@ pub struct Transmit<'a> {
     pub src_ip: Option<IpAddr>,
 }
 
-/// Log at most 1 IO error per minute
-#[cfg(not(wasm_browser))]
-const IO_ERROR_LOG_INTERVAL: Duration = std::time::Duration::from_secs(60);
-
-/// Logs a warning message when sendmsg fails
-///
-/// Logging will only be performed if at least [`IO_ERROR_LOG_INTERVAL`]
-/// has elapsed since the last error was logged.
 #[cfg(all(not(wasm_browser), any(feature = "tracing", feature = "direct-log")))]
-fn log_sendmsg_error(
-    last_send_error: &Mutex<Instant>,
-    err: impl core::fmt::Debug,
-    transmit: &Transmit,
-) {
-    let now = Instant::now();
-    let last_send_error = &mut *last_send_error.lock().expect("poisend lock");
-    if now.saturating_duration_since(*last_send_error) > IO_ERROR_LOG_INTERVAL {
-        *last_send_error = now;
-        log::warn!(
-        "sendmsg error: {:?}, Transmit: {{ destination: {:?}, src_ip: {:?}, ecn: {:?}, len: {:?}, segment_size: {:?} }}",
-            err, transmit.destination, transmit.src_ip, transmit.ecn, transmit.contents.len(), transmit.segment_size);
-    }
+const IO_ERROR_LOG_INTERVAL: Duration = Duration::from_secs(60);
+
+/// Helper for rate-limiting the logging of sendmsg errors.
+#[derive(Debug)]
+struct SendErrorSink {
+    #[cfg(all(not(wasm_browser), any(feature = "tracing", feature = "direct-log")))]
+    last_send_error: Mutex<Instant>,
 }
 
-// No-op
-#[cfg(not(any(wasm_browser, feature = "tracing", feature = "direct-log")))]
-fn log_sendmsg_error(_: &Mutex<Instant>, _: impl core::fmt::Debug, _: &Transmit) {}
+impl SendErrorSink {
+    /// Produces a record the last send error in socket state for error reporting
+    fn new() -> Self {
+        Self {
+            #[cfg(all(not(wasm_browser), any(feature = "tracing", feature = "direct-log")))]
+            last_send_error: {
+                let now = Instant::now();
+                Mutex::new(now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now))
+            },
+        }
+    }
+
+    /// Logs a warning message when sendmsg fails
+    ///
+    /// Logging will only be performed if at least [`IO_ERROR_LOG_INTERVAL`]
+    /// has elapsed since the last error was logged.
+    #[allow(unused_variables)]
+    fn log_sendmsg_error(&self, err: impl core::fmt::Debug, transmit: &Transmit) {
+        #[cfg(all(not(wasm_browser), any(feature = "tracing", feature = "direct-log")))]
+        {
+            let now = Instant::now();
+            let last_send_error = &mut *self.last_send_error.lock().expect("poisoned lock");
+            if now.saturating_duration_since(*last_send_error) > IO_ERROR_LOG_INTERVAL {
+                *last_send_error = now;
+                log::warn!(
+                    "sendmsg error: {:?}, Transmit: {{ destination: {:?}, src_ip: {:?}, ecn: {:?}, len: {:?}, segment_size: {:?} }}",
+                    err, transmit.destination, transmit.src_ip, transmit.ecn, transmit.contents.len(), transmit.segment_size);
+            }
+        }
+    }
+}
 
 /// A borrowed UDP socket
 ///
diff --git a/quinn-udp/src/unix.rs b/quinn-udp/src/unix.rs
@@ -5,18 +5,12 @@ use std::{
     mem::{self, MaybeUninit},
     net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6},
     os::unix::io::AsRawFd,
-    sync::{
-        atomic::{AtomicBool, AtomicUsize, Ordering},
-        Mutex,
-    },
-    time::Instant,
+    sync::atomic::{AtomicBool, AtomicUsize, Ordering},
 };
 
 use socket2::SockRef;
 
-use super::{
-    cmsg, log_sendmsg_error, EcnCodepoint, RecvMeta, Transmit, UdpSockRef, IO_ERROR_LOG_INTERVAL,
-};
+use super::{cmsg, EcnCodepoint, RecvMeta, SendErrorSink, Transmit, UdpSockRef};
 
 // Adapted from https://github.com/apple-oss-distributions/xnu/blob/8d741a5de7ff4191bf97d57b9f54c2f6d4a15585/bsd/sys/socket_private.h
 #[cfg(apple_fast)]
@@ -69,7 +63,7 @@ type IpTosTy = libc::c_int;
 /// platforms.
 #[derive(Debug)]
 pub struct UdpSocketState {
-    last_send_error: Mutex<Instant>,
+    last_send_error: SendErrorSink,
     max_gso_segments: AtomicUsize,
     gro_segments: usize,
     may_fragment: bool,
@@ -183,9 +177,8 @@ impl UdpSocketState {
                 !set_socket_option_supported(&*io, libc::IPPROTO_IPV6, IPV6_DONTFRAG, OPTION_ON)?;
         }
 
-        let now = Instant::now();
         Ok(Self {
-            last_send_error: Mutex::new(now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now)),
+            last_send_error: SendErrorSink::new(),
             max_gso_segments: AtomicUsize::new(gso::max_gso_segments()),
             gro_segments: gro::gro_segments(),
             may_fragment,
@@ -209,7 +202,7 @@ impl UdpSocketState {
             Ok(()) => Ok(()),
             Err(e) if e.kind() == io::ErrorKind::WouldBlock => Err(e),
             Err(e) => {
-                log_sendmsg_error(&self.last_send_error, e, transmit);
+                self.last_send_error.log_sendmsg_error(e, transmit);
 
                 Ok(())
             }
diff --git a/quinn-udp/src/windows.rs b/quinn-udp/src/windows.rs
@@ -4,8 +4,6 @@ use std::{
     net::{IpAddr, Ipv4Addr},
     os::windows::io::AsRawSocket,
     ptr,
-    sync::Mutex,
-    time::Instant,
 };
 
 use libc::{c_int, c_uint};
@@ -15,15 +13,15 @@ use windows_sys::Win32::Networking::WinSock;
 use crate::{
     cmsg::{self, CMsgHdr},
     log::debug,
-    log_sendmsg_error, EcnCodepoint, RecvMeta, Transmit, UdpSockRef, IO_ERROR_LOG_INTERVAL,
+    EcnCodepoint, RecvMeta, SendErrorSink, Transmit, UdpSockRef,
 };
 
 /// QUIC-friendly UDP socket for Windows
 ///
 /// Unlike a standard Windows UDP socket, this allows ECN bits to be read and written.
 #[derive(Debug)]
 pub struct UdpSocketState {
-    last_send_error: Mutex<Instant>,
+    last_send_error: SendErrorSink,
 }
 
 impl UdpSocketState {
@@ -112,9 +110,8 @@ impl UdpSocketState {
             )?;
         }
 
-        let now = Instant::now();
         Ok(Self {
-            last_send_error: Mutex::new(now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now)),
+            last_send_error: SendErrorSink::new(),
         })
     }
 
@@ -156,7 +153,7 @@ impl UdpSocketState {
             Ok(()) => Ok(()),
             Err(e) if e.kind() == io::ErrorKind::WouldBlock => Err(e),
             Err(e) => {
-                log_sendmsg_error(&self.last_send_error, e, transmit);
+                self.last_send_error.log_sendmsg_error(e, transmit);
 
                 Ok(())
             }
