mirror of
https://seed.radicle.dev/rad:zwTxygwuz5LDGBq255RA2CbNGrz8.git
synced 2026-10-18 19:30:29 +08:00
fix: make reading child output not block forever, if there is none
Spawn a thread that sleeps until timeout, then notifies the RealtimeLines condition variable, which wakes up the reader thread and lets it eventually finish. Signed-off-by: Lars Wirzenius <liw@liw.fi>
This commit is contained in:
+44
-6
@@ -161,6 +161,9 @@ pub struct ChildProcess {
|
||||
stdout_thread: JoinHandle<Result<(), TimeoutError>>,
|
||||
stderr_thread: JoinHandle<Result<Vec<u8>, TimeoutError>>,
|
||||
arc: Arc<Mutex<RealtimeLines>>,
|
||||
|
||||
#[allow(dead_code)]
|
||||
timeout_thread: JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl ChildProcess {
|
||||
@@ -216,6 +219,16 @@ impl ChildProcess {
|
||||
.ok_or(TimeoutError::TakeHandle("stderr"))?;
|
||||
let stderr_thread = Self::capture(stderr)?;
|
||||
|
||||
// Launch a thread that notifies the `CondVar` when the child
|
||||
// has been running too long.
|
||||
let timeout_thread = {
|
||||
let mut lines = stdout_lines.clone();
|
||||
spawn(move || {
|
||||
sleep(timeout);
|
||||
lines.finish();
|
||||
})
|
||||
};
|
||||
|
||||
Ok(Self {
|
||||
deadline: Some(
|
||||
Instant::now()
|
||||
@@ -228,6 +241,7 @@ impl ChildProcess {
|
||||
stdout_thread,
|
||||
stderr_thread,
|
||||
arc,
|
||||
timeout_thread,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -417,16 +431,25 @@ impl RealtimeLines {
|
||||
// Lock the mutex to get access to unlocked buffer.
|
||||
let mut buf = mutex.lock().expect("lock to wait for line");
|
||||
|
||||
let started = Instant::now();
|
||||
loop {
|
||||
match buf.line() {
|
||||
// We've been reading for too long.
|
||||
_ if self.started.elapsed() > self.max_duration => return None,
|
||||
|
||||
let timed_out = self.started.elapsed() > self.max_duration;
|
||||
if timed_out {
|
||||
eprintln!("timed out");
|
||||
return None;
|
||||
}
|
||||
let line = buf.line(); // this doesn't block but if there is no output, do we time out?
|
||||
eprintln!(
|
||||
"{} trying to read line: {line:?}",
|
||||
started.elapsed().as_millis()
|
||||
);
|
||||
match line {
|
||||
// We didn't get a line, but the input stream has
|
||||
// finished. Return final partial line, if there is one,
|
||||
// or None.
|
||||
None if buf.is_finished() => {
|
||||
let line = buf.line();
|
||||
eprintln!("return line after finished {line:?}");
|
||||
return line;
|
||||
}
|
||||
|
||||
@@ -434,11 +457,13 @@ impl RealtimeLines {
|
||||
// assume the new input results in either a complete
|
||||
// line or the input stream finishing, so we loop.
|
||||
None => {
|
||||
buf = var.wait(buf).expect("wait for line");
|
||||
eprintln!("wait for line");
|
||||
buf = var.wait(buf).expect("wait for line"); // var is not notified if we time out
|
||||
}
|
||||
|
||||
// We got a line: return it.
|
||||
Some(line) => {
|
||||
eprintln!("return line {line:?}");
|
||||
return Some(line);
|
||||
}
|
||||
}
|
||||
@@ -680,7 +705,20 @@ mod tests {
|
||||
fn yes_to_stdout_while_reading_with_realtimelines() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let (running, mut stdout) = setup("exec yes", SHORT_TIMEOUT, None)?;
|
||||
while stdout.line().is_some() {}
|
||||
assert!(matches!(running.wait(), Err(TimeoutError::TimedOut)));
|
||||
let result = running.wait();
|
||||
eprintln!("result: {result:#?}");
|
||||
assert!(matches!(result, Err(TimeoutError::TimedOut)));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sleep_for_too_long_while_reading_with_realtimelines(
|
||||
) -> Result<(), Box<dyn std::error::Error>> {
|
||||
let (running, mut stdout) = setup("exec sleep 1000", SHORT_TIMEOUT, None)?;
|
||||
while stdout.line().is_some() {}
|
||||
let result = running.wait();
|
||||
eprintln!("result: {result:#?}");
|
||||
assert!(matches!(result, Err(TimeoutError::TimedOut)));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user