Skip to content

Commit fcb6301

Browse files
author
Scott Hutton
committed
Track last error atomically
Eliminate the need for UdpSocket to be passed mutably into send_mmsg().
1 parent 4b1e816 commit fcb6301

File tree

2 files changed

+63
-26
lines changed

2 files changed

+63
-26
lines changed

src/lib.rs

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,10 @@
22
use std::{
33
net::{IpAddr, Ipv6Addr, SocketAddr},
44
sync::atomic::{AtomicUsize, Ordering},
5-
time::{Duration, Instant},
65
};
76

87
pub use crate::cmsg::{AsPtr, EcnCodepoint, Source, Transmit};
8+
use imp::LastSendError;
99
use tracing::warn;
1010

1111
mod cmsg;
@@ -94,20 +94,18 @@ impl Default for RecvMeta {
9494
}
9595

9696
/// Log at most 1 IO error per minute
97-
const IO_ERROR_LOG_INTERVAL: Duration = std::time::Duration::from_secs(60);
97+
const IO_ERROR_LOG_INTERVAL: u64 = 60;
9898

9999
/// Logs a warning message when sendmsg fails
100100
///
101101
/// Logging will only be performed if at least [`IO_ERROR_LOG_INTERVAL`]
102102
/// has elapsed since the last error was logged.
103103
fn log_sendmsg_error<B: AsPtr<u8>>(
104-
last_send_error: &mut Instant,
104+
last_send_error: LastSendError,
105105
err: impl core::fmt::Debug,
106106
transmit: &Transmit<B>,
107107
) {
108-
let now = Instant::now();
109-
if now.saturating_duration_since(*last_send_error) > IO_ERROR_LOG_INTERVAL {
110-
*last_send_error = now;
108+
if last_send_error.should_log() {
111109
warn!(
112110
"sendmsg error: {:?}, Transmit: {{ destination: {:?}, src_ip: {:?}, enc: {:?}, len: {:?}, segment_size: {:?} }}",
113111
err, transmit.dst, transmit.src, transmit.ecn, transmit.contents.len(), transmit.segment_size);

src/unix.rs

Lines changed: 59 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,12 @@ use std::{
44
mem::{self, MaybeUninit},
55
net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6},
66
os::{fd::AsFd, unix::io::AsRawFd},
7-
sync::atomic::AtomicUsize,
7+
sync::{
8+
atomic::{AtomicU64, AtomicUsize, Ordering},
9+
Arc,
10+
},
811
task::{Context, Poll},
9-
time::Instant,
12+
time::SystemTime,
1013
};
1114

1215
use crate::cmsg::{AsPtr, EcnCodepoint, Source, Transmit};
@@ -31,7 +34,7 @@ type IpTosTy = libc::c_int;
3134
#[derive(Debug)]
3235
pub struct UdpSocket {
3336
io: tokio::net::UdpSocket,
34-
last_send_error: Instant,
37+
last_send_error: LastSendError,
3538
}
3639

3740
impl AsRawFd for UdpSocket {
@@ -46,16 +49,51 @@ impl AsFd for UdpSocket {
4649
}
4750
}
4851

52+
#[derive(Clone, Debug)]
53+
pub(crate) struct LastSendError(Arc<AtomicU64>);
54+
55+
impl Default for LastSendError {
56+
fn default() -> Self {
57+
let now = Self::now();
58+
Self(Arc::new(AtomicU64::new(
59+
now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now),
60+
)))
61+
}
62+
}
63+
64+
impl LastSendError {
65+
fn now() -> u64 {
66+
SystemTime::now()
67+
.duration_since(SystemTime::UNIX_EPOCH)
68+
.unwrap()
69+
.as_secs()
70+
}
71+
72+
/// Determine whether the last error was more than `IO_ERROR_LOG_INTERVAL`
73+
/// seconds ago. If so, update the last error time and return true.
74+
///
75+
/// Note: if the system clock regresses more tha `IO_ERROR_LOG_INTERVAL`,
76+
/// this function may impose an additional delay on log message emission.
77+
/// Similarly, if it advances, messages may be emitted prematurely.
78+
pub(crate) fn should_log(&self) -> bool {
79+
let now = Self::now();
80+
self.0
81+
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |cur| {
82+
(now.saturating_sub(cur) > IO_ERROR_LOG_INTERVAL).then_some(now)
83+
})
84+
.is_ok()
85+
}
86+
}
87+
4988
impl UdpSocket {
5089
/// Creates a new UDP socket from a previously created `std::net::UdpSocket`
5190
pub fn from_std(socket: std::net::UdpSocket) -> io::Result<UdpSocket> {
5291
socket.set_nonblocking(true)?;
5392

5493
init(SockRef::from(&socket))?;
55-
let now = Instant::now();
5694
Ok(UdpSocket {
5795
io: tokio::net::UdpSocket::from_std(socket)?,
58-
last_send_error: now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now),
96+
last_send_error: LastSendError::default(),
5997
})
6098
}
6199

@@ -67,10 +105,9 @@ impl UdpSocket {
67105
pub async fn bind<A: ToSocketAddrs>(addr: A) -> io::Result<UdpSocket> {
68106
let io = tokio::net::UdpSocket::bind(addr).await?;
69107
init(SockRef::from(&io))?;
70-
let now = Instant::now();
71108
Ok(UdpSocket {
72109
io,
73-
last_send_error: now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now),
110+
last_send_error: LastSendError::default(),
74111
})
75112
}
76113

@@ -195,13 +232,13 @@ impl UdpSocket {
195232
///
196233
/// [`sendmmsg`]: https://linux.die.net/man/2/sendmmsg
197234
pub async fn send_mmsg<B: AsPtr<u8>>(
198-
&mut self,
235+
&self,
199236
state: &UdpState,
200237
transmits: &[Transmit<B>],
201238
) -> Result<usize, io::Error> {
202239
let n = loop {
203240
self.io.writable().await?;
204-
let last_send_error = &mut self.last_send_error;
241+
let last_send_error = self.last_send_error.clone();
205242
let io = &self.io;
206243
match io.try_io(Interest::WRITABLE, || {
207244
send(state, SockRef::from(io), last_send_error, transmits)
@@ -278,11 +315,15 @@ impl UdpSocket {
278315
transmits: &[Transmit<B>],
279316
) -> Poll<io::Result<usize>> {
280317
loop {
281-
let last_send_error = &mut self.last_send_error;
282318
ready!(self.io.poll_send_ready(cx))?;
283319
let io = &self.io;
284320
if let Ok(res) = io.try_io(Interest::WRITABLE, || {
285-
send(state, SockRef::from(io), last_send_error, transmits)
321+
send(
322+
state,
323+
SockRef::from(io),
324+
self.last_send_error.clone(),
325+
transmits,
326+
)
286327
}) {
287328
return Poll::Ready(Ok(res));
288329
}
@@ -353,7 +394,7 @@ pub mod sync {
353394
#[derive(Debug)]
354395
pub struct UdpSocket {
355396
io: std::net::UdpSocket,
356-
last_send_error: Instant,
397+
last_send_error: LastSendError,
357398
}
358399

359400
impl AsRawFd for UdpSocket {
@@ -372,21 +413,19 @@ pub mod sync {
372413
pub fn from_std(socket: std::net::UdpSocket) -> io::Result<Self> {
373414
init(SockRef::from(&socket))?;
374415
socket.set_nonblocking(false)?;
375-
let now = Instant::now();
376416
Ok(Self {
377417
io: socket,
378-
last_send_error: now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now),
418+
last_send_error: LastSendError::default(),
379419
})
380420
}
381421
/// create a new UDP socket and attempt to bind to `addr`
382422
pub fn bind<A: std::net::ToSocketAddrs>(addr: A) -> io::Result<Self> {
383423
let io = std::net::UdpSocket::bind(addr)?;
384424
init(SockRef::from(&io))?;
385425
io.set_nonblocking(false)?;
386-
let now = Instant::now();
387426
Ok(Self {
388427
io,
389-
last_send_error: now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now),
428+
last_send_error: LastSendError::default(),
390429
})
391430
}
392431
/// sets nonblocking mode
@@ -474,7 +513,7 @@ pub mod sync {
474513
send(
475514
state,
476515
SockRef::from(&self.io),
477-
&mut self.last_send_error,
516+
self.last_send_error.clone(),
478517
transmits,
479518
)
480519
}
@@ -681,7 +720,7 @@ fn send_msg<B: AsPtr<u8>>(
681720
fn send<B: AsPtr<u8>>(
682721
state: &UdpState,
683722
io: SockRef<'_>,
684-
last_send_error: &mut Instant,
723+
last_send_error: LastSendError,
685724
transmits: &[Transmit<B>],
686725
) -> io::Result<usize> {
687726
use std::ptr;
@@ -802,7 +841,7 @@ fn send_msg<B: AsPtr<u8>>(
802841
fn send<B: AsPtr<u8>>(
803842
_state: &UdpState,
804843
io: SockRef<'_>,
805-
last_send_error: &mut Instant,
844+
last_send_error: LastSendError,
806845
transmits: &[Transmit<B>],
807846
) -> io::Result<usize> {
808847
let mut hdr: libc::msghdr = unsafe { mem::zeroed() };
@@ -828,7 +867,7 @@ fn send<B: AsPtr<u8>>(
828867
// Those are not fatal errors, since the
829868
// configuration can be dynamically changed.
830869
// - Destination unreachable errors have been observed for other
831-
log_sendmsg_error(last_send_error, e, &transmits[sent]);
870+
log_sendmsg_error(last_send_error.clone(), e, &transmits[sent]);
832871
sent += 1;
833872
}
834873
}

0 commit comments

Comments
 (0)