[WIP] Add native Windows back-end

.travis.yml

@@ -6,19 +6,13 @@ rust:
 
 os:
   - linux
-  - osx
 
 env:
   global:
     - RUST_BACKTRACE=1
   matrix:
     - FEATURES="unstable"
-    - FEATURES="unstable force-inprocess"
-
-matrix:
-  include:
-    - os: linux
-      env: FEATURES="unstable memfd"
+    - FEATURES="unstable memfd"
 
 notifications:
   webhooks: http://build.servo.org:54856/travis

Cargo.toml

@@ -11,6 +11,7 @@ force-inprocess = []
 memfd = ["sc"]
 unstable = []
 async = ["futures"]
+win32-trace = []
 
 [dependencies]
 bincode = "1"
@@ -30,3 +31,7 @@ futures = { version = "0.1", optional = true }
 
 [dev-dependencies]
 crossbeam = "0.2"
+
+[target.'cfg(target_os = "windows")'.dependencies]
+winapi = "0.2"
+kernel32-sys = "0.2"

appveyor.yml

@@ -3,8 +3,7 @@ environment:
   RUST_BACKTRACE: 1
   matrix:
   - TARGET: x86_64-pc-windows-msvc
-  - TARGET: i686-pc-windows-msvc
-  - TARGET: i686-pc-windows-gnu
+    FEATURES: "unstable"
 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"
@@ -14,4 +13,4 @@ install:
 build: false
 
 test_script:
-  - 'cargo test --verbose --features "unstable"'
+  - cargo test --verbose --features "%FEATURES%"

src/lib.rs

@@ -7,7 +7,7 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-#![cfg_attr(any(feature = "force-inprocess", target_os = "windows", target_os = "android", target_os = "ios"),
+#![cfg_attr(any(feature = "force-inprocess", target_os = "android", target_os = "ios"),
 			feature(mpsc_select))]
 #![cfg_attr(all(feature = "unstable", test), feature(specialization))]
 
@@ -18,6 +18,7 @@ extern crate bincode;
 extern crate libc;
 extern crate rand;
 extern crate serde;
+
 #[cfg(any(feature = "force-inprocess", target_os = "windows", target_os = "android", target_os = "ios"))]
 extern crate uuid;
 extern crate tempfile;
@@ -38,6 +39,11 @@ extern crate sc;
 extern crate futures;
 
 
+#[cfg(all(not(feature = "force-inprocess"), target_os = "windows"))]
+extern crate winapi;
+#[cfg(all(not(feature = "force-inprocess"), target_os = "windows"))]
+extern crate kernel32;
+
 pub mod ipc;
 pub mod platform;
 pub mod router;

src/platform/mod.rs

@@ -25,9 +25,16 @@ mod os {
     pub use super::macos::*;
 }
 
-#[cfg(any(feature = "force-inprocess", target_os = "windows", target_os = "android", target_os = "ios"))]
+#[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 = "android", target_os = "ios"))]
 mod inprocess;
-#[cfg(any(feature = "force-inprocess", target_os = "windows", target_os = "android", target_os = "ios"))]
+#[cfg(any(feature = "force-inprocess", target_os = "android", target_os = "ios"))]
 mod os {
     pub use super::inprocess::*;
 }

src/platform/test.rs

@@ -14,11 +14,15 @@ 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 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 test::{fork, Wait};
+#[cfg(not(any(feature = "force-inprocess", target_os = "android", target_os = "ios")))]
+use test::{get_channel_name_arg, spawn_server};
 
 #[test]
 fn simple() {
@@ -173,9 +177,15 @@ fn with_n_fds(n: usize, size: usize) {
     let (super_tx, super_rx) = platform::channel().unwrap();
 
     let data: Vec<u8> = (0..size).map(|i| (i % 251) as u8).collect();
-    super_tx.send(&data[..], sender_fds, vec![]).unwrap();
+    let thread = {
+        let data = data.clone();
+        thread::spawn(move || {
+            super_tx.send(&data[..], sender_fds, vec![]).unwrap();
+        })
+    };
     let (received_data, received_channels, received_shared_memory_regions) =
         super_rx.recv().unwrap();
+    thread.join().unwrap();
 
     assert_eq!(received_data.len(), data.len());
     assert_eq!(&received_data[..], &data[..]);
@@ -196,7 +206,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 platform;
     use super::with_n_fds;
@@ -643,9 +654,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";
 
@@ -661,9 +695,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(|| {
@@ -678,7 +745,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();
 
@@ -968,3 +1035,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![]));
+}