refactor(fspy): use shared memory IPC on Windows

crates/fspy/src/ipc.rs

@@ -0,0 +1,42 @@
+use std::io;
+
+use bincode::borrow_decode_from_slice;
+use fspy_shared::ipc::{
+    BINCODE_CONFIG, PathAccess,
+    channel::{Receiver, ReceiverLockGuard},
+};
+use tokio::task::spawn_blocking;
+
+// 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;
+
+#[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>,
+}
+
+impl OwnedReceiverLockGuard {
+    pub fn lock(receiver: Receiver) -> io::Result<Self> {
+        OwnedReceiverLockGuard::try_new(receiver, |receiver| receiver.lock())
+    }
+
+    pub async fn lock_async(receiver: Receiver) -> io::Result<Self> {
+        spawn_blocking(move || Self::lock(receiver)).await.expect("lock task panicked")
+    }
+
+    pub fn iter_path_accesses(&self) -> impl Iterator<Item = PathAccess<'_>> {
+        self.borrow_lock_guard().iter_frames().map(|frame| {
+            let (path_access, decoded_size) =
+                borrow_decode_from_slice::<PathAccess<'_>, _>(frame, BINCODE_CONFIG).unwrap();
+            assert_eq!(decoded_size, frame.len());
+            path_access
+        })
+    }
+}

crates/fspy/src/lib.rs

@@ -4,6 +4,8 @@
 // Persist the injected DLL/shared library somewhere in the filesystem.
 mod fixture;
 
+mod ipc;
+
 #[cfg(unix)]
 #[path = "./unix/mod.rs"]
 mod os_impl;
@@ -12,6 +14,7 @@ mod os_impl;
 #[path = "./windows/mod.rs"]
 mod os_impl;
 
+#[cfg(unix)]
 mod arena;
 mod command;
 

crates/fspy/src/unix/mod.rs

@@ -1,7 +1,3 @@
-// use std::{os::unix::process::CommandExt};
-
-// use tokio::process::Command;
-
 #[cfg(target_os = "linux")]
 mod syscall_handler;
 
@@ -10,13 +6,9 @@ mod macos_fixtures;
 
 use std::{io, path::Path};
 
-use bincode::borrow_decode_from_slice;
 #[cfg(target_os = "linux")]
 use fspy_seccomp_unotify::supervisor::supervise;
-use fspy_shared::ipc::{
-    BINCODE_CONFIG, NativeString, PathAccess,
-    channel::{Receiver, ReceiverLockGuard, channel},
-};
+use fspy_shared::ipc::{NativeString, PathAccess, channel::channel};
 #[cfg(target_os = "macos")]
 use fspy_shared_unix::payload::Fixtures;
 use fspy_shared_unix::{
@@ -27,9 +19,12 @@ use fspy_shared_unix::{
 use futures_util::FutureExt;
 #[cfg(target_os = "linux")]
 use syscall_handler::SyscallHandler;
-use tokio::task::spawn_blocking;
 
-use crate::{Command, TrackedChild, arena::PathAccessArena};
+use crate::{
+    Command, TrackedChild,
+    arena::PathAccessArena,
+    ipc::{OwnedReceiverLockGuard, SHM_CAPACITY},
+};
 
 #[derive(Debug, Clone)]
 pub struct SpyInner {
@@ -71,16 +66,6 @@ impl SpyInner {
     }
 }
 
-#[ouroboros::self_referencing]
-struct OwnedReceiverLockGuard {
-    /// Owns the shared memory
-    receiver: Receiver,
-    /// Borrows the shared memory and owns the file lock
-    #[borrows(receiver)]
-    #[covariant]
-    lock_guard: ReceiverLockGuard<'this>,
-}
-
 pub struct PathAccessIterable {
     arenas: Vec<PathAccessArena>,
     ipc_receiver_lock_guard: OwnedReceiverLockGuard,
@@ -91,22 +76,11 @@ impl PathAccessIterable {
         let accesses_in_arena =
             self.arenas.iter().flat_map(|arena| arena.borrow_accesses().iter()).copied();
 
-        let accesses_in_shm =
-            self.ipc_receiver_lock_guard.borrow_lock_guard().iter_frames().map(|frame| {
-                let (path_access, decoded_size) =
-                    borrow_decode_from_slice::<PathAccess<'_>, _>(frame, BINCODE_CONFIG).unwrap();
-                assert_eq!(decoded_size, frame.len());
-                path_access
-            });
+        let accesses_in_shm = self.ipc_receiver_lock_guard.iter_path_accesses();
         accesses_in_shm.chain(accesses_in_arena)
     }
 }
 
-// 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.
-const SHM_CAPACITY: usize = 4 * 1024 * 1024 * 1024;
-
 pub(crate) async fn spawn_impl(mut command: Command) -> io::Result<TrackedChild> {
     #[cfg(target_os = "linux")]
     let supervisor = supervise::<SyscallHandler>()?;
@@ -170,10 +144,7 @@ pub(crate) async fn spawn_impl(mut command: Command) -> io::Result<TrackedChild>
     let accesses_future = async move {
         let arenas = arenas_future.await?;
         // `receiver.lock()` blocks. Run it inside `spawn_blocking` to avoid blocking the tokio runtime.
-        let ipc_receiver_lock_guard = spawn_blocking(move || {
-            OwnedReceiverLockGuard::try_new(ipc_receiver, |receiver| receiver.lock())
-        })
-        .await??;
+        let ipc_receiver_lock_guard = OwnedReceiverLockGuard::lock_async(ipc_receiver).await?;
         Ok(PathAccessIterable { arenas, ipc_receiver_lock_guard })
     }
     .boxed();

crates/fspy/src/windows/mod.rs

@@ -1,40 +1,31 @@
 use std::{
-    ffi::{CStr, c_char, c_void},
-    fs::OpenOptions,
+    ffi::{CStr, c_char},
     io,
     os::windows::{ffi::OsStrExt, io::AsRawHandle, process::ChildExt as _},
     path::Path,
-    ptr::null_mut,
     sync::Arc,
 };
 
-use bincode::borrow_decode_from_slice;
 use const_format::formatcp;
 use fspy_detours_sys::{DetourCopyPayloadToProcess, DetourUpdateProcessWithDll};
 use fspy_shared::{
-    ipc::{BINCODE_CONFIG, PathAccess},
+    ipc::{BINCODE_CONFIG, PathAccess, channel::channel},
     windows::{PAYLOAD_ID, Payload},
 };
 use futures_util::FutureExt;
-use tokio::{
-    io::AsyncReadExt,
-    net::windows::named_pipe::{PipeMode, ServerOptions},
-};
-// use detours_sys2::{DetourAttach,};
 use winapi::{
-    shared::minwindef::{FALSE, TRUE},
-    um::{
-        handleapi::DuplicateHandle,
-        processthreadsapi::{GetCurrentProcess, ResumeThread},
-        winbase::CREATE_SUSPENDED,
-        winnt::DUPLICATE_SAME_ACCESS,
-    },
+    shared::minwindef::TRUE,
+    um::{processthreadsapi::ResumeThread, winbase::CREATE_SUSPENDED},
 };
-// use windows_sys::Win32::System::Threading::{CREATE_SUSPENDED, ResumeThread};
 use winsafe::co::{CP, WC};
 use xxhash_rust::const_xxh3::xxh3_128;
 
-use crate::{TrackedChild, arena::PathAccessArena, command::Command, fixture::Fixture};
+use crate::{
+    TrackedChild,
+    command::Command,
+    fixture::Fixture,
+    ipc::{OwnedReceiverLockGuard, SHM_CAPACITY},
+};
 
 const PRELOAD_CDYLIB_BINARY: &[u8] = include_bytes!(env!("CARGO_CDYLIB_FILE_FSPY_PRELOAD_WINDOWS"));
 const INTERPOSE_CDYLIB: Fixture = Fixture::new(
@@ -43,38 +34,14 @@ const INTERPOSE_CDYLIB: Fixture = Fixture::new(
     formatcp!("{:x}", xxh3_128(PRELOAD_CDYLIB_BINARY)),
 );
 
-fn luid() -> io::Result<u64> {
-    let mut luid = unsafe { std::mem::zeroed::<winapi::um::winnt::LUID>() };
-    let ret = unsafe { winapi::um::securitybaseapi::AllocateLocallyUniqueId(&mut luid) };
-    if ret == 0 {
-        return Err(io::Error::last_os_error());
-    }
-    Ok((u64::from(luid.HighPart as u32)) << 32 | u64::from(luid.LowPart))
-}
-
 pub struct PathAccessIterable {
-    arena: PathAccessArena,
-    // pipe_receiver: NamedPipeServer,
+    ipc_receiver_lock_guard: OwnedReceiverLockGuard,
 }
 
-const MESSAGE_MAX_LEN: usize = 4096;
-
 impl PathAccessIterable {
     pub fn iter(&self) -> impl Iterator<Item = PathAccess<'_>> {
-        self.arena.borrow_accesses().iter().copied()
+        self.ipc_receiver_lock_guard.iter_path_accesses()
     }
-    //     pub async fn next<'a>(&mut self, buf: &'a mut Vec<u8>) -> io::Result<Option<PathAccess<'a>>> {
-    //         buf.resize(MESSAGE_MAX_LEN, 0);
-    //         let n = self.pipe_receiver.read(buf.as_mut_slice()).await?;
-    //         if n == 0 {
-    //             return Ok(None);
-    //         }
-    //         let msg = &buf[..n];
-    //         let (path_access, decoded_len) =
-    //             borrow_decode_from_slice::<'_, PathAccess, _>(msg, BINCODE_CONFIG).unwrap();
-    //         assert_eq!(decoded_len, msg.len());
-    //         Ok(Some(path_access))
-    //     }
 }
 
 // pub struct TracedProcess {
@@ -110,46 +77,13 @@ pub(crate) async fn spawn_impl(command: Command) -> io::Result<TrackedChild> {
 
     command.creation_flags(CREATE_SUSPENDED);
 
-    let pipe_name = format!(r"\\.\pipe\fspy_ipc_{:x}", luid()?);
-
-    let mut pipe_receiver = ServerOptions::new()
-        .pipe_mode(PipeMode::Message)
-        .access_outbound(false)
-        .access_inbound(true)
-        .in_buffer_size(1024)
-        // .out_buffer_size(100 * 1024 * 1024)
-        .create(&pipe_name)?;
-
-    let connect_fut = pipe_receiver.connect();
-
-    let pipe_sender = OpenOptions::new().write(true).open(&pipe_name).unwrap();
-
-    connect_fut.await?;
-
-    // Temporary workaround before switching to shared memory IPC on Windows.
-    // The shared memory IPC requires `accesses_future` to be polled after the child process has exited.
-    // The test code has updated but the Windows implementation here has not yet adopted that model.
-    // As a result, accesses_future didn't get polled making the pipe_sender to block indefinitely.
-    // To unblock the pipe_sender, we spawn a separate task to keep reading pipe_receiver.
-    // This workaround can be removed once the shared memory IPC is used on Windows.
-    let accesses_future = tokio::task::spawn(async move {
-        let mut arena = PathAccessArena::default();
-
-        let mut buf = [0u8; MESSAGE_MAX_LEN];
-        loop {
-            let n = pipe_receiver.read(&mut buf).await?;
-            if n == 0 {
-                break;
-            }
-            let msg = &buf[..n];
-            let (path_access, decoded_len) =
-                borrow_decode_from_slice::<'_, PathAccess, _>(msg, BINCODE_CONFIG).unwrap();
-            assert_eq!(decoded_len, msg.len());
-            arena.add(path_access);
-        }
-        io::Result::Ok(PathAccessIterable { arena })
-    });
-    let accesses_future = async move { accesses_future.await.unwrap() }.boxed();
+    let (channel_conf, receiver) = channel(SHM_CAPACITY)?;
+
+    let accesses_future = async move {
+        let ipc_receiver_lock_guard = OwnedReceiverLockGuard::lock_async(receiver).await?;
+        io::Result::Ok(PathAccessIterable { ipc_receiver_lock_guard })
+    }
+    .boxed();
 
     // let path_access_stream = PathAccessIterable { pipe_receiver };
 
@@ -163,24 +97,8 @@ pub(crate) async fn spawn_impl(command: Command) -> io::Result<TrackedChild> {
             return Err(io::Error::last_os_error());
         }
 
-        let mut handle_in_child: *mut c_void = null_mut();
-        let ret = unsafe {
-            DuplicateHandle(
-                GetCurrentProcess(),
-                pipe_sender.as_raw_handle(),
-                process_handle,
-                &mut handle_in_child,
-                0,
-                FALSE,
-                DUPLICATE_SAME_ACCESS,
-            )
-        };
-        if ret == 0 {
-            return Err(io::Error::last_os_error());
-        }
-
         let payload = Payload {
-            pipe_handle: handle_in_child.addr(),
+            channel_conf: channel_conf.clone(),
             asni_dll_path_with_nul: asni_dll_path_with_nul.to_bytes(),
         };
         let payload_bytes = bincode::encode_to_vec(payload, BINCODE_CONFIG).unwrap();
@@ -207,6 +125,5 @@ pub(crate) async fn spawn_impl(command: Command) -> io::Result<TrackedChild> {
         Ok(std_child)
     })?;
 
-    drop(pipe_sender);
     Ok(TrackedChild { tokio_child: child, accesses_future })
 }

crates/fspy_preload_windows/Cargo.toml

@@ -26,6 +26,3 @@ winsafe = { workspace = true }
 
 [target.'cfg(target_os = "windows")'.dev-dependencies]
 tempfile = { workspace = true }
-
-[dependencies]
-dashmap = { workspace = true }