timestamped_socket/socket/
linux.rs

1use std::{
2    marker::PhantomData,
3    net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6},
4    sync::Arc,
5};
6
7use tokio::io::{unix::AsyncFd, Interest};
8
9use crate::{
10    control_message::{ControlMessage, MessageQueue, EXPECTED_MAX_CMSG_SIZE},
11    interface::{lookup_phc, InterfaceName},
12    networkaddress::{sealed::PrivateToken, EthernetAddress, MacAddress, NetworkAddress},
13    raw_socket::RawSocket,
14    socket::TimestampData,
15};
16
17use super::{InterfaceTimestampMode, Open, Socket};
18
19const SOF_TIMESTAMPING_BIND_PHC: libc::c_uint = 1 << 15;
20
21impl<A: NetworkAddress, S> Socket<A, S> {
22    pub(super) async fn fetch_send_timestamp(
23        socket: Arc<AsyncFd<RawSocket>>,
24    ) -> std::io::Result<(u32, TimestampData)> {
25        let try_read = |socket: &RawSocket| fetch_send_timestamp_try_read(socket);
26
27        loop {
28            // the timestamp being available triggers the error interest
29            if let Some((counter, timestamp_data)) =
30                socket.async_io(Interest::ERROR, try_read).await?
31            {
32                break Ok((counter, timestamp_data));
33            }
34        }
35    }
36}
37
38/// This function tries to fetch a send timestamp from a single error queue message.
39///
40/// We assume that we get error queue messages for exactly two reasons:
41///  - When the driver has pushed a timestamp set up, in which timestamps will be present
42///  - When something has gone wrong in the send process after the return of the send*
43///    family of functions (such as an ICMP error).
44///
45/// In particular, we assume that the second scenario occurs independent of whether the first
46/// occurs, and therefore we fully ignore those messages, only logging them but never
47/// returning the message index, to avoid upper layers of interpreting the error as a marker
48/// that the timestamp is definitively unavailable.
49///
50/// Note that this means that we don't expect, and therefore don't report, any signal from
51/// the kernel that the message has been sent but there won't be timestamps available. This
52/// scenario is handled in the higher layers through timeouts.
53///
54/// The above are assumptions for us as the kernel does not clearly document precisely how
55/// this works, and we haven't had the capacity to do a full deep dive into the kernel source.
56fn fetch_send_timestamp_try_read(
57    socket: &RawSocket,
58) -> std::io::Result<Option<(u32, TimestampData)>> {
59    let mut control_buf = [0; EXPECTED_MAX_CMSG_SIZE];
60
61    // NOTE: this read could block!
62    let (_, control_messages, _) =
63        socket.receive_message(&mut [], &mut control_buf, MessageQueue::Error)?;
64
65    let mut send_ts = None;
66    let mut counter = None;
67    for msg in control_messages {
68        match msg {
69            ControlMessage::Timestamping { software, hardware } => {
70                send_ts = Some(TimestampData {
71                    software,
72                    hardware,
73                    timestamp_mode: InterfaceTimestampMode::default(),
74                });
75            }
76
77            ControlMessage::ReceiveError(error) => {
78                // the timestamping does not set a message; if there is a message, that means
79                // something else is wrong, and we want to know about it.
80                #[expect(
81                    clippy::cast_possible_wrap,
82                    reason = "The error constants can have a different type from the ee_errno field."
83                )]
84                if error.ee_errno as libc::c_int != libc::ENOMSG {
85                    tracing::debug!(error.ee_data, "error message on the MSG_ERRQUEUE");
86                }
87
88                counter = Some(error.ee_data);
89            }
90
91            ControlMessage::DestinationIp(_) => {
92                tracing::debug!("unexpected destination ip control message");
93            }
94
95            ControlMessage::Other(msg) => {
96                tracing::debug!(
97                    msg.cmsg_level,
98                    msg.cmsg_type,
99                    "unexpected message on the MSG_ERRQUEUE",
100                );
101            }
102        }
103    }
104
105    Ok(counter.zip(send_ts))
106}
107
108pub(super) fn configure_timestamping(
109    socket: &RawSocket,
110    interface: Option<InterfaceName>,
111    mode: InterfaceTimestampMode,
112    mut bind_phc: Option<u32>,
113) -> std::io::Result<()> {
114    // Check if the phc is not the interface-native phc.
115    if let Some(interface) = interface {
116        if lookup_phc(interface) == bind_phc {
117            bind_phc = None;
118        }
119    }
120
121    let options = match mode {
122        InterfaceTimestampMode::HardwareAll | InterfaceTimestampMode::HardwarePTPAll => {
123            libc::SOF_TIMESTAMPING_RAW_HARDWARE
124                | libc::SOF_TIMESTAMPING_RX_SOFTWARE
125                | libc::SOF_TIMESTAMPING_TX_SOFTWARE
126                | libc::SOF_TIMESTAMPING_RX_HARDWARE
127                | libc::SOF_TIMESTAMPING_TX_HARDWARE
128                | libc::SOF_TIMESTAMPING_OPT_TSONLY
129                | libc::SOF_TIMESTAMPING_OPT_ID
130                | bind_phc
131                    .map(|_| SOF_TIMESTAMPING_BIND_PHC)
132                    .unwrap_or_default()
133        }
134        InterfaceTimestampMode::HardwareRecv | InterfaceTimestampMode::HardwarePTPRecv => {
135            libc::SOF_TIMESTAMPING_RAW_HARDWARE
136                | libc::SOF_TIMESTAMPING_RX_SOFTWARE
137                | libc::SOF_TIMESTAMPING_RX_HARDWARE
138                | bind_phc
139                    .map(|_| SOF_TIMESTAMPING_BIND_PHC)
140                    .unwrap_or_default()
141        }
142        InterfaceTimestampMode::SoftwareAll => {
143            libc::SOF_TIMESTAMPING_SOFTWARE
144                | libc::SOF_TIMESTAMPING_RX_SOFTWARE
145                | libc::SOF_TIMESTAMPING_TX_SOFTWARE
146                | libc::SOF_TIMESTAMPING_OPT_TSONLY
147                | libc::SOF_TIMESTAMPING_OPT_ID
148        }
149        InterfaceTimestampMode::SoftwareRecv => {
150            libc::SOF_TIMESTAMPING_SOFTWARE | libc::SOF_TIMESTAMPING_RX_SOFTWARE
151        }
152        InterfaceTimestampMode::None => return Ok(()),
153    };
154
155    socket.so_timestamping(options, bind_phc.unwrap_or_default())
156}
157
158/// Open a network interface for udp traffic over both ipv4 and ipv6
159///
160/// # Errors
161/// May fail if the interface is unknown or cannot be opened.
162pub fn open_interface_udp(
163    interface: InterfaceName,
164    port: u16,
165    timestamping: InterfaceTimestampMode,
166    bind_phc: Option<u32>,
167) -> std::io::Result<Socket<SocketAddr, Open>> {
168    // Setup the socket
169    let socket = RawSocket::open(libc::PF_INET6, libc::SOCK_DGRAM, libc::IPPROTO_UDP)?;
170    socket.enable_destination_ipv4()?;
171    socket.enable_destination_ipv6()?;
172    socket.reuse_addr()?;
173    socket.ipv6_v6only(false)?;
174    socket.bind(SocketAddrV6::new(Ipv6Addr::UNSPECIFIED, port, 0, 0).to_sockaddr(PrivateToken))?;
175    socket.bind_to_device(interface)?;
176    socket.ipv6_multicast_if(interface)?;
177    socket.ipv6_multicast_loop(false)?;
178    configure_timestamping(&socket, Some(interface), timestamping, bind_phc)?;
179    match timestamping {
180        InterfaceTimestampMode::HardwareAll | InterfaceTimestampMode::HardwareRecv => {
181            socket.driver_enable_hardware_timestamping(
182                interface,
183                libc::HWTSTAMP_FILTER_ALL.cast_signed(),
184            )?;
185        }
186        InterfaceTimestampMode::HardwarePTPAll | InterfaceTimestampMode::HardwarePTPRecv => socket
187            .driver_enable_hardware_timestamping(
188                interface,
189                libc::HWTSTAMP_FILTER_PTP_V2_L4_EVENT.cast_signed(),
190            )?,
191        InterfaceTimestampMode::None
192        | InterfaceTimestampMode::SoftwareAll
193        | InterfaceTimestampMode::SoftwareRecv => {}
194    }
195    socket.set_nonblocking(true)?;
196
197    let local_addr = SocketAddr::from_sockaddr(socket.getsockname()?, PrivateToken)
198        .ok_or::<std::io::Error>(std::io::ErrorKind::Other.into())?;
199
200    // Safety: We own the file descriptor and so can transfer it, and it is
201    // guaranteed to be open by the earlier error checking.
202    Ok(Socket {
203        timestamp_mode: timestamping,
204        raw: Arc::new(unsafe { AsyncFd::register(socket) }?),
205        send_counter: std::sync::Mutex::new(0),
206        local_addr,
207        _state: PhantomData,
208    })
209}
210
211/// Open a network interface for udp traffic over ipv4
212///
213/// # Errors
214/// May fail if the interface is unknown or cannot be opened.
215pub fn open_interface_udp4(
216    interface: InterfaceName,
217    port: u16,
218    timestamping: InterfaceTimestampMode,
219    bind_phc: Option<u32>,
220) -> std::io::Result<Socket<SocketAddrV4, Open>> {
221    // Setup the socket
222    let socket = RawSocket::open(libc::PF_INET, libc::SOCK_DGRAM, libc::IPPROTO_UDP)?;
223    socket.enable_destination_ipv4()?;
224    socket.reuse_addr()?;
225    socket.bind(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, port).to_sockaddr(PrivateToken))?;
226    socket.bind_to_device(interface)?;
227    socket.ip_multicast_if(interface)?;
228    socket.ip_multicast_loop(false)?;
229    configure_timestamping(&socket, Some(interface), timestamping, bind_phc)?;
230    match timestamping {
231        InterfaceTimestampMode::HardwareAll | InterfaceTimestampMode::HardwareRecv => {
232            socket.driver_enable_hardware_timestamping(
233                interface,
234                libc::HWTSTAMP_FILTER_ALL.cast_signed(),
235            )?;
236        }
237        InterfaceTimestampMode::HardwarePTPAll | InterfaceTimestampMode::HardwarePTPRecv => socket
238            .driver_enable_hardware_timestamping(
239                interface,
240                libc::HWTSTAMP_FILTER_PTP_V2_L4_EVENT.cast_signed(),
241            )?,
242        InterfaceTimestampMode::None
243        | InterfaceTimestampMode::SoftwareAll
244        | InterfaceTimestampMode::SoftwareRecv => {}
245    }
246    socket.set_nonblocking(true)?;
247
248    let local_addr = SocketAddrV4::from_sockaddr(socket.getsockname()?, PrivateToken)
249        .ok_or::<std::io::Error>(std::io::ErrorKind::Other.into())?;
250
251    // Safety: We own the file descriptor and so can transfer it, and it is
252    // guaranteed to be open by the earlier error checking.
253    Ok(Socket {
254        timestamp_mode: timestamping,
255        raw: Arc::new(unsafe { AsyncFd::register(socket) }?),
256        send_counter: std::sync::Mutex::new(0),
257        local_addr,
258        _state: PhantomData,
259    })
260}
261
262/// Open a network interface for udp traffic over ipv6
263///
264/// # Errors
265/// May fail if the interface is unknown or cannot be opened.
266pub fn open_interface_udp6(
267    interface: InterfaceName,
268    port: u16,
269    timestamping: InterfaceTimestampMode,
270    bind_phc: Option<u32>,
271) -> std::io::Result<Socket<SocketAddrV6, Open>> {
272    // Setup the socket
273    let socket = RawSocket::open(libc::PF_INET6, libc::SOCK_DGRAM, libc::IPPROTO_UDP)?;
274    socket.enable_destination_ipv6()?;
275    socket.reuse_addr()?;
276    socket.ipv6_v6only(true)?;
277    socket.bind(SocketAddrV6::new(Ipv6Addr::UNSPECIFIED, port, 0, 0).to_sockaddr(PrivateToken))?;
278    socket.bind_to_device(interface)?;
279    socket.ipv6_multicast_if(interface)?;
280    socket.ipv6_multicast_loop(false)?;
281    configure_timestamping(&socket, Some(interface), timestamping, bind_phc)?;
282    match timestamping {
283        InterfaceTimestampMode::HardwareAll | InterfaceTimestampMode::HardwareRecv => {
284            socket.driver_enable_hardware_timestamping(
285                interface,
286                libc::HWTSTAMP_FILTER_ALL.cast_signed(),
287            )?;
288        }
289        InterfaceTimestampMode::HardwarePTPAll | InterfaceTimestampMode::HardwarePTPRecv => socket
290            .driver_enable_hardware_timestamping(
291                interface,
292                libc::HWTSTAMP_FILTER_PTP_V2_L4_EVENT.cast_signed(),
293            )?,
294        InterfaceTimestampMode::None
295        | InterfaceTimestampMode::SoftwareAll
296        | InterfaceTimestampMode::SoftwareRecv => {}
297    }
298    socket.set_nonblocking(true)?;
299
300    let local_addr = SocketAddrV6::from_sockaddr(socket.getsockname()?, PrivateToken)
301        .ok_or::<std::io::Error>(std::io::ErrorKind::Other.into())?;
302
303    // Safety: We own the file descriptor and so can transfer it, and it is
304    // guaranteed to be open by the earlier error checking.
305    Ok(Socket {
306        timestamp_mode: timestamping,
307        raw: Arc::new(unsafe { AsyncFd::register(socket) }?),
308        send_counter: std::sync::Mutex::new(0),
309        local_addr,
310        _state: PhantomData,
311    })
312}
313
314/// Open a network interface for ethernet traffic.
315///
316/// # Errors
317/// May error if the interface is not known, or cannot be opened.
318pub fn open_interface_ethernet(
319    interface: InterfaceName,
320    protocol: u16,
321    timestamping: InterfaceTimestampMode,
322    bind_phc: Option<u32>,
323) -> std::io::Result<Socket<EthernetAddress, Open>> {
324    let socket = RawSocket::open(
325        libc::AF_PACKET,
326        libc::SOCK_DGRAM,
327        u16::from_ne_bytes(protocol.to_be_bytes()).into(),
328    )?;
329    socket.bind(
330        EthernetAddress::new(
331            u16::from_ne_bytes(protocol.to_le_bytes()),
332            MacAddress::new([0; 6]),
333            interface
334                .get_index()
335                .ok_or(std::io::ErrorKind::InvalidInput)?
336                .cast_signed(),
337        )
338        .to_sockaddr(PrivateToken),
339    )?;
340    configure_timestamping(&socket, Some(interface), timestamping, bind_phc)?;
341    match timestamping {
342        InterfaceTimestampMode::HardwareAll | InterfaceTimestampMode::HardwareRecv => {
343            socket.driver_enable_hardware_timestamping(
344                interface,
345                libc::HWTSTAMP_FILTER_ALL.cast_signed(),
346            )?;
347        }
348        InterfaceTimestampMode::HardwarePTPAll | InterfaceTimestampMode::HardwarePTPRecv => socket
349            .driver_enable_hardware_timestamping(
350                interface,
351                libc::HWTSTAMP_FILTER_PTP_V2_L2_EVENT.cast_signed(),
352            )?,
353        InterfaceTimestampMode::None
354        | InterfaceTimestampMode::SoftwareAll
355        | InterfaceTimestampMode::SoftwareRecv => {}
356    }
357    socket.set_nonblocking(true)?;
358
359    let local_addr = EthernetAddress::from_sockaddr(socket.getsockname()?, PrivateToken)
360        .ok_or::<std::io::Error>(std::io::ErrorKind::Other.into())?;
361
362    // Safety: We own the file descriptor and so can transfer it, and it is
363    // guaranteed to be open by the earlier error checking.
364    Ok(Socket {
365        timestamp_mode: timestamping,
366        raw: Arc::new(unsafe { AsyncFd::register(socket) }?),
367        send_counter: std::sync::Mutex::new(0),
368        local_addr,
369        _state: PhantomData,
370    })
371}
372
373#[cfg(test)]
374mod tests {
375    use std::net::IpAddr;
376
377    use statime_base::Timestamp;
378
379    use crate::socket::{connect_address, open_ip, GeneralTimestampMode};
380
381    use super::*;
382
383    #[tokio::test]
384    async fn test_open_udp() {
385        use std::str::FromStr;
386        let a = open_interface_udp(
387            InterfaceName::from_str("lo").unwrap(),
388            5128,
389            super::InterfaceTimestampMode::None,
390            None,
391        )
392        .unwrap();
393
394        let mut b = connect_address(
395            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 5128),
396            GeneralTimestampMode::None,
397        )
398        .unwrap();
399        assert!(b.send(&[1, 2, 3]).await.is_ok());
400        let mut buf = [0; 4];
401        let recv_result = a.recv(&mut buf).await.unwrap();
402        assert_eq!(recv_result.bytes_read, 3);
403        assert_eq!(&buf[0..3], &[1, 2, 3]);
404        assert_eq!(
405            recv_result.local_addr,
406            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 5128)
407        );
408
409        let mut b = connect_address(
410            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 1)), 5128),
411            GeneralTimestampMode::None,
412        )
413        .unwrap();
414        assert!(b.send(&[1, 2, 3]).await.is_ok());
415        let mut buf = [0; 4];
416        let recv_result = a.recv(&mut buf).await.unwrap();
417        assert_eq!(recv_result.bytes_read, 3);
418        assert_eq!(&buf[0..3], &[1, 2, 3]);
419        assert_eq!(
420            recv_result.local_addr,
421            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 1)), 5128)
422        );
423    }
424
425    #[tokio::test]
426    async fn test_open_ip_reuse_addr_after_interface() {
427        use std::str::FromStr;
428        let _a = open_interface_udp(
429            InterfaceName::from_str("lo").unwrap(),
430            5132,
431            super::InterfaceTimestampMode::None,
432            None,
433        )
434        .unwrap();
435        let _b = open_ip(
436            SocketAddr::new(Ipv4Addr::UNSPECIFIED.into(), 5132),
437            GeneralTimestampMode::None,
438            true,
439        )
440        .unwrap();
441    }
442
443    #[tokio::test]
444    async fn test_open_ip_reuse_addr_before_interface() {
445        use std::str::FromStr;
446        let _a = open_ip(
447            SocketAddr::new(Ipv4Addr::UNSPECIFIED.into(), 5133),
448            GeneralTimestampMode::None,
449            true,
450        )
451        .unwrap();
452        let _b = open_interface_udp(
453            InterfaceName::from_str("lo").unwrap(),
454            5133,
455            super::InterfaceTimestampMode::None,
456            None,
457        )
458        .unwrap();
459    }
460
461    #[tokio::test]
462    async fn test_open_udp6() {
463        use std::str::FromStr;
464        let mut a = open_interface_udp6(
465            InterfaceName::from_str("lo").unwrap(),
466            5123,
467            super::InterfaceTimestampMode::None,
468            None,
469        )
470        .unwrap();
471        let mut b = connect_address(
472            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 5123),
473            GeneralTimestampMode::None,
474        )
475        .unwrap();
476        assert!(b.send(&[1, 2, 3]).await.is_ok());
477        let mut buf = [0; 4];
478        let recv_result = a.recv(&mut buf).await.unwrap();
479        assert_eq!(recv_result.bytes_read, 3);
480        assert_eq!(&buf[0..3], &[1, 2, 3]);
481        assert_eq!(
482            recv_result.local_addr,
483            SocketAddrV6::new(Ipv6Addr::LOCALHOST, 5123, 0, 0)
484        );
485        assert!(a.send_to(&[4, 5, 6], recv_result.remote_addr).await.is_ok());
486        let recv_result = b.recv(&mut buf).await.unwrap();
487        assert_eq!(recv_result.bytes_read, 3);
488        assert_eq!(&buf[0..3], &[4, 5, 6]);
489    }
490
491    #[tokio::test]
492    async fn test_open_udp4() {
493        use std::str::FromStr;
494        let mut a = open_interface_udp4(
495            InterfaceName::from_str("lo").unwrap(),
496            5124,
497            super::InterfaceTimestampMode::None,
498            None,
499        )
500        .unwrap();
501        let mut b = connect_address(
502            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5124),
503            GeneralTimestampMode::None,
504        )
505        .unwrap();
506        assert!(b.send(&[1, 2, 3]).await.is_ok());
507        let mut buf = [0; 4];
508        let recv_result = a.recv(&mut buf).await.unwrap();
509        assert_eq!(recv_result.bytes_read, 3);
510        assert_eq!(&buf[0..3], &[1, 2, 3]);
511        assert_eq!(
512            recv_result.local_addr,
513            SocketAddrV4::new(Ipv4Addr::LOCALHOST, 5124)
514        );
515        assert!(a.send_to(&[4, 5, 6], recv_result.remote_addr).await.is_ok());
516        let recv_result = b.recv(&mut buf).await.unwrap();
517        assert_eq!(recv_result.bytes_read, 3);
518        assert_eq!(&buf[0..3], &[4, 5, 6]);
519    }
520
521    #[tokio::test]
522    async fn test_software_timestamping() {
523        use std::time::SystemTime;
524
525        let a = open_ip(
526            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5126),
527            GeneralTimestampMode::SoftwareAll,
528            false,
529        )
530        .unwrap();
531        let mut b = connect_address(
532            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5126),
533            GeneralTimestampMode::SoftwareAll,
534        )
535        .unwrap();
536
537        let before = SystemTime::now();
538        let send_ts = b
539            .send(&[1, 2, 3])
540            .await
541            .unwrap()
542            .selected_timestamp()
543            .unwrap();
544        let after = SystemTime::now();
545
546        let mut buf = [0; 4];
547        let recv_result = a.recv(&mut buf).await.unwrap();
548        let recv_ts = recv_result.timestamp_data.selected_timestamp().unwrap();
549
550        let before = before
551            .duration_since(SystemTime::UNIX_EPOCH)
552            .unwrap()
553            .as_secs_f64();
554        let after = after
555            .duration_since(SystemTime::UNIX_EPOCH)
556            .unwrap()
557            .as_secs_f64();
558        assert!(((send_ts.as_utc(37) - Timestamp::UNIX_EPOCH).as_seconds() - before).abs() < 2.0);
559        assert!(((send_ts.as_utc(37) - Timestamp::UNIX_EPOCH).as_seconds() - after).abs() < 2.0);
560
561        assert!((send_ts.as_utc(37) - recv_ts.as_utc(37)).as_nanos().abs() < 10_000_000);
562    }
563}
timestamped_socket/socket/linux.rs

timestamped_socket/socket/
linux.rs
