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}