std/io/
pipe.rs

1use crate::io;
2use crate::sys::anonymous_pipe::{AnonPipe, pipe as pipe_inner};
3
4/// Create an anonymous pipe.
5///
6/// # Behavior
7///
8/// A pipe is a one-way data channel provided by the OS, which works across processes. A pipe is
9/// typically used to communicate between two or more separate processes, as there are better,
10/// faster ways to communicate within a single process.
11///
12/// In particular:
13///
14/// * A read on a [`PipeReader`] blocks until the pipe is non-empty.
15/// * A write on a [`PipeWriter`] blocks when the pipe is full.
16/// * When all copies of a [`PipeWriter`] are closed, a read on the corresponding [`PipeReader`]
17///   returns EOF.
18/// * [`PipeWriter`] can be shared, and multiple processes or threads can write to it at once, but
19///   writes (above a target-specific threshold) may have their data interleaved.
20/// * [`PipeReader`] can be shared, and multiple processes or threads can read it at once. Any
21///   given byte will only get consumed by one reader. There are no guarantees about data
22///   interleaving.
23/// * Portable applications cannot assume any atomicity of messages larger than a single byte.
24///
25/// # Platform-specific behavior
26///
27/// This function currently corresponds to the `pipe` function on Unix and the
28/// `CreatePipe` function on Windows.
29///
30/// Note that this [may change in the future][changes].
31///
32/// # Capacity
33///
34/// Pipe capacity is platform dependent. To quote the Linux [man page]:
35///
36/// > Different implementations have different limits for the pipe capacity. Applications should
37/// > not rely on a particular capacity: an application should be designed so that a reading process
38/// > consumes data as soon as it is available, so that a writing process does not remain blocked.
39///
40/// # Examples
41///
42/// ```no_run
43/// #![feature(anonymous_pipe)]
44/// # #[cfg(miri)] fn main() {}
45/// # #[cfg(not(miri))]
46/// # fn main() -> std::io::Result<()> {
47/// use std::process::Command;
48/// use std::io::{pipe, Read, Write};
49/// let (ping_rx, mut ping_tx) = pipe()?;
50/// let (mut pong_rx, pong_tx) = pipe()?;
51///
52/// // Spawn a process that echoes its input.
53/// let mut echo_server = Command::new("cat").stdin(ping_rx).stdout(pong_tx).spawn()?;
54///
55/// ping_tx.write_all(b"hello")?;
56/// // Close to unblock echo_server's reader.
57/// drop(ping_tx);
58///
59/// let mut buf = String::new();
60/// // Block until echo_server's writer is closed.
61/// pong_rx.read_to_string(&mut buf)?;
62/// assert_eq!(&buf, "hello");
63///
64/// echo_server.wait()?;
65/// # Ok(())
66/// # }
67/// ```
68/// [changes]: io#platform-specific-behavior
69/// [man page]: https://man7.org/linux/man-pages/man7/pipe.7.html
70#[unstable(feature = "anonymous_pipe", issue = "127154")]
71#[inline]
72pub fn pipe() -> io::Result<(PipeReader, PipeWriter)> {
73    pipe_inner().map(|(reader, writer)| (PipeReader(reader), PipeWriter(writer)))
74}
75
76/// Read end of an anonymous pipe.
77#[unstable(feature = "anonymous_pipe", issue = "127154")]
78#[derive(Debug)]
79pub struct PipeReader(pub(crate) AnonPipe);
80
81/// Write end of an anonymous pipe.
82#[unstable(feature = "anonymous_pipe", issue = "127154")]
83#[derive(Debug)]
84pub struct PipeWriter(pub(crate) AnonPipe);
85
86impl PipeReader {
87    /// Create a new [`PipeReader`] instance that shares the same underlying file description.
88    ///
89    /// # Examples
90    ///
91    /// ```no_run
92    /// #![feature(anonymous_pipe)]
93    /// # #[cfg(miri)] fn main() {}
94    /// # #[cfg(not(miri))]
95    /// # fn main() -> std::io::Result<()> {
96    /// use std::fs;
97    /// use std::io::{pipe, Write};
98    /// use std::process::Command;
99    /// const NUM_SLOT: u8 = 2;
100    /// const NUM_PROC: u8 = 5;
101    /// const OUTPUT: &str = "work.txt";
102    ///
103    /// let mut jobs = vec![];
104    /// let (reader, mut writer) = pipe()?;
105    ///
106    /// // Write NUM_SLOT characters the pipe.
107    /// writer.write_all(&[b'|'; NUM_SLOT as usize])?;
108    ///
109    /// // Spawn several processes that read a character from the pipe, do some work, then
110    /// // write back to the pipe. When the pipe is empty, the processes block, so only
111    /// // NUM_SLOT processes can be working at any given time.
112    /// for _ in 0..NUM_PROC {
113    ///     jobs.push(
114    ///         Command::new("bash")
115    ///             .args(["-c",
116    ///                 &format!(
117    ///                      "read -n 1\n\
118    ///                       echo -n 'x' >> '{OUTPUT}'\n\
119    ///                       echo -n '|'",
120    ///                 ),
121    ///             ])
122    ///             .stdin(reader.try_clone()?)
123    ///             .stdout(writer.try_clone()?)
124    ///             .spawn()?,
125    ///     );
126    /// }
127    ///
128    /// // Wait for all jobs to finish.
129    /// for mut job in jobs {
130    ///     job.wait()?;
131    /// }
132    ///
133    /// // Check our work and clean up.
134    /// let xs = fs::read_to_string(OUTPUT)?;
135    /// fs::remove_file(OUTPUT)?;
136    /// assert_eq!(xs, "x".repeat(NUM_PROC.into()));
137    /// # Ok(())
138    /// # }
139    /// ```
140    #[unstable(feature = "anonymous_pipe", issue = "127154")]
141    pub fn try_clone(&self) -> io::Result<Self> {
142        self.0.try_clone().map(Self)
143    }
144}
145
146impl PipeWriter {
147    /// Create a new [`PipeWriter`] instance that shares the same underlying file description.
148    ///
149    /// # Examples
150    ///
151    /// ```no_run
152    /// #![feature(anonymous_pipe)]
153    /// # #[cfg(miri)] fn main() {}
154    /// # #[cfg(not(miri))]
155    /// # fn main() -> std::io::Result<()> {
156    /// use std::process::Command;
157    /// use std::io::{pipe, Read};
158    /// let (mut reader, writer) = pipe()?;
159    ///
160    /// // Spawn a process that writes to stdout and stderr.
161    /// let mut peer = Command::new("bash")
162    ///     .args([
163    ///         "-c",
164    ///         "echo -n foo\n\
165    ///          echo -n bar >&2"
166    ///     ])
167    ///     .stdout(writer.try_clone()?)
168    ///     .stderr(writer)
169    ///     .spawn()?;
170    ///
171    /// // Read and check the result.
172    /// let mut msg = String::new();
173    /// reader.read_to_string(&mut msg)?;
174    /// assert_eq!(&msg, "foobar");
175    ///
176    /// peer.wait()?;
177    /// # Ok(())
178    /// # }
179    /// ```
180    #[unstable(feature = "anonymous_pipe", issue = "127154")]
181    pub fn try_clone(&self) -> io::Result<Self> {
182        self.0.try_clone().map(Self)
183    }
184}
185
186#[unstable(feature = "anonymous_pipe", issue = "127154")]
187impl io::Read for &PipeReader {
188    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
189        self.0.read(buf)
190    }
191    fn read_vectored(&mut self, bufs: &mut [io::IoSliceMut<'_>]) -> io::Result<usize> {
192        self.0.read_vectored(bufs)
193    }
194    #[inline]
195    fn is_read_vectored(&self) -> bool {
196        self.0.is_read_vectored()
197    }
198    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
199        self.0.read_to_end(buf)
200    }
201    fn read_buf(&mut self, buf: io::BorrowedCursor<'_>) -> io::Result<()> {
202        self.0.read_buf(buf)
203    }
204}
205
206#[unstable(feature = "anonymous_pipe", issue = "127154")]
207impl io::Read for PipeReader {
208    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
209        self.0.read(buf)
210    }
211    fn read_vectored(&mut self, bufs: &mut [io::IoSliceMut<'_>]) -> io::Result<usize> {
212        self.0.read_vectored(bufs)
213    }
214    #[inline]
215    fn is_read_vectored(&self) -> bool {
216        self.0.is_read_vectored()
217    }
218    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
219        self.0.read_to_end(buf)
220    }
221    fn read_buf(&mut self, buf: io::BorrowedCursor<'_>) -> io::Result<()> {
222        self.0.read_buf(buf)
223    }
224}
225
226#[unstable(feature = "anonymous_pipe", issue = "127154")]
227impl io::Write for &PipeWriter {
228    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
229        self.0.write(buf)
230    }
231    #[inline]
232    fn flush(&mut self) -> io::Result<()> {
233        Ok(())
234    }
235    fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
236        self.0.write_vectored(bufs)
237    }
238    #[inline]
239    fn is_write_vectored(&self) -> bool {
240        self.0.is_write_vectored()
241    }
242}
243
244#[unstable(feature = "anonymous_pipe", issue = "127154")]
245impl io::Write for PipeWriter {
246    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
247        self.0.write(buf)
248    }
249    #[inline]
250    fn flush(&mut self) -> io::Result<()> {
251        Ok(())
252    }
253    fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
254        self.0.write_vectored(bufs)
255    }
256    #[inline]
257    fn is_write_vectored(&self) -> bool {
258        self.0.is_write_vectored()
259    }
260}