Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 33 additions & 9 deletions crates/kernel/src/kernel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,9 @@ struct Io {
/// host-managed server stops when it reaches EOF, which also happens
/// when this process dies, so a crashed shell leaves no kernel behind.
lifeline: Option<tokio::process::ChildStdin>,
/// The task copying the child's stderr into the [`Tail`]; it ends at
/// the end of the pipe, after the kernel's last line.
stderr: Option<tokio::task::JoinHandle<()>>,
}

/// Keeps the last lines the kernel wrote to stderr, to say why it exited.
Expand Down Expand Up @@ -116,7 +119,7 @@ fn start(launch: &Launch, network: &Network, log: &LogSink, tail: &Tail) -> Resu
return Err("the kernel's stdin/stdout are not piped".into());
};
// The kernel logs to stderr; it never carries protocol frames.
if let Some(stderr) = child.stderr.take() {
let stderr = child.stderr.take().map(|stderr| {
let log = log.clone();
let tail = tail.clone();
tokio::spawn(async move {
Expand All @@ -126,14 +129,14 @@ fn start(launch: &Launch, network: &Network, log: &LogSink, tail: &Tail) -> Resu
(log)(&format!("octos: {line}"));
tail.push(line);
}
});
}
})
});
let lines = BufReader::new(Box::pin(stdout) as Pin<Box<dyn AsyncRead + Send>>).lines();
Ok(if shared {
// Frames go over the WebSocket once it is up (`prepare_network`).
Io { writer: Box::pin(tokio::io::sink()), lines, running: Running::Child(child), lifeline: Some(stdin) }
Io { writer: Box::pin(tokio::io::sink()), lines, running: Running::Child(child), lifeline: Some(stdin), stderr }
} else {
Io { writer: Box::pin(stdin), lines, running: Running::Child(child), lifeline: None }
Io { writer: Box::pin(stdin), lines, running: Running::Child(child), lifeline: None, stderr }
})
}
#[cfg(target_env = "ohos")]
Expand All @@ -153,6 +156,7 @@ fn start(launch: &Launch, network: &Network, log: &LogSink, tail: &Tail) -> Resu
lines: BufReader::new(Box::pin(reader) as Pin<Box<dyn AsyncRead + Send>>).lines(),
running: Running::Embedded(task),
lifeline: None,
stderr: None,
})
}
#[cfg(not(target_env = "ohos"))]
Expand Down Expand Up @@ -266,7 +270,7 @@ async fn exited(running: &mut Running) -> String {
async fn stop(io: Io) {
// Closing stdin stops either kind of child: the stdio pipe drains owned
// turns on EOF; a host-managed server drains its connections and exits.
let Io { writer, lines, mut running, lifeline } = io;
let Io { writer, lines, mut running, lifeline, .. } = io;
let grace = if lifeline.is_some() { SHARED_GRACE } else { STDIO_GRACE };
drop(writer);
drop(lines);
Expand Down Expand Up @@ -401,16 +405,18 @@ pub(crate) async fn supervise(
}
}
}
Ok(None) => break CloseReason::Exited(with_tail("the kernel closed its output", &tail)),
Ok(None) => break CloseReason::Exited("the kernel closed its output".into()),
Err(e) => break CloseReason::Exited(format!("reading from the kernel failed: {e}")),
},
why = exited(&mut io.running) => break CloseReason::Exited(with_tail(&why, &tail)),
why = exited(&mut io.running) => break CloseReason::Exited(why),
}
}};
(log)(&format!("octos-core: stopping kernel {generation}: {reason}"));
let stderr = io.stderr.take();
stop(io).await;
match reason {
CloseReason::Failed(why) => CloseReason::Failed(with_tail(&why, &tail)),
CloseReason::Failed(why) => CloseReason::Failed(last_words(&why, stderr, &tail).await),
CloseReason::Exited(why) => CloseReason::Exited(last_words(&why, stderr, &tail).await),
other => other,
}
}
Expand Down Expand Up @@ -464,6 +470,24 @@ fn tool_list_reply(text: &str) -> Option<String> {
})
}

/// How long a stopped kernel's stderr may take to reach its end. The pipe
/// closes with the kernel; only a process it started that inherited the
/// pipe can hold it open longer.
const LAST_WORDS_GRACE: Duration = Duration::from_secs(2);

/// `why` with the kernel's last words (the [`Tail`]), once `stderr`, the
/// task reading them, has reached the end of the pipe. That task runs
/// beside the supervisor, so lines the kernel wrote just before its stdout
/// closed or it exited can still be on their way to the tail: read at once,
/// the reason said "the kernel closed its output: fake kernel up" without
/// the "asked to exit" the kernel wrote last.
async fn last_words(why: &str, stderr: Option<tokio::task::JoinHandle<()>>, tail: &Tail) -> String {
if let Some(reader) = stderr {
let _ = tokio::time::timeout(LAST_WORDS_GRACE, reader).await;
}
with_tail(why, tail)
}

fn with_tail(why: &str, tail: &Tail) -> String {
let tail = tail.text();
if tail.is_empty() {
Expand Down
4 changes: 3 additions & 1 deletion crates/kernel/tests/core.rs
Original file line number Diff line number Diff line change
Expand Up @@ -238,7 +238,9 @@ async fn a_kernel_that_dies_closes_its_connections_with_its_last_words() {
call(&mut a, "1", "session/list", json!({})).await;
a.send(json!({"jsonrpc": "2.0", "method": "test/exit", "params": {}}).to_string()).unwrap();
match next(&mut a).await {
Err(CloseReason::Exited(why)) => assert!(why.contains("asked to exit") || why.contains("exit"), "{why}"),
// The line it wrote to stderr last, whichever end the core saw
// first: its output closing or its exit status.
Err(CloseReason::Exited(why)) => assert!(why.contains("asked to exit"), "{why}"),
other => panic!("expected Exited, got {other:?}"),
}
for _ in 0..100 {
Expand Down
Loading