Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Stabilize anonymous_pipe #135822

Closed
wants to merge 5 commits into from
Closed

Stabilize anonymous_pipe #135822

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
68e983f
Move `std::io::pipe` code into its own file
tbu- Jan 17, 2025
ad28cbb
Update `std::io::{pipe, PipeReader, PipeWriter}` docs the new location
tbu- Jan 17, 2025
e1f9370
pipe: Remove unnecessary miri exclusions from no_run code
tbu- Jan 21, 2025
604faa6
Stabilize `anonymous_pipe`
tbu- Jan 21, 2025
0778f24
Move trait impls for `Pipe{Reader,Writer}` to `std::io::pipe`
tbu- Jan 26, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
258 changes: 3 additions & 255 deletions library/std/src/io/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,6 +310,8 @@ pub use self::error::RawOsError;
pub use self::error::SimpleMessage;
#[unstable(feature = "io_const_error", issue = "133448")]
pub use self::error::const_error;
#[stable(feature = "anonymous_pipe", since = "CURRENT_RUSTC_VERSION")]
pub use self::pipe::{PipeReader, PipeWriter, pipe};
#[stable(feature = "is_terminal", since = "1.70.0")]
pub use self::stdio::IsTerminal;
pub(crate) use self::stdio::attempt_print_to_stderr;
Expand All @@ -330,14 +332,14 @@ pub use self::{
};
use crate::mem::take;
use crate::ops::{Deref, DerefMut};
use crate::sys::anonymous_pipe::{AnonPipe, pipe as pipe_inner};
use crate::{cmp, fmt, slice, str, sys};

mod buffered;
pub(crate) mod copy;
mod cursor;
mod error;
mod impls;
mod pipe;
pub mod prelude;
mod stdio;
mod util;
Expand Down Expand Up @@ -3251,257 +3253,3 @@ impl<B: BufRead> Iterator for Lines<B> {
}
}
}

/// Create an anonymous pipe that is close-on-exec and blocking.
///
/// # 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.
///
/// # 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.
///
/// # Examples
///
/// ```no_run
/// #![feature(anonymous_pipe)]
/// # #[cfg(miri)] fn main() {}
/// # #[cfg(not(miri))]
/// # fn main() -> std::io::Result<()> {
/// # use std::process::Command;
/// # use std::io::{Read, Write};
/// let (ping_rx, mut ping_tx) = std::io::pipe()?;
/// let (mut pong_rx, pong_tx) = std::io::pipe()?;
///
/// // Spawn a process that echoes its input.
/// let mut echo_server = Command::new("cat").stdin(ping_rx).stdout(pong_tx).spawn()?;
///
/// ping_tx.write_all(b"hello")?;
/// // Close to unblock echo_server's reader.
/// drop(ping_tx);
///
/// let mut buf = String::new();
/// // Block until echo_server's writer is closed.
/// pong_rx.read_to_string(&mut buf)?;
/// assert_eq!(&buf, "hello");
///
/// echo_server.wait()?;
/// # Ok(())
/// # }
/// ```
/// [man page]: https://man7.org/linux/man-pages/man7/pipe.7.html
#[unstable(feature = "anonymous_pipe", issue = "127154")]
#[inline]
pub fn pipe() -> Result<(PipeReader, PipeWriter)> {
pipe_inner().map(|(reader, writer)| (PipeReader(reader), PipeWriter(writer)))
}

/// Read end of an anonymous pipe.
#[unstable(feature = "anonymous_pipe", issue = "127154")]
#[derive(Debug)]
pub struct PipeReader(pub(crate) AnonPipe);

/// Write end of an anonymous pipe.
#[unstable(feature = "anonymous_pipe", issue = "127154")]
#[derive(Debug)]
pub struct PipeWriter(pub(crate) AnonPipe);

impl PipeReader {
/// Create a new [`PipeReader`] instance that shares the same underlying file description.
///
/// # Examples
///
/// ```no_run
/// #![feature(anonymous_pipe)]
/// # #[cfg(miri)] fn main() {}
/// # #[cfg(not(miri))]
/// # fn main() -> std::io::Result<()> {
/// # use std::fs;
/// # use std::io::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) = std::io::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(())
/// # }
/// ```
#[unstable(feature = "anonymous_pipe", issue = "127154")]
pub fn try_clone(&self) -> Result<Self> {
self.0.try_clone().map(Self)
}
}

impl PipeWriter {
/// Create a new [`PipeWriter`] instance that shares the same underlying file description.
///
/// # Examples
///
/// ```no_run
/// #![feature(anonymous_pipe)]
/// # #[cfg(miri)] fn main() {}
/// # #[cfg(not(miri))]
/// # fn main() -> std::io::Result<()> {
/// # use std::process::Command;
/// # use std::io::Read;
/// let (mut reader, writer) = std::io::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(())
/// # }
/// ```
#[unstable(feature = "anonymous_pipe", issue = "127154")]
pub fn try_clone(&self) -> Result<Self> {
self.0.try_clone().map(Self)
}
}

#[unstable(feature = "anonymous_pipe", issue = "127154")]
impl Read for &PipeReader {
fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
self.0.read(buf)
}
fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> 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>) -> Result<usize> {
self.0.read_to_end(buf)
}
fn read_buf(&mut self, buf: BorrowedCursor<'_>) -> Result<()> {
self.0.read_buf(buf)
}
}

#[unstable(feature = "anonymous_pipe", issue = "127154")]
impl Read for PipeReader {
fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
self.0.read(buf)
}
fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> 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>) -> Result<usize> {
self.0.read_to_end(buf)
}
fn read_buf(&mut self, buf: BorrowedCursor<'_>) -> Result<()> {
self.0.read_buf(buf)
}
}

#[unstable(feature = "anonymous_pipe", issue = "127154")]
impl Write for &PipeWriter {
fn write(&mut self, buf: &[u8]) -> Result<usize> {
self.0.write(buf)
}
#[inline]
fn flush(&mut self) -> Result<()> {
Ok(())
}

fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
self.0.write_vectored(bufs)
}

#[inline]
fn is_write_vectored(&self) -> bool {
self.0.is_write_vectored()
}
}

#[unstable(feature = "anonymous_pipe", issue = "127154")]
impl Write for PipeWriter {
fn write(&mut self, buf: &[u8]) -> Result<usize> {
self.0.write(buf)
}
#[inline]
fn flush(&mut self) -> Result<()> {
Ok(())
}

fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
self.0.write_vectored(bufs)
}

#[inline]
fn is_write_vectored(&self) -> bool {
self.0.is_write_vectored()
}
}
Loading
Loading