src/librustuv/process.rs

   1 // Copyright 2013 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 use libc::c_int;
  12 use libc;
  13 use std::ptr;
  14 use std::c_str::CString;
  15 use std::rt::rtio;
  16 use std::rt::rtio::IoResult;
  17 use std::rt::task::BlockedTask;
  18
  19 use homing::{HomingIO, HomeHandle};
  20 use pipe::PipeWatcher;
  21 use super::{UvHandle, UvError, uv_error_to_io_error,
  22             wait_until_woken_after, wakeup, Loop};
  23 use timer::TimerWatcher;
  24 use uvio::UvIoFactory;
  25 use uvll;
  26
  27 pub struct Process {
  28     handle: *uvll::uv_process_t,
  29     home: HomeHandle,
  30
  31     /// Task to wake up (may be null) for when the process exits
  32     to_wake: Option<BlockedTask>,
  33
  34     /// Collected from the exit_cb
  35     exit_status: Option<rtio::ProcessExit>,
  36
  37     /// Lazily initialized timeout timer
  38     timer: Option<Box<TimerWatcher>>,
  39     timeout_state: TimeoutState,
  40 }
  41
  42 enum TimeoutState {
  43     NoTimeout,
  44     TimeoutPending,
  45     TimeoutElapsed,
  46 }
  47
  48 impl Process {
  49     /// Spawn a new process inside the specified event loop.
  50     ///
  51     /// Returns either the corresponding process object or an error which
  52     /// occurred.
  53     pub fn spawn(io_loop: &mut UvIoFactory, cfg: rtio::ProcessConfig)
  54                 -> Result<(Box<Process>, Vec<Option<PipeWatcher>>), UvError> {
  55         let mut io = vec![cfg.stdin, cfg.stdout, cfg.stderr];
  56         for slot in cfg.extra_io.iter() {
  57             io.push(*slot);
  58         }
  59         let mut stdio = Vec::<uvll::uv_stdio_container_t>::with_capacity(io.len());
  60         let mut ret_io = Vec::with_capacity(io.len());
  61         unsafe {
  62             stdio.set_len(io.len());
  63             for (slot, other) in stdio.iter().zip(io.iter()) {
  64                 let io = set_stdio(slot as *uvll::uv_stdio_container_t, other,
  65                                    io_loop);
  66                 ret_io.push(io);
  67             }
  68         }
  69
  70         let ret = with_argv(cfg.program, cfg.args, |argv| {
  71             with_env(cfg.env, |envp| {
  72                 let mut flags = 0;
  73                 if cfg.uid.is_some() {
  74                     flags |= uvll::PROCESS_SETUID;
  75                 }
  76                 if cfg.gid.is_some() {
  77                     flags |= uvll::PROCESS_SETGID;
  78                 }
  79                 if cfg.detach {
  80                     flags |= uvll::PROCESS_DETACHED;
  81                 }
  82                 let options = uvll::uv_process_options_t {
  83                     exit_cb: on_exit,
  84                     file: unsafe { *argv },
  85                     args: argv,
  86                     env: envp,
  87                     cwd: match cfg.cwd {
  88                         Some(cwd) => cwd.with_ref(|p| p),
  89                         None => ptr::null(),
  90                     },
  91                     flags: flags as libc::c_uint,
  92                     stdio_count: stdio.len() as libc::c_int,
  93                     stdio: stdio.as_ptr(),
  94                     uid: cfg.uid.unwrap_or(0) as uvll::uv_uid_t,
  95                     gid: cfg.gid.unwrap_or(0) as uvll::uv_gid_t,
  96                 };
  97
  98                 let handle = UvHandle::alloc(None::<Process>, uvll::UV_PROCESS);
  99                 let process = box Process {
 100                     handle: handle,
 101                     home: io_loop.make_handle(),
 102                     to_wake: None,
 103                     exit_status: None,
 104                     timer: None,
 105                     timeout_state: NoTimeout,
 106                 };
 107                 match unsafe {
 108                     uvll::uv_spawn(io_loop.uv_loop(), handle, &options)
 109                 } {
 110                     0 => Ok(process.install()),
 111                     err => Err(UvError(err)),
 112                 }
 113             })
 114         });
 115
 116         match ret {
 117             Ok(p) => Ok((p, ret_io)),
 118             Err(e) => Err(e),
 119         }
 120     }
 121
 122     pub fn kill(pid: libc::pid_t, signum: int) -> Result<(), UvError> {
 123         match unsafe {
 124             uvll::uv_kill(pid as libc::c_int, signum as libc::c_int)
 125         } {
 126             0 => Ok(()),
 127             n => Err(UvError(n))
 128         }
 129     }
 130 }
 131
 132 extern fn on_exit(handle: *uvll::uv_process_t,
 133                   exit_status: i64,
 134                   term_signal: libc::c_int) {
 135     let p: &mut Process = unsafe { UvHandle::from_uv_handle(&handle) };
 136
 137     assert!(p.exit_status.is_none());
 138     p.exit_status = Some(match term_signal {
 139         0 => rtio::ExitStatus(exit_status as int),
 140         n => rtio::ExitSignal(n as int),
 141     });
 142
 143     if p.to_wake.is_none() { return }
 144     wakeup(&mut p.to_wake);
 145 }
 146
 147 unsafe fn set_stdio(dst: *uvll::uv_stdio_container_t,
 148                     io: &rtio::StdioContainer,
 149                     io_loop: &mut UvIoFactory) -> Option<PipeWatcher> {
 150     match *io {
 151         rtio::Ignored => {
 152             uvll::set_stdio_container_flags(dst, uvll::STDIO_IGNORE);
 153             None
 154         }
 155         rtio::InheritFd(fd) => {
 156             uvll::set_stdio_container_flags(dst, uvll::STDIO_INHERIT_FD);
 157             uvll::set_stdio_container_fd(dst, fd);
 158             None
 159         }
 160         rtio::CreatePipe(readable, writable) => {
 161             let mut flags = uvll::STDIO_CREATE_PIPE as libc::c_int;
 162             if readable {
 163                 flags |= uvll::STDIO_READABLE_PIPE as libc::c_int;
 164             }
 165             if writable {
 166                 flags |= uvll::STDIO_WRITABLE_PIPE as libc::c_int;
 167             }
 168             let pipe = PipeWatcher::new(io_loop, false);
 169             uvll::set_stdio_container_flags(dst, flags);
 170             uvll::set_stdio_container_stream(dst, pipe.handle());
 171             Some(pipe)
 172         }
 173     }
 174 }
 175
 176 /// Converts the program and arguments to the argv array expected by libuv.
 177 fn with_argv<T>(prog: &CString, args: &[CString], cb: |**libc::c_char| -> T) -> T {
 178     let mut ptrs: Vec<*libc::c_char> = Vec::with_capacity(args.len()+1);
 179
 180     // Convert the CStrings into an array of pointers. Note: the
 181     // lifetime of the various CStrings involved is guaranteed to be
 182     // larger than the lifetime of our invocation of cb, but this is
 183     // technically unsafe as the callback could leak these pointers
 184     // out of our scope.
 185     ptrs.push(prog.with_ref(|buf| buf));
 186     ptrs.extend(args.iter().map(|tmp| tmp.with_ref(|buf| buf)));
 187
 188     // Add a terminating null pointer (required by libc).
 189     ptrs.push(ptr::null());
 190
 191     cb(ptrs.as_ptr())
 192 }
 193
 194 /// Converts the environment to the env array expected by libuv
 195 fn with_env<T>(env: Option<&[(CString, CString)]>, cb: |**libc::c_char| -> T) -> T {
 196     // We can pass a char** for envp, which is a null-terminated array
 197     // of "k=v\0" strings. Since we must create these strings locally,
 198     // yet expose a raw pointer to them, we create a temporary vector
 199     // to own the CStrings that outlives the call to cb.
 200     match env {
 201         Some(env) => {
 202             let mut tmps = Vec::with_capacity(env.len());
 203
 204             for pair in env.iter() {
 205                 let mut kv = Vec::new();
 206                 kv.push_all(pair.ref0().as_bytes_no_nul());
 207                 kv.push('=' as u8);
 208                 kv.push_all(pair.ref1().as_bytes()); // includes terminal \0
 209                 tmps.push(kv);
 210             }
 211
 212             // As with `with_argv`, this is unsafe, since cb could leak the pointers.
 213             let mut ptrs: Vec<*libc::c_char> =
 214                 tmps.iter()
 215                     .map(|tmp| tmp.as_ptr() as *libc::c_char)
 216                     .collect();
 217             ptrs.push(ptr::null());
 218
 219             cb(ptrs.as_ptr())
 220         }
 221         _ => cb(ptr::null())
 222     }
 223 }
 224
 225 impl HomingIO for Process {
 226     fn home<'r>(&'r mut self) -> &'r mut HomeHandle { &mut self.home }
 227 }
 228
 229 impl UvHandle<uvll::uv_process_t> for Process {
 230     fn uv_handle(&self) -> *uvll::uv_process_t { self.handle }
 231 }
 232
 233 impl rtio::RtioProcess for Process {
 234     fn id(&self) -> libc::pid_t {
 235         unsafe { uvll::process_pid(self.handle) as libc::pid_t }
 236     }
 237
 238     fn kill(&mut self, signal: int) -> IoResult<()> {
 239         let _m = self.fire_homing_missile();
 240         match unsafe {
 241             uvll::uv_process_kill(self.handle, signal as libc::c_int)
 242         } {
 243             0 => Ok(()),
 244             err => Err(uv_error_to_io_error(UvError(err)))
 245         }
 246     }
 247
 248     fn wait(&mut self) -> IoResult<rtio::ProcessExit> {
 249         // Make sure (on the home scheduler) that we have an exit status listed
 250         let _m = self.fire_homing_missile();
 251         match self.exit_status {
 252             Some(status) => return Ok(status),
 253             None => {}
 254         }
 255
 256         // If there's no exit code previously listed, then the process's exit
 257         // callback has yet to be invoked. We just need to deschedule ourselves
 258         // and wait to be reawoken.
 259         match self.timeout_state {
 260             NoTimeout | TimeoutPending => {
 261                 wait_until_woken_after(&mut self.to_wake, &self.uv_loop(), || {});
 262             }
 263             TimeoutElapsed => {}
 264         }
 265
 266         // If there's still no exit status listed, then we timed out, and we
 267         // need to return.
 268         match self.exit_status {
 269             Some(status) => Ok(status),
 270             None => Err(uv_error_to_io_error(UvError(uvll::ECANCELED)))
 271         }
 272     }
 273
 274     fn set_timeout(&mut self, timeout: Option<u64>) {
 275         let _m = self.fire_homing_missile();
 276         self.timeout_state = NoTimeout;
 277         let ms = match timeout {
 278             Some(ms) => ms,
 279             None => {
 280                 match self.timer {
 281                     Some(ref mut timer) => timer.stop(),
 282                     None => {}
 283                 }
 284                 return
 285             }
 286         };
 287         if self.timer.is_none() {
 288             let loop_ = Loop::wrap(unsafe {
 289                 uvll::get_loop_for_uv_handle(self.uv_handle())
 290             });
 291             let mut timer = box TimerWatcher::new_home(&loop_, self.home().clone());
 292             unsafe {
 293                 timer.set_data(self as *mut _ as *Process);
 294             }
 295             self.timer = Some(timer);
 296         }
 297
 298         let timer = self.timer.get_mut_ref();
 299         timer.stop();
 300         timer.start(timer_cb, ms, 0);
 301         self.timeout_state = TimeoutPending;
 302
 303         extern fn timer_cb(timer: *uvll::uv_timer_t) {
 304             let p: &mut Process = unsafe {
 305                 &mut *(uvll::get_data_for_uv_handle(timer) as *mut Process)
 306             };
 307             p.timeout_state = TimeoutElapsed;
 308             match p.to_wake.take() {
 309                 Some(task) => { let _t = task.wake().map(|t| t.reawaken()); }
 310                 None => {}
 311             }
 312         }
 313     }
 314 }
 315
 316 impl Drop for Process {
 317     fn drop(&mut self) {
 318         let _m = self.fire_homing_missile();
 319         assert!(self.to_wake.is_none());
 320         self.close();
 321     }
 322 }