//! A cross-platform library for opening OS pipes, like those from //! [`pipe`](https://man7.org/linux/man-pages/man2/pipe.2.html) on Linux //! or //! [`CreatePipe`](https://docs.microsoft.com/en-us/windows/win32/api/namedpipeapi/nf-namedpipeapi-createpipe) //! on Windows. The Rust standard library provides //! [`Stdio::piped`](https://doc.rust-lang.org/std/process/struct.Stdio.html#method.piped) //! for simple use cases involving child processes, but it doesn't //! support creating pipes directly. This crate fills that gap. //! //! - [Docs](https://docs.rs/os_pipe) //! - [Crate](https://crates.io/crates/os_pipe) //! - [Repo](https://github.com/oconnor663/os_pipe.rs) //! //! # Common deadlocks related to pipes //! //! When you work with pipes, you often end up debugging a deadlock at //! some point. These can be confusing if you don't know why they //! happen. Here are two things you need to know: //! //! 1. Pipe reads will block waiting for input as long as there's at //! least one writer still open. **If you forget to close a writer, //! reads will block forever.** This includes writers that you give //! to child processes. //! 2. Pipes have an internal buffer of some fixed size. On Linux for //! example, pipe buffers are 64 KiB by default. When the buffer is //! full, writes will block waiting for space. **If the buffer is //! full and there aren't any readers, writes will block forever.** //! //! Deadlocks caused by a forgotten writer usually show up immediately, //! which makes them relatively easy to fix once you know what to look //! for. (See "Avoid a deadlock!" in the example code below.) However, //! deadlocks caused by full pipe buffers are trickier. These might only //! show up for larger inputs, and they might be timing-dependent or //! platform-dependent. If you find that writing to a pipe deadlocks //! sometimes, think about who's supposed to be reading from that pipe, //! and whether that thread or process might be blocked on something //! else. For more on this, see the [Gotchas //! Doc](https://github.com/oconnor663/duct.py/blob/master/gotchas.md#using-io-threads-to-avoid-blocking-children) //! from the [`duct`](https://github.com/oconnor663/duct.rs) crate. (And //! consider whether [`duct`](https://github.com/oconnor663/duct.rs) //! might be a good fit for your use case.) //! //! # Examples //! //! Here we write a single byte into a pipe and read it back out: //! //! ```rust //! # fn main() -> Result<(), Box> { //! use std::io::prelude::*; //! //! let (mut reader, mut writer) = os_pipe::pipe()?; //! // XXX: If this write blocks, we'll never get to the read. //! writer.write_all(b"x")?; //! let mut output = [0]; //! reader.read_exact(&mut output)?; //! assert_eq!(b"x", &output); //! # Ok(()) //! # } //! ``` //! //! This is a minimal working example, but as discussed in the section //! above, reading and writing on the same thread like this is //! deadlock-prone. If we wrote 100 KB instead of just one byte, this //! example would block on `write_all`, it would never make it to //! `read_exact`, and that would be a deadlock. Doing the read and write //! from different threads or different processes would fix the //! deadlock. //! //! For a more complex example, here we join the stdout and stderr of a //! child process into a single pipe. To do that we open a pipe, clone //! its writer, and set that pair of writers as the child's stdout and //! stderr. (This is possible because `PipeWriter` implements //! `Into`.) Then we can read interleaved output from the pipe //! reader. This example is deadlock-free, but note the comment about //! closing the writers. //! //! ```rust //! # use std::io::prelude::*; //! # fn main() -> Result<(), Box> { //! // We're going to spawn a child process that prints "foo" to stdout //! // and "bar" to stderr, and we'll combine these into a single pipe. //! let mut command = std::process::Command::new("python"); //! command.args(&["-c", r#" //! import sys //! sys.stdout.write("foo") //! sys.stdout.flush() //! sys.stderr.write("bar") //! sys.stderr.flush() //! "#]); //! //! // Here's the interesting part. Open a pipe, clone its writer, and //! // set that pair of writers as the child's stdout and stderr. //! let (mut reader, writer) = os_pipe::pipe()?; //! let writer_clone = writer.try_clone()?; //! command.stdout(writer); //! command.stderr(writer_clone); //! //! // Now start the child process running. //! let mut handle = command.spawn()?; //! //! // Avoid a deadlock! This parent process is still holding open pipe //! // writers inside the Command object, and we have to close those //! // before we read. Here we do this by dropping the Command object. //! drop(command); //! //! // Finally we can read all the output and clean up the child. //! let mut output = String::new(); //! reader.read_to_string(&mut output)?; //! handle.wait()?; //! assert_eq!(output, "foobar"); //! # Ok(()) //! # } //! ``` //! //! Note that the [`duct`](https://github.com/oconnor663/duct.rs) crate //! can reproduce the example above in a single line of code, with no //! risk of deadlocks and no risk of leaking [zombie //! children](https://en.wikipedia.org/wiki/Zombie_process). use std::fs::File; use std::io; use std::process::Stdio; #[cfg(not(windows))] #[path = "unix.rs"] mod sys; #[cfg(windows)] #[path = "windows.rs"] mod sys; /// The reading end of a pipe, returned by [`pipe`](fn.pipe.html). /// /// `PipeReader` implements `Into`, so you can pass it as an argument to /// `Command::stdin` to spawn a child process that reads from the pipe. #[derive(Debug)] pub struct PipeReader( // We use std::fs::File here for two reasons: OwnedFd and OwnedHandle are platform-specific, // and this gives us read/write/flush for free. File, ); impl PipeReader { pub fn try_clone(&self) -> io::Result { self.0.try_clone().map(PipeReader) } } impl io::Read for PipeReader { fn read(&mut self, buf: &mut [u8]) -> io::Result { self.0.read(buf) } } impl<'a> io::Read for &'a PipeReader { fn read(&mut self, buf: &mut [u8]) -> io::Result { (&self.0).read(buf) } } impl From for Stdio { fn from(p: PipeReader) -> Stdio { p.0.into() } } /// The writing end of a pipe, returned by [`pipe`](fn.pipe.html). /// /// `PipeWriter` implements `Into`, so you can pass it as an argument to /// `Command::stdout` or `Command::stderr` to spawn a child process that writes /// to the pipe. #[derive(Debug)] pub struct PipeWriter(File); impl PipeWriter { pub fn try_clone(&self) -> io::Result { self.0.try_clone().map(PipeWriter) } } impl io::Write for PipeWriter { fn write(&mut self, buf: &[u8]) -> io::Result { self.0.write(buf) } fn flush(&mut self) -> io::Result<()> { self.0.flush() } } impl<'a> io::Write for &'a PipeWriter { fn write(&mut self, buf: &[u8]) -> io::Result { (&self.0).write(buf) } fn flush(&mut self) -> io::Result<()> { (&self.0).flush() } } impl From for Stdio { fn from(p: PipeWriter) -> Stdio { p.0.into() } } /// Open a new pipe and return a [`PipeReader`] and [`PipeWriter`] pair. /// /// This corresponds to the `pipe2` library call on Posix and the /// `CreatePipe` library call on Windows (though these implementation /// details might change). These pipes are non-inheritable, so new child /// processes won't receive a copy of them unless they're explicitly /// passed as stdin/stdout/stderr. /// /// [`PipeReader`]: struct.PipeReader.html /// [`PipeWriter`]: struct.PipeWriter.html pub fn pipe() -> io::Result<(PipeReader, PipeWriter)> { sys::pipe() } /// Get a duplicated copy of the current process's standard input, as a /// [`PipeReader`]. /// /// Reading directly from this pipe isn't recommended, because it's not /// synchronized with [`std::io::stdin`]. [`PipeReader`] implements /// [`Into`], so it can be passed directly to [`Command::stdin`]. This is /// equivalent to [`Stdio::inherit`], though, so it's usually not necessary /// unless you need a collection of different pipes. /// /// [`std::io::stdin`]: https://doc.rust-lang.org/std/io/fn.stdin.html /// [`PipeReader`]: struct.PipeReader.html /// [`Into`]: https://doc.rust-lang.org/std/process/struct.Stdio.html /// [`Command::stdin`]: https://doc.rust-lang.org/std/process/struct.Command.html#method.stdin /// [`Stdio::inherit`]: https://doc.rust-lang.org/std/process/struct.Stdio.html#method.inherit pub fn dup_stdin() -> io::Result { sys::dup(io::stdin()).map(PipeReader::from) } /// Get a duplicated copy of the current process's standard output, as a /// [`PipeWriter`](struct.PipeWriter.html). /// /// Writing directly to this pipe isn't recommended, because it's not /// synchronized with [`std::io::stdout`]. [`PipeWriter`] implements /// [`Into`], so it can be passed directly to [`Command::stdout`] or /// [`Command::stderr`]. This can be useful if you want the child's stderr to go /// to the parent's stdout. /// /// [`std::io::stdout`]: https://doc.rust-lang.org/std/io/fn.stdout.html /// [`PipeWriter`]: struct.PipeWriter.html /// [`Into`]: https://doc.rust-lang.org/std/process/struct.Stdio.html /// [`Command::stdout`]: https://doc.rust-lang.org/std/process/struct.Command.html#method.stdout /// [`Command::stderr`]: https://doc.rust-lang.org/std/process/struct.Command.html#method.stderr /// [`Stdio::inherit`]: https://doc.rust-lang.org/std/process/struct.Stdio.html#method.inherit pub fn dup_stdout() -> io::Result { sys::dup(io::stdout()).map(PipeWriter::from) } /// Get a duplicated copy of the current process's standard error, as a /// [`PipeWriter`](struct.PipeWriter.html). /// /// Writing directly to this pipe isn't recommended, because it's not /// synchronized with [`std::io::stderr`]. [`PipeWriter`] implements /// [`Into`], so it can be passed directly to [`Command::stdout`] or /// [`Command::stderr`]. This can be useful if you want the child's stdout to go /// to the parent's stderr. /// /// [`std::io::stderr`]: https://doc.rust-lang.org/std/io/fn.stderr.html /// [`PipeWriter`]: struct.PipeWriter.html /// [`Into`]: https://doc.rust-lang.org/std/process/struct.Stdio.html /// [`Command::stdout`]: https://doc.rust-lang.org/std/process/struct.Command.html#method.stdout /// [`Command::stderr`]: https://doc.rust-lang.org/std/process/struct.Command.html#method.stderr /// [`Stdio::inherit`]: https://doc.rust-lang.org/std/process/struct.Stdio.html#method.inherit pub fn dup_stderr() -> io::Result { sys::dup(io::stderr()).map(PipeWriter::from) } #[cfg(test)] mod tests { use std::env::consts::EXE_EXTENSION; use std::io::prelude::*; use std::path::{Path, PathBuf}; use std::process::Command; use std::sync::Once; use std::thread; fn path_to_exe(name: &str) -> PathBuf { // This project defines some associated binaries for testing, and we shell out to them in // these tests. `cargo test` doesn't automatically build associated binaries, so this // function takes care of building them explicitly, with the right debug/release flavor. static CARGO_BUILD_ONCE: Once = Once::new(); CARGO_BUILD_ONCE.call_once(|| { let mut build_command = Command::new("cargo"); build_command.args(&["build", "--quiet"]); if !cfg!(debug_assertions) { build_command.arg("--release"); } let build_status = build_command.status().unwrap(); assert!( build_status.success(), "Cargo failed to build associated binaries." ); }); let flavor = if cfg!(debug_assertions) { "debug" } else { "release" }; Path::new("target") .join(flavor) .join(name) .with_extension(EXE_EXTENSION) } #[test] fn test_pipe_some_data() { let (mut reader, mut writer) = crate::pipe().unwrap(); // A small write won't fill the pipe buffer, so it won't block this thread. writer.write_all(b"some stuff").unwrap(); drop(writer); let mut out = String::new(); reader.read_to_string(&mut out).unwrap(); assert_eq!(out, "some stuff"); } #[test] fn test_pipe_some_data_with_refs() { // As with `File`, there's a second set of impls for shared // refs. Test those. let (reader, writer) = crate::pipe().unwrap(); let mut reader_ref = &reader; { let mut writer_ref = &writer; // A small write won't fill the pipe buffer, so it won't block this thread. writer_ref.write_all(b"some stuff").unwrap(); } drop(writer); let mut out = String::new(); reader_ref.read_to_string(&mut out).unwrap(); assert_eq!(out, "some stuff"); } #[test] fn test_pipe_no_data() { let (mut reader, writer) = crate::pipe().unwrap(); drop(writer); let mut out = String::new(); reader.read_to_string(&mut out).unwrap(); assert_eq!(out, ""); } #[test] fn test_pipe_a_megabyte_of_data_from_another_thread() { let data = vec![0xff; 1_000_000]; let data_copy = data.clone(); let (mut reader, mut writer) = crate::pipe().unwrap(); let joiner = thread::spawn(move || { writer.write_all(&data_copy).unwrap(); // This drop happens automatically, so writing it out here is mostly // just for clarity. For what it's worth, it also guards against // accidentally forgetting to drop if we switch to scoped threads or // something like that and change this to a non-moving closure. The // explicit drop forces `writer` to move. drop(writer); }); let mut out = Vec::new(); reader.read_to_end(&mut out).unwrap(); joiner.join().unwrap(); assert_eq!(out, data); } #[test] fn test_pipes_are_not_inheritable() { // Create pipes for a child process. let (input_reader, mut input_writer) = crate::pipe().unwrap(); let (mut output_reader, output_writer) = crate::pipe().unwrap(); // Create a bunch of duplicated copies, which we'll close later. This // tests that duplication preserves non-inheritability. let ir_dup = input_reader.try_clone().unwrap(); let iw_dup = input_writer.try_clone().unwrap(); let or_dup = output_reader.try_clone().unwrap(); let ow_dup = output_writer.try_clone().unwrap(); // Spawn the child. Note that this temporary Command object takes // ownership of our copies of the child's stdin and stdout, and then // closes them immediately when it drops. That stops us from blocking // our own read below. We use our own simple implementation of cat for // compatibility with Windows. let mut child = Command::new(path_to_exe("cat")) .stdin(input_reader) .stdout(output_writer) .spawn() .unwrap(); // Drop all the dups now that the child is spawned. drop(ir_dup); drop(iw_dup); drop(or_dup); drop(ow_dup); // Write to the child's stdin. This is a small write, so it shouldn't // block. input_writer.write_all(b"hello").unwrap(); drop(input_writer); // Read from the child's stdout. If this child has accidentally // inherited the write end of its own stdin, then it will never exit, // and this read will block forever. That's what this test is all // about. let mut output = Vec::new(); output_reader.read_to_end(&mut output).unwrap(); child.wait().unwrap(); // Confirm that we got the right bytes. assert_eq!(b"hello", &*output); } #[test] fn test_parent_handles() { // This test invokes the `swap` test program, which uses parent_stdout() and // parent_stderr() to swap the outputs for another child that it spawns. // Create pipes for a child process. let (reader, mut writer) = crate::pipe().unwrap(); // Write input. This shouldn't block because it's small. Then close the write end, or else // the child will hang. writer.write_all(b"quack").unwrap(); drop(writer); // Use `swap` to run `cat_both`. `cat_both will read "quack" from stdin // and write it to stdout and stderr with different tags. But because we // run it inside `swap`, the tags in the output should be backwards. let output = Command::new(path_to_exe("swap")) .arg(path_to_exe("cat_both")) .stdin(reader) .output() .unwrap(); // Check for a clean exit. assert!( output.status.success(), "child process returned {:#?}", output ); // Confirm that we got the right bytes. assert_eq!(b"stderr: quack", &*output.stdout); assert_eq!(b"stdout: quack", &*output.stderr); } #[test] fn test_parent_handles_dont_close() { // Open and close each parent pipe multiple times. If this closes the // original, subsequent opens should fail. let stdin = crate::dup_stdin().unwrap(); drop(stdin); let stdin = crate::dup_stdin().unwrap(); drop(stdin); let stdout = crate::dup_stdout().unwrap(); drop(stdout); let stdout = crate::dup_stdout().unwrap(); drop(stdout); let stderr = crate::dup_stderr().unwrap(); drop(stderr); let stderr = crate::dup_stderr().unwrap(); drop(stderr); } #[test] fn test_try_clone() { let (reader, writer) = crate::pipe().unwrap(); let mut reader_clone = reader.try_clone().unwrap(); let mut writer_clone = writer.try_clone().unwrap(); // A small write won't fill the pipe buffer, so it won't block this thread. writer_clone.write_all(b"some stuff").unwrap(); drop(writer); drop(writer_clone); let mut out = String::new(); reader_clone.read_to_string(&mut out).unwrap(); assert_eq!(out, "some stuff"); } #[test] fn test_debug() { let (reader, writer) = crate::pipe().unwrap(); _ = format!("{:?} {:?}", reader, writer); } }