1use crate::io;
2use crate::sys::anonymous_pipe::{AnonPipe, pipe as pipe_inner};
3use crate::sys_common::{FromInner, IntoInner};
4
5/// Create 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/// # Examples
42///
43/// ```no_run
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#[stable(feature = "anonymous_pipe", since = "1.87.0")]
71#[inline]
72pub fn pipe() -> io::Result<(PipeReader, PipeWriter)> {
73 pipe_inner().map(|(reader: FileDesc, writer: FileDesc)| (PipeReader(reader), PipeWriter(writer)))
74}
75
76/// Read end of an anonymous pipe.
77#[stable(feature = "anonymous_pipe", since = "1.87.0")]
78#[derive(Debug)]
79pub struct PipeReader(pub(crate) AnonPipe);
80
81/// Write end of an anonymous pipe.
82#[stable(feature = "anonymous_pipe", since = "1.87.0")]
83#[derive(Debug)]
84pub struct PipeWriter(pub(crate) AnonPipe);
85
86impl FromInner<AnonPipe> for PipeReader {
87 fn from_inner(inner: AnonPipe) -> Self {
88 Self(inner)
89 }
90}
91
92impl IntoInner<AnonPipe> for PipeReader {
93 fn into_inner(self) -> AnonPipe {
94 self.0
95 }
96}
97
98impl FromInner<AnonPipe> for PipeWriter {
99 fn from_inner(inner: AnonPipe) -> Self {
100 Self(inner)
101 }
102}
103
104impl IntoInner<AnonPipe> for PipeWriter {
105 fn into_inner(self) -> AnonPipe {
106 self.0
107 }
108}
109
110impl PipeReader {
111 /// Create a new [`PipeReader`] instance that shares the same underlying file description.
112 ///
113 /// # Examples
114 ///
115 /// ```no_run
116 /// # #[cfg(miri)] fn main() {}
117 /// # #[cfg(not(miri))]
118 /// # fn main() -> std::io::Result<()> {
119 /// use std::fs;
120 /// use std::io::{pipe, Write};
121 /// use std::process::Command;
122 /// const NUM_SLOT: u8 = 2;
123 /// const NUM_PROC: u8 = 5;
124 /// const OUTPUT: &str = "work.txt";
125 ///
126 /// let mut jobs = vec![];
127 /// let (reader, mut writer) = pipe()?;
128 ///
129 /// // Write NUM_SLOT characters the pipe.
130 /// writer.write_all(&[b'|'; NUM_SLOT as usize])?;
131 ///
132 /// // Spawn several processes that read a character from the pipe, do some work, then
133 /// // write back to the pipe. When the pipe is empty, the processes block, so only
134 /// // NUM_SLOT processes can be working at any given time.
135 /// for _ in 0..NUM_PROC {
136 /// jobs.push(
137 /// Command::new("bash")
138 /// .args(["-c",
139 /// &format!(
140 /// "read -n 1\n\
141 /// echo -n 'x' >> '{OUTPUT}'\n\
142 /// echo -n '|'",
143 /// ),
144 /// ])
145 /// .stdin(reader.try_clone()?)
146 /// .stdout(writer.try_clone()?)
147 /// .spawn()?,
148 /// );
149 /// }
150 ///
151 /// // Wait for all jobs to finish.
152 /// for mut job in jobs {
153 /// job.wait()?;
154 /// }
155 ///
156 /// // Check our work and clean up.
157 /// let xs = fs::read_to_string(OUTPUT)?;
158 /// fs::remove_file(OUTPUT)?;
159 /// assert_eq!(xs, "x".repeat(NUM_PROC.into()));
160 /// # Ok(())
161 /// # }
162 /// ```
163 #[stable(feature = "anonymous_pipe", since = "1.87.0")]
164 pub fn try_clone(&self) -> io::Result<Self> {
165 self.0.try_clone().map(Self)
166 }
167}
168
169impl PipeWriter {
170 /// Create a new [`PipeWriter`] instance that shares the same underlying file description.
171 ///
172 /// # Examples
173 ///
174 /// ```no_run
175 /// # #[cfg(miri)] fn main() {}
176 /// # #[cfg(not(miri))]
177 /// # fn main() -> std::io::Result<()> {
178 /// use std::process::Command;
179 /// use std::io::{pipe, Read};
180 /// let (mut reader, writer) = pipe()?;
181 ///
182 /// // Spawn a process that writes to stdout and stderr.
183 /// let mut peer = Command::new("bash")
184 /// .args([
185 /// "-c",
186 /// "echo -n foo\n\
187 /// echo -n bar >&2"
188 /// ])
189 /// .stdout(writer.try_clone()?)
190 /// .stderr(writer)
191 /// .spawn()?;
192 ///
193 /// // Read and check the result.
194 /// let mut msg = String::new();
195 /// reader.read_to_string(&mut msg)?;
196 /// assert_eq!(&msg, "foobar");
197 ///
198 /// peer.wait()?;
199 /// # Ok(())
200 /// # }
201 /// ```
202 #[stable(feature = "anonymous_pipe", since = "1.87.0")]
203 pub fn try_clone(&self) -> io::Result<Self> {
204 self.0.try_clone().map(Self)
205 }
206}
207
208#[stable(feature = "anonymous_pipe", since = "1.87.0")]
209impl io::Read for &PipeReader {
210 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
211 self.0.read(buf)
212 }
213 fn read_vectored(&mut self, bufs: &mut [io::IoSliceMut<'_>]) -> io::Result<usize> {
214 self.0.read_vectored(bufs)
215 }
216 #[inline]
217 fn is_read_vectored(&self) -> bool {
218 self.0.is_read_vectored()
219 }
220 fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
221 self.0.read_to_end(buf)
222 }
223 fn read_buf(&mut self, buf: io::BorrowedCursor<'_>) -> io::Result<()> {
224 self.0.read_buf(cursor:buf)
225 }
226}
227
228#[stable(feature = "anonymous_pipe", since = "1.87.0")]
229impl io::Read for PipeReader {
230 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
231 self.0.read(buf)
232 }
233 fn read_vectored(&mut self, bufs: &mut [io::IoSliceMut<'_>]) -> io::Result<usize> {
234 self.0.read_vectored(bufs)
235 }
236 #[inline]
237 fn is_read_vectored(&self) -> bool {
238 self.0.is_read_vectored()
239 }
240 fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
241 self.0.read_to_end(buf)
242 }
243 fn read_buf(&mut self, buf: io::BorrowedCursor<'_>) -> io::Result<()> {
244 self.0.read_buf(cursor:buf)
245 }
246}
247
248#[stable(feature = "anonymous_pipe", since = "1.87.0")]
249impl io::Write for &PipeWriter {
250 fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
251 self.0.write(buf)
252 }
253 #[inline]
254 fn flush(&mut self) -> io::Result<()> {
255 Ok(())
256 }
257 fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
258 self.0.write_vectored(bufs)
259 }
260 #[inline]
261 fn is_write_vectored(&self) -> bool {
262 self.0.is_write_vectored()
263 }
264}
265
266#[stable(feature = "anonymous_pipe", since = "1.87.0")]
267impl io::Write for PipeWriter {
268 fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
269 self.0.write(buf)
270 }
271 #[inline]
272 fn flush(&mut self) -> io::Result<()> {
273 Ok(())
274 }
275 fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
276 self.0.write_vectored(bufs)
277 }
278 #[inline]
279 fn is_write_vectored(&self) -> bool {
280 self.0.is_write_vectored()
281 }
282}
283