Implement process bindings to libuv

Closes #6436

Author: alexcrichton

mk/rt.mk

@@ -271,3 +271,10 @@ $(LIBUV_MAKEFILE): $(LIBUV_GYP)
 $(foreach stage,$(STAGES), \
 	$(foreach target,$(CFG_TARGET_TRIPLES), \
 	 $(eval $(call DEF_RUNTIME_TARGETS,$(target),$(stage)))))
+
+$(LIBUV_GYP):
+	mkdir -p $(S)src/libuv/build
+	git clone https://git.chromium.org/external/gyp.git $(S)src/libuv/build/gyp
+
+$(LIBUV_MAKEFILE): $(LIBUV_GYP)
+	(cd $(S)src/libuv/ && ./gyp_uv -f make)

src/compiletest/procsrv.rs

@@ -54,10 +54,10 @@ pub fn run(lib_path: &str,
         in_fd: None,
         out_fd: None,
         err_fd: None
-    });
+    }).unwrap();
 
     for input in input.iter() {
-        proc.input().write_str(*input);
+        proc.input().write(input.as_bytes());
     }
     let output = proc.finish_with_output();
 

src/compiletest/runtest.rs

@@ -20,41 +20,16 @@ use procsrv;
 use util;
 use util::logv;
 
-use std::cell::Cell;
 use std::io;
 use std::os;
 use std::str;
-use std::task::{spawn_sched, SingleThreaded};
 use std::vec;
-use std::unstable::running_on_valgrind;
 
 use extra::test::MetricMap;
 
 pub fn run(config: config, testfile: ~str) {
-    let config = Cell::new(config);
-    let testfile = Cell::new(testfile);
-    // FIXME #6436: Creating another thread to run the test because this
-    // is going to call waitpid. The new scheduler has some strange
-    // interaction between the blocking tasks and 'friend' schedulers
-    // that destroys parallelism if we let normal schedulers block.
-    // It should be possible to remove this spawn once std::run is
-    // rewritten to be non-blocking.
-    //
-    // We do _not_ create another thread if we're running on V because
-    // it serializes all threads anyways.
-    if running_on_valgrind() {
-        let config = config.take();
-        let testfile = testfile.take();
-        let mut _mm = MetricMap::new();
-        run_metrics(config, testfile, &mut _mm);
-    } else {
-        do spawn_sched(SingleThreaded) {
-            let config = config.take();
-            let testfile = testfile.take();
-            let mut _mm = MetricMap::new();
-            run_metrics(config, testfile, &mut _mm);
-        }
-    }
+    let mut _mm = MetricMap::new();
+    run_metrics(config, testfile, &mut _mm);
 }
 
 pub fn run_metrics(config: config, testfile: ~str, mm: &mut MetricMap) {

src/librustdoc/markdown_writer.rs

@@ -104,14 +104,14 @@ fn pandoc_writer(
     ];
 
     do generic_writer |markdown| {
-        use std::io::WriterUtil;
-
         debug!("pandoc cmd: %s", pandoc_cmd);
         debug!("pandoc args: %s", pandoc_args.connect(" "));
 
-        let mut proc = run::Process::new(pandoc_cmd, pandoc_args, run::ProcessOptions::new());
+        let proc = run::Process::new(pandoc_cmd, pandoc_args,
+                                     run::ProcessOptions::new());
+        let mut proc = proc.unwrap();
 
-        proc.input().write_str(markdown);
+        proc.input().write(markdown.as_bytes());
         let output = proc.finish_with_output();
 
         debug!("pandoc result: %i", output.status);

src/librustpkg/source_control.rs

@@ -89,7 +89,7 @@ pub fn git_clone_general(source: &str, target: &Path, v: &Version) -> bool {
 
 fn process_output_in_cwd(prog: &str, args: &[~str], cwd: &Path) -> ProcessOutput {
     let mut prog = Process::new(prog, args, ProcessOptions{ dir: Some(cwd)
-                                ,..ProcessOptions::new()});
+                                ,..ProcessOptions::new()}).unwrap();
     prog.finish_with_output()
 }
 

src/librustpkg/tests.rs

@@ -112,13 +112,14 @@ fn mk_temp_workspace(short_name: &Path, version: &Version) -> Path {
 
 fn run_git(args: &[~str], env: Option<~[(~str, ~str)]>, cwd: &Path, err_msg: &str) {
     let cwd = (*cwd).clone();
-    let mut prog = run::Process::new("git", args, run::ProcessOptions {
+    let prog = run::Process::new("git", args, run::ProcessOptions {
         env: env,
         dir: Some(&cwd),
         in_fd: None,
         out_fd: None,
         err_fd: None
     });
+    let mut prog = prog.unwrap();
     let rslt = prog.finish_with_output();
     if rslt.status != 0 {
         fail!("%s [git returned %?, output = %s, error = %s]", err_msg,
@@ -226,7 +227,7 @@ fn command_line_test_with_env(args: &[~str], cwd: &Path, env: Option<~[(~str, ~s
         in_fd: None,
         out_fd: None,
         err_fd: None
-    });
+    }).unwrap();
     let output = prog.finish_with_output();
     debug!("Output from command %s with args %? was %s {%s}[%?]",
                     cmd, args, str::from_bytes(output.output),
@@ -1024,16 +1025,17 @@ fn test_extern_mod() {
                      test_sysroot().to_str(),
                      exec_file.to_str());
 
-    let mut prog = run::Process::new(rustc.to_str(), [main_file.to_str(),
-                                                      ~"--sysroot", test_sysroot().to_str(),
-                                               ~"-o", exec_file.to_str()],
-                                     run::ProcessOptions {
+    let prog = run::Process::new(rustc.to_str(), [main_file.to_str(),
+                                                  ~"--sysroot", test_sysroot().to_str(),
+                                                  ~"-o", exec_file.to_str()],
+                                 run::ProcessOptions {
         env: env,
         dir: Some(&dir),
         in_fd: None,
         out_fd: None,
         err_fd: None
     });
+    let mut prog = prog.unwrap();
     let outp = prog.finish_with_output();
     if outp.status != 0 {
         fail!("output was %s, error was %s",

src/libstd/rt/io/file.rs

@@ -71,9 +71,6 @@ pub struct FileStream {
     last_nread: int,
 }
 
-impl FileStream {
-}
-
 impl Reader for FileStream {
     fn read(&mut self, buf: &mut [u8]) -> Option<uint> {
         match self.fd.read(buf) {

src/libstd/rt/io/mod.rs

@@ -268,6 +268,9 @@ pub use self::extensions::WriterByteConversions;
 /// Synchronous, non-blocking file I/O.
 pub mod file;
 
+/// Synchronous, in-memory I/O.
+pub mod pipe;
+
 /// Synchronous, non-blocking network I/O.
 pub mod net {
     pub mod tcp;

src/libstd/rt/io/net/tcp.rs

@@ -16,7 +16,7 @@ use rt::io::{io_error, read_error, EndOfFile};
 use rt::rtio::{IoFactory, IoFactoryObject,
                RtioSocket, RtioTcpListener,
                RtioTcpListenerObject, RtioTcpStream,
-               RtioTcpStreamObject};
+               RtioTcpStreamObject, RtioStream};
 use rt::local::Local;
 
 pub struct TcpStream(~RtioTcpStreamObject);
@@ -69,7 +69,7 @@ impl TcpStream {
 
 impl Reader for TcpStream {
     fn read(&mut self, buf: &mut [u8]) -> Option<uint> {
-        match (**self).read(buf) {
+        match (***self).read(buf) {
             Ok(read) => Some(read),
             Err(ioerr) => {
                 // EOF is indicated by returning None
@@ -86,7 +86,7 @@ impl Reader for TcpStream {
 
 impl Writer for TcpStream {
     fn write(&mut self, buf: &[u8]) {
-        match (**self).write(buf) {
+        match (***self).write(buf) {
             Ok(_) => (),
             Err(ioerr) => io_error::cond.raise(ioerr),
         }

src/libstd/rt/io/pipe.rs

@@ -0,0 +1,77 @@
+// Copyright 2013 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+//! Synchronous, in-memory pipes.
+//!
+//! Currently these aren't particularly useful, there only exists bindings
+//! enough so that pipes can be created to child processes.
+
+use prelude::*;
+use super::{Reader, Writer};
+use rt::io::{io_error, read_error, EndOfFile};
+use rt::local::Local;
+use rt::rtio::{RtioPipeObject, RtioStream, IoFactoryObject, IoFactory};
+use rt::uv::pipe;
+
+pub struct PipeStream(~RtioPipeObject);
+
+impl PipeStream {
+    /// Creates a new pipe initialized, but not bound to any particular
+    /// source/destination
+    pub fn new() -> Option<PipeStream> {
+        let pipe = unsafe {
+            let io: *mut IoFactoryObject = Local::unsafe_borrow();
+            (*io).pipe_init(false)
+        };
+        match pipe {
+            Ok(p) => Some(PipeStream(p)),
+            Err(ioerr) => {
+                io_error::cond.raise(ioerr);
+                None
+            }
+        }
+    }
+
+    /// Extracts the underlying libuv pipe to be bound to another source.
+    pub fn uv_pipe(&self) -> pipe::Pipe {
+        // Did someone say multiple layers of indirection?
+        (**self).uv_pipe()
+    }
+}
+
+impl Reader for PipeStream {
+    fn read(&mut self, buf: &mut [u8]) -> Option<uint> {
+        match (***self).read(buf) {
+            Ok(read) => Some(read),
+            Err(ioerr) => {
+                // EOF is indicated by returning None
+                if ioerr.kind != EndOfFile {
+                    read_error::cond.raise(ioerr);
+                }
+                return None;
+            }
+        }
+    }
+
+    fn eof(&mut self) -> bool { fail!() }
+}
+
+impl Writer for PipeStream {
+    fn write(&mut self, buf: &[u8]) {
+        match (***self).write(buf) {
+            Ok(_) => (),
+            Err(ioerr) => {
+                io_error::cond.raise(ioerr);
+            }
+        }
+    }
+
+    fn flush(&mut self) { fail!() }
+}

src/libstd/rt/rtio.rs

@@ -8,12 +8,14 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use libc;
 use option::*;
 use result::*;
 use libc::c_int;
 
 use rt::io::IoError;
 use super::io::net::ip::{IpAddr, SocketAddr};
+use rt::uv;
 use rt::uv::uvio;
 use path::Path;
 use super::io::support::PathLike;
@@ -30,6 +32,9 @@ pub type RtioTcpListenerObject = uvio::UvTcpListener;
 pub type RtioUdpSocketObject = uvio::UvUdpSocket;
 pub type RtioTimerObject = uvio::UvTimer;
 pub type PausibleIdleCallback = uvio::UvPausibleIdleCallback;
+pub type RtioPipeObject = uvio::UvPipeStream;
+pub type RtioProcessObject = uvio::UvProcess;
+pub type RtioProcessConfig<'self> = uv::process::Config<'self>;
 
 pub trait EventLoop {
     fn run(&mut self);
@@ -72,6 +77,13 @@ pub trait IoFactory {
     fn fs_open<P: PathLike>(&mut self, path: &P, fm: FileMode, fa: FileAccess)
         -> Result<~RtioFileStream, IoError>;
     fn fs_unlink<P: PathLike>(&mut self, path: &P) -> Result<(), IoError>;
+    fn pipe_init(&mut self, ipc: bool) -> Result<~RtioPipeObject, IoError>;
+    fn spawn(&mut self, config: &RtioProcessConfig) -> Result<~RtioProcessObject, IoError>;
+}
+
+pub trait RtioStream {
+    fn read(&mut self, buf: &mut [u8]) -> Result<uint, IoError>;
+    fn write(&mut self, buf: &[u8]) -> Result<(), IoError>;
 }
 
 pub trait RtioTcpListener : RtioSocket {
@@ -80,9 +92,7 @@ pub trait RtioTcpListener : RtioSocket {
     fn dont_accept_simultaneously(&mut self) -> Result<(), IoError>;
 }
 
-pub trait RtioTcpStream : RtioSocket {
-    fn read(&mut self, buf: &mut [u8]) -> Result<uint, IoError>;
-    fn write(&mut self, buf: &[u8]) -> Result<(), IoError>;
+pub trait RtioTcpStream : RtioSocket + RtioStream {
     fn peer_name(&mut self) -> Result<SocketAddr, IoError>;
     fn control_congestion(&mut self) -> Result<(), IoError>;
     fn nodelay(&mut self) -> Result<(), IoError>;
@@ -124,3 +134,9 @@ pub trait RtioFileStream {
     fn tell(&self) -> Result<u64, IoError>;
     fn flush(&mut self) -> Result<(), IoError>;
 }
+
+pub trait RtioProcess {
+    fn id(&self) -> libc::pid_t;
+    fn kill(&mut self, signal: int) -> Result<(), IoError>;
+    fn wait(&mut self) -> int;
+}

src/libstd/rt/uv/async.rs

@@ -34,7 +34,7 @@ impl AsyncWatcher {
 
         extern fn async_cb(handle: *uvll::uv_async_t, status: c_int) {
             let mut watcher: AsyncWatcher = NativeHandle::from_native_handle(handle);
-            let status = status_to_maybe_uv_error(watcher, status);
+            let status = status_to_maybe_uv_error(status);
             let data = watcher.get_watcher_data();
             let cb = data.async_cb.get_ref();
             (*cb)(watcher, status);