Auto merge of #272 - tdgne:windows-rebased-20210227, r=jdm

Windows (rebased and updated)

This is a rebase of #233 with some updates, namely
 * fixed unresolved dependencies
 * impls of error conversions
 * remaining work explained in #233 (comment)

Existing tests passed on my Windows 10 PC with toolchain `nightly-x86_64-pc-windows-msvc` running `cargo test --features="windows-shared-memory-equality,unstable"`, `cargo test --features="windows-shared-memory-equality"` and `cargo test --features="force-inprocess,windows-shared-memory-equality"` so far.

Happened to need this, hope it helps...

Author: bors-servo

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ipc-channel"
-version = "0.14.1"
+version = "0.15.0"
 description = "A multiprocess drop-in replacement for Rust channels"
 authors = ["The Servo Project Developers"]
 license = "MIT/Apache-2.0"
@@ -12,6 +12,8 @@ force-inprocess = []
 memfd = ["sc"]
 unstable = []
 async = ["futures-preview", "futures-test-preview"]
+win32-trace = []
+windows-shared-memory-equality = []
 
 [dependencies]
 bincode = "1"
@@ -32,3 +34,6 @@ sc = { version = "0.2.2", optional = true }
 
 [dev-dependencies]
 crossbeam-utils = "0.7"
+
+[target.'cfg(target_os = "windows")'.dependencies]
+winapi = {version = "0.3.7", features = ["minwindef", "ioapiset", "memoryapi", "namedpipeapi", "handleapi", "fileapi", "impl-default"]}

README.md

@@ -21,5 +21,3 @@ In order to bootstrap an IPC connection across processes, you create an instance
 ## Major missing features
 
 * Servers only accept one client at a time. This is fine if you simply want to use this API to split your application up into a fixed number of mutually untrusting processes, but it's not suitable for implementing a system service. An API for multiple clients may be added later if demand exists for it.
-
-* No Windows support exists yet. The right way to implement this will likely be with named pipes and `DuplicateHandle`.

appveyor.yml

@@ -3,7 +3,23 @@ environment:
   RUST_BACKTRACE: 1
   matrix:
   - TARGET: x86_64-pc-windows-msvc
+    FEATURES: "unstable"
   - TARGET: i686-pc-windows-msvc
+    FEATURES: "unstable"
+  - TARGET: i686-pc-windows-gnu
+    FEATURES: "unstable"
+  - TARGET: x86_64-pc-windows-msvc
+    FEATURES: "unstable force-inprocess"
+  - TARGET: i686-pc-windows-msvc
+    FEATURES: "unstable force-inprocess"
+  - TARGET: i686-pc-windows-gnu
+    FEATURES: "unstable force-inprocess"
+  - TARGET: x86_64-pc-windows-msvc
+    FEATURES: "unstable windows-shared-memory-equality"
+  - TARGET: i686-pc-windows-msvc
+    FEATURES: "unstable windows-shared-memory-equality"
+  - TARGET: i686-pc-windows-gnu
+    FEATURES: "unstable windows-shared-memory-equality"
 install:
   - ps: Start-FileDownload "https://static.rust-lang.org/dist/rust-nightly-${env:TARGET}.exe"
   - rust-nightly-%TARGET%.exe /VERYSILENT /NORESTART /DIR="C:\Program Files (x86)\Rust"
@@ -13,4 +29,4 @@ install:
 build: false
 
 test_script:
-  - 'cargo test --verbose --features "unstable"'
+  - cargo test --verbose --features "%FEATURES%"

src/ipc.rs

@@ -454,9 +454,11 @@ impl IpcReceiverSet {
 /// let shmem = IpcSharedMemory::from_bytes(&data);
 /// tx.send(shmem.clone()).unwrap();
 /// # let rx_shmem = rx.recv().unwrap();
+/// # #[cfg(any(not(target_os = "windows"), all(target_os = "windows", feature = "windows-shared-memory-equality")))]
 /// # assert_eq!(shmem, rx_shmem);
 /// ```
-#[derive(Clone, Debug, PartialEq)]
+#[derive(Clone, Debug)]
+#[cfg_attr(any(not(target_os = "windows"), all(target_os = "windows", feature = "windows-shared-memory-equality")), derive(PartialEq))]
 pub struct IpcSharedMemory {
     os_shared_memory: OsIpcSharedMemory,
 }

src/lib.rs

@@ -79,6 +79,10 @@ extern crate futures_test;
 #[cfg(feature = "async")]
 pub mod asynch;
 
+#[cfg(all(not(feature = "force-inprocess"), target_os = "windows"))]
+extern crate winapi;
+
+
 pub mod ipc;
 pub mod platform;
 pub mod router;

src/platform/mod.rs

@@ -25,14 +25,21 @@ mod os {
     pub use super::macos::*;
 }
 
+#[cfg(all(not(feature = "force-inprocess"), target_os = "windows"))]
+mod windows;
+#[cfg(all(not(feature = "force-inprocess"), target_os = "windows"))]
+mod os {
+    pub use super::windows::*;
+}
+
 #[cfg(any(
     feature = "force-inprocess",
-    target_os = "windows", target_os = "android", target_os = "ios", target_os = "wasi", target_os = "unknown"
+    target_os = "android", target_os = "ios", target_os = "wasi", target_os = "unknown"
 ))]
 mod inprocess;
 #[cfg(any(
     feature = "force-inprocess",
-    target_os = "windows", target_os = "android", target_os = "ios", target_os = "wasi", target_os = "unknown"
+    target_os = "android", target_os = "ios", target_os = "wasi", target_os = "unknown"
 ))]
 mod os {
     pub use super::inprocess::*;

src/platform/test.rs

@@ -7,19 +7,29 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+// Most of this file won't compile without `windows-shared-memory-equality` feature on Windows since `PartialEq` won't be implemented for `IpcSharedMemory`.
+#![cfg(any(not(target_os = "windows"), all(target_os = "windows", feature = "windows-shared-memory-equality")))]
+
 use crate::platform::{self, OsIpcChannel, OsIpcReceiverSet};
 use crate::platform::{OsIpcSharedMemory};
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use std::thread;
 
+#[cfg(not(any(feature = "force-inprocess", target_os = "android", target_os = "ios")))]
+use libc;
 use crate::platform::{OsIpcSender, OsIpcOneShotServer};
 #[cfg(not(any(feature = "force-inprocess", target_os = "windows", target_os = "android", target_os = "ios")))]
 use libc::{kill, SIGSTOP, SIGCONT};
 #[cfg(not(any(feature = "force-inprocess", target_os = "windows", target_os = "android", target_os = "ios")))]
 use crate::test::{fork, Wait};
 
+// Helper to get a channel_name argument passed in; used for the
+// cross-process spawn server tests.
+#[cfg(not(any(feature = "force-inprocess", target_os = "android", target_os = "ios")))]
+use crate::test::{get_channel_name_arg, spawn_server};
+
 #[test]
 fn simple() {
     let (tx, rx) = platform::channel().unwrap();
@@ -209,7 +219,8 @@ fn with_n_fds(n: usize, size: usize) {
 
 // These tests only apply to platforms that need fragmentation.
 #[cfg(all(not(feature = "force-inprocess"), any(target_os = "linux",
-                                                target_os = "freebsd")))]
+                                                target_os = "freebsd",
+                                                target_os = "windows")))]
 mod fragment_tests {
     use crate::platform;
     use super::with_n_fds;
@@ -656,9 +667,32 @@ fn server_connect_first() {
                (data, vec![], vec![]));
 }
 
+#[cfg(not(any(feature = "force-inprocess", target_os = "android", target_os = "ios")))]
+#[test]
+fn cross_process_spawn() {
+    let data: &[u8] = b"1234567";
+
+    let channel_name = get_channel_name_arg("server");
+    if let Some(channel_name) = channel_name {
+        let tx = OsIpcSender::connect(channel_name).unwrap();
+        tx.send(data, vec![], vec![]).unwrap();
+
+        unsafe { libc::exit(0); }
+    }
+
+    let (server, name) = OsIpcOneShotServer::new().unwrap();
+    let mut child_pid = spawn_server("cross_process_spawn", &[("server", &*name)]);
+
+    let (_, received_data, received_channels, received_shared_memory_regions) =
+        server.accept().unwrap();
+    child_pid.wait().expect("failed to wait on child");
+    assert_eq!((&received_data[..], received_channels, received_shared_memory_regions),
+               (data, vec![], vec![]));
+}
+
 #[cfg(not(any(feature = "force-inprocess", target_os = "windows", target_os = "android", target_os = "ios")))]
 #[test]
-fn cross_process() {
+fn cross_process_fork() {
     let (server, name) = OsIpcOneShotServer::new().unwrap();
     let data: &[u8] = b"1234567";
 
@@ -674,9 +708,42 @@ fn cross_process() {
                (data, vec![], vec![]));
 }
 
+#[cfg(not(any(feature = "force-inprocess", target_os = "android", target_os = "ios")))]
+#[test]
+fn cross_process_sender_transfer_spawn() {
+    let channel_name = get_channel_name_arg("server");
+    if let Some(channel_name) = channel_name {
+        let super_tx = OsIpcSender::connect(channel_name).unwrap();
+        let (sub_tx, sub_rx) = platform::channel().unwrap();
+        let data: &[u8] = b"foo";
+        super_tx.send(data, vec![OsIpcChannel::Sender(sub_tx)], vec![]).unwrap();
+        sub_rx.recv().unwrap();
+        let data: &[u8] = b"bar";
+        super_tx.send(data, vec![], vec![]).unwrap();
+
+        unsafe { libc::exit(0); }
+    }
+
+    let (server, name) = OsIpcOneShotServer::new().unwrap();
+    let mut child_pid = spawn_server("cross_process_sender_transfer_spawn", &[("server", &*name)]);
+
+    let (super_rx, _, mut received_channels, _) = server.accept().unwrap();
+    assert_eq!(received_channels.len(), 1);
+    let sub_tx = received_channels[0].to_sender();
+    let data: &[u8] = b"baz";
+    sub_tx.send(data, vec![], vec![]).unwrap();
+
+    let data: &[u8] = b"bar";
+    let (received_data, received_channels, received_shared_memory_regions) =
+        super_rx.recv().unwrap();
+    child_pid.wait().expect("failed to wait on child");
+    assert_eq!((&received_data[..], received_channels, received_shared_memory_regions),
+               (data, vec![], vec![]));
+}
+
 #[cfg(not(any(feature = "force-inprocess", target_os = "windows", target_os = "android", target_os = "ios")))]
 #[test]
-fn cross_process_sender_transfer() {
+fn cross_process_sender_transfer_fork() {
     let (server, name) = OsIpcOneShotServer::new().unwrap();
 
     let child_pid = unsafe { fork(|| {
@@ -691,7 +758,7 @@ fn cross_process_sender_transfer() {
 
     let (super_rx, _, mut received_channels, _) = server.accept().unwrap();
     assert_eq!(received_channels.len(), 1);
-    let sub_tx = received_channels.pop().unwrap().to_sender();
+    let sub_tx = received_channels[0].to_sender();
     let data: &[u8] = b"baz";
     sub_tx.send(data, vec![], vec![]).unwrap();
 
@@ -981,3 +1048,82 @@ mod sync_test {
         platform::OsIpcSender::test_not_sync();
     }
 }
+
+// This test panics on Windows, because the other process will panic
+// when it detects that it receives handles that are intended for another
+// process.  It's marked as ignore/known-fail on Windows for this reason.
+//
+// TODO -- this fails on OSX as well with a MACH_SEND_INVALID_RIGHT!
+// Needs investigation.  It may be a similar underlying issue, just done by
+// the kernel instead of explicitly (ports in a message that's already
+// buffered are intended for only one process).
+#[cfg(not(any(feature = "force-inprocess", target_os = "android", target_os = "ios")))]
+#[cfg_attr(any(target_os = "windows", target_os = "macos"), ignore)]
+#[test]
+fn cross_process_two_step_transfer_spawn() {
+    let cookie: &[u8] = b"cookie";
+
+    let channel_name = get_channel_name_arg("server");
+    if let Some(channel_name) = channel_name {
+        // connect by name to our other process
+        let super_tx = OsIpcSender::connect(channel_name).unwrap();
+
+        // create a channel for real communication between the two processes
+        let (sub_tx, sub_rx) = platform::channel().unwrap();
+
+        // send the other process the tx side, so it can send us the channels
+        super_tx.send(&[], vec![OsIpcChannel::Sender(sub_tx)], vec![]).unwrap();
+
+        // get two_rx from the other process
+        let (_, mut received_channels, _) = sub_rx.recv().unwrap();
+        assert_eq!(received_channels.len(), 1);
+        let two_rx = received_channels[0].to_receiver();
+
+        // get one_rx from two_rx's buffer
+        let (_, mut received_channels, _) = two_rx.recv().unwrap();
+        assert_eq!(received_channels.len(), 1);
+        let one_rx = received_channels[0].to_receiver();
+
+        // get a cookie from one_rx
+        let (data, _, _) = one_rx.recv().unwrap();
+        assert_eq!(&data[..], cookie);
+
+        // finally, send a cookie back
+        super_tx.send(&data, vec![], vec![]).unwrap();
+
+        // terminate
+        unsafe { libc::exit(0); }
+    }
+
+    // create channel 1
+    let (one_tx, one_rx) = platform::channel().unwrap();
+    // put data in channel 1's pipe
+    one_tx.send(cookie, vec![], vec![]).unwrap();
+
+    // create channel 2
+    let (two_tx, two_rx) = platform::channel().unwrap();
+    // put channel 1's rx end in channel 2's pipe
+    two_tx.send(&[], vec![OsIpcChannel::Receiver(one_rx)], vec![]).unwrap();
+
+    // create a one-shot server, and spawn another process
+    let (server, name) = OsIpcOneShotServer::new().unwrap();
+    let mut child_pid = spawn_server("cross_process_two_step_transfer_spawn",
+                                     &[("server", &*name)]);
+
+    // The other process will have sent us a transmit channel in received channels
+    let (super_rx, _, mut received_channels, _) = server.accept().unwrap();
+    assert_eq!(received_channels.len(), 1);
+    let sub_tx = received_channels[0].to_sender();
+
+    // Send the outer payload channel, so the server can use it to
+    // retrive the inner payload and the cookie
+    sub_tx.send(&[], vec![OsIpcChannel::Receiver(two_rx)], vec![]).unwrap();
+
+    // Then we wait for the cookie to make its way back to us
+    let (received_data, received_channels, received_shared_memory_regions) =
+        super_rx.recv().unwrap();
+    let child_exit_code = child_pid.wait().expect("failed to wait on child");
+    assert!(child_exit_code.success());
+    assert_eq!((&received_data[..], received_channels, received_shared_memory_regions),
+               (cookie, vec![], vec![]));
+}