rust/crates/stdx/src/process.rs

268 lines
8.0 KiB
Rust
Raw Normal View History

//! Read both stdout and stderr of child without deadlocks.
//!
//! <https://github.com/rust-lang/cargo/blob/905af549966f23a9288e9993a85d1249a5436556/crates/cargo-util/src/read2.rs>
//! <https://github.com/rust-lang/cargo/blob/58a961314437258065e23cb6316dfc121d96fb71/crates/cargo-util/src/process_builder.rs#L231>
use std::{
io,
process::{ChildStderr, ChildStdout, Command, Output, Stdio},
};
use crate::JodChild;
pub fn streaming_output(
out: ChildStdout,
err: ChildStderr,
on_stdout_line: &mut dyn FnMut(&str),
on_stderr_line: &mut dyn FnMut(&str),
) -> io::Result<(Vec<u8>, Vec<u8>)> {
let mut stdout = Vec::new();
let mut stderr = Vec::new();
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);
}
}
}
})?;
Ok((stdout, stderr))
}
pub fn spawn_with_streaming_output(
mut cmd: Command,
on_stdout_line: &mut dyn FnMut(&str),
on_stderr_line: &mut dyn FnMut(&str),
) -> io::Result<Output> {
let cmd = cmd.stdout(Stdio::piped()).stderr(Stdio::piped()).stdin(Stdio::null());
let mut child = JodChild(cmd.spawn()?);
let (stdout, stderr) = streaming_output(
child.stdout.take().unwrap(),
child.stderr.take().unwrap(),
on_stdout_line,
on_stderr_line,
)?;
let status = 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<u8>, 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<u8>,
overlapped: Overlapped,
pipe: NamedPipe,
done: bool,
}
pub(crate) fn read2(
out_pipe: ChildStdout,
err_pipe: ChildStderr,
data: &mut dyn FnMut(bool, &mut Vec<u8>, 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: IntoRawHandle>(p: P, dst: &'a mut Vec<u8>) -> 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<u8>) -> &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<u8>, bool),
) -> io::Result<()> {
panic!("no processes on wasm")
}
}