1 //! Read both stdout and stderr of child without deadlocks.
2 //!
3 //! <https://github.com/rust-lang/cargo/blob/905af549966f23a9288e9993a85d1249a5436556/crates/cargo-util/src/read2.rs>
4 //! <https://github.com/rust-lang/cargo/blob/58a961314437258065e23cb6316dfc121d96fb71/crates/cargo-util/src/process_builder.rs#L231>
5
6 use std::{
7 io,
8 process::{ChildStderr, ChildStdout, Command, Output, Stdio},
9 };
10
11 use crate::JodChild;
12
streaming_output( out: ChildStdout, err: ChildStderr, on_stdout_line: &mut dyn FnMut(&str), on_stderr_line: &mut dyn FnMut(&str), ) -> io::Result<(Vec<u8>, Vec<u8>)>13 pub fn streaming_output(
14 out: ChildStdout,
15 err: ChildStderr,
16 on_stdout_line: &mut dyn FnMut(&str),
17 on_stderr_line: &mut dyn FnMut(&str),
18 ) -> io::Result<(Vec<u8>, Vec<u8>)> {
19 let mut stdout = Vec::new();
20 let mut stderr = Vec::new();
21
22 imp::read2(out, err, &mut |is_out, data, eof| {
23 let idx = if eof {
24 data.len()
25 } else {
26 match data.iter().rposition(|b| *b == b'\n') {
27 Some(i) => i + 1,
28 None => return,
29 }
30 };
31 {
32 // scope for new_lines
33 let new_lines = {
34 let dst = if is_out { &mut stdout } else { &mut stderr };
35 let start = dst.len();
36 let data = data.drain(..idx);
37 dst.extend(data);
38 &dst[start..]
39 };
40 for line in String::from_utf8_lossy(new_lines).lines() {
41 if is_out {
42 on_stdout_line(line);
43 } else {
44 on_stderr_line(line);
45 }
46 }
47 }
48 })?;
49
50 Ok((stdout, stderr))
51 }
52
spawn_with_streaming_output( mut cmd: Command, on_stdout_line: &mut dyn FnMut(&str), on_stderr_line: &mut dyn FnMut(&str), ) -> io::Result<Output>53 pub fn spawn_with_streaming_output(
54 mut cmd: Command,
55 on_stdout_line: &mut dyn FnMut(&str),
56 on_stderr_line: &mut dyn FnMut(&str),
57 ) -> io::Result<Output> {
58 let cmd = cmd.stdout(Stdio::piped()).stderr(Stdio::piped()).stdin(Stdio::null());
59
60 let mut child = JodChild(cmd.spawn()?);
61 let (stdout, stderr) = streaming_output(
62 child.stdout.take().unwrap(),
63 child.stderr.take().unwrap(),
64 on_stdout_line,
65 on_stderr_line,
66 )?;
67 let status = child.wait()?;
68 Ok(Output { status, stdout, stderr })
69 }
70
71 #[cfg(unix)]
72 mod imp {
73 use std::{
74 io::{self, prelude::*},
75 mem,
76 os::unix::prelude::*,
77 process::{ChildStderr, ChildStdout},
78 };
79
read2( mut out_pipe: ChildStdout, mut err_pipe: ChildStderr, data: &mut dyn FnMut(bool, &mut Vec<u8>, bool), ) -> io::Result<()>80 pub(crate) fn read2(
81 mut out_pipe: ChildStdout,
82 mut err_pipe: ChildStderr,
83 data: &mut dyn FnMut(bool, &mut Vec<u8>, bool),
84 ) -> io::Result<()> {
85 unsafe {
86 libc::fcntl(out_pipe.as_raw_fd(), libc::F_SETFL, libc::O_NONBLOCK);
87 libc::fcntl(err_pipe.as_raw_fd(), libc::F_SETFL, libc::O_NONBLOCK);
88 }
89
90 let mut out_done = false;
91 let mut err_done = false;
92 let mut out = Vec::new();
93 let mut err = Vec::new();
94
95 let mut fds: [libc::pollfd; 2] = unsafe { mem::zeroed() };
96 fds[0].fd = out_pipe.as_raw_fd();
97 fds[0].events = libc::POLLIN;
98 fds[1].fd = err_pipe.as_raw_fd();
99 fds[1].events = libc::POLLIN;
100 let mut nfds = 2;
101 let mut errfd = 1;
102
103 while nfds > 0 {
104 // wait for either pipe to become readable using `select`
105 let r = unsafe { libc::poll(fds.as_mut_ptr(), nfds, -1) };
106 if r == -1 {
107 let err = io::Error::last_os_error();
108 if err.kind() == io::ErrorKind::Interrupted {
109 continue;
110 }
111 return Err(err);
112 }
113
114 // Read as much as we can from each pipe, ignoring EWOULDBLOCK or
115 // EAGAIN. If we hit EOF, then this will happen because the underlying
116 // reader will return Ok(0), in which case we'll see `Ok` ourselves. In
117 // this case we flip the other fd back into blocking mode and read
118 // whatever's leftover on that file descriptor.
119 let handle = |res: io::Result<_>| match res {
120 Ok(_) => Ok(true),
121 Err(e) => {
122 if e.kind() == io::ErrorKind::WouldBlock {
123 Ok(false)
124 } else {
125 Err(e)
126 }
127 }
128 };
129 if !err_done && fds[errfd].revents != 0 && handle(err_pipe.read_to_end(&mut err))? {
130 err_done = true;
131 nfds -= 1;
132 }
133 data(false, &mut err, err_done);
134 if !out_done && fds[0].revents != 0 && handle(out_pipe.read_to_end(&mut out))? {
135 out_done = true;
136 fds[0].fd = err_pipe.as_raw_fd();
137 errfd = 0;
138 nfds -= 1;
139 }
140 data(true, &mut out, out_done);
141 }
142 Ok(())
143 }
144 }
145
146 #[cfg(windows)]
147 mod imp {
148 use std::{
149 io,
150 os::windows::prelude::*,
151 process::{ChildStderr, ChildStdout},
152 slice,
153 };
154
155 use miow::{
156 iocp::{CompletionPort, CompletionStatus},
157 pipe::NamedPipe,
158 Overlapped,
159 };
160 use winapi::shared::winerror::ERROR_BROKEN_PIPE;
161
162 struct Pipe<'a> {
163 dst: &'a mut Vec<u8>,
164 overlapped: Overlapped,
165 pipe: NamedPipe,
166 done: bool,
167 }
168
read2( out_pipe: ChildStdout, err_pipe: ChildStderr, data: &mut dyn FnMut(bool, &mut Vec<u8>, bool), ) -> io::Result<()>169 pub(crate) fn read2(
170 out_pipe: ChildStdout,
171 err_pipe: ChildStderr,
172 data: &mut dyn FnMut(bool, &mut Vec<u8>, bool),
173 ) -> io::Result<()> {
174 let mut out = Vec::new();
175 let mut err = Vec::new();
176
177 let port = CompletionPort::new(1)?;
178 port.add_handle(0, &out_pipe)?;
179 port.add_handle(1, &err_pipe)?;
180
181 unsafe {
182 let mut out_pipe = Pipe::new(out_pipe, &mut out);
183 let mut err_pipe = Pipe::new(err_pipe, &mut err);
184
185 out_pipe.read()?;
186 err_pipe.read()?;
187
188 let mut status = [CompletionStatus::zero(), CompletionStatus::zero()];
189
190 while !out_pipe.done || !err_pipe.done {
191 for status in port.get_many(&mut status, None)? {
192 if status.token() == 0 {
193 out_pipe.complete(status);
194 data(true, out_pipe.dst, out_pipe.done);
195 out_pipe.read()?;
196 } else {
197 err_pipe.complete(status);
198 data(false, err_pipe.dst, err_pipe.done);
199 err_pipe.read()?;
200 }
201 }
202 }
203
204 Ok(())
205 }
206 }
207
208 impl<'a> Pipe<'a> {
new<P: IntoRawHandle>(p: P, dst: &'a mut Vec<u8>) -> Pipe<'a>209 unsafe fn new<P: IntoRawHandle>(p: P, dst: &'a mut Vec<u8>) -> Pipe<'a> {
210 Pipe {
211 dst,
212 pipe: NamedPipe::from_raw_handle(p.into_raw_handle()),
213 overlapped: Overlapped::zero(),
214 done: false,
215 }
216 }
217
read(&mut self) -> io::Result<()>218 unsafe fn read(&mut self) -> io::Result<()> {
219 let dst = slice_to_end(self.dst);
220 match self.pipe.read_overlapped(dst, self.overlapped.raw()) {
221 Ok(_) => Ok(()),
222 Err(e) => {
223 if e.raw_os_error() == Some(ERROR_BROKEN_PIPE as i32) {
224 self.done = true;
225 Ok(())
226 } else {
227 Err(e)
228 }
229 }
230 }
231 }
232
complete(&mut self, status: &CompletionStatus)233 unsafe fn complete(&mut self, status: &CompletionStatus) {
234 let prev = self.dst.len();
235 self.dst.set_len(prev + status.bytes_transferred() as usize);
236 if status.bytes_transferred() == 0 {
237 self.done = true;
238 }
239 }
240 }
241
slice_to_end(v: &mut Vec<u8>) -> &mut [u8]242 unsafe fn slice_to_end(v: &mut Vec<u8>) -> &mut [u8] {
243 if v.capacity() == 0 {
244 v.reserve(16);
245 }
246 if v.capacity() == v.len() {
247 v.reserve(1);
248 }
249 slice::from_raw_parts_mut(v.as_mut_ptr().add(v.len()), v.capacity() - v.len())
250 }
251 }
252
253 #[cfg(target_arch = "wasm32")]
254 mod imp {
255 use std::{
256 io,
257 process::{ChildStderr, ChildStdout},
258 };
259
read2( _out_pipe: ChildStdout, _err_pipe: ChildStderr, _data: &mut dyn FnMut(bool, &mut Vec<u8>, bool), ) -> io::Result<()>260 pub(crate) fn read2(
261 _out_pipe: ChildStdout,
262 _err_pipe: ChildStderr,
263 _data: &mut dyn FnMut(bool, &mut Vec<u8>, bool),
264 ) -> io::Result<()> {
265 panic!("no processes on wasm")
266 }
267 }
268