dial9_perf_self_profile/
socket_accept_queues.rs

1//! TCP listener accept queue snapshots sampled from Linux sock_diag, exposed as
2//! a dial9 [`Source`](dial9_core::source::Source).
3
4use std::time::Duration;
5
6const DEFAULT_SAMPLE_INTERVAL: Duration = Duration::from_millis(400);
7
8/// Wire-format event for a TCP listener accept queue snapshot.
9#[derive(Debug, dial9_trace_format::TraceEvent)]
10#[traceevent(wire_slot)]
11#[cfg_attr(not(feature = "unstable-events"), non_exhaustive)]
12pub struct TcpAcceptQueueEvent {
13    /// Monotonic timestamp in nanoseconds.
14    #[traceevent(timestamp)]
15    pub timestamp_ns: u64,
16    /// Linux socket cookie reported by sock_diag.
17    pub socket_cookie: u64,
18    /// Linux socket inode reported by sock_diag.
19    pub socket_inode: u64,
20    /// IP version for `local_addr`: 4 or 6.
21    pub ip_version: u8,
22    /// Local listener address.
23    pub local_addr: String,
24    /// Local listener port.
25    pub local_port: u16,
26    /// Completed connections waiting to be accepted.
27    #[traceevent(kind = "gauge")]
28    pub pending_connections: u32,
29    /// Effective accept backlog limit.
30    pub backlog_limit: u32,
31}
32
33/// Configuration for socket accept queue sampling.
34///
35/// Build via `SocketAcceptQueuesConfig::builder()...build()`, then plug the
36/// [`SocketAcceptQueuesSource`] into a dial9 recorder.
37///
38/// # Performance
39///
40/// Full scans can be expensive because they walk `/proc/self/fd` to find this
41/// process's listeners. The cost grows with the number of open file descriptors
42/// in this process, including accepted sockets, open files, pipes, and similar
43/// handles.
44///
45/// To avoid that cost on every sample, this source caches the classification of
46/// TCP listeners visible in the current network namespace. While that listener
47/// set is stable, samples do not need a full file descriptor scan and should be
48/// cheap.
49///
50/// # Reliability
51///
52/// Listeners classified as foreign are cached as foreign. If such a listener is
53/// later transferred into this process with `SCM_RIGHTS`, it will not be tracked
54/// while it keeps the same kernel socket identity.
55#[derive(Debug, Clone, bon::Builder)]
56pub struct SocketAcceptQueuesConfig {
57    /// Minimum time between samples.
58    #[builder(default = DEFAULT_SAMPLE_INTERVAL)]
59    sample_interval: Duration,
60}
61
62impl Default for SocketAcceptQueuesConfig {
63    fn default() -> Self {
64        Self::builder().build()
65    }
66}
67
68impl SocketAcceptQueuesConfig {
69    /// Minimum time between samples.
70    pub fn sample_interval(&self) -> Duration {
71        self.sample_interval
72    }
73}
74
75#[cfg(test)]
76mod tests {
77    use super::*;
78
79    #[test]
80    fn default_sample_interval_is_400ms() {
81        assert_eq!(
82            SocketAcceptQueuesConfig::default().sample_interval(),
83            Duration::from_millis(400)
84        );
85    }
86
87    #[test]
88    fn pending_connections_is_a_gauge() {
89        use dial9_trace_format::TraceEvent;
90
91        let entry = TcpAcceptQueueEvent::schema_entry();
92        let kind_annotations = entry
93            .annotations()
94            .iter()
95            .filter(|annotation| annotation.key() == "kind")
96            .map(|annotation| {
97                (
98                    entry.fields()[annotation.field_index() as usize].name(),
99                    annotation.value(),
100                )
101            })
102            .collect::<Vec<_>>();
103
104        assert_eq!(kind_annotations, vec![("pending_connections", "gauge")]);
105    }
106}
107
108#[cfg(target_os = "linux")]
109mod linux {
110    use super::{SocketAcceptQueuesConfig, TcpAcceptQueueEvent};
111    use dial9_core::clock::clock_monotonic_ns;
112    use dial9_core::rate_limited;
113    use dial9_core::source::{FlushContext, Source};
114    use netlink_packet_core::{
115        NLM_F_DUMP, NLM_F_REQUEST, NetlinkBuffer, NetlinkMessage, NetlinkPayload,
116    };
117    use netlink_packet_sock_diag::inet::{ExtensionFlags, InetRequest, SocketId, StateFlags};
118    use netlink_packet_sock_diag::{AF_INET, AF_INET6, IPPROTO_TCP, SockDiagMessage, TCP_LISTEN};
119    use netlink_sys::{Socket, SocketAddr, protocols::NETLINK_SOCK_DIAG};
120    use std::collections::{HashMap, HashSet};
121    use std::fs;
122    use std::io;
123    use std::net::IpAddr;
124    use std::os::fd::AsRawFd;
125    use std::path::{Path, PathBuf};
126    use std::time::{Duration, Instant};
127
128    const NETLINK_RECEIVE_TIMEOUT: Duration = Duration::from_millis(100);
129
130    /// Flush-thread source that samples TCP listener accept queue depth.
131    ///
132    /// See [issue #501](https://github.com/dial9-rs/dial9/issues/501) for details.
133    #[derive(Debug)]
134    pub struct SocketAcceptQueuesSource {
135        config: SocketAcceptQueuesConfig,
136        last_sample: Option<Instant>,
137        cache: SocketAcceptQueueCache,
138    }
139
140    #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
141    struct SocketKey {
142        inode: u64,
143        cookie: u64,
144    }
145
146    #[derive(Debug, Default)]
147    struct SocketAcceptQueueCache {
148        owned: HashMap<SocketKey, PathBuf>,
149        foreign: HashSet<SocketKey>,
150    }
151
152    #[derive(Debug, Clone, PartialEq, Eq)]
153    struct TcpAcceptQueueSnapshot {
154        key: SocketKey,
155        ip_version: u8,
156        local_addr: IpAddr,
157        local_port: u16,
158        pending_connections: u32,
159        backlog_limit: u32,
160    }
161
162    impl SocketAcceptQueuesSource {
163        pub fn new(config: SocketAcceptQueuesConfig) -> Self {
164            Self {
165                config,
166                last_sample: None,
167                cache: SocketAcceptQueueCache::default(),
168            }
169        }
170    }
171
172    impl SocketAcceptQueueCache {
173        fn prune(&mut self, active_keys: &HashSet<SocketKey>) {
174            self.owned.retain(|key, _| active_keys.contains(key));
175            self.foreign.retain(|key| active_keys.contains(key));
176        }
177    }
178
179    impl Source for SocketAcceptQueuesSource {
180        fn flush(&mut self, ctx: &FlushContext<'_>) {
181            let now = Instant::now();
182            if let Some(last_sample) = self.last_sample
183                && now.duration_since(last_sample) < self.config.sample_interval
184            {
185                return;
186            }
187            self.last_sample = Some(now);
188
189            match collect_socket_accept_queues(&mut self.cache) {
190                Ok(snapshots) => {
191                    let timestamp_ns = clock_monotonic_ns();
192                    for snapshot in snapshots {
193                        ctx.record_event(&snapshot.into_event(timestamp_ns));
194                    }
195                }
196                Err(e) => rate_limited!(Duration::from_secs(60), {
197                    tracing::warn!("failed to read socket accept queues via sock_diag: {e}");
198                }),
199            }
200        }
201
202        fn name(&self) -> &'static str {
203            "socket_accept_queues"
204        }
205    }
206
207    impl TcpAcceptQueueSnapshot {
208        fn into_event(self, timestamp_ns: u64) -> TcpAcceptQueueEvent {
209            TcpAcceptQueueEvent {
210                timestamp_ns,
211                socket_cookie: self.key.cookie,
212                socket_inode: self.key.inode,
213                ip_version: self.ip_version,
214                local_addr: self.local_addr.to_string(),
215                local_port: self.local_port,
216                pending_connections: self.pending_connections,
217                backlog_limit: self.backlog_limit,
218            }
219        }
220    }
221
222    fn collect_socket_accept_queues(
223        cache: &mut SocketAcceptQueueCache,
224    ) -> io::Result<Vec<TcpAcceptQueueSnapshot>> {
225        let mut socket = Socket::new(NETLINK_SOCK_DIAG)?;
226        set_socket_receive_timeout(&socket, NETLINK_RECEIVE_TIMEOUT)?;
227        let _local_addr = socket.bind_auto()?;
228        let kernel_addr = SocketAddr::new(0, 0);
229        socket.connect(&kernel_addr)?;
230
231        let mut snapshots = Vec::new();
232        dump_tcp_listeners(&socket, AF_INET, 1, &mut snapshots)?;
233        dump_tcp_listeners(&socket, AF_INET6, 2, &mut snapshots)?;
234        classify_process_listeners(
235            snapshots,
236            cache,
237            read_socket_inode_for_fd_path,
238            scan_process_socket_fds,
239        )
240    }
241
242    fn dump_tcp_listeners(
243        socket: &Socket,
244        family: u8,
245        sequence_number: u32,
246        snapshots: &mut Vec<TcpAcceptQueueSnapshot>,
247    ) -> io::Result<()> {
248        let socket_id = match family {
249            AF_INET => SocketId::new_v4(),
250            AF_INET6 => SocketId::new_v6(),
251            _ => {
252                return Err(io::Error::new(
253                    io::ErrorKind::InvalidInput,
254                    format!("unsupported inet socket family {family}"),
255                ));
256            }
257        };
258
259        let request = InetRequest {
260            family,
261            protocol: IPPROTO_TCP,
262            extensions: ExtensionFlags::empty(),
263            states: StateFlags::LISTEN,
264            socket_id,
265        };
266        let mut message = NetlinkMessage::from(SockDiagMessage::InetRequest(request));
267        message.header.flags = NLM_F_REQUEST | NLM_F_DUMP;
268        message.header.sequence_number = sequence_number;
269        message.finalize();
270
271        let mut request_bytes = vec![0; message.buffer_len()];
272        message.serialize(&mut request_bytes);
273        let sent = socket.send(&request_bytes, 0)?;
274        if sent != request_bytes.len() {
275            return Err(io::Error::new(
276                io::ErrorKind::WriteZero,
277                format!(
278                    "short netlink sock_diag request write: sent {sent} of {} bytes",
279                    request_bytes.len()
280                ),
281            ));
282        }
283
284        loop {
285            let (response_bytes, _sender) = socket.recv_from_full()?;
286            if response_bytes.is_empty() {
287                return Err(io::Error::new(
288                    io::ErrorKind::UnexpectedEof,
289                    "sock_diag netlink socket returned an empty response",
290                ));
291            }
292
293            let done = parse_response_datagram(&response_bytes, sequence_number, snapshots)?;
294            if done {
295                return Ok(());
296            }
297        }
298    }
299
300    fn parse_response_datagram(
301        bytes: &[u8],
302        sequence_number: u32,
303        snapshots: &mut Vec<TcpAcceptQueueSnapshot>,
304    ) -> io::Result<bool> {
305        let mut offset = 0;
306        let mut done = false;
307
308        while offset < bytes.len() {
309            let remaining = &bytes[offset..];
310            let packet = NetlinkBuffer::new_checked(&remaining).map_err(decode_error)?;
311            let packet_len = packet.length() as usize;
312            if packet_len == 0 {
313                return Err(io::Error::new(
314                    io::ErrorKind::InvalidData,
315                    "sock_diag returned zero-length netlink message",
316                ));
317            }
318            let message = NetlinkMessage::<SockDiagMessage>::deserialize(&remaining[..packet_len])
319                .map_err(decode_error)?;
320
321            if message.header.sequence_number == sequence_number {
322                match message.payload {
323                    NetlinkPayload::Done(_) => done = true,
324                    NetlinkPayload::Error(error) => {
325                        if error.code.is_some() {
326                            return Err(error.into());
327                        }
328                    }
329                    NetlinkPayload::InnerMessage(SockDiagMessage::InetResponse(response)) => {
330                        if let Some(snapshot) = snapshot_from_response(&response) {
331                            snapshots.push(snapshot);
332                        }
333                    }
334                    _ => {}
335                }
336            }
337
338            offset += nlmsg_align(packet_len);
339        }
340
341        Ok(done)
342    }
343
344    fn snapshot_from_response(
345        response: &netlink_packet_sock_diag::inet::InetResponse,
346    ) -> Option<TcpAcceptQueueSnapshot> {
347        let header = &response.header;
348        let socket_inode = u64::from(header.inode);
349        if header.state != TCP_LISTEN {
350            return None;
351        }
352
353        Some(TcpAcceptQueueSnapshot {
354            key: SocketKey {
355                inode: socket_inode,
356                cookie: socket_cookie_from_diag(header.socket_id.cookie),
357            },
358            ip_version: match header.family {
359                AF_INET => 4,
360                AF_INET6 => 6,
361                _ => return None,
362            },
363            local_addr: header.socket_id.source_address,
364            local_port: header.socket_id.source_port,
365            pending_connections: header.recv_queue,
366            backlog_limit: header.send_queue,
367        })
368    }
369
370    fn classify_process_listeners(
371        snapshots: Vec<TcpAcceptQueueSnapshot>,
372        cache: &mut SocketAcceptQueueCache,
373        mut read_fd_inode: impl FnMut(&Path) -> io::Result<Option<u64>>,
374        mut scan_socket_fds: impl FnMut(&HashSet<u64>) -> io::Result<HashMap<u64, PathBuf>>,
375    ) -> io::Result<Vec<TcpAcceptQueueSnapshot>> {
376        // The expensive operation is walking /proc/self/fd. Do it only when
377        // sock_diag reports a listener we have not classified before.
378        let active_keys = snapshots
379            .iter()
380            .map(|snapshot| snapshot.key)
381            .collect::<HashSet<_>>();
382        cache.prune(&active_keys);
383
384        let mut owned_snapshots = Vec::new();
385        let mut unknown_snapshots = Vec::new();
386        let mut unknown_inodes = HashSet::new();
387
388        for snapshot in snapshots {
389            let key = snapshot.key;
390
391            if let Some(fd_path) = cache.owned.get(&key) {
392                if read_fd_inode(fd_path)? == Some(key.inode) {
393                    owned_snapshots.push(snapshot);
394                    continue;
395                }
396                cache.owned.remove(&key);
397            }
398
399            if cache.foreign.contains(&key) {
400                continue;
401            }
402
403            unknown_inodes.insert(key.inode);
404            unknown_snapshots.push(snapshot);
405        }
406
407        if unknown_snapshots.is_empty() {
408            return Ok(owned_snapshots);
409        }
410
411        let owned_fds_by_inode = scan_socket_fds(&unknown_inodes)?;
412        for snapshot in unknown_snapshots {
413            if let Some(fd_path) = owned_fds_by_inode.get(&snapshot.key.inode) {
414                cache.owned.insert(snapshot.key, fd_path.clone());
415                owned_snapshots.push(snapshot);
416            } else {
417                cache.foreign.insert(snapshot.key);
418            }
419        }
420
421        Ok(owned_snapshots)
422    }
423
424    fn set_socket_receive_timeout(socket: &Socket, timeout: Duration) -> io::Result<()> {
425        let timeout = libc::timeval {
426            tv_sec: timeout.as_secs().try_into().map_err(|_| {
427                io::Error::new(
428                    io::ErrorKind::InvalidInput,
429                    format!("socket receive timeout {timeout:?} exceeds timeval.tv_sec"),
430                )
431            })?,
432            tv_usec: timeout.subsec_micros().into(),
433        };
434        let optlen: libc::socklen_t = std::mem::size_of_val(&timeout).try_into().map_err(|_| {
435            io::Error::new(
436                io::ErrorKind::InvalidInput,
437                "timeval size exceeds libc::socklen_t",
438            )
439        })?;
440
441        // SAFETY: `socket.as_raw_fd()` is a live netlink socket owned by `socket`,
442        // `timeout` points to a properly initialized `timeval`, and the length
443        // matches the pointed-to value for the duration of the syscall.
444        let result = unsafe {
445            libc::setsockopt(
446                socket.as_raw_fd(),
447                libc::SOL_SOCKET,
448                libc::SO_RCVTIMEO,
449                &timeout as *const libc::timeval as *const libc::c_void,
450                optlen,
451            )
452        };
453        if result == -1 {
454            return Err(io::Error::last_os_error());
455        }
456        Ok(())
457    }
458
459    fn scan_process_socket_fds(target_inodes: &HashSet<u64>) -> io::Result<HashMap<u64, PathBuf>> {
460        let mut fds_by_inode = HashMap::new();
461        if target_inodes.is_empty() {
462            return Ok(fds_by_inode);
463        }
464
465        for entry in fs::read_dir("/proc/self/fd")? {
466            let entry = entry?;
467            let path = entry.path();
468            if let Some(inode) = read_socket_inode_for_fd_path(&path)?
469                && target_inodes.contains(&inode)
470            {
471                fds_by_inode.entry(inode).or_insert(path);
472                if fds_by_inode.len() == target_inodes.len() {
473                    break;
474                }
475            }
476        }
477        Ok(fds_by_inode)
478    }
479
480    fn read_socket_inode_for_fd_path(path: &Path) -> io::Result<Option<u64>> {
481        match fs::read_link(path) {
482            Ok(target) => parse_socket_inode(&target),
483            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
484            Err(e) => Err(e),
485        }
486    }
487
488    fn parse_socket_inode(path: &Path) -> io::Result<Option<u64>> {
489        let Some(text) = path.to_str() else {
490            return Ok(None);
491        };
492        let Some(inode) = text
493            .strip_prefix("socket:[")
494            .and_then(|value| value.strip_suffix(']'))
495        else {
496            return Ok(None);
497        };
498
499        inode.parse::<u64>().map(Some).map_err(|e| {
500            io::Error::new(
501                io::ErrorKind::InvalidData,
502                format!("invalid /proc/self/fd socket inode {inode:?}: {e}"),
503            )
504        })
505    }
506
507    fn socket_cookie_from_diag(bytes: [u8; 8]) -> u64 {
508        let low = u32::from_ne_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as u64;
509        let high = u32::from_ne_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]) as u64;
510        (high << 32) | low
511    }
512
513    fn decode_error(error: impl std::fmt::Display) -> io::Error {
514        io::Error::new(io::ErrorKind::InvalidData, error.to_string())
515    }
516
517    fn nlmsg_align(len: usize) -> usize {
518        (len + 3) & !3
519    }
520
521    #[cfg(test)]
522    mod tests {
523        use super::*;
524        use std::cell::Cell;
525        use std::collections::HashMap;
526        use std::path::PathBuf;
527
528        #[test]
529        fn parse_socket_inode_extracts_socket_symlink_inode() {
530            assert_eq!(
531                parse_socket_inode(Path::new("socket:[12345]")).unwrap(),
532                Some(12345)
533            );
534        }
535
536        #[test]
537        fn parse_socket_inode_ignores_non_socket_symlink() {
538            assert_eq!(parse_socket_inode(Path::new("/dev/null")).unwrap(), None);
539        }
540
541        #[test]
542        fn cached_owned_listener_does_not_scan_process_fds_again() {
543            let key = SocketKey {
544                inode: 123,
545                cookie: 456,
546            };
547            let snapshot = snapshot_with_key(key);
548            let fd_path = PathBuf::from("/proc/self/fd/7");
549            let mut cache = SocketAcceptQueueCache::default();
550            cache.owned.insert(key, fd_path.clone());
551            let scans = Cell::new(0);
552
553            let snapshots = classify_process_listeners(
554                vec![snapshot.clone()],
555                &mut cache,
556                |path| {
557                    assert_eq!(path, fd_path.as_path());
558                    Ok(Some(key.inode))
559                },
560                |_| {
561                    scans.set(scans.get() + 1);
562                    Ok(HashMap::new())
563                },
564            )
565            .unwrap();
566
567            assert_eq!(snapshots, vec![snapshot]);
568            assert_eq!(scans.get(), 0);
569        }
570
571        #[test]
572        fn unknown_listener_scans_once_and_caches_owned_listener() {
573            let key = SocketKey {
574                inode: 123,
575                cookie: 456,
576            };
577            let snapshot = snapshot_with_key(key);
578            let fd_path = PathBuf::from("/proc/self/fd/7");
579            let mut cache = SocketAcceptQueueCache::default();
580            let scans = Cell::new(0);
581
582            let snapshots = classify_process_listeners(
583                vec![snapshot.clone()],
584                &mut cache,
585                |_| panic!("unknown listeners should not validate cached fd paths"),
586                |target_inodes| {
587                    scans.set(scans.get() + 1);
588                    assert!(target_inodes.contains(&key.inode));
589                    Ok(HashMap::from([(key.inode, fd_path.clone())]))
590                },
591            )
592            .unwrap();
593
594            assert_eq!(snapshots, vec![snapshot]);
595            assert_eq!(cache.owned.get(&key), Some(&fd_path));
596            assert_eq!(scans.get(), 1);
597        }
598
599        #[test]
600        fn cached_foreign_listener_does_not_rescan() {
601            let key = SocketKey {
602                inode: 123,
603                cookie: 456,
604            };
605            let mut cache = SocketAcceptQueueCache::default();
606            cache.foreign.insert(key);
607            let scans = Cell::new(0);
608
609            let snapshots = classify_process_listeners(
610                vec![snapshot_with_key(key)],
611                &mut cache,
612                |_| panic!("foreign listeners should not validate fd paths"),
613                |_| {
614                    scans.set(scans.get() + 1);
615                    Ok(HashMap::new())
616                },
617            )
618            .unwrap();
619
620            assert!(snapshots.is_empty());
621            assert_eq!(scans.get(), 0);
622        }
623
624        #[test]
625        fn inactive_cache_entries_are_pruned_before_classification() {
626            let old_key = SocketKey {
627                inode: 123,
628                cookie: 456,
629            };
630            let new_key = SocketKey {
631                inode: 123,
632                cookie: 789,
633            };
634            let fd_path = PathBuf::from("/proc/self/fd/7");
635            let mut cache = SocketAcceptQueueCache::default();
636            cache.foreign.insert(old_key);
637
638            let snapshots = classify_process_listeners(
639                vec![snapshot_with_key(new_key)],
640                &mut cache,
641                |_| panic!("new listener should be unknown, not cached"),
642                |_| Ok(HashMap::from([(new_key.inode, fd_path.clone())])),
643            )
644            .unwrap();
645
646            assert_eq!(snapshots, vec![snapshot_with_key(new_key)]);
647            assert!(!cache.foreign.contains(&old_key));
648            assert_eq!(cache.owned.get(&new_key), Some(&fd_path));
649        }
650
651        #[test]
652        fn socket_cookie_from_diag_combines_kernel_cookie_words() {
653            let low = 0x5566_7788_u32.to_ne_bytes();
654            let high = 0x1122_3344_u32.to_ne_bytes();
655
656            assert_eq!(
657                socket_cookie_from_diag([
658                    low[0], low[1], low[2], low[3], high[0], high[1], high[2], high[3],
659                ]),
660                0x1122_3344_5566_7788
661            );
662        }
663
664        fn snapshot_with_key(key: SocketKey) -> TcpAcceptQueueSnapshot {
665            TcpAcceptQueueSnapshot {
666                key,
667                ip_version: 4,
668                local_addr: "127.0.0.1".parse().unwrap(),
669                local_port: 8080,
670                pending_connections: 1,
671                backlog_limit: 128,
672            }
673        }
674    }
675}
676
677#[cfg(target_os = "linux")]
678pub use linux::SocketAcceptQueuesSource;
dial9_perf_self_profile/socket_accept_queues.rs

dial9_perf_self_profile/
socket_accept_queues.rs
