perf(fspy): report accesses over pipe sockets

Cargo.toml

@@ -78,7 +78,6 @@ fspy_detours_sys = { path = "crates/fspy_detours_sys" }
 fspy_preload_unix = { path = "crates/fspy_preload_unix", artifact = "cdylib", target = "target" }
 fspy_preload_windows = { path = "crates/fspy_preload_windows", artifact = "cdylib", target = "target" }
 fspy_seccomp_unotify = { path = "crates/fspy_seccomp_unotify" }
-fspy_shm = { path = "crates/fspy_shm" }
 fspy_shared = { path = "crates/fspy_shared" }
 fspy_shared_unix = { path = "crates/fspy_shared_unix" }
 futures = "0.3.31"

crates/fspy/Cargo.toml

@@ -15,6 +15,7 @@ fspy_shared = { workspace = true }
 futures-util = { workspace = true }
 libc = { workspace = true }
 ouroboros = { workspace = true }
+pipe_socket = { workspace = true }
 rustc-hash = { workspace = true }
 tempfile = { workspace = true }
 thiserror = { workspace = true }

crates/fspy/src/error.rs

@@ -16,8 +16,8 @@ pub enum SpawnError {
     #[error("failed to initialize seccomp_unotify supervisor: {0}")]
     Supervisor(std::io::Error),
 
-    #[error("failed to create IPC channel: {0}")]
-    ChannelCreation(std::io::Error),
+    #[error("failed to create IPC server: {0}")]
+    IpcServer(std::io::Error),
 
     /// On unix systems, the injection happens before the spawn actually occurs on.
     /// On Windows, the injection happens after the spawn but before resuming the process.

crates/fspy/src/ipc.rs

@@ -1,38 +1,184 @@
 use std::io;
 
-use fspy_shared::ipc::{
-    PathAccess,
-    channel::{Receiver, ReceiverLockGuard},
+use fspy_shared::ipc::{NativeStr, PathAccess};
+use pipe_socket::{Server, ServerConnection};
+use tokio::{
+    io::{AsyncReadExt as _, BufReader},
+    task::{JoinHandle, JoinSet},
 };
-use tokio::task::spawn_blocking;
+use tokio_util::sync::CancellationToken;
 
-// Shared memory size for storing path accesses.
-// 4 GiB is large enough to store path accesses in almost any realistic scenario.
-// This doesn't allocate physical memory until it's actually used.
-pub const SHM_CAPACITY: usize = 4 * 1024 * 1024 * 1024;
+use crate::arena::PathAccessArena;
 
-#[ouroboros::self_referencing]
-pub struct OwnedReceiverLockGuard {
-    /// Owns the shared memory
-    receiver: Receiver,
-    /// Borrows the shared memory and owns the file lock
-    #[borrows(receiver)]
-    #[covariant]
-    lock_guard: ReceiverLockGuard<'this>,
+const FRAME_HEADER_LEN: usize = size_of::<u32>();
+
+pub struct IpcSupervisor {
+    server_name: Box<NativeStr>,
+    cancellation_token: CancellationToken,
+    task: Option<JoinHandle<io::Result<Vec<PathAccessArena>>>>,
+}
+
+impl IpcSupervisor {
+    pub fn bind() -> io::Result<Self> {
+        let server = Server::bind()?;
+        let server_name = server.name().into();
+        let cancellation_token = CancellationToken::new();
+        let task = tokio::spawn(run_server(server, cancellation_token.clone()));
+        Ok(Self { server_name, cancellation_token, task: Some(task) })
+    }
+
+    pub fn server_name(&self) -> &NativeStr {
+        &self.server_name
+    }
+
+    pub async fn stop(mut self) -> io::Result<Vec<PathAccessArena>> {
+        self.cancellation_token.cancel();
+        self.task.take().expect("IPC supervisor task is missing").await.map_err(io::Error::other)?
+    }
+}
+
+impl Drop for IpcSupervisor {
+    fn drop(&mut self) {
+        self.cancellation_token.cancel();
+    }
+}
+
+async fn run_server(
+    mut server: Server,
+    cancellation_token: CancellationToken,
+) -> io::Result<Vec<PathAccessArena>> {
+    let mut readers = JoinSet::new();
+    let mut arenas = Vec::new();
+
+    loop {
+        tokio::select! {
+            biased;
+            () = cancellation_token.cancelled() => break,
+            result = readers.join_next(), if !readers.is_empty() => {
+                collect_reader(result.expect("reader set is not empty"), &mut arenas)?;
+            }
+            connection = server.accept() => {
+                readers.spawn(read_connection(connection?));
+            }
+        }
+    }
+
+    // Dropping the server prevents any new clients from connecting. Existing
+    // connections remain alive in their reader tasks and are drained to EOF.
+    drop(server);
+    while let Some(result) = readers.join_next().await {
+        collect_reader(result, &mut arenas)?;
+    }
+    Ok(arenas)
 }
 
-impl OwnedReceiverLockGuard {
-    pub fn lock(receiver: Receiver) -> io::Result<Self> {
-        Self::try_new(receiver, fspy_shared::ipc::channel::Receiver::lock)
+fn collect_reader(
+    result: Result<io::Result<PathAccessArena>, tokio::task::JoinError>,
+    arenas: &mut Vec<PathAccessArena>,
+) -> io::Result<()> {
+    arenas.push(result.map_err(io::Error::other)??);
+    Ok(())
+}
+
+async fn read_connection(connection: ServerConnection) -> io::Result<PathAccessArena> {
+    let mut connection = BufReader::new(connection);
+    let mut arena = PathAccessArena::default();
+    let mut frame = Vec::new();
+    let mut header = [0; FRAME_HEADER_LEN];
+
+    loop {
+        if let Err(error) = connection.read_exact(&mut header).await {
+            if connection_closed(&error) {
+                return Ok(arena);
+            }
+            return Err(error);
+        }
+
+        let frame_len = u32::from_le_bytes(header) as usize;
+        frame.resize(frame_len, 0);
+        if let Err(error) = connection.read_exact(&mut frame).await {
+            if connection_closed(&error) {
+                return Ok(arena);
+            }
+            return Err(error);
+        }
+
+        let access: PathAccess<'_> = wincode::deserialize_exact(&frame)
+            .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
+        arena.add(access);
+    }
+}
+
+fn connection_closed(error: &io::Error) -> bool {
+    matches!(
+        error.kind(),
+        io::ErrorKind::UnexpectedEof
+            | io::ErrorKind::BrokenPipe
+            | io::ErrorKind::ConnectionAborted
+            | io::ErrorKind::ConnectionReset
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{sync::mpsc, time::Duration};
+
+    use fspy_shared::ipc::{AccessMode, NativePath, PathAccessSender};
+
+    use super::*;
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn stop_rejects_new_connections_and_waits_for_existing_ones() {
+        let supervisor = IpcSupervisor::bind().unwrap();
+        let server_name = supervisor.server_name().to_cow_os_str().into_owned();
+        let client_server_name = server_name.clone();
+        let (connected_tx, connected_rx) = tokio::sync::oneshot::channel();
+        let (close_tx, close_rx) = mpsc::channel();
+
+        let client = tokio::task::spawn_blocking(move || {
+            let mut sender = PathAccessSender::connect(&client_server_name).unwrap();
+            sender.send(PathAccess { mode: AccessMode::READ, path: test_path() }).unwrap();
+            connected_tx.send(()).unwrap();
+            close_rx.recv().unwrap();
+        });
+        connected_rx.await.unwrap();
+
+        let mut stop = tokio::spawn(supervisor.stop());
+        assert!(tokio::time::timeout(Duration::from_millis(50), &mut stop).await.is_err());
+        let rejected = tokio::task::spawn_blocking(move || PathAccessSender::connect(&server_name))
+            .await
+            .unwrap();
+        assert!(rejected.is_err());
+
+        close_tx.send(()).unwrap();
+        client.await.unwrap();
+        let arenas = stop.await.unwrap().unwrap();
+        let accesses =
+            arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).collect::<Vec<_>>();
+        assert_eq!(accesses.len(), 1);
+        assert_eq!(accesses[0].mode, AccessMode::READ);
+        assert_eq!(accesses[0].path, test_path());
     }
 
-    pub async fn lock_async(receiver: Receiver) -> io::Result<Self> {
-        spawn_blocking(move || Self::lock(receiver)).await.expect("lock task panicked")
+    #[cfg(unix)]
+    fn test_path() -> &'static NativePath {
+        std::path::Path::new("/fspy-pipe-socket-test").into()
     }
 
-    pub fn iter_path_accesses(&self) -> impl Iterator<Item = PathAccess<'_>> {
-        self.borrow_lock_guard()
-            .iter_frames()
-            .map(|frame| wincode::deserialize_exact(frame).unwrap())
+    #[cfg(windows)]
+    fn test_path() -> &'static NativePath {
+        NativePath::from_wide(&[
+            b'\\' as u16,
+            b'?' as u16,
+            b'?' as u16,
+            b'\\' as u16,
+            b'C' as u16,
+            b':' as u16,
+            b'\\' as u16,
+            b't' as u16,
+            b'e' as u16,
+            b's' as u16,
+            b't' as u16,
+        ])
     }
 }

crates/fspy/src/lib.rs

@@ -13,7 +13,6 @@ mod os_impl;
 #[path = "./windows/mod.rs"]
 mod os_impl;
 
-#[cfg(unix)]
 mod arena;
 mod command;
 

crates/fspy/src/unix/mod.rs

@@ -8,9 +8,9 @@ use std::{io, path::Path};
 
 #[cfg(target_os = "linux")]
 use fspy_seccomp_unotify::supervisor::supervise;
-use fspy_shared::ipc::PathAccess;
 #[cfg(not(target_env = "musl"))]
-use fspy_shared::ipc::{NativeStr, channel::channel};
+use fspy_shared::ipc::NativeStr;
+use fspy_shared::ipc::PathAccess;
 #[cfg(target_os = "macos")]
 use fspy_shared_unix::payload::Artifacts;
 use fspy_shared_unix::{
@@ -25,7 +25,7 @@ use tokio::task::spawn_blocking;
 use tokio_util::sync::CancellationToken;
 
 #[cfg(not(target_env = "musl"))]
-use crate::ipc::{OwnedReceiverLockGuard, SHM_CAPACITY};
+use crate::ipc::IpcSupervisor;
 use crate::{ChildTermination, Command, TrackedChild, arena::PathAccessArena, error::SpawnError};
 
 #[derive(Debug)]
@@ -78,12 +78,11 @@ impl SpyImpl {
         let supervisor = supervise::<SyscallHandler>().map_err(SpawnError::Supervisor)?;
 
         #[cfg(not(target_env = "musl"))]
-        let (ipc_channel_conf, ipc_receiver) =
-            channel(SHM_CAPACITY).map_err(SpawnError::ChannelCreation)?;
+        let ipc_supervisor = IpcSupervisor::bind().map_err(SpawnError::IpcServer)?;
 
         let payload = Payload {
             #[cfg(not(target_env = "musl"))]
-            ipc_channel_conf,
+            server_name: ipc_supervisor.server_name().to_cow_os_str().into_owned().into(),
 
             #[cfg(target_os = "macos")]
             artifacts: self.artifacts.clone(),
@@ -136,7 +135,7 @@ impl SpyImpl {
             stdout: child.stdout.take(),
             stderr: child.stderr.take(),
             // Keep polling for the child to exit in the background even if `wait_handle` is not awaited,
-            // because we need to stop the supervisor and lock the channel as soon as the child exits.
+            // because we need to stop accepting IPC connections as soon as the child exits.
             wait_handle: tokio::spawn(async move {
                 let status = tokio::select! {
                     status = child.wait() => status?,
@@ -146,6 +145,9 @@ impl SpyImpl {
                     }
                 };
 
+                #[cfg(not(target_env = "musl"))]
+                let mut ipc_arenas = ipc_supervisor.stop().await?;
+
                 let arenas = std::iter::once(exec_resolve_accesses);
                 // Stop the supervisor and collect path accesses from it.
                 #[cfg(target_os = "linux")]
@@ -157,17 +159,12 @@ impl SpyImpl {
                         .map(syscall_handler::SyscallHandler::into_arena),
                 );
                 let arenas = arenas.collect::<Vec<_>>();
-
-                // Lock the ipc channel after the child has exited.
-                // We are not interested in path accesses from descendants after the main child has exited.
                 #[cfg(not(target_env = "musl"))]
-                let ipc_receiver_lock_guard =
-                    OwnedReceiverLockGuard::lock_async(ipc_receiver).await?;
-                let path_accesses = PathAccessIterable {
-                    arenas,
-                    #[cfg(not(target_env = "musl"))]
-                    ipc_receiver_lock_guard,
+                let arenas = {
+                    ipc_arenas.extend(arenas);
+                    ipc_arenas
                 };
+                let path_accesses = PathAccessIterable { arenas };
 
                 io::Result::Ok(ChildTermination { status, path_accesses })
             })
@@ -179,23 +176,10 @@ impl SpyImpl {
 
 pub struct PathAccessIterable {
     arenas: Vec<PathAccessArena>,
-    #[cfg(not(target_env = "musl"))]
-    ipc_receiver_lock_guard: OwnedReceiverLockGuard,
 }
 
 impl PathAccessIterable {
     pub fn iter(&self) -> impl Iterator<Item = PathAccess<'_>> {
-        let accesses_in_arena =
-            self.arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).copied();
-
-        #[cfg(not(target_env = "musl"))]
-        {
-            let accesses_in_shm = self.ipc_receiver_lock_guard.iter_path_accesses();
-            accesses_in_shm.chain(accesses_in_arena)
-        }
-        #[cfg(target_env = "musl")]
-        {
-            accesses_in_arena
-        }
+        self.arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).copied()
     }
 }