//! Read both stdout and stderr of child without deadlocks. //! //! //! use std::{ io, process::{Command, Output, Stdio}, }; pub fn streaming_output( mut cmd: Command, on_stdout_line: &mut dyn FnMut(&str), on_stderr_line: &mut dyn FnMut(&str), ) -> io::Result { let mut stdout = Vec::new(); let mut stderr = Vec::new(); let cmd = cmd.stdout(Stdio::piped()).stderr(Stdio::piped()).stdin(Stdio::null()); let status = { let mut child = cmd.spawn()?; let out = child.stdout.take().unwrap(); let err = child.stderr.take().unwrap(); imp::read2(out, err, &mut |is_out, data, eof| { let idx = if eof { data.len() } else { match data.iter().rposition(|b| *b == b'\n') { Some(i) => i + 1, None => return, } }; { // scope for new_lines let new_lines = { let dst = if is_out { &mut stdout } else { &mut stderr }; let start = dst.len(); let data = data.drain(..idx); dst.extend(data); &dst[start..] }; for line in String::from_utf8_lossy(new_lines).lines() { if is_out { on_stdout_line(line) } else { on_stderr_line(line) } } } })?; child.wait()? }; Ok(Output { status, stdout, stderr }) } #[cfg(unix)] mod imp { use std::{ io::{self, prelude::*}, mem, os::unix::prelude::*, process::{ChildStderr, ChildStdout}, }; pub(crate) fn read2( mut out_pipe: ChildStdout, mut err_pipe: ChildStderr, data: &mut dyn FnMut(bool, &mut Vec, bool), ) -> io::Result<()> { unsafe { libc::fcntl(out_pipe.as_raw_fd(), libc::F_SETFL, libc::O_NONBLOCK); libc::fcntl(err_pipe.as_raw_fd(), libc::F_SETFL, libc::O_NONBLOCK); } let mut out_done = false; let mut err_done = false; let mut out = Vec::new(); let mut err = Vec::new(); let mut fds: [libc::pollfd; 2] = unsafe { mem::zeroed() }; fds[0].fd = out_pipe.as_raw_fd(); fds[0].events = libc::POLLIN; fds[1].fd = err_pipe.as_raw_fd(); fds[1].events = libc::POLLIN; let mut nfds = 2; let mut errfd = 1; while nfds > 0 { // wait for either pipe to become readable using `select` let r = unsafe { libc::poll(fds.as_mut_ptr(), nfds, -1) }; if r == -1 { let err = io::Error::last_os_error(); if err.kind() == io::ErrorKind::Interrupted { continue; } return Err(err); } // Read as much as we can from each pipe, ignoring EWOULDBLOCK or // EAGAIN. If we hit EOF, then this will happen because the underlying // reader will return Ok(0), in which case we'll see `Ok` ourselves. In // this case we flip the other fd back into blocking mode and read // whatever's leftover on that file descriptor. let handle = |res: io::Result<_>| match res { Ok(_) => Ok(true), Err(e) => { if e.kind() == io::ErrorKind::WouldBlock { Ok(false) } else { Err(e) } } }; if !err_done && fds[errfd].revents != 0 && handle(err_pipe.read_to_end(&mut err))? { err_done = true; nfds -= 1; } data(false, &mut err, err_done); if !out_done && fds[0].revents != 0 && handle(out_pipe.read_to_end(&mut out))? { out_done = true; fds[0].fd = err_pipe.as_raw_fd(); errfd = 0; nfds -= 1; } data(true, &mut out, out_done); } Ok(()) } } #[cfg(windows)] mod imp { use std::{ io, os::windows::prelude::*, process::{ChildStderr, ChildStdout}, slice, }; use miow::{ iocp::{CompletionPort, CompletionStatus}, pipe::NamedPipe, Overlapped, }; use winapi::shared::winerror::ERROR_BROKEN_PIPE; struct Pipe<'a> { dst: &'a mut Vec, overlapped: Overlapped, pipe: NamedPipe, done: bool, } pub(crate) fn read2( out_pipe: ChildStdout, err_pipe: ChildStderr, data: &mut dyn FnMut(bool, &mut Vec, bool), ) -> io::Result<()> { let mut out = Vec::new(); let mut err = Vec::new(); let port = CompletionPort::new(1)?; port.add_handle(0, &out_pipe)?; port.add_handle(1, &err_pipe)?; unsafe { let mut out_pipe = Pipe::new(out_pipe, &mut out); let mut err_pipe = Pipe::new(err_pipe, &mut err); out_pipe.read()?; err_pipe.read()?; let mut status = [CompletionStatus::zero(), CompletionStatus::zero()]; while !out_pipe.done || !err_pipe.done { for status in port.get_many(&mut status, None)? { if status.token() == 0 { out_pipe.complete(status); data(true, out_pipe.dst, out_pipe.done); out_pipe.read()?; } else { err_pipe.complete(status); data(false, err_pipe.dst, err_pipe.done); err_pipe.read()?; } } } Ok(()) } } impl<'a> Pipe<'a> { unsafe fn new(p: P, dst: &'a mut Vec) -> Pipe<'a> { Pipe { dst, pipe: NamedPipe::from_raw_handle(p.into_raw_handle()), overlapped: Overlapped::zero(), done: false, } } unsafe fn read(&mut self) -> io::Result<()> { let dst = slice_to_end(self.dst); match self.pipe.read_overlapped(dst, self.overlapped.raw()) { Ok(_) => Ok(()), Err(e) => { if e.raw_os_error() == Some(ERROR_BROKEN_PIPE as i32) { self.done = true; Ok(()) } else { Err(e) } } } } unsafe fn complete(&mut self, status: &CompletionStatus) { let prev = self.dst.len(); self.dst.set_len(prev + status.bytes_transferred() as usize); if status.bytes_transferred() == 0 { self.done = true; } } } unsafe fn slice_to_end(v: &mut Vec) -> &mut [u8] { if v.capacity() == 0 { v.reserve(16); } if v.capacity() == v.len() { v.reserve(1); } slice::from_raw_parts_mut(v.as_mut_ptr().add(v.len()), v.capacity() - v.len()) } } #[cfg(target_arch = "wasm32")] mod imp { use std::{ io, process::{ChildStderr, ChildStdout}, }; pub(crate) fn read2( _out_pipe: ChildStdout, _err_pipe: ChildStderr, _data: &mut dyn FnMut(bool, &mut Vec, bool), ) -> io::Result<()> { panic!("no processes on wasm") } }