|
| 1 | +use std::{ |
| 2 | + fs::File, |
| 3 | + io::{self, Read}, |
| 4 | + os::fd::AsFd, |
| 5 | + process::{Command, ExitStatus, Stdio}, |
| 6 | + thread, |
| 7 | + time::Duration, |
| 8 | +}; |
| 9 | + |
| 10 | +use nix::{ |
| 11 | + fcntl::{FcntlArg, OFlag, fcntl}, |
| 12 | + poll::{PollFd, PollFlags, PollTimeout, poll}, |
| 13 | + sys::socket::{ |
| 14 | + AddressFamily, SockFlag, SockType, setsockopt, socketpair, |
| 15 | + sockopt::{RcvBuf, SndBuf}, |
| 16 | + }, |
| 17 | +}; |
| 18 | + |
| 19 | +const REQUESTED_SOCKET_BUFFER: usize = 1024; |
| 20 | +const DRAIN_CHUNK: usize = 1024; |
| 21 | + |
| 22 | +#[derive(Debug, Default)] |
| 23 | +struct Options { |
| 24 | + digest: Option<(usize, usize)>, |
| 25 | + command: Vec<String>, |
| 26 | +} |
| 27 | + |
| 28 | +/// Run a command with a deliberately non-blocking, backpressured stdout. |
| 29 | +/// |
| 30 | +/// The reader consumes one small chunk whenever the channel fills. That lets a |
| 31 | +/// blocking writer advance while keeping enough pressure for a non-blocking |
| 32 | +/// writer to encounter `EAGAIN`. |
| 33 | +pub fn run(args: &[String]) -> Result<(), Box<dyn std::error::Error>> { |
| 34 | + let options = parse_options(args)?; |
| 35 | + let (reader_fd, writer_fd) = |
| 36 | + socketpair(AddressFamily::Unix, SockType::Stream, None, SockFlag::empty())?; |
| 37 | + setsockopt(&reader_fd, RcvBuf, &REQUESTED_SOCKET_BUFFER)?; |
| 38 | + setsockopt(&writer_fd, SndBuf, &REQUESTED_SOCKET_BUFFER)?; |
| 39 | + |
| 40 | + let mut reader = File::from(reader_fd); |
| 41 | + let writer = File::from(writer_fd); |
| 42 | + set_nonblocking(&writer, true)?; |
| 43 | + |
| 44 | + let retained_writer = writer.try_clone()?; |
| 45 | + |
| 46 | + let mut child = Command::new(&options.command[0]) |
| 47 | + .args(&options.command[1..]) |
| 48 | + .stdout(Stdio::from(writer)) |
| 49 | + .stderr(Stdio::piped()) |
| 50 | + .spawn()?; |
| 51 | + |
| 52 | + let stderr = child.stderr.take().ok_or("failed to capture child stderr")?; |
| 53 | + let stderr_thread = thread::spawn(move || -> io::Result<Vec<u8>> { |
| 54 | + let mut stderr = stderr; |
| 55 | + let mut captured = Vec::new(); |
| 56 | + stderr.read_to_end(&mut captured)?; |
| 57 | + Ok(captured) |
| 58 | + }); |
| 59 | + |
| 60 | + let (status, stdout, saw_nonblocking_backpressure) = |
| 61 | + capture_backpressured_stdout(&mut child, &mut reader, retained_writer)?; |
| 62 | + let stderr = stderr_thread.join().map_err(|_| "stderr reader thread panicked")??; |
| 63 | + |
| 64 | + if saw_nonblocking_backpressure { |
| 65 | + eprintln!("backpressure-run detected truncated child output under stdio backpressure"); |
| 66 | + std::process::exit(1); |
| 67 | + } |
| 68 | + |
| 69 | + replay("stdout", &stdout, options.digest); |
| 70 | + replay("stderr", &stderr, options.digest); |
| 71 | + |
| 72 | + std::process::exit(status.code().unwrap_or(1)); |
| 73 | +} |
| 74 | + |
| 75 | +fn parse_options(args: &[String]) -> Result<Options, Box<dyn std::error::Error>> { |
| 76 | + let Some(separator) = args.iter().position(|arg| arg == "--") else { |
| 77 | + return Err(usage().into()); |
| 78 | + }; |
| 79 | + |
| 80 | + let mut options = Options { command: args[separator + 1..].to_vec(), ..Options::default() }; |
| 81 | + if options.command.is_empty() { |
| 82 | + return Err(usage().into()); |
| 83 | + } |
| 84 | + |
| 85 | + let mut index = 0; |
| 86 | + while index < separator { |
| 87 | + match args[index].as_str() { |
| 88 | + "--digest" => { |
| 89 | + let value = args.get(index + 1).ok_or_else(usage)?; |
| 90 | + options.digest = Some(parse_digest(value)?); |
| 91 | + index += 2; |
| 92 | + } |
| 93 | + _ => return Err(usage().into()), |
| 94 | + } |
| 95 | + } |
| 96 | + |
| 97 | + Ok(options) |
| 98 | +} |
| 99 | + |
| 100 | +fn usage() -> String { |
| 101 | + "Usage: vpt backpressure-run [--digest <head>,<tail>] -- <command> [args...]".to_owned() |
| 102 | +} |
| 103 | + |
| 104 | +fn parse_digest(value: &str) -> Result<(usize, usize), Box<dyn std::error::Error>> { |
| 105 | + let Some((head, tail)) = value.split_once(',') else { |
| 106 | + return Err("--digest must be <head>,<tail>".into()); |
| 107 | + }; |
| 108 | + Ok((head.parse()?, tail.parse()?)) |
| 109 | +} |
| 110 | + |
| 111 | +fn set_nonblocking(fd: &impl AsFd, nonblocking: bool) -> io::Result<()> { |
| 112 | + let flags = fcntl(fd, FcntlArg::F_GETFL).map_err(io::Error::from)?; |
| 113 | + let mut flags = OFlag::from_bits_retain(flags); |
| 114 | + flags.set(OFlag::O_NONBLOCK, nonblocking); |
| 115 | + fcntl(fd, FcntlArg::F_SETFL(flags)).map_err(io::Error::from)?; |
| 116 | + Ok(()) |
| 117 | +} |
| 118 | + |
| 119 | +fn capture_backpressured_stdout( |
| 120 | + child: &mut std::process::Child, |
| 121 | + reader: &mut File, |
| 122 | + retained_writer: File, |
| 123 | +) -> io::Result<(ExitStatus, Vec<u8>, bool)> { |
| 124 | + let mut retained_writer = Some(retained_writer); |
| 125 | + let mut captured = Vec::new(); |
| 126 | + let mut saw_nonblocking_backpressure = false; |
| 127 | + |
| 128 | + loop { |
| 129 | + if let Some(status) = child.try_wait()? { |
| 130 | + retained_writer.take(); |
| 131 | + reader.read_to_end(&mut captured)?; |
| 132 | + return Ok((status, captured, saw_nonblocking_backpressure)); |
| 133 | + } |
| 134 | + |
| 135 | + let writer = retained_writer.as_ref().expect("writer is retained until child exit"); |
| 136 | + if !is_writable(writer)? { |
| 137 | + saw_nonblocking_backpressure |= is_nonblocking(writer)?; |
| 138 | + |
| 139 | + let mut chunk = [0_u8; DRAIN_CHUNK]; |
| 140 | + let read = reader.read(&mut chunk)?; |
| 141 | + if read == 0 { |
| 142 | + return Err(io::Error::new( |
| 143 | + io::ErrorKind::UnexpectedEof, |
| 144 | + "stdout socket closed before the child exited", |
| 145 | + )); |
| 146 | + } |
| 147 | + captured.extend_from_slice(&chunk[..read]); |
| 148 | + } else { |
| 149 | + thread::sleep(Duration::from_millis(1)); |
| 150 | + } |
| 151 | + } |
| 152 | +} |
| 153 | + |
| 154 | +fn is_writable(fd: &impl AsFd) -> io::Result<bool> { |
| 155 | + let mut fds = [PollFd::new(fd.as_fd(), PollFlags::POLLOUT)]; |
| 156 | + loop { |
| 157 | + match poll(&mut fds, PollTimeout::ZERO) { |
| 158 | + Ok(ready) => return Ok(ready > 0), |
| 159 | + Err(nix::errno::Errno::EINTR) => {} |
| 160 | + Err(error) => return Err(io::Error::from(error)), |
| 161 | + } |
| 162 | + } |
| 163 | +} |
| 164 | + |
| 165 | +fn is_nonblocking(fd: &impl AsFd) -> io::Result<bool> { |
| 166 | + let flags = fcntl(fd, FcntlArg::F_GETFL).map_err(io::Error::from)?; |
| 167 | + Ok(OFlag::from_bits_retain(flags).contains(OFlag::O_NONBLOCK)) |
| 168 | +} |
| 169 | + |
| 170 | +fn replay(name: &str, bytes: &[u8], digest: Option<(usize, usize)>) { |
| 171 | + println!("--- {name} ---"); |
| 172 | + let text = String::from_utf8_lossy(bytes); |
| 173 | + if let Some((head, tail)) = digest { |
| 174 | + let lines: Vec<_> = text.lines().collect(); |
| 175 | + println!("{name}: {} lines", lines.len()); |
| 176 | + if lines.len() <= head + tail { |
| 177 | + for line in lines { |
| 178 | + println!("{line}"); |
| 179 | + } |
| 180 | + } else { |
| 181 | + for line in &lines[..head] { |
| 182 | + println!("{line}"); |
| 183 | + } |
| 184 | + println!("... {} lines elided ...", lines.len() - head - tail); |
| 185 | + for line in &lines[lines.len() - tail..] { |
| 186 | + println!("{line}"); |
| 187 | + } |
| 188 | + } |
| 189 | + } else { |
| 190 | + print!("{text}"); |
| 191 | + if !text.is_empty() && !text.ends_with('\n') { |
| 192 | + println!(); |
| 193 | + } |
| 194 | + } |
| 195 | +} |
0 commit comments