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: *mut 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.mut_iter().zip(io.iter()) {
  64                 let io = set_stdio(slot as *mut 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 mut 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_mut_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, &mut 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: *mut 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: *mut 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],
 178                 cb: |*const *const libc::c_char| -> T) -> T {
 179     let mut ptrs: Vec<*const libc::c_char> = Vec::with_capacity(args.len()+1);
 180
 181     // Convert the CStrings into an array of pointers. Note: the
 182     // lifetime of the various CStrings involved is guaranteed to be
 183     // larger than the lifetime of our invocation of cb, but this is
 184     // technically unsafe as the callback could leak these pointers
 185     // out of our scope.
 186     ptrs.push(prog.with_ref(|buf| buf));
 187     ptrs.extend(args.iter().map(|tmp| tmp.with_ref(|buf| buf)));
 188
 189     // Add a terminating null pointer (required by libc).
 190     ptrs.push(ptr::null());
 191
 192     cb(ptrs.as_ptr())
 193 }
 194
 195 /// Converts the environment to the env array expected by libuv
 196 fn with_env<T>(env: Option<&[(CString, CString)]>,
 197                cb: |*const *const libc::c_char| -> T) -> T {
 198     // We can pass a char** for envp, which is a null-terminated array
 199     // of "k=v\0" strings. Since we must create these strings locally,
 200     // yet expose a raw pointer to them, we create a temporary vector
 201     // to own the CStrings that outlives the call to cb.
 202     match env {
 203         Some(env) => {
 204             let mut tmps = Vec::with_capacity(env.len());
 205
 206             for pair in env.iter() {
 207                 let mut kv = Vec::new();
 208                 kv.push_all(pair.ref0().as_bytes_no_nul());
 209                 kv.push('=' as u8);
 210                 kv.push_all(pair.ref1().as_bytes()); // includes terminal \0
 211                 tmps.push(kv);
 212             }
 213
 214             // As with `with_argv`, this is unsafe, since cb could leak the pointers.
 215             let mut ptrs: Vec<*const libc::c_char> =
 216                 tmps.iter()
 217                     .map(|tmp| tmp.as_ptr() as *const libc::c_char)
 218                     .collect();
 219             ptrs.push(ptr::null());
 220
 221             cb(ptrs.as_ptr())
 222         }
 223         _ => cb(ptr::null())
 224     }
 225 }
 226
 227 impl HomingIO for Process {
 228     fn home<'r>(&'r mut self) -> &'r mut HomeHandle { &mut self.home }
 229 }
 230
 231 impl UvHandle<uvll::uv_process_t> for Process {
 232     fn uv_handle(&self) -> *mut uvll::uv_process_t { self.handle }
 233 }
 234
 235 impl rtio::RtioProcess for Process {
 236     fn id(&self) -> libc::pid_t {
 237         unsafe { uvll::process_pid(self.handle) as libc::pid_t }
 238     }
 239
 240     fn kill(&mut self, signal: int) -> IoResult<()> {
 241         let _m = self.fire_homing_missile();
 242         match unsafe {
 243             uvll::uv_process_kill(self.handle, signal as libc::c_int)
 244         } {
 245             0 => Ok(()),
 246             err => Err(uv_error_to_io_error(UvError(err)))
 247         }
 248     }
 249
 250     fn wait(&mut self) -> IoResult<rtio::ProcessExit> {
 251         // Make sure (on the home scheduler) that we have an exit status listed
 252         let _m = self.fire_homing_missile();
 253         match self.exit_status {
 254             Some(status) => return Ok(status),
 255             None => {}
 256         }
 257
 258         // If there's no exit code previously listed, then the process's exit
 259         // callback has yet to be invoked. We just need to deschedule ourselves
 260         // and wait to be reawoken.
 261         match self.timeout_state {
 262             NoTimeout | TimeoutPending => {
 263                 wait_until_woken_after(&mut self.to_wake, &self.uv_loop(), || {});
 264             }
 265             TimeoutElapsed => {}
 266         }
 267
 268         // If there's still no exit status listed, then we timed out, and we
 269         // need to return.
 270         match self.exit_status {
 271             Some(status) => Ok(status),
 272             None => Err(uv_error_to_io_error(UvError(uvll::ECANCELED)))
 273         }
 274     }
 275
 276     fn set_timeout(&mut self, timeout: Option<u64>) {
 277         let _m = self.fire_homing_missile();
 278         self.timeout_state = NoTimeout;
 279         let ms = match timeout {
 280             Some(ms) => ms,
 281             None => {
 282                 match self.timer {
 283                     Some(ref mut timer) => timer.stop(),
 284                     None => {}
 285                 }
 286                 return
 287             }
 288         };
 289         if self.timer.is_none() {
 290             let loop_ = Loop::wrap(unsafe {
 291                 uvll::get_loop_for_uv_handle(self.uv_handle())
 292             });
 293             let mut timer = box TimerWatcher::new_home(&loop_, self.home().clone());
 294             unsafe {
 295                 timer.set_data(self as *mut _);
 296             }
 297             self.timer = Some(timer);
 298         }
 299
 300         let timer = self.timer.get_mut_ref();
 301         timer.stop();
 302         timer.start(timer_cb, ms, 0);
 303         self.timeout_state = TimeoutPending;
 304
 305         extern fn timer_cb(timer: *mut uvll::uv_timer_t) {
 306             let p: &mut Process = unsafe {
 307                 &mut *(uvll::get_data_for_uv_handle(timer) as *mut Process)
 308             };
 309             p.timeout_state = TimeoutElapsed;
 310             match p.to_wake.take() {
 311                 Some(task) => { let _t = task.wake().map(|t| t.reawaken()); }
 312                 None => {}
 313             }
 314         }
 315     }
 316 }
 317
 318 impl Drop for Process {
 319     fn drop(&mut self) {
 320         let _m = self.fire_homing_missile();
 321         assert!(self.to_wake.is_none());
 322         self.close();
 323     }
 324 }