• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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