std/io/

pipe.rs

use crate::io;
use crate::sys::{FromInner, IntoInner, pipe as imp};

/// Creates an anonymous pipe.
///
/// # Behavior
///
/// A pipe is a one-way data channel provided by the OS, which works across processes. A pipe is
/// typically used to communicate between two or more separate processes, as there are better,
/// faster ways to communicate within a single process.
///
/// In particular:
///
/// * A read on a [`PipeReader`] blocks until the pipe is non-empty.
/// * A write on a [`PipeWriter`] blocks when the pipe is full.
/// * When all copies of a [`PipeWriter`] are closed, a read on the corresponding [`PipeReader`]
///   returns EOF.
/// * [`PipeWriter`] can be shared, and multiple processes or threads can write to it at once, but
///   writes (above a target-specific threshold) may have their data interleaved.
/// * [`PipeReader`] can be shared, and multiple processes or threads can read it at once. Any
///   given byte will only get consumed by one reader. There are no guarantees about data
///   interleaving.
/// * Portable applications cannot assume any atomicity of messages larger than a single byte.
///
/// # Platform-specific behavior
///
/// This function currently corresponds to the `pipe` function on Unix and the
/// `CreatePipe` function on Windows.
///
/// Note that this [may change in the future][changes].
///
/// # Capacity
///
/// Pipe capacity is platform dependent. To quote the Linux [man page]:
///
/// > Different implementations have different limits for the pipe capacity. Applications should
/// > not rely on a particular capacity: an application should be designed so that a reading process
/// > consumes data as soon as it is available, so that a writing process does not remain blocked.
///
/// # Example
///
/// ```no_run
/// # #[cfg(miri)] fn main() {}
/// # #[cfg(not(miri))]
/// # fn main() -> std::io::Result<()> {
/// use std::io::{Read, Write, pipe};
/// use std::process::Command;
/// let (ping_reader, mut ping_writer) = pipe()?;
/// let (mut pong_reader, pong_writer) = pipe()?;
///
/// // Spawn a child process that echoes its input.
/// let mut echo_command = Command::new("cat");
/// echo_command.stdin(ping_reader);
/// echo_command.stdout(pong_writer);
/// let mut echo_child = echo_command.spawn()?;
///
/// // Send input to the child process. Note that because we're writing all the input before we
/// // read any output, this could deadlock if the child's input and output pipe buffers both
/// // filled up. Those buffers are usually at least a few KB, so "hello" is fine, but for longer
/// // inputs we'd need to read and write at the same time, e.g. using threads.
/// ping_writer.write_all(b"hello")?;
///
/// // `cat` exits when it reads EOF from stdin, but that can't happen while any ping writer
/// // remains open. We need to drop our ping writer, or read_to_string will deadlock below.
/// drop(ping_writer);
///
/// // The pong reader can't report EOF while any pong writer remains open. Our Command object is
/// // holding a pong writer, and again read_to_string will deadlock if we don't drop it.
/// drop(echo_command);
///
/// let mut buf = String::new();
/// // Block until `cat` closes its stdout (a pong writer).
/// pong_reader.read_to_string(&mut buf)?;
/// assert_eq!(&buf, "hello");
///
/// // At this point we know `cat` has exited, but we still need to wait to clean up the "zombie".
/// echo_child.wait()?;
/// # Ok(())
/// # }
/// ```
/// [changes]: io#platform-specific-behavior
/// [man page]: https://man7.org/linux/man-pages/man7/pipe.7.html
#[stable(feature = "anonymous_pipe", since = "1.87.0")]
#[inline]
pub fn pipe() -> io::Result<(PipeReader, PipeWriter)> {
    imp::pipe().map(|(reader, writer)| (PipeReader(reader), PipeWriter(writer)))
}

/// Read end of an anonymous pipe.
#[stable(feature = "anonymous_pipe", since = "1.87.0")]
#[derive(Debug)]
pub struct PipeReader(pub(crate) imp::Pipe);

/// Write end of an anonymous pipe.
#[stable(feature = "anonymous_pipe", since = "1.87.0")]
#[derive(Debug)]
pub struct PipeWriter(pub(crate) imp::Pipe);

impl FromInner<imp::Pipe> for PipeReader {
    fn from_inner(inner: imp::Pipe) -> Self {
        Self(inner)
    }
}

impl IntoInner<imp::Pipe> for PipeReader {
    fn into_inner(self) -> imp::Pipe {
        self.0
    }
}

impl FromInner<imp::Pipe> for PipeWriter {
    fn from_inner(inner: imp::Pipe) -> Self {
        Self(inner)
    }
}

impl IntoInner<imp::Pipe> for PipeWriter {
    fn into_inner(self) -> imp::Pipe {
        self.0
    }
}

impl PipeReader {
    /// Creates a new [`PipeReader`] instance that shares the same underlying file description.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[cfg(miri)] fn main() {}
    /// # #[cfg(not(miri))]
    /// # fn main() -> std::io::Result<()> {
    /// use std::fs;
    /// use std::io::{pipe, Write};
    /// use std::process::Command;
    /// const NUM_SLOT: u8 = 2;
    /// const NUM_PROC: u8 = 5;
    /// const OUTPUT: &str = "work.txt";
    ///
    /// let mut jobs = vec![];
    /// let (reader, mut writer) = pipe()?;
    ///
    /// // Write NUM_SLOT characters the pipe.
    /// writer.write_all(&[b'|'; NUM_SLOT as usize])?;
    ///
    /// // Spawn several processes that read a character from the pipe, do some work, then
    /// // write back to the pipe. When the pipe is empty, the processes block, so only
    /// // NUM_SLOT processes can be working at any given time.
    /// for _ in 0..NUM_PROC {
    ///     jobs.push(
    ///         Command::new("bash")
    ///             .args(["-c",
    ///                 &format!(
    ///                      "read -n 1\n\
    ///                       echo -n 'x' >> '{OUTPUT}'\n\
    ///                       echo -n '|'",
    ///                 ),
    ///             ])
    ///             .stdin(reader.try_clone()?)
    ///             .stdout(writer.try_clone()?)
    ///             .spawn()?,
    ///     );
    /// }
    ///
    /// // Wait for all jobs to finish.
    /// for mut job in jobs {
    ///     job.wait()?;
    /// }
    ///
    /// // Check our work and clean up.
    /// let xs = fs::read_to_string(OUTPUT)?;
    /// fs::remove_file(OUTPUT)?;
    /// assert_eq!(xs, "x".repeat(NUM_PROC.into()));
    /// # Ok(())
    /// # }
    /// ```
    #[stable(feature = "anonymous_pipe", since = "1.87.0")]
    pub fn try_clone(&self) -> io::Result<Self> {
        self.0.try_clone().map(Self)
    }
}

impl PipeWriter {
    /// Creates a new [`PipeWriter`] instance that shares the same underlying file description.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[cfg(miri)] fn main() {}
    /// # #[cfg(not(miri))]
    /// # fn main() -> std::io::Result<()> {
    /// use std::process::Command;
    /// use std::io::{pipe, Read};
    /// let (mut reader, writer) = pipe()?;
    ///
    /// // Spawn a process that writes to stdout and stderr.
    /// let mut peer = Command::new("bash")
    ///     .args([
    ///         "-c",
    ///         "echo -n foo\n\
    ///          echo -n bar >&2"
    ///     ])
    ///     .stdout(writer.try_clone()?)
    ///     .stderr(writer)
    ///     .spawn()?;
    ///
    /// // Read and check the result.
    /// let mut msg = String::new();
    /// reader.read_to_string(&mut msg)?;
    /// assert_eq!(&msg, "foobar");
    ///
    /// peer.wait()?;
    /// # Ok(())
    /// # }
    /// ```
    #[stable(feature = "anonymous_pipe", since = "1.87.0")]
    pub fn try_clone(&self) -> io::Result<Self> {
        self.0.try_clone().map(Self)
    }
}

#[stable(feature = "anonymous_pipe", since = "1.87.0")]
impl io::Read for &PipeReader {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.0.read(buf)
    }
    fn read_vectored(&mut self, bufs: &mut [io::IoSliceMut<'_>]) -> io::Result<usize> {
        self.0.read_vectored(bufs)
    }
    #[inline]
    fn is_read_vectored(&self) -> bool {
        self.0.is_read_vectored()
    }
    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
        self.0.read_to_end(buf)
    }
    fn read_buf(&mut self, buf: io::BorrowedCursor<'_>) -> io::Result<()> {
        self.0.read_buf(buf)
    }
}

#[stable(feature = "anonymous_pipe", since = "1.87.0")]
impl io::Read for PipeReader {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.0.read(buf)
    }
    fn read_vectored(&mut self, bufs: &mut [io::IoSliceMut<'_>]) -> io::Result<usize> {
        self.0.read_vectored(bufs)
    }
    #[inline]
    fn is_read_vectored(&self) -> bool {
        self.0.is_read_vectored()
    }
    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
        self.0.read_to_end(buf)
    }
    fn read_buf(&mut self, buf: io::BorrowedCursor<'_>) -> io::Result<()> {
        self.0.read_buf(buf)
    }
}

#[stable(feature = "anonymous_pipe", since = "1.87.0")]
impl io::Write for &PipeWriter {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.0.write(buf)
    }
    #[inline]
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
    fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
        self.0.write_vectored(bufs)
    }
    #[inline]
    fn is_write_vectored(&self) -> bool {
        self.0.is_write_vectored()
    }
}

#[stable(feature = "anonymous_pipe", since = "1.87.0")]
impl io::Write for PipeWriter {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.0.write(buf)
    }
    #[inline]
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
    fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
        self.0.write_vectored(bufs)
    }
    #[inline]
    fn is_write_vectored(&self) -> bool {
        self.0.is_write_vectored()
    }
}