timestamped_socket/
socket.rs

1use std::{
2    marker::PhantomData,
3    net::{SocketAddr, SocketAddrV4, SocketAddrV6},
4    os::fd::AsRawFd,
5    sync::Arc,
6};
7
8use tokio::io::{unix::AsyncFd, Interest};
9
10use crate::{
11    control_message::{ControlMessage, MessageQueue, EXPECTED_MAX_CMSG_SIZE},
12    interface::InterfaceName,
13    networkaddress::{sealed::PrivateToken, MulticastJoinable, NetworkAddress},
14    raw_socket::RawSocket,
15};
16
17#[cfg(not(any(target_os = "linux", target_os = "freebsd", target_os = "macos")))]
18mod fallback;
19#[cfg(target_os = "freebsd")]
20mod freebsd;
21#[cfg(target_os = "linux")]
22mod linux;
23#[cfg(target_os = "macos")]
24mod macos;
25
26#[cfg(not(any(target_os = "linux", target_os = "freebsd", target_os = "macos")))]
27use self::fallback::*;
28#[cfg(target_os = "freebsd")]
29use self::freebsd::*;
30#[cfg(target_os = "linux")]
31pub use self::linux::*;
32#[cfg(target_os = "macos")]
33use self::macos::*;
34
35#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, Default)]
36pub struct TimestampData {
37    /// The timestamping mode requested by the user when creating the socket.
38    ///
39    /// If this was not available, this is set to `None`.
40    pub timestamp_mode: InterfaceTimestampMode,
41    /// The hardware based timestamp returned by the OS
42    pub hardware: Option<Timestamp>,
43    /// The software based timestamp returned by the OS
44    pub software: Option<Timestamp>,
45}
46
47impl TimestampData {
48    /// Creates a new `TimestampData` with the given requested timestamping mode, but no timestamps.
49    fn empty(timestamp_mode: InterfaceTimestampMode) -> Self {
50        Self::default().with_timestamp_mode(timestamp_mode)
51    }
52
53    /// Returns the timestamp type selected by the requested timestamping mode.
54    ///
55    /// If the requested timestamping mode is `None` or if the requested timestamp
56    /// is not available, this function will return `None`.
57    pub fn selected_timestamp(self) -> Option<Timestamp> {
58        use InterfaceTimestampMode::*;
59
60        match self.timestamp_mode {
61            SoftwareAll | SoftwareRecv => self.software,
62            HardwareAll | HardwareRecv | HardwarePTPAll | HardwarePTPRecv => self.hardware,
63            None => Option::None,
64        }
65    }
66
67    /// Create a new `TimestampData` with the given requested timestamping mode.
68    fn with_timestamp_mode(self, timestamp_mode: InterfaceTimestampMode) -> Self {
69        Self {
70            timestamp_mode,
71            ..self
72        }
73    }
74}
75
76#[derive(Debug, Clone, Copy, Eq, PartialEq, PartialOrd, Ord, Hash, Default)]
77pub struct Timestamp {
78    pub seconds: i64,
79    pub nanos: u32,
80}
81
82impl Timestamp {
83    #[cfg_attr(target_os = "macos", allow(unused))] // macos does not do nanoseconds
84    pub(crate) fn from_timespec(timespec: libc::timespec) -> Self {
85        Self {
86            seconds: timespec.tv_sec as _,
87            nanos: timespec.tv_nsec as _,
88        }
89    }
90
91    pub(crate) fn from_timeval(timeval: libc::timeval) -> Self {
92        Self {
93            seconds: timeval.tv_sec as _,
94            nanos: (1000 * timeval.tv_usec) as _,
95        }
96    }
97}
98
99#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
100pub enum GeneralTimestampMode {
101    SoftwareAll,
102    SoftwareRecv,
103    #[default]
104    None,
105}
106
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
108pub enum InterfaceTimestampMode {
109    HardwareAll,
110    HardwareRecv,
111    HardwarePTPAll,
112    HardwarePTPRecv,
113    SoftwareAll,
114    SoftwareRecv,
115    #[default]
116    None,
117}
118
119impl From<GeneralTimestampMode> for InterfaceTimestampMode {
120    fn from(value: GeneralTimestampMode) -> Self {
121        match value {
122            GeneralTimestampMode::SoftwareAll => InterfaceTimestampMode::SoftwareAll,
123            GeneralTimestampMode::SoftwareRecv => InterfaceTimestampMode::SoftwareRecv,
124            GeneralTimestampMode::None => InterfaceTimestampMode::None,
125        }
126    }
127}
128
129#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
130pub struct RecvResult<A> {
131    pub bytes_read: usize,
132    pub remote_addr: A,
133    pub local_addr: A,
134    pub timestamp_data: TimestampData,
135}
136
137#[derive(Debug)]
138pub struct Socket<A, S> {
139    timestamp_mode: InterfaceTimestampMode,
140    // FIXME: Remove the arc once tokio also allows polling asyncfds for the Error interest
141    socket: Arc<AsyncFd<RawSocket>>,
142    #[cfg(target_os = "linux")]
143    send_counter: std::sync::Mutex<u32>,
144    local_addr: A,
145    _state: PhantomData<S>,
146}
147
148#[non_exhaustive]
149pub struct Open;
150#[non_exhaustive]
151pub struct Connected;
152
153#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
154pub struct SendTimestampToken(u32);
155
156impl<A: NetworkAddress, S> Socket<A, S> {
157    pub fn local_addr(&self) -> A {
158        self.local_addr
159    }
160
161    fn inner_recv(&self, buf: &mut [u8], socket: &RawSocket) -> std::io::Result<RecvResult<A>> {
162        let mut control_buf = [0; EXPECTED_MAX_CMSG_SIZE];
163
164        // loops for when we receive an interrupt during the recv
165        let (bytes_read, control_messages, remote_address) =
166            socket.receive_message(buf, &mut control_buf, MessageQueue::Normal)?;
167
168        let mut timestamp_data = TimestampData::empty(self.timestamp_mode);
169        let mut local_addr = self.local_addr;
170
171        // Loops through the control messages, but we should only get a single message
172        // in practice
173        for msg in control_messages {
174            match msg {
175                ControlMessage::Timestamping { software, hardware } => {
176                    tracing::trace!("Timestamps: {:?} {:?}", software, hardware);
177
178                    // Keep the first timestamp of each kind
179                    timestamp_data.software = timestamp_data.software.or(software);
180                    timestamp_data.hardware = timestamp_data.hardware.or(hardware);
181                }
182
183                #[cfg(target_os = "linux")]
184                ControlMessage::ReceiveError(error) => {
185                    tracing::debug!(
186                        "unexpected error control message on receive: {}",
187                        error.ee_errno
188                    );
189                }
190
191                ControlMessage::DestinationIp(addr) => {
192                    if let Some(addr) = A::from_ip_and_port(addr, self.local_addr.port()) {
193                        local_addr = addr;
194                    }
195                }
196
197                ControlMessage::Other(msg) => {
198                    tracing::debug!(
199                        "unexpected control message on receive: {} {}",
200                        msg.cmsg_level,
201                        msg.cmsg_type,
202                    );
203                }
204            }
205        }
206
207        let remote_addr =
208            A::from_sockaddr(remote_address, PrivateToken).ok_or(std::io::ErrorKind::Other)?;
209
210        Ok(RecvResult {
211            bytes_read,
212            remote_addr,
213            local_addr,
214            timestamp_data,
215        })
216    }
217
218    /// Poll to receive a packet on the socket.
219    ///
220    /// Note that on multiple calls to [`poll_recv`](Socket::poll_recv), only
221    /// the [`Waker`](std::task::Waker) from the [`Context`](std::task::Context)
222    /// on the most recent call is scheduled to receive a wakeup.
223    pub fn poll_recv(
224        &self,
225        buf: &mut [u8],
226        cx: &mut std::task::Context<'_>,
227    ) -> std::task::Poll<std::io::Result<RecvResult<A>>> {
228        match self.socket.poll_read_ready(cx) {
229            std::task::Poll::Ready(Ok(mut guard)) => {
230                match guard.try_io(|inner| self.inner_recv(buf, inner.get_ref())) {
231                    Ok(result) => std::task::Poll::Ready(result),
232                    Err(_) => std::task::Poll::Pending,
233                }
234            }
235            std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)),
236            std::task::Poll::Pending => std::task::Poll::Pending,
237        }
238    }
239
240    pub async fn recv(&self, buf: &mut [u8]) -> std::io::Result<RecvResult<A>> {
241        self.socket
242            .async_io(Interest::READABLE, |socket| self.inner_recv(buf, socket))
243            .await
244    }
245
246    fn send_inner(
247        &self,
248        send_call: impl FnOnce() -> std::io::Result<()>,
249    ) -> std::io::Result<Option<SendTimestampToken>> {
250        if matches!(
251            self.timestamp_mode,
252            InterfaceTimestampMode::HardwarePTPAll | InterfaceTimestampMode::SoftwareAll
253        ) {
254            #[cfg(target_os = "linux")]
255            {
256                let mut counter = self.send_counter.lock().unwrap();
257                send_call()?;
258                let token = SendTimestampToken(*counter);
259                *counter = counter.wrapping_add(1);
260                Ok(Some(token))
261            }
262
263            #[cfg(not(target_os = "linux"))]
264            {
265                unreachable!("Should not be able to create send timestamping sockets on platforms other than linux")
266            }
267        } else {
268            send_call()?;
269            Ok(None)
270        }
271    }
272
273    /// Wait for the next send timestamp to be returned.
274    ///
275    /// Note: There is no poll variant for this function, as that is currently
276    /// impossible to implement with the tools tokio provides. To compensate,
277    /// the future returned here is independent of the self reference and can
278    /// be stored to simulate the existence of a poll function.
279    pub fn get_send_timestamp(
280        &self,
281    ) -> impl std::future::Future<Output = std::io::Result<(SendTimestampToken, TimestampData)>>
282           + Send
283           + Sync
284           + 'static {
285        let timestamp_mode = self.timestamp_mode;
286        let socket = self.socket.clone();
287        async move {
288            if matches!(
289                timestamp_mode,
290                InterfaceTimestampMode::HardwarePTPAll | InterfaceTimestampMode::SoftwareAll
291            ) {
292                #[cfg(target_os = "linux")]
293                {
294                    let (counter, timestamp) = Self::fetch_send_timestamp(socket).await?;
295                    Ok((
296                        SendTimestampToken(counter),
297                        timestamp.with_timestamp_mode(timestamp_mode),
298                    ))
299                }
300
301                #[cfg(not(target_os = "linux"))]
302                {
303                    let _ = socket;
304                    unreachable!("Should not be able to create send timestamping sockets on platforms other than linux")
305                }
306            } else {
307                Err(std::io::ErrorKind::Unsupported.into())
308            }
309        }
310    }
311
312    /// Retrieves the timestamp for a given SendTimestampToken.
313    ///
314    /// If multiple tokens are still pending for a send timestamp, this function may
315    /// drop the timestamps for those tokens . Therefore, this should not be used in
316    /// contexts where the socket may also be polled or there may otherwise be multiple
317    /// tokens in use.
318    async fn send_timestamp_for_transmit(
319        &mut self,
320        token: SendTimestampToken,
321    ) -> std::io::Result<TimestampData> {
322        use std::time::Duration;
323
324        const TIMEOUT: Duration = Duration::from_millis(200);
325
326        tokio::time::timeout(TIMEOUT, async {
327            loop {
328                let (cur_token, timestamp_data) = self.get_send_timestamp().await?;
329                if cur_token == token {
330                    return Ok(timestamp_data.with_timestamp_mode(self.timestamp_mode));
331                }
332            }
333        })
334        .await
335        .unwrap_or(Ok(TimestampData::empty(self.timestamp_mode)))
336    }
337}
338
339impl<A: NetworkAddress> Socket<A, Open> {
340    /// Send a packet to a given receiver on the socket.
341    ///
342    /// Note that on multiple calls to `poll_send_*`, only
343    /// the [`Waker`](std::task::Waker) from the [`Context`](std::task::Context)
344    /// on the most recent call is scheduled to receive a wakeup.
345    pub fn poll_send_to(
346        &self,
347        buf: &[u8],
348        addr: A,
349        cx: &mut std::task::Context<'_>,
350    ) -> std::task::Poll<std::io::Result<Option<SendTimestampToken>>> {
351        let addr = addr.to_sockaddr(PrivateToken);
352
353        match self.socket.poll_write_ready(cx) {
354            std::task::Poll::Ready(Ok(mut guard)) => {
355                match guard.try_io(|inner| self.send_inner(|| inner.get_ref().send_to(buf, addr))) {
356                    Ok(result) => std::task::Poll::Ready(result),
357                    Err(_) => std::task::Poll::Pending,
358                }
359            }
360            std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)),
361            std::task::Poll::Pending => std::task::Poll::Pending,
362        }
363    }
364
365    /// Send a packet to a given receiver on the socket.
366    ///
367    /// When used in combination with the polling send functions, if there are timestamps
368    /// pending for some `SendTimestampToken`, these may be skipped and become unavailable
369    /// when calling this function.
370    pub async fn send_to(&mut self, buf: &[u8], addr: A) -> std::io::Result<TimestampData> {
371        let addr = addr.to_sockaddr(PrivateToken);
372
373        if let Some(token) = self
374            .socket
375            .async_io(Interest::WRITABLE, |socket| {
376                self.send_inner(|| socket.send_to(buf, addr))
377            })
378            .await?
379        {
380            self.send_timestamp_for_transmit(token).await
381        } else {
382            Ok(TimestampData::empty(self.timestamp_mode))
383        }
384    }
385
386    /// Send a packet to a given receiver on the socket, using the specified origin address.
387    ///
388    /// Note that on multiple calls to `poll_send_*`, only
389    /// the [`Waker`](std::task::Waker) from the [`Context`](std::task::Context)
390    /// on the most recent call is scheduled to receive a wakeup.
391    pub fn poll_send_from_to(
392        &self,
393        buf: &[u8],
394        from: A,
395        to: A,
396        cx: &mut std::task::Context<'_>,
397    ) -> std::task::Poll<std::io::Result<Option<SendTimestampToken>>> {
398        let from = from.to_sockaddr(PrivateToken);
399        let to = to.to_sockaddr(PrivateToken);
400
401        match self.socket.poll_write_ready(cx) {
402            std::task::Poll::Ready(Ok(mut guard)) => match guard
403                .try_io(|inner| self.send_inner(|| inner.get_ref().send_from_to(buf, from, to)))
404            {
405                Ok(result) => std::task::Poll::Ready(result),
406                Err(_) => std::task::Poll::Pending,
407            },
408            std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)),
409            std::task::Poll::Pending => std::task::Poll::Pending,
410        }
411    }
412
413    /// Send a packet to a given receiver on the socket, using the specified origin address.
414    ///
415    /// When used in combination with the polling send functions, if there are timestamps
416    /// pending for some `SendTimestampToken`, these may be skipped and become unavailable
417    /// when calling this function.
418    pub async fn send_from_to(
419        &mut self,
420        buf: &[u8],
421        from: A,
422        to: A,
423    ) -> std::io::Result<TimestampData> {
424        let from = from.to_sockaddr(PrivateToken);
425        let to = to.to_sockaddr(PrivateToken);
426
427        if let Some(token) = self
428            .socket
429            .async_io(Interest::WRITABLE, |socket| {
430                self.send_inner(|| socket.send_from_to(buf, from, to))
431            })
432            .await?
433        {
434            self.send_timestamp_for_transmit(token).await
435        } else {
436            Ok(TimestampData::empty(self.timestamp_mode))
437        }
438    }
439
440    pub fn connect(self, addr: A) -> std::io::Result<Socket<A, Connected>> {
441        let addr = addr.to_sockaddr(PrivateToken);
442        self.socket.get_ref().connect(addr)?;
443        Ok(Socket {
444            timestamp_mode: self.timestamp_mode,
445            socket: self.socket,
446            #[cfg(target_os = "linux")]
447            send_counter: self.send_counter,
448            local_addr: self.local_addr,
449            _state: PhantomData,
450        })
451    }
452}
453
454impl<A: NetworkAddress> Socket<A, Connected> {
455    pub fn peer_addr(&self) -> std::io::Result<A> {
456        let addr = self.socket.get_ref().getpeername()?;
457        A::from_sockaddr(addr, PrivateToken).ok_or_else(|| std::io::ErrorKind::Other.into())
458    }
459
460    /// Send a packet on the socket.
461    ///
462    /// Note that on multiple calls to `poll_send_*`, only
463    /// the [`Waker`](std::task::Waker) from the [`Context`](std::task::Context)
464    /// on the most recent call is scheduled to receive a wakeup.
465    pub fn poll_send(
466        &self,
467        buf: &[u8],
468        cx: &mut std::task::Context<'_>,
469    ) -> std::task::Poll<std::io::Result<Option<SendTimestampToken>>> {
470        match self.socket.poll_write_ready(cx) {
471            std::task::Poll::Ready(Ok(mut guard)) => {
472                match guard.try_io(|inner| self.send_inner(|| inner.get_ref().send(buf))) {
473                    Ok(result) => std::task::Poll::Ready(result),
474                    Err(_) => std::task::Poll::Pending,
475                }
476            }
477            std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)),
478            std::task::Poll::Pending => std::task::Poll::Pending,
479        }
480    }
481
482    /// Send a packet on the socket.
483    ///
484    /// When used in combination with the polling send functions, if there are timestamps
485    /// pending for some `SendTimestampToken`, these may be skipped and become unavailable
486    /// when calling this function.
487    pub async fn send(&mut self, buf: &[u8]) -> std::io::Result<TimestampData> {
488        if let Some(token) = self
489            .socket
490            .async_io(Interest::WRITABLE, |socket| {
491                self.send_inner(|| socket.send(buf))
492            })
493            .await?
494        {
495            self.send_timestamp_for_transmit(token).await
496        } else {
497            Ok(TimestampData::empty(self.timestamp_mode))
498        }
499    }
500
501    /// Send a packet on the socket.
502    ///
503    /// Note that on multiple calls to `poll_send_*`, only
504    /// the [`Waker`](std::task::Waker) from the [`Context`](std::task::Context)
505    /// on the most recent call is scheduled to receive a wakeup.
506    pub fn poll_send_from(
507        &self,
508        buf: &[u8],
509        from: A,
510        cx: &mut std::task::Context<'_>,
511    ) -> std::task::Poll<std::io::Result<Option<SendTimestampToken>>> {
512        let from = from.to_sockaddr(PrivateToken);
513
514        match self.socket.poll_write_ready(cx) {
515            std::task::Poll::Ready(Ok(mut guard)) => match guard
516                .try_io(|inner| self.send_inner(|| inner.get_ref().send_from(buf, from)))
517            {
518                Ok(result) => std::task::Poll::Ready(result),
519                Err(_) => std::task::Poll::Pending,
520            },
521            std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)),
522            std::task::Poll::Pending => std::task::Poll::Pending,
523        }
524    }
525
526    /// Send a packet on the socket, with the given local address.
527    ///
528    /// When used in combination with the polling send functions, if there are timestamps
529    /// pending for some `SendTimestampToken`, these may be skipped and become unavailable
530    /// when calling this function.
531    pub async fn send_from(&mut self, buf: &[u8], from: A) -> std::io::Result<TimestampData> {
532        let from = from.to_sockaddr(PrivateToken);
533
534        if let Some(token) = self
535            .socket
536            .async_io(Interest::WRITABLE, |socket| {
537                self.send_inner(|| socket.send_from(buf, from))
538            })
539            .await?
540        {
541            self.send_timestamp_for_transmit(token).await
542        } else {
543            Ok(TimestampData::empty(self.timestamp_mode))
544        }
545    }
546}
547
548impl<A: MulticastJoinable, S> Socket<A, S> {
549    pub fn join_multicast(&self, addr: A, interface: InterfaceName) -> std::io::Result<()> {
550        addr.join_multicast(self.socket.get_ref().as_raw_fd(), interface, PrivateToken)
551    }
552
553    pub fn leave_multicast(&self, addr: A, interface: InterfaceName) -> std::io::Result<()> {
554        addr.leave_multicast(self.socket.get_ref().as_raw_fd(), interface, PrivateToken)
555    }
556}
557
558pub fn open_ip(
559    addr: SocketAddr,
560    timestamping: GeneralTimestampMode,
561    #[cfg_attr(not(target_os = "linux"), expect(unused))] reuse_addr: bool,
562) -> std::io::Result<Socket<SocketAddr, Open>> {
563    // Setup the socket
564    let socket = match addr {
565        SocketAddr::V4(_) => RawSocket::open(libc::PF_INET, libc::SOCK_DGRAM, libc::IPPROTO_UDP),
566        SocketAddr::V6(_) => RawSocket::open(libc::PF_INET6, libc::SOCK_DGRAM, libc::IPPROTO_UDP),
567    }?;
568    match addr {
569        SocketAddr::V4(_) => socket.enable_destination_ipv4()?,
570        SocketAddr::V6(_) => socket.enable_destination_ipv6()?,
571    }
572    #[cfg(target_os = "linux")]
573    if reuse_addr {
574        socket.reuse_addr()?;
575    }
576    socket.bind(addr.to_sockaddr(PrivateToken))?;
577    socket.set_nonblocking(true)?;
578    configure_timestamping(&socket, None, timestamping.into(), None)?;
579
580    let local_addr = SocketAddr::from_sockaddr(socket.getsockname()?, PrivateToken)
581        .ok_or::<std::io::Error>(std::io::ErrorKind::Other.into())?;
582
583    Ok(Socket {
584        timestamp_mode: timestamping.into(),
585        socket: Arc::new(AsyncFd::new(socket)?),
586        #[cfg(target_os = "linux")]
587        send_counter: std::sync::Mutex::new(0),
588        local_addr,
589        _state: PhantomData,
590    })
591}
592
593pub fn open_ipv4(
594    addr: SocketAddrV4,
595    timestamping: GeneralTimestampMode,
596    #[cfg_attr(not(target_os = "linux"), expect(unused))] reuse_addr: bool,
597) -> std::io::Result<Socket<SocketAddrV4, Open>> {
598    // Setup the socket
599    let socket = RawSocket::open(libc::PF_INET, libc::SOCK_DGRAM, libc::IPPROTO_UDP)?;
600    socket.enable_destination_ipv4()?;
601    #[cfg(target_os = "linux")]
602    if reuse_addr {
603        socket.reuse_addr()?;
604    }
605    socket.bind(addr.to_sockaddr(PrivateToken))?;
606    socket.set_nonblocking(true)?;
607    configure_timestamping(&socket, None, timestamping.into(), None)?;
608
609    let local_addr = SocketAddrV4::from_sockaddr(socket.getsockname()?, PrivateToken)
610        .ok_or::<std::io::Error>(std::io::ErrorKind::Other.into())?;
611
612    Ok(Socket {
613        timestamp_mode: timestamping.into(),
614        socket: Arc::new(AsyncFd::new(socket)?),
615        #[cfg(target_os = "linux")]
616        send_counter: std::sync::Mutex::new(0),
617        local_addr,
618        _state: PhantomData,
619    })
620}
621
622pub fn open_ipv6(
623    addr: SocketAddrV6,
624    timestamping: GeneralTimestampMode,
625    #[cfg_attr(not(target_os = "linux"), expect(unused))] reuse_addr: bool,
626) -> std::io::Result<Socket<SocketAddrV6, Open>> {
627    // Setup the socket
628    let socket = RawSocket::open(libc::PF_INET6, libc::SOCK_DGRAM, libc::IPPROTO_UDP)?;
629    socket.ipv6_only()?;
630    socket.enable_destination_ipv6()?;
631    #[cfg(target_os = "linux")]
632    if reuse_addr {
633        socket.reuse_addr()?;
634    }
635    socket.bind(addr.to_sockaddr(PrivateToken))?;
636    socket.set_nonblocking(true)?;
637    configure_timestamping(&socket, None, timestamping.into(), None)?;
638
639    let local_addr = SocketAddrV6::from_sockaddr(socket.getsockname()?, PrivateToken)
640        .ok_or::<std::io::Error>(std::io::ErrorKind::Other.into())?;
641
642    Ok(Socket {
643        timestamp_mode: timestamping.into(),
644        socket: Arc::new(AsyncFd::new(socket)?),
645        #[cfg(target_os = "linux")]
646        send_counter: std::sync::Mutex::new(0),
647        local_addr,
648        _state: PhantomData,
649    })
650}
651
652pub fn connect_address(
653    addr: SocketAddr,
654    timestamping: GeneralTimestampMode,
655) -> std::io::Result<Socket<SocketAddr, Connected>> {
656    // Setup the socket
657    let socket = match addr {
658        SocketAddr::V4(_) => RawSocket::open(libc::PF_INET, libc::SOCK_DGRAM, libc::IPPROTO_UDP),
659        SocketAddr::V6(_) => RawSocket::open(libc::PF_INET6, libc::SOCK_DGRAM, libc::IPPROTO_UDP),
660    }?;
661    match addr {
662        SocketAddr::V4(_) => socket.enable_destination_ipv4()?,
663        SocketAddr::V6(_) => socket.enable_destination_ipv6()?,
664    }
665    socket.connect(addr.to_sockaddr(PrivateToken))?;
666    socket.set_nonblocking(true)?;
667    configure_timestamping(&socket, None, timestamping.into(), None)?;
668
669    let local_addr = SocketAddr::from_sockaddr(socket.getsockname()?, PrivateToken)
670        .ok_or::<std::io::Error>(std::io::ErrorKind::Other.into())?;
671
672    Ok(Socket {
673        timestamp_mode: timestamping.into(),
674        socket: Arc::new(AsyncFd::new(socket)?),
675        #[cfg(target_os = "linux")]
676        send_counter: std::sync::Mutex::new(0),
677        local_addr,
678        _state: PhantomData,
679    })
680}
681
682#[cfg(test)]
683mod tests {
684    use super::*;
685    use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
686
687    #[tokio::test]
688    async fn test_open_ip() {
689        let mut a = open_ip(
690            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5125),
691            GeneralTimestampMode::None,
692            false,
693        )
694        .unwrap();
695        let mut b = connect_address(
696            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5125),
697            GeneralTimestampMode::None,
698        )
699        .unwrap();
700        assert!(b.send(&[1, 2, 3]).await.is_ok());
701        let mut buf = [0; 4];
702        let recv_result = a.recv(&mut buf).await.unwrap();
703        assert_eq!(recv_result.bytes_read, 3);
704        assert_eq!(&buf[0..3], &[1, 2, 3]);
705        assert!(a.send_to(&[4, 5, 6], recv_result.remote_addr).await.is_ok());
706        let recv_result = b.recv(&mut buf).await.unwrap();
707        assert_eq!(recv_result.bytes_read, 3);
708        assert_eq!(&buf[0..3], &[4, 5, 6]);
709    }
710
711    #[tokio::test]
712    async fn test_open_ip_dest_addr() {
713        let a = open_ip(
714            SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 5127),
715            GeneralTimestampMode::None,
716            false,
717        )
718        .unwrap();
719        let mut b = connect_address(
720            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5127),
721            GeneralTimestampMode::None,
722        )
723        .unwrap();
724        assert!(b.send(&[1, 2, 3]).await.is_ok());
725        let mut buf = [0; 4];
726        let recv_result = a.recv(&mut buf).await.unwrap();
727        assert_eq!(recv_result.bytes_read, 3);
728        assert_eq!(&buf[0..3], &[1, 2, 3]);
729        assert_eq!(
730            recv_result.local_addr,
731            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5127)
732        );
733        assert_ne!(a.local_addr().ip(), IpAddr::V4(Ipv4Addr::LOCALHOST));
734
735        let a = open_ip(
736            SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 5129),
737            GeneralTimestampMode::None,
738            false,
739        )
740        .unwrap();
741        let mut b = connect_address(
742            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 5129),
743            GeneralTimestampMode::None,
744        )
745        .unwrap();
746        assert!(b.send(&[1, 2, 3]).await.is_ok());
747        let mut buf = [0; 4];
748        let recv_result = a.recv(&mut buf).await.unwrap();
749        assert_eq!(recv_result.bytes_read, 3);
750        assert_eq!(&buf[0..3], &[1, 2, 3]);
751        assert_eq!(
752            recv_result.local_addr,
753            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 5129)
754        );
755        assert_ne!(a.local_addr().ip(), IpAddr::V6(Ipv6Addr::LOCALHOST));
756    }
757
758    #[tokio::test]
759    async fn test_send_from() {
760        let mut a = open_ip(
761            SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 5130),
762            GeneralTimestampMode::None,
763            false,
764        )
765        .unwrap();
766        let mut b = connect_address(
767            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5130),
768            GeneralTimestampMode::None,
769        )
770        .unwrap();
771        b.send_from(
772            &[1, 2, 3],
773            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0),
774        )
775        .await
776        .unwrap();
777        let mut buf = [0; 4];
778        let recv_result = a.recv(&mut buf).await.unwrap();
779        assert_eq!(recv_result.bytes_read, 3);
780        assert_eq!(&buf[0..3], &[1, 2, 3]);
781        assert_eq!(
782            recv_result.remote_addr.ip(),
783            IpAddr::V4(Ipv4Addr::LOCALHOST)
784        );
785
786        a.send_from_to(
787            &[1, 2, 3],
788            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0),
789            dbg!(b.local_addr()),
790        )
791        .await
792        .unwrap();
793        let mut buf = [0; 4];
794        let recv_result = b.recv(&mut buf).await.unwrap();
795        assert_eq!(recv_result.bytes_read, 3);
796        assert_eq!(&buf[0..3], &[1, 2, 3]);
797        assert_eq!(
798            recv_result.remote_addr.ip(),
799            IpAddr::V4(Ipv4Addr::LOCALHOST)
800        );
801    }
802
803    #[tokio::test]
804    async fn test_send_from_v6() {
805        let mut a = open_ip(
806            SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 5131),
807            GeneralTimestampMode::None,
808            false,
809        )
810        .unwrap();
811        let mut b = connect_address(
812            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 5131),
813            GeneralTimestampMode::None,
814        )
815        .unwrap();
816        b.send_from(
817            &[1, 2, 3],
818            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0),
819        )
820        .await
821        .unwrap();
822        let mut buf = [0; 4];
823        let recv_result = a.recv(&mut buf).await.unwrap();
824        assert_eq!(recv_result.bytes_read, 3);
825        assert_eq!(&buf[0..3], &[1, 2, 3]);
826        assert_eq!(
827            recv_result.remote_addr.ip(),
828            IpAddr::V6(Ipv6Addr::LOCALHOST)
829        );
830
831        a.send_from_to(
832            &[1, 2, 3],
833            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0),
834            dbg!(b.local_addr()),
835        )
836        .await
837        .unwrap();
838        let mut buf = [0; 4];
839        let recv_result = b.recv(&mut buf).await.unwrap();
840        assert_eq!(recv_result.bytes_read, 3);
841        assert_eq!(&buf[0..3], &[1, 2, 3]);
842        assert_eq!(
843            recv_result.remote_addr.ip(),
844            IpAddr::V6(Ipv6Addr::LOCALHOST)
845        );
846    }
847}
timestamped_socket/socket.rs

timestamped_socket/
socket.rs
