Skip to content

Commit

Permalink
Remove dependency on pipe, unless parallel
Browse files Browse the repository at this point in the history
  • Loading branch information
@dpaoliello
dpaoliello committed Feb 2, 2024
1 parent 2b52daf commit 0c8151d
Show file tree
Hide file tree
Showing 7 changed files with 210 additions and 315 deletions.
4 changes: 2 additions & 2 deletions Cargo.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,10 @@ rust-version = "1.53"
[target.'cfg(unix)'.dependencies]
# Don't turn on the feature "std" for this, see https://github.com/rust-lang/cargo/issues/4866
# which is still an issue with `resolver = "1"`.
libc = { version = "0.2.62", default-features = false }
libc = { version = "0.2.62", default-features = false, optional = true }

[features]
parallel = []
parallel = ["libc"]

[dev-dependencies]
tempfile = "3"
Expand Down
260 changes: 172 additions & 88 deletions src/command_helpers.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,12 @@ use std::{
collections::hash_map,
ffi::OsString,
fmt::Display,
fs::{self, File},
fs,
hash::Hasher,
io::{self, BufRead, BufReader, Read, Write},
io::{self, Read, Write},
path::Path,
process::{Child, Command, Stdio},
process::{Child, ChildStderr, Command, Stdio},
sync::Arc,
thread::{self, JoinHandle},
};

use crate::{Error, ErrorKind, Object};
Expand Down Expand Up @@ -41,83 +40,175 @@ impl CargoOutput {
}
}

pub(crate) fn print_thread(&self) -> Result<Option<PrintThread>, Error> {
self.warnings.then(PrintThread::new).transpose()
fn stdio_for_warnings(&self) -> Stdio {
if self.warnings {
Stdio::piped()
} else {
Stdio::null()
}
}
}

pub(crate) struct PrintThread {
handle: Option<JoinHandle<()>>,
pipe_writer: Option<File>,
pub(crate) struct StderrForwarder {
inner: Option<(ChildStderr, Vec<u8>)>,
#[cfg(feature = "parallel")]
is_non_blocking: bool,
}

impl PrintThread {
pub(crate) fn new() -> Result<Self, Error> {
let (pipe_reader, pipe_writer) = crate::os_pipe::pipe()?;

// Capture the standard error coming from compilation, and write it out
// with cargo:warning= prefixes. Note that this is a bit wonky to avoid
// requiring the output to be UTF-8, we instead just ship bytes from one
// location to another.
let print = thread::spawn(move || {
let mut stderr = BufReader::with_capacity(4096, pipe_reader);
let mut line = Vec::with_capacity(20);
let stdout = io::stdout();
#[cfg(all(feature = "parallel", not(unix), not(windows)))]
compile_error!("Only unix and windows support non-blocking pipes! For other OSes, disable the parallel feature.");

const MIN_BUFFER_CAPACITY: usize = 100;

impl StderrForwarder {
pub(crate) fn new(child: &mut Child) -> Self {
Self {
inner: child
.stderr
.take()
.map(|stderr| (stderr, Vec::with_capacity(MIN_BUFFER_CAPACITY))),
#[cfg(feature = "parallel")]
is_non_blocking: false,
}
}

// read_until returns 0 on Eof
while stderr.read_until(b'\n', &mut line).unwrap() != 0 {
{
let mut stdout = stdout.lock();
fn forward_available(&mut self) -> bool {
if let Some((stderr, buffer)) = self.inner.as_mut() {
let stdout = io::stdout();
let write_warning = move |line: &[u8]| {
let mut stdout = stdout.lock();
stdout.write_all(b"cargo:warning=").unwrap();
stdout.write_all(line).unwrap();
stdout.write_all(b"\n").unwrap();
};

stdout.write_all(b"cargo:warning=").unwrap();
stdout.write_all(&line).unwrap();
stdout.write_all(b"\n").unwrap();
#[cfg(all(windows, feature = "parallel"))]
let is_non_blocking = self.is_non_blocking;
let mut read_stderr = move |buf: &mut [u8]| -> Result<usize, io::Error> {
// On Unix, the pipe is non-blocking, so we can just read.
// On Windows, take a peek at the pipe to see if there's data.
#[cfg(all(windows, feature = "parallel"))]
if is_non_blocking {
use crate::windows::windows_sys::PeekNamedPipe;
use std::os::windows::io::AsRawHandle;
use std::ptr::null_mut;
let mut bytes_available = 0;
unsafe {
if PeekNamedPipe(
stderr.as_raw_handle(),
null_mut(),
0,
null_mut(),
&mut bytes_available,
null_mut(),
) == 0
{
return Err(io::Error::last_os_error());
}
}
if bytes_available == 0 {
return Err(io::Error::new(
io::ErrorKind::WouldBlock,
"The pipe is empty",
));
}
}

// read_until does not clear the buffer
line.clear();
}
});
stderr.read(buf)
};

Ok(Self {
handle: Some(print),
pipe_writer: Some(pipe_writer),
})
loop {
buffer.reserve(MIN_BUFFER_CAPACITY);

let old_data_end = buffer.len();
buffer.resize(buffer.capacity(), 0);
match read_stderr(&mut buffer[old_data_end..]) {
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
// No data currently, yield back.
buffer.truncate(old_data_end);
return false;
}
Err(err) if err.kind() == std::io::ErrorKind::Interrupted => {
// Interrupted, try again.
buffer.truncate(old_data_end);
}
Ok(0) | Err(_) => {
// End of stream: flush remaining data and bail.
if old_data_end > 0 {
write_warning(&buffer[..old_data_end]);
}
return true;
}
Ok(bytes_read) => {
buffer.truncate(old_data_end + bytes_read);
let mut consumed = 0;
for line in buffer.split_inclusive(|&b| b == b'\n') {
// Only forward complete lines, leave the rest in the buffer.
if let Some((b'\n', line)) = line.split_last() {
consumed += line.len() + 1;
write_warning(line);
}
}
buffer.drain(..consumed);
}
}
}
} else {
true
}
}

/// # Panics
///
/// Will panic if the pipe writer has already been taken.
pub(crate) fn take_pipe_writer(&mut self) -> File {
self.pipe_writer.take().unwrap()
}
#[cfg(feature = "parallel")]
pub(crate) fn set_non_blocking(&mut self) -> Result<(), Error> {
assert!(!self.is_non_blocking);

// On Unix, switch the pipe to non-blocking mode.
// On Windows, we have a different way to be non-blocking.
#[cfg(unix)]
if let Some((stderr, _)) = self.inner.as_mut() {
use std::os::unix::io::AsRawFd;
let fd = stderr.as_raw_fd();
let flags = unsafe { libc::fcntl(fd, libc::F_GETFL, 0) };
if flags < 0 {
return Err(Error::new(
ErrorKind::IOError,
format!(
"Failed to get flags for child stderr: {}",
io::Error::last_os_error()
),
));
}

/// # Panics
///
/// Will panic if the pipe writer has already been taken.
pub(crate) fn clone_pipe_writer(&self) -> Result<File, Error> {
self.try_clone_pipe_writer().map(Option::unwrap)
}
if unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } != 0 {
return Err(Error::new(
ErrorKind::IOError,
format!(
"Failed to set flags for child stderr: {}",
io::Error::last_os_error()
),
));
}
}

pub(crate) fn try_clone_pipe_writer(&self) -> Result<Option<File>, Error> {
self.pipe_writer
.as_ref()
.map(File::try_clone)
.transpose()
.map_err(From::from)
self.is_non_blocking = true;
Ok(())
}
}

impl Drop for PrintThread {
fn drop(&mut self) {
// Drop pipe_writer first to avoid deadlock
self.pipe_writer.take();
#[cfg(feature = "parallel")]
fn forward_all(&mut self) {
while !self.forward_available() {}
}

self.handle.take().unwrap().join().unwrap();
#[cfg(not(feature = "parallel"))]
fn forward_all(&mut self) {
let forward_result = self.forward_available();
assert!(forward_result, "Should have consumed all data");
}
}

fn wait_on_child(cmd: &Command, program: &str, child: &mut Child) -> Result<(), Error> {
StderrForwarder::new(child).forward_all();

let status = match child.wait() {
Ok(s) => s,
Err(e) => {
Expand Down Expand Up @@ -193,20 +284,13 @@ pub(crate) fn objects_from_files(files: &[Arc<Path>], dst: &Path) -> Result<Vec<
Ok(objects)
}

fn run_inner(cmd: &mut Command, program: &str, pipe_writer: Option<File>) -> Result<(), Error> {
let mut child = spawn(cmd, program, pipe_writer)?;
wait_on_child(cmd, program, &mut child)
}

pub(crate) fn run(
cmd: &mut Command,
program: &str,
print: Option<&PrintThread>,
cargo_output: &CargoOutput,
) -> Result<(), Error> {
let pipe_writer = print.map(PrintThread::clone_pipe_writer).transpose()?;
run_inner(cmd, program, pipe_writer)?;

Ok(())
let mut child = spawn(cmd, program, cargo_output)?;
wait_on_child(cmd, program, &mut child)
}

pub(crate) fn run_output(
Expand All @@ -216,12 +300,7 @@ pub(crate) fn run_output(
) -> Result<Vec<u8>, Error> {
cmd.stdout(Stdio::piped());

let mut print = cargo_output.print_thread()?;
let mut child = spawn(
cmd,
program,
print.as_mut().map(PrintThread::take_pipe_writer),
)?;
let mut child = spawn(cmd, program, cargo_output)?;

let mut stdout = vec![];
child
Expand All @@ -239,7 +318,7 @@ pub(crate) fn run_output(
pub(crate) fn spawn(
cmd: &mut Command,
program: &str,
pipe_writer: Option<File>,
cargo_output: &CargoOutput,
) -> Result<Child, Error> {
struct ResetStderr<'cmd>(&'cmd mut Command);

Expand All @@ -254,10 +333,7 @@ pub(crate) fn spawn(
println!("running: {:?}", cmd);

let cmd = ResetStderr(cmd);
let child = cmd
.0
.stderr(pipe_writer.map_or_else(Stdio::null, Stdio::from))
.spawn();
let child = cmd.0.stderr(cargo_output.stdio_for_warnings()).spawn();
match child {
Ok(child) => Ok(child),
Err(ref e) if e.kind() == io::ErrorKind::NotFound => {
Expand Down Expand Up @@ -307,9 +383,14 @@ pub(crate) fn try_wait_on_child(
program: &str,
child: &mut Child,
stdout: &mut dyn io::Write,
stderr_forwarder: &mut StderrForwarder,
) -> Result<Option<()>, Error> {
stderr_forwarder.forward_available();

match child.try_wait() {
Ok(Some(status)) => {
stderr_forwarder.forward_all();

let _ = writeln!(stdout, "{}", status);

if status.success() {
Expand All @@ -325,12 +406,15 @@ pub(crate) fn try_wait_on_child(
}
}
Ok(None) => Ok(None),
Err(e) => Err(Error::new(
ErrorKind::ToolExecError,
format!(
"Failed to wait on spawned child process, command {:?} with args {:?}: {}.",
cmd, program, e
),
)),
Err(e) => {
stderr_forwarder.forward_all();
Err(Error::new(
ErrorKind::ToolExecError,
format!(
"Failed to wait on spawned child process, command {:?} with args {:?}: {}.",
cmd, program, e
),
))
}
}
}
Loading

0 comments on commit 0c8151d

Please sign in to comment.