[DNM] macos: read the command line of each process

src/macos.rs

@@ -115,6 +115,7 @@ pub fn record(
         collect(&mut session, &mut collector, &mut |message| {
             tracker.handle(message)
         })?;
+        collector.tick(&mut |message| tracker.handle(message))?;
         if let Some(ended) = ended
             && ended.elapsed() > DRAIN
         {
@@ -138,6 +139,13 @@ pub fn record(
         let _ = session.wait(READ_INTERVAL.as_millis() as u64);
     }
     collector.flush(&mut |message| tracker.handle(message))?;
+    if collector.unnamed > 0 {
+        note!(
+            "{} processes ended before their command line could be read; they are named after \
+             their program alone",
+            collector.unnamed
+        );
+    }
     if collector.dropped > 0 {
         note!(
             "The kernel dropped events {} times while the build ran; the trace may be missing \

src/macos/event.rs

@@ -20,8 +20,9 @@ pub enum Event {
     Fork {
         child: i32,
     },
-    /// The program a process is now running.
+    /// What a process is now running.
     Exec {
+        argv: Vec<String>,
         executable: String,
     },
     /// A `wait(2)` status.

src/macos/kdebug.rs

@@ -12,9 +12,11 @@
 //! `fs_usage(1)` and `ktrace(1)`. It can change between releases.
 
 use std::io;
+use std::time::{Duration, SystemTime};
 
 const CTL_KERN: i32 = 1;
 const KERN_KDEBUG: i32 = 24;
+const KERN_PROCARGS2: i32 = 49;
 
 const KERN_KDENABLE: i32 = 3;
 const KERN_KDSETBUF: i32 = 4;
@@ -206,6 +208,64 @@ fn describe(error: io::Error) -> io::Error {
     }
 }
 
+/// The command line of a running process, as `(executable, argv)`.
+///
+/// Fails with `EINVAL` while a process is still setting up its new image after
+/// `exec`, and once it has exited, so callers retry briefly.
+pub fn command_line(pid: i32, buffer: &mut Vec<u8>) -> io::Result<(String, Vec<String>)> {
+    let mut length = buffer.len();
+    sysctl(
+        &mut [CTL_KERN, KERN_PROCARGS2, pid],
+        buffer.as_mut_ptr().cast(),
+        &mut length,
+    )?;
+    let data = &buffer[..length.min(buffer.len())];
+    if data.len() < 4 {
+        return Err(io::Error::new(io::ErrorKind::InvalidData, "short procargs"));
+    }
+    // [argc][executable path][padding][argv][envp]
+    let count = i32::from_ne_bytes(data[0..4].try_into().expect("4 bytes")).max(0) as usize;
+    let rest = &data[4..];
+    let executable_end = rest
+        .iter()
+        .position(|byte| *byte == 0)
+        .unwrap_or(rest.len());
+    let executable = String::from_utf8_lossy(&rest[..executable_end]).into_owned();
+    let rest = &rest[executable_end..];
+    let argv_start = rest
+        .iter()
+        .position(|byte| *byte != 0)
+        .unwrap_or(rest.len());
+    let argv = rest[argv_start..]
+        .split(|byte| *byte == 0)
+        .take(count)
+        .map(|arg| String::from_utf8_lossy(arg).into_owned())
+        .collect();
+    Ok((executable, argv))
+}
+
+/// When a process started, which identifies it apart from its pid: pids are
+/// reused, and a build can burn through thousands of them.
+pub fn start_time(pid: i32) -> io::Result<SystemTime> {
+    let mut info: libc::proc_bsdinfo = unsafe { std::mem::zeroed() };
+    let size = size_of::<libc::proc_bsdinfo>() as i32;
+    // SAFETY: `info` is a valid, correctly sized destination for the call.
+    let written = unsafe {
+        libc::proc_pidinfo(
+            pid,
+            libc::PROC_PIDTBSDINFO,
+            0,
+            std::ptr::from_mut(&mut info).cast(),
+            size,
+        )
+    };
+    if written != size {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(SystemTime::UNIX_EPOCH
+        + Duration::new(info.pbi_start_tvsec, info.pbi_start_tvusec as u32 * 1_000))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -224,4 +284,30 @@ mod tests {
         assert_eq!(event.code(), 0x0401_0004);
         assert!(event.is_end());
     }
+
+    #[test]
+    fn reads_own_command_line() {
+        let mut buffer = vec![0_u8; 1 << 16];
+        let (executable, argv) =
+            command_line(std::process::id() as i32, &mut buffer).expect("own argv");
+        assert!(
+            executable.contains("kdebug") || executable.contains("buildprof"),
+            "{executable}"
+        );
+        assert!(!argv.is_empty());
+    }
+
+    #[test]
+    fn own_start_time_is_in_the_past() {
+        let started = start_time(std::process::id() as i32).expect("own start time");
+        assert!(started <= SystemTime::now());
+        assert!(start_time(-1).is_err());
+    }
+
+    #[test]
+    fn missing_process_is_an_error() {
+        let mut buffer = vec![0_u8; 1 << 16];
+        // Pid 0 is the kernel, which has no argv to read.
+        assert!(command_line(0, &mut buffer).is_err());
+    }
 }

src/macos/recon.rs

@@ -5,13 +5,15 @@
 //!
 //! The kernel reports every process on the machine, so this follows the ones
 //! the build started: a process is part of the build if the process that
-//! created it was.
+//! created it was. Command lines are not in the trace at all, and are read from
+//! the process itself as soon as its execution appears.
 
 use super::event::{Event, Message};
 use super::kdebug;
 use super::tracker::Clock;
 use std::collections::{HashMap, HashSet};
 use std::io;
+use std::time::Duration;
 
 const NEWTHREAD: u32 = 0x0700_0004;
 const DATA_EXEC: u32 = 0x0700_0008;
@@ -23,6 +25,12 @@ const PROC_EXIT: u32 = 0x0401_0004;
 /// also reports an exit when a process replaces its image, and the new image
 /// then goes on doing things; anything further from the pid cancels the exit.
 const EXIT_HOLD_NS: u64 = 10_000_000;
+/// Attempts to read a command line before giving up on it. A read lands
+/// before the new image is ready surprisingly often.
+const ARGV_ATTEMPTS: u32 = 5;
+/// How much later than its exec a process may claim to have started before it
+/// is taken to be a different process that reused the pid.
+const REUSE_TOLERANCE: Duration = Duration::from_millis(2);
 
 struct HeldExit {
     pid: i32,
@@ -36,12 +44,17 @@ pub struct Collector {
     clock: Clock,
     processes: HashSet<i32>,
     threads: HashMap<u64, i32>,
-    /// Executions seen but not yet reported: the kernel names the new program
-    /// in a second event.
-    starting: HashMap<i32, u64>,
+    /// The program each process is running, for the ones whose command line
+    /// could not be read.
+    names: HashMap<i32, String>,
     held_exits: Vec<HeldExit>,
+    argv_retries: Vec<(i32, u64, u32)>,
+    argv_buffer: Vec<u8>,
     /// Events the kernel dropped because the buffer filled.
     pub dropped: u64,
+    /// Processes whose command line was gone before it could be read, and which
+    /// are named after their program alone.
+    pub unnamed: u64,
 }
 
 impl Collector {
@@ -51,9 +64,12 @@ impl Collector {
             clock,
             processes: HashSet::from([root]),
             threads: HashMap::new(),
-            starting: HashMap::new(),
+            names: HashMap::new(),
             held_exits: Vec::new(),
+            argv_retries: Vec::new(),
+            argv_buffer: vec![0; 1 << 18],
             dropped: 0,
+            unnamed: 0,
         }
     }
 
@@ -114,21 +130,10 @@ impl Collector {
         }
 
         match code {
-            DATA_EXEC => {
-                self.starting.insert(pid, event.timestamp);
-            }
-            // The program's name follows the execution that started it.
+            DATA_EXEC => return self.exec(pid, event.timestamp, out),
+            // The kernel names the new program right after the execution.
             STRING_EXEC => {
-                if let Some(mach_time) = self.starting.remove(&pid) {
-                    out(Message {
-                        mach_time,
-                        pid,
-                        ppid: 0,
-                        event: Event::Exec {
-                            executable: decode_string(&args),
-                        },
-                    })?;
-                }
+                self.names.insert(pid, decode_string(&args));
             }
             PROC_EXIT if event.is_end() => {
                 self.held_exits.push(HeldExit {
@@ -143,12 +148,17 @@ impl Collector {
         Ok(())
     }
 
+    /// Follows up the command lines that were not ready yet.
+    pub fn tick(&mut self, out: &mut impl FnMut(Message) -> io::Result<()>) -> io::Result<()> {
+        self.retry_command_lines(out, false)
+    }
+
     /// Emits everything still held back, at the end of a recording.
     pub fn flush(&mut self, out: &mut impl FnMut(Message) -> io::Result<()>) -> io::Result<()> {
         for exit in std::mem::take(&mut self.held_exits) {
             self.emit_exit(exit, out)?;
         }
-        Ok(())
+        self.retry_command_lines(out, true)
     }
 
     /// Pids the build started that have not been seen to exit.
@@ -160,6 +170,84 @@ impl Collector {
         (nanoseconds as f64 * self.clock.ticks_per_ns()) as u64
     }
 
+    /// Reads a command line, refusing one from a process that started after the
+    /// exec: that pid now belongs to someone else.
+    fn command_line(&mut self, pid: i32, mach_time: u64) -> io::Result<(String, Vec<String>)> {
+        let started = kdebug::start_time(pid)?;
+        if started > self.clock.wall(mach_time) + REUSE_TOLERANCE {
+            return Err(io::Error::new(
+                io::ErrorKind::NotFound,
+                "the pid was reused before its command line could be read",
+            ));
+        }
+        kdebug::command_line(pid, &mut self.argv_buffer)
+    }
+
+    fn exec(
+        &mut self,
+        pid: i32,
+        mach_time: u64,
+        out: &mut impl FnMut(Message) -> io::Result<()>,
+    ) -> io::Result<()> {
+        match self.command_line(pid, mach_time) {
+            Ok((executable, argv)) => self.emit_exec(pid, mach_time, executable, argv, out),
+            // Either the image is still being set up, or the process is already
+            // gone. Come back to it: the kernel names the program in the event
+            // after this one, so even giving up reads better later.
+            Err(_) => {
+                self.argv_retries.push((pid, mach_time, 1));
+                Ok(())
+            }
+        }
+    }
+
+    fn retry_command_lines(
+        &mut self,
+        out: &mut impl FnMut(Message) -> io::Result<()>,
+        last_chance: bool,
+    ) -> io::Result<()> {
+        for (pid, mach_time, attempt) in std::mem::take(&mut self.argv_retries) {
+            match self.command_line(pid, mach_time) {
+                Ok((executable, argv)) => self.emit_exec(pid, mach_time, executable, argv, out)?,
+                Err(_) if attempt < ARGV_ATTEMPTS && !last_chance => {
+                    self.argv_retries.push((pid, mach_time, attempt + 1));
+                }
+                // Gone before its command line could be read.
+                Err(_) => self.emit_named_exec(pid, mach_time, out)?,
+            }
+        }
+        Ok(())
+    }
+
+    /// Reports an execution the kernel named but whose arguments are lost.
+    fn emit_named_exec(
+        &mut self,
+        pid: i32,
+        mach_time: u64,
+        out: &mut impl FnMut(Message) -> io::Result<()>,
+    ) -> io::Result<()> {
+        self.unnamed += 1;
+        let executable = self.names.get(&pid).cloned().unwrap_or_default();
+        let argv = vec![executable.clone()];
+        self.emit_exec(pid, mach_time, executable, argv, out)
+    }
+
+    fn emit_exec(
+        &mut self,
+        pid: i32,
+        mach_time: u64,
+        executable: String,
+        argv: Vec<String>,
+        out: &mut impl FnMut(Message) -> io::Result<()>,
+    ) -> io::Result<()> {
+        out(Message {
+            mach_time,
+            pid,
+            ppid: 0,
+            event: Event::Exec { argv, executable },
+        })
+    }
+
     fn cancel_exit(&mut self, pid: i32) {
         self.held_exits.retain(|exit| exit.pid != pid);
     }
@@ -188,7 +276,7 @@ impl Collector {
         if exit.pid != self.root {
             self.processes.remove(&exit.pid);
         }
-        self.starting.remove(&exit.pid);
+        self.names.remove(&exit.pid);
         out(Message {
             mach_time: exit.mach_time,
             pid: exit.pid,
@@ -277,28 +365,32 @@ mod tests {
             origin: 0,
             numer: 1,
             denom: 1,
+            origin_wall: std::time::SystemTime::now(),
         };
         let mut collector = Collector::new(100, clock);
         collector.seed_threads([(10, 100)]);
         collector
     }
 
     #[test]
-    fn reports_a_process_and_what_it_runs() {
+    fn an_execution_whose_command_line_is_gone_is_named_after_its_program() {
         let mut stream = Stream::new();
         stream.exec(10, 100, "make");
-        let messages = stream.collect(&mut collector());
+        let mut collector = collector();
+        let messages = stream.collect(&mut collector);
         assert_eq!(
             messages,
             [Message {
                 mach_time: 1_010,
                 pid: 100,
                 ppid: 0,
                 event: Event::Exec {
+                    argv: vec!["make".into()],
                     executable: "make".into(),
                 },
             }]
         );
+        assert_eq!(collector.unnamed, 1);
     }
 
     #[test]
@@ -312,7 +404,7 @@ mod tests {
         assert_eq!(messages[0].pid, 100);
         assert!(matches!(
             &messages[1].event,
-            Event::Exec { executable } if executable == "cc"
+            Event::Exec { executable, .. } if executable == "cc"
         ));
         assert_eq!(messages[1].pid, 200);
     }