src/libstd/sys/windows/pipe.rs

   1 use crate::os::windows::prelude::*;
   2
   3 use crate::ffi::OsStr;
   4 use crate::io::{self, IoSlice, IoSliceMut};
   5 use crate::mem;
   6 use crate::path::Path;
   7 use crate::ptr;
   8 use crate::slice;
   9 use crate::sync::atomic::Ordering::SeqCst;
  10 use crate::sync::atomic::AtomicUsize;
  11 use crate::sys::c;
  12 use crate::sys::fs::{File, OpenOptions};
  13 use crate::sys::handle::Handle;
  14 use crate::sys::hashmap_random_keys;
  15
  16 ////////////////////////////////////////////////////////////////////////////////
  17 // Anonymous pipes
  18 ////////////////////////////////////////////////////////////////////////////////
  19
  20 pub struct AnonPipe {
  21     inner: Handle,
  22 }
  23
  24 pub struct Pipes {
  25     pub ours: AnonPipe,
  26     pub theirs: AnonPipe,
  27 }
  28
  29 /// Although this looks similar to `anon_pipe` in the Unix module it's actually
  30 /// subtly different. Here we'll return two pipes in the `Pipes` return value,
  31 /// but one is intended for "us" where as the other is intended for "someone
  32 /// else".
  33 ///
  34 /// Currently the only use case for this function is pipes for stdio on
  35 /// processes in the standard library, so "ours" is the one that'll stay in our
  36 /// process whereas "theirs" will be inherited to a child.
  37 ///
  38 /// The ours/theirs pipes are *not* specifically readable or writable. Each
  39 /// one only supports a read or a write, but which is which depends on the
  40 /// boolean flag given. If `ours_readable` is `true`, then `ours` is readable and
  41 /// `theirs` is writable. Conversely, if `ours_readable` is `false`, then `ours`
  42 /// is writable and `theirs` is readable.
  43 ///
  44 /// Also note that the `ours` pipe is always a handle opened up in overlapped
  45 /// mode. This means that technically speaking it should only ever be used
  46 /// with `OVERLAPPED` instances, but also works out ok if it's only ever used
  47 /// once at a time (which we do indeed guarantee).
  48 pub fn anon_pipe(ours_readable: bool) -> io::Result<Pipes> {
  49     // Note that we specifically do *not* use `CreatePipe` here because
  50     // unfortunately the anonymous pipes returned do not support overlapped
  51     // operations. Instead, we create a "hopefully unique" name and create a
  52     // named pipe which has overlapped operations enabled.
  53     //
  54     // Once we do this, we connect do it as usual via `CreateFileW`, and then
  55     // we return those reader/writer halves. Note that the `ours` pipe return
  56     // value is always the named pipe, whereas `theirs` is just the normal file.
  57     // This should hopefully shield us from child processes which assume their
  58     // stdout is a named pipe, which would indeed be odd!
  59     unsafe {
  60         let ours;
  61         let mut name;
  62         let mut tries = 0;
  63         let mut reject_remote_clients_flag = c::PIPE_REJECT_REMOTE_CLIENTS;
  64         loop {
  65             tries += 1;
  66             name = format!(r"\\.\pipe\__rust_anonymous_pipe1__.{}.{}",
  67                            c::GetCurrentProcessId(),
  68                            random_number());
  69             let wide_name = OsStr::new(&name)
  70                                   .encode_wide()
  71                                   .chain(Some(0))
  72                                   .collect::<Vec<_>>();
  73             let mut flags = c::FILE_FLAG_FIRST_PIPE_INSTANCE |
  74                 c::FILE_FLAG_OVERLAPPED;
  75             if ours_readable {
  76                 flags |= c::PIPE_ACCESS_INBOUND;
  77             } else {
  78                 flags |= c::PIPE_ACCESS_OUTBOUND;
  79             }
  80
  81             let handle = c::CreateNamedPipeW(wide_name.as_ptr(),
  82                                              flags,
  83                                              c::PIPE_TYPE_BYTE |
  84                                              c::PIPE_READMODE_BYTE |
  85                                              c::PIPE_WAIT |
  86                                              reject_remote_clients_flag,
  87                                              1,
  88                                              4096,
  89                                              4096,
  90                                              0,
  91                                              ptr::null_mut());
  92
  93             // We pass the `FILE_FLAG_FIRST_PIPE_INSTANCE` flag above, and we're
  94             // also just doing a best effort at selecting a unique name. If
  95             // `ERROR_ACCESS_DENIED` is returned then it could mean that we
  96             // accidentally conflicted with an already existing pipe, so we try
  97             // again.
  98             //
  99             // Don't try again too much though as this could also perhaps be a
 100             // legit error.
 101             // If `ERROR_INVALID_PARAMETER` is returned, this probably means we're
 102             // running on pre-Vista version where `PIPE_REJECT_REMOTE_CLIENTS` is
 103             // not supported, so we continue retrying without it. This implies
 104             // reduced security on Windows versions older than Vista by allowing
 105             // connections to this pipe from remote machines.
 106             // Proper fix would increase the number of FFI imports and introduce
 107             // significant amount of Windows XP specific code with no clean
 108             // testing strategy
 109             // For more info, see https://github.com/rust-lang/rust/pull/37677.
 110             if handle == c::INVALID_HANDLE_VALUE {
 111                 let err = io::Error::last_os_error();
 112                 let raw_os_err = err.raw_os_error();
 113                 if tries < 10 {
 114                     if raw_os_err == Some(c::ERROR_ACCESS_DENIED as i32) {
 115                         continue
 116                     } else if reject_remote_clients_flag != 0 &&
 117                         raw_os_err == Some(c::ERROR_INVALID_PARAMETER as i32) {
 118                         reject_remote_clients_flag = 0;
 119                         tries -= 1;
 120                         continue
 121                     }
 122                 }
 123                 return Err(err)
 124             }
 125             ours = Handle::new(handle);
 126             break
 127         }
 128
 129         // Connect to the named pipe we just created. This handle is going to be
 130         // returned in `theirs`, so if `ours` is readable we want this to be
 131         // writable, otherwise if `ours` is writable we want this to be
 132         // readable.
 133         //
 134         // Additionally we don't enable overlapped mode on this because most
 135         // client processes aren't enabled to work with that.
 136         let mut opts = OpenOptions::new();
 137         opts.write(ours_readable);
 138         opts.read(!ours_readable);
 139         opts.share_mode(0);
 140         let theirs = File::open(Path::new(&name), &opts)?;
 141         let theirs = AnonPipe { inner: theirs.into_handle() };
 142
 143         Ok(Pipes {
 144             ours: AnonPipe { inner: ours },
 145             theirs: AnonPipe { inner: theirs.into_handle() },
 146         })
 147     }
 148 }
 149
 150 fn random_number() -> usize {
 151     static N: AtomicUsize = AtomicUsize::new(0);
 152     loop {
 153         if N.load(SeqCst) != 0 {
 154             return N.fetch_add(1, SeqCst)
 155         }
 156
 157         N.store(hashmap_random_keys().0 as usize, SeqCst);
 158     }
 159 }
 160
 161 impl AnonPipe {
 162     pub fn handle(&self) -> &Handle { &self.inner }
 163     pub fn into_handle(self) -> Handle { self.inner }
 164
 165     pub fn read(&self, buf: &mut [u8]) -> io::Result<usize> {
 166         self.inner.read(buf)
 167     }
 168
 169     pub fn read_vectored(&self, bufs: &mut [IoSliceMut<'_>]) -> io::Result<usize> {
 170         self.inner.read_vectored(bufs)
 171     }
 172
 173     pub fn write(&self, buf: &[u8]) -> io::Result<usize> {
 174         self.inner.write(buf)
 175     }
 176
 177     pub fn write_vectored(&self, bufs: &[IoSlice<'_>]) -> io::Result<usize> {
 178         self.inner.write_vectored(bufs)
 179     }
 180 }
 181
 182 pub fn read2(p1: AnonPipe,
 183              v1: &mut Vec<u8>,
 184              p2: AnonPipe,
 185              v2: &mut Vec<u8>) -> io::Result<()> {
 186     let p1 = p1.into_handle();
 187     let p2 = p2.into_handle();
 188
 189     let mut p1 = AsyncPipe::new(p1, v1)?;
 190     let mut p2 = AsyncPipe::new(p2, v2)?;
 191     let objs = [p1.event.raw(), p2.event.raw()];
 192
 193     // In a loop we wait for either pipe's scheduled read operation to complete.
 194     // If the operation completes with 0 bytes, that means EOF was reached, in
 195     // which case we just finish out the other pipe entirely.
 196     //
 197     // Note that overlapped I/O is in general super unsafe because we have to
 198     // be careful to ensure that all pointers in play are valid for the entire
 199     // duration of the I/O operation (where tons of operations can also fail).
 200     // The destructor for `AsyncPipe` ends up taking care of most of this.
 201     loop {
 202         let res = unsafe {
 203             c::WaitForMultipleObjects(2, objs.as_ptr(), c::FALSE, c::INFINITE)
 204         };
 205         if res == c::WAIT_OBJECT_0 {
 206             if !p1.result()? || !p1.schedule_read()? {
 207                 return p2.finish()
 208             }
 209         } else if res == c::WAIT_OBJECT_0 + 1 {
 210             if !p2.result()? || !p2.schedule_read()? {
 211                 return p1.finish()
 212             }
 213         } else {
 214             return Err(io::Error::last_os_error())
 215         }
 216     }
 217 }
 218
 219 struct AsyncPipe<'a> {
 220     pipe: Handle,
 221     event: Handle,
 222     overlapped: Box<c::OVERLAPPED>, // needs a stable address
 223     dst: &'a mut Vec<u8>,
 224     state: State,
 225 }
 226
 227 #[derive(PartialEq, Debug)]
 228 enum State {
 229     NotReading,
 230     Reading,
 231     Read(usize),
 232 }
 233
 234 impl<'a> AsyncPipe<'a> {
 235     fn new(pipe: Handle, dst: &'a mut Vec<u8>) -> io::Result<AsyncPipe<'a>> {
 236         // Create an event which we'll use to coordinate our overlapped
 237         // operations, this event will be used in WaitForMultipleObjects
 238         // and passed as part of the OVERLAPPED handle.
 239         //
 240         // Note that we do a somewhat clever thing here by flagging the
 241         // event as being manually reset and setting it initially to the
 242         // signaled state. This means that we'll naturally fall through the
 243         // WaitForMultipleObjects call above for pipes created initially,
 244         // and the only time an even will go back to "unset" will be once an
 245         // I/O operation is successfully scheduled (what we want).
 246         let event = Handle::new_event(true, true)?;
 247         let mut overlapped: Box<c::OVERLAPPED> = unsafe {
 248             Box::new(mem::zeroed())
 249         };
 250         overlapped.hEvent = event.raw();
 251         Ok(AsyncPipe {
 252             pipe,
 253             overlapped,
 254             event,
 255             dst,
 256             state: State::NotReading,
 257         })
 258     }
 259
 260     /// Executes an overlapped read operation.
 261     ///
 262     /// Must not currently be reading, and returns whether the pipe is currently
 263     /// at EOF or not. If the pipe is not at EOF then `result()` must be called
 264     /// to complete the read later on (may block), but if the pipe is at EOF
 265     /// then `result()` should not be called as it will just block forever.
 266     fn schedule_read(&mut self) -> io::Result<bool> {
 267         assert_eq!(self.state, State::NotReading);
 268         let amt = unsafe {
 269             let slice = slice_to_end(self.dst);
 270             self.pipe.read_overlapped(slice, &mut *self.overlapped)?
 271         };
 272
 273         // If this read finished immediately then our overlapped event will
 274         // remain signaled (it was signaled coming in here) and we'll progress
 275         // down to the method below.
 276         //
 277         // Otherwise the I/O operation is scheduled and the system set our event
 278         // to not signaled, so we flag ourselves into the reading state and move
 279         // on.
 280         self.state = match amt {
 281             Some(0) => return Ok(false),
 282             Some(amt) => State::Read(amt),
 283             None => State::Reading,
 284         };
 285         Ok(true)
 286     }
 287
 288     /// Wait for the result of the overlapped operation previously executed.
 289     ///
 290     /// Takes a parameter `wait` which indicates if this pipe is currently being
 291     /// read whether the function should block waiting for the read to complete.
 292     ///
 293     /// Returns values:
 294     ///
 295     /// * `true` - finished any pending read and the pipe is not at EOF (keep
 296     ///            going)
 297     /// * `false` - finished any pending read and pipe is at EOF (stop issuing
 298     ///             reads)
 299     fn result(&mut self) -> io::Result<bool> {
 300         let amt = match self.state {
 301             State::NotReading => return Ok(true),
 302             State::Reading => {
 303                 self.pipe.overlapped_result(&mut *self.overlapped, true)?
 304             }
 305             State::Read(amt) => amt,
 306         };
 307         self.state = State::NotReading;
 308         unsafe {
 309             let len = self.dst.len();
 310             self.dst.set_len(len + amt);
 311         }
 312         Ok(amt != 0)
 313     }
 314
 315     /// Finishes out reading this pipe entirely.
 316     ///
 317     /// Waits for any pending and schedule read, and then calls `read_to_end`
 318     /// if necessary to read all the remaining information.
 319     fn finish(&mut self) -> io::Result<()> {
 320         while self.result()? && self.schedule_read()? {
 321             // ...
 322         }
 323         Ok(())
 324     }
 325 }
 326
 327 impl<'a> Drop for AsyncPipe<'a> {
 328     fn drop(&mut self) {
 329         match self.state {
 330             State::Reading => {}
 331             _ => return,
 332         }
 333
 334         // If we have a pending read operation, then we have to make sure that
 335         // it's *done* before we actually drop this type. The kernel requires
 336         // that the `OVERLAPPED` and buffer pointers are valid for the entire
 337         // I/O operation.
 338         //
 339         // To do that, we call `CancelIo` to cancel any pending operation, and
 340         // if that succeeds we wait for the overlapped result.
 341         //
 342         // If anything here fails, there's not really much we can do, so we leak
 343         // the buffer/OVERLAPPED pointers to ensure we're at least memory safe.
 344         if self.pipe.cancel_io().is_err() || self.result().is_err() {
 345             let buf = mem::take(self.dst);
 346             let overlapped = Box::new(unsafe { mem::zeroed() });
 347             let overlapped = mem::replace(&mut self.overlapped, overlapped);
 348             mem::forget((buf, overlapped));
 349         }
 350     }
 351 }
 352
 353 unsafe fn slice_to_end(v: &mut Vec<u8>) -> &mut [u8] {
 354     if v.capacity() == 0 {
 355         v.reserve(16);
 356     }
 357     if v.capacity() == v.len() {
 358         v.reserve(1);
 359     }
 360     slice::from_raw_parts_mut(v.as_mut_ptr().add(v.len()),
 361                               v.capacity() - v.len())
 362 }