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