diff options
Diffstat (limited to 'library/std/src/io')
| -rw-r--r-- | library/std/src/io/buffered/bufreader/buffer.rs | 2 | ||||
| -rw-r--r-- | library/std/src/io/error.rs | 4 | ||||
| -rw-r--r-- | library/std/src/io/error/repr_bitpacked.rs | 7 | ||||
| -rw-r--r-- | library/std/src/io/mod.rs | 249 | ||||
| -rw-r--r-- | library/std/src/io/tests.rs | 17 |
5 files changed, 273 insertions, 6 deletions
diff --git a/library/std/src/io/buffered/bufreader/buffer.rs b/library/std/src/io/buffered/bufreader/buffer.rs index 17721090db5..5251cc302cb 100644 --- a/library/std/src/io/buffered/bufreader/buffer.rs +++ b/library/std/src/io/buffered/bufreader/buffer.rs @@ -50,7 +50,7 @@ impl Buffer { pub fn buffer(&self) -> &[u8] { // SAFETY: self.pos and self.cap are valid, and self.cap => self.pos, and // that region is initialized because those are all invariants of this type. - unsafe { MaybeUninit::slice_assume_init_ref(self.buf.get_unchecked(self.pos..self.filled)) } + unsafe { self.buf.get_unchecked(self.pos..self.filled).assume_init_ref() } } #[inline] diff --git a/library/std/src/io/error.rs b/library/std/src/io/error.rs index 476c403c21f..38b72336617 100644 --- a/library/std/src/io/error.rs +++ b/library/std/src/io/error.rs @@ -339,7 +339,7 @@ pub enum ErrorKind { #[stable(feature = "io_error_a_bit_more", since = "1.83.0")] NotSeekable, /// Filesystem quota or some other kind of quota was exceeded. - #[stable(feature = "io_error_quota_exceeded", since = "CURRENT_RUSTC_VERSION")] + #[stable(feature = "io_error_quota_exceeded", since = "1.85.0")] QuotaExceeded, /// File larger than allowed or supported. /// @@ -364,7 +364,7 @@ pub enum ErrorKind { #[stable(feature = "io_error_a_bit_more", since = "1.83.0")] Deadlock, /// Cross-device or cross-filesystem (hard) link or rename. - #[stable(feature = "io_error_crosses_devices", since = "CURRENT_RUSTC_VERSION")] + #[stable(feature = "io_error_crosses_devices", since = "1.85.0")] CrossesDevices, /// Too many (hard) links to the same filesystem object. /// diff --git a/library/std/src/io/error/repr_bitpacked.rs b/library/std/src/io/error/repr_bitpacked.rs index f958a938646..716da37168d 100644 --- a/library/std/src/io/error/repr_bitpacked.rs +++ b/library/std/src/io/error/repr_bitpacked.rs @@ -103,7 +103,8 @@ //! the time. use core::marker::PhantomData; -use core::ptr::{self, NonNull}; +use core::num::NonZeroUsize; +use core::ptr::NonNull; use super::{Custom, ErrorData, ErrorKind, RawOsError, SimpleMessage}; @@ -176,7 +177,7 @@ impl Repr { let utagged = ((code as usize) << 32) | TAG_OS; // Safety: `TAG_OS` is not zero, so the result of the `|` is not 0. let res = Self( - unsafe { NonNull::new_unchecked(ptr::without_provenance_mut(utagged)) }, + NonNull::without_provenance(unsafe { NonZeroUsize::new_unchecked(utagged) }), PhantomData, ); // quickly smoke-check we encoded the right thing (This generally will @@ -193,7 +194,7 @@ impl Repr { let utagged = ((kind as usize) << 32) | TAG_SIMPLE; // Safety: `TAG_SIMPLE` is not zero, so the result of the `|` is not 0. let res = Self( - unsafe { NonNull::new_unchecked(ptr::without_provenance_mut(utagged)) }, + NonNull::without_provenance(unsafe { NonZeroUsize::new_unchecked(utagged) }), PhantomData, ); // quickly smoke-check we encoded the right thing (This generally will diff --git a/library/std/src/io/mod.rs b/library/std/src/io/mod.rs index 7912f969bbd..231c8712ebd 100644 --- a/library/std/src/io/mod.rs +++ b/library/std/src/io/mod.rs @@ -330,6 +330,7 @@ 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; @@ -3250,3 +3251,251 @@ impl<B: BufRead> Iterator for Lines<B> { } } } + +/// Create anonymous pipe that is close-on-exec and blocking. +/// +/// # Behavior +/// +/// A pipe is a synchronous, unidirectional data channel between two or more processes, like an +/// interprocess [`mpsc`](crate::sync::mpsc) provided by the OS. 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. +/// * [`PipeReader`] can be shared, but only one process will consume the data in the pipe. +/// +/// # 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(()) +/// # } +/// ``` +/// [pipe]: https://man7.org/linux/man-pages/man2/pipe.2.html +/// [CreatePipe]: https://learn.microsoft.com/en-us/windows/win32/api/namedpipeapi/nf-namedpipeapi-createpipe +/// [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 the anonymous pipe. +#[unstable(feature = "anonymous_pipe", issue = "127154")] +#[derive(Debug)] +pub struct PipeReader(pub(crate) AnonPipe); + +/// Write end of the 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() + } +} diff --git a/library/std/src/io/tests.rs b/library/std/src/io/tests.rs index 47cbb9614af..85098b3bb18 100644 --- a/library/std/src/io/tests.rs +++ b/library/std/src/io/tests.rs @@ -823,3 +823,20 @@ fn try_oom_error() { let io_err = io::Error::from(reserve_err); assert_eq!(io::ErrorKind::OutOfMemory, io_err.kind()); } + +#[test] +#[cfg(all(windows, unix, not(miri)))] +fn pipe_creation_clone_and_rw() { + let (rx, tx) = std::io::pipe().unwrap(); + + tx.try_clone().unwrap().write_all(b"12345").unwrap(); + drop(tx); + + let mut rx2 = rx.try_clone().unwrap(); + drop(rx); + + let mut s = String::new(); + rx2.read_to_string(&mut s).unwrap(); + drop(rx2); + assert_eq!(s, "12345"); +} |