diff --git a/nvx-rs/README.md b/nvx-rs/README.md index 9ab0128be..f87896aef 100644 --- a/nvx-rs/README.md +++ b/nvx-rs/README.md @@ -161,7 +161,9 @@ frames, the `EXEC` payload that the host encodes and the guest agent decodes, and the control features that a guest image advertises. Beside version 1's frames, `protocol::app::v2` holds the codec of version 2's, which carry every control path, with version 1's payloads, and stream an execution's standard -input, output and error. +input, output and error. The guest loop that nvx-init and the managed agent +share, and the fake OpenVMM, serve both versions, so `v2` follows their +features, `guest` and `testing`, until the host speaks it too. The [conformance vectors](tests/vectors/README.md) of version 1, `tests/vectors/nvxc-v1.txt`, and OpenVMM's vectors and cases of the `NVXS` diff --git a/nvx-rs/src/guest/managed_agent/dispatch.rs b/nvx-rs/src/guest/managed_agent/dispatch.rs index af0749ead..bcc1144bb 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch.rs @@ -12,7 +12,7 @@ use crate::protocol::{ use super::diagnostics; use super::mappings::{HostMappings, MappingError}; use super::seam::{Paths, RealSystem, System}; -use super::session::{Event, ReceiveFailure, Session}; +use super::session::{Event, ReceiveFailure, Request, Session}; use super::{AgentConfig, exec, launch, tty}; impl AgentConfig { @@ -108,6 +108,9 @@ pub(crate) enum Outcome { } /// Serves control requests until STOP or a fatal session error. +/// +/// The control descriptor is nonblocking, so that the session's send queue never waits for the +/// console while an execution of NVXC version 2 runs. pub(crate) fn run( system: &mut impl System, session: &mut Session, @@ -117,6 +120,7 @@ pub(crate) fn run( if session.stopping() { return Ok(Outcome::Stopped); } + session.descriptor().set_nonblocking()?; if let Err(error) = session.send_attach() { return if session.stopping() { Ok(Outcome::Stopped) @@ -193,7 +197,9 @@ fn handle_data( payload: &[u8], ) -> Result, Errno> { let request = match session.accept_request(payload) { - Ok(request) => request, + Ok(Request::App(request)) => request, + // The input of an execution that ended takes no more bytes. + Ok(Request::Stdin(..)) => return Ok(None), Err(failure) => { diagnostics::error(system, failure.kind.stage(), failure.status); return Err(failure.into_error()); @@ -259,7 +265,18 @@ fn handle_data( } }; session.start_activation()?; - exec::run(system, session, config, request.request_id, &execution)?; + let mut stop = None; + exec::run( + system, + session, + config, + request.request_id, + &execution, + &mut stop, + )?; + if let Some(request_id) = stop { + return stopped(system, session, request_id); + } if session.capture_pending() { return session.finish_capture(); } @@ -293,9 +310,7 @@ fn handle_data( if !request.payload.is_empty() { return Err(Errno::EPROTO); } - session.send_app(APP_STOPPED, request.request_id, 0, &[])?; - tty::drain(system, session.descriptor()); - return Ok(Some(Outcome::Stopped)); + return stopped(system, session, request.request_id); } _ => session.send_error( request.request_id, @@ -306,5 +321,17 @@ fn handle_data( Ok(None) } +/// Acknowledges the `STOP` of `request_id`, and drains the control terminal before the session +/// ends. +fn stopped( + system: &mut impl System, + session: &mut Session, + request_id: u64, +) -> Result, Errno> { + session.send_app(APP_STOPPED, request_id, 0, &[])?; + tty::drain(system, session.descriptor()); + Ok(Some(Outcome::Stopped)) +} + #[cfg(all(test, feature = "cli"))] mod tests; diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs index 4eb84a829..ef4010e94 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs @@ -225,6 +225,7 @@ fn read_outer(host: &mut UnixStream) -> (OuterHeader, Vec) { } mod runtime; +mod streams; fn response(kind: u8, request_id: u64, status: i32, payload: &[u8]) -> Response { Response { diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs index 262a6ed36..85d85d27c 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs @@ -237,7 +237,7 @@ fn prelaunch_termination(helper_ends: bool) { let Ok(Event::Data(payload)) = session.next_event() else { panic!("the complete EXEC record was not received"); }; - let Ok(request) = session.accept_request(&payload) else { + let Ok(Request::App(request)) = session.accept_request(&payload) else { panic!("the complete EXEC request was not accepted"); }; assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT); @@ -258,6 +258,7 @@ fn prelaunch_termination(helper_ends: bool) { &config, request.request_id, &execution, + &mut None, ) .err(); drop(session); diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/streams.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/streams.rs new file mode 100644 index 000000000..00795b901 --- /dev/null +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/streams.rs @@ -0,0 +1,461 @@ +//! NVXC version 2 through the shared loop: its requests beside version 1's, with `READY`'s +//! version; one execution at a time, with `BUSY` for another, and `CANCEL` and `STOP` while it +//! runs; its standard input, which the agent acknowledges once its pipe took it, under the +//! window, until its explicit end; its output past version 1's cap; and the send queue, while +//! which is full the agent reads no output and still serves control requests. + +use std::os::fd::AsFd; +use std::time::{Duration, Instant}; + +use crate::guest::activation::ActivationAction; +use crate::protocol::v2::{self, Frame, STDIN_WINDOW, VERSION}; + +use super::*; + +/// A frame of version 2 that the agent sent, with its request ID. +#[derive(Debug, PartialEq, Eq)] +enum Reply { + Ready(u64, Vec), + Stdout(u64, Vec), + Stderr(u64, Vec), + Exit(u64, i32, v2::Outcome), + Stopped(u64), + Ack(u64, u32), + Busy(u64), + Error(u64, i32, Vec), +} + +impl Harness { + /// Sends a request of version 2. The agent's credit for it comes among its replies. + fn send_v2(&mut self, request_id: u64, frame: Frame<'_>) { + let frame = frame.encode(request_id).unwrap(); + let record = encode_outer( + OUTER_DATA, + &INSTANCE, + self.epoch, + self.host_sequence, + &frame, + ); + self.host.write_all(&record.unwrap()).unwrap(); + self.host_sequence += 1; + } + + /// Reads the agent's next frame, of version 2, after the credits before it. + fn next_v2(&mut self) -> Reply { + loop { + if let Some(reply) = self.next_record() { + return reply; + } + } + } + + /// Reads the agent's next frame, as [`next_v2`](Self::next_v2) does, unless none arrives + /// within `timeout`. + fn next_v2_within(&mut self, timeout: Duration) -> Option { + let deadline = Instant::now() + timeout; + loop { + let left = deadline + .saturating_duration_since(Instant::now()) + .as_millis(); + let ready = sys::readable_or_hung_up(&[self.host.as_fd()], left as u16).unwrap(); + if !ready[0] { + return None; + } + if let Some(reply) = self.next_record() { + return Some(reply); + } + } + } + + /// Reads the agent's next record: a credit, as `None`, or a frame of version 2. + fn next_record(&mut self) -> Option { + let (header, payload) = self.read_outer(); + assert_eq!( + (header.epoch, header.sequence), + (self.epoch, self.guest_sequence) + ); + self.guest_sequence += 1; + if header.record_type == OUTER_CREDIT { + return None; + } + assert_eq!(header.record_type, OUTER_DATA); + let (id, frame) = Frame::decode(&payload).unwrap(); + Some(match frame { + Frame::Ready { + version, + activation, + } => { + assert_eq!(version, VERSION); + Reply::Ready(id, activation.to_vec()) + } + Frame::Stdout(bytes) => Reply::Stdout(id, bytes.to_vec()), + Frame::Stderr(bytes) => Reply::Stderr(id, bytes.to_vec()), + Frame::Exit { status, outcome } => Reply::Exit(id, status, outcome), + Frame::Stopped => Reply::Stopped(id), + Frame::StdinAck(count) => Reply::Ack(id, count), + Frame::Busy => Reply::Busy(id), + Frame::Error { status, category } => Reply::Error(id, status, category.to_vec()), + frame => panic!("the agent sent a request: {frame:?}"), + }) + } + + /// Reads the output of execution `id` up to its first line, which its shell's `echo $$` + /// wrote, and returns that process, and what followed that line in the same reads. + fn shell_process(&mut self, id: u64) -> (sys::ProcessId, Vec) { + let mut output = Vec::new(); + let end = loop { + if let Some(end) = output.iter().position(|&byte| byte == b'\n') { + break end; + } + let Reply::Stdout(reply, bytes) = self.next_v2() else { + panic!("the shell did not report its process ID"); + }; + assert_eq!(reply, id); + output.extend(bytes); + }; + let rest = output.split_off(end + 1); + let process = std::str::from_utf8(&output).unwrap().trim_end().parse(); + ( + sys::ProcessId::from_positive(process.unwrap()).unwrap(), + rest, + ) + } + + /// Stops the agent over version 2, and returns its system. + fn stop_v2(mut self, request_id: u64) -> TestSystem { + self.send_v2(request_id, Frame::Stop); + assert_eq!(self.next_v2(), Reply::Stopped(request_id)); + let (outcome, system) = self.thread.join().unwrap(); + assert!(matches!(outcome, Ok(Outcome::Stopped))); + system + } +} + +/// Waits until `process` has ended and its status was collected, or fails the test. +fn until_gone(process: sys::ProcessId) { + let deadline = Instant::now() + Duration::from_secs(20); + while sys::process_exists(process) { + assert!(Instant::now() < deadline, "the workload did not end"); + thread::sleep(Duration::from_millis(5)); + } +} + +/// Returns the payload of an `EXEC` request for `script` in `/bin/sh`. +fn shell(script: &str) -> Vec { + workload_payload(&["/bin/sh", "-c", script], None).unwrap() +} + +#[test] +fn version_2_serves_ping_and_the_mapping_table_beside_version_1() { + let (temporary, system) = TestSystem::with_host_export(); + let export = system.paths.hostfs_dir.join("root").canonicalize().unwrap(); + let source = temporary.path().canonicalize().unwrap().join("guest/src"); + let maps = maps_payload( + 0, + &[( + b"0/src", + source.as_os_str().as_bytes(), + protocol::MAP_READ_ONLY, + )], + None, + ); + let mut harness = Harness::with_system(system, 1); + harness.send_v2(1, Frame::Ping); + assert_eq!(harness.next_v2(), Reply::Ready(1, Vec::new())); + harness.send_v2(2, Frame::Exec(&exec_payload(None))); + assert_eq!( + harness.next_v2(), + Reply::Error(2, Errno::EAGAIN as i32, b"mappings-incomplete".to_vec()) + ); + harness.send_v2(3, Frame::Maps(&maps)); + assert_eq!(harness.next_v2(), Reply::Ready(3, Vec::new())); + // The same session still serves version 1, its features request included. + assert_eq!( + harness.request(APP_PING, &[]), + response(APP_READY, 42, 0, b"") + ); + let features = harness.request(APP_FEATURES, &[]); + assert_eq!( + GuestFeatures::decode(&features.payload), + Ok(GuestFeatures::DIRECT_AGENT) + ); + let system = harness.stop_v2(4); + assert_eq!( + system.kernel.calls(), + attach_calls(&export.join("0/src"), &source, Access::ReadOnly) + ); + assert_eq!(system.tty_drained, 1); + assert!(system.diagnostics.is_empty()); +} + +#[test] +fn version_2_has_no_features_request() { + let frame = [&b"NVXC\x02\x05\0\0"[..], &42u64.to_le_bytes(), &[0; 8]].concat(); + let diagnostics = + Harness::new().fail(&encode_outer(OUTER_DATA, &INSTANCE, 1, 8, &frame).unwrap()); + let line = + |stage: &str, status: i32| format!("NVX-MANAGED-ERROR: stage={stage} status={status}\n"); + assert_eq!( + diagnostics, + [ + line("data-application", 24), + line("data", Errno::EPROTO as i32) + ] + .concat() + .as_bytes() + ); +} + +#[test] +fn version_2_carries_capture_and_activation_with_their_payloads() { + let config = AgentConfig::for_test(); + let mappings = HostMappings::announced(0); + let system = TestSystem::new(Paths::default()); + let mut harness = + Harness::spawn_with_capture(system, config, mappings, Some(Capture::fixture(true))); + harness.reset(1, 7); + harness.send_v2(1, Frame::Ping); + assert_eq!( + harness.next_v2(), + Reply::Error(1, Errno::ENOTSUP as i32, b"unsupported-operation".to_vec()) + ); + // Outside PID 1, the capture's preparation fails, which the agent reports as version 1 + // does. + harness.send_v2(2, Frame::Capture(&[0])); + assert_eq!( + harness.next_v2(), + Reply::Error(2, Errno::EPERM as i32, b"capture-failed".to_vec()) + ); + assert!(harness.stop_v2(3).diagnostics.is_empty()); + + // A restored session's activate request gets the READY of version 2, with what Ready + // reports about the activation. + let config = AgentConfig::for_test(); + let mappings = HostMappings::announced(0); + let (mut capture, payload) = Capture::activation_fixture(); + let restored = capture.restored.as_mut().unwrap(); + let now = Instant::now(); + assert!(matches!( + restored.accept(&payload, now).unwrap().0, + ActivationAction::Apply(_) + )); + let ready = restored.activation.complete(now, Ok(())).unwrap().encode(); + let system = TestSystem::new(Paths::default()); + let mut harness = Harness::spawn_with_capture(system, config, mappings, Some(capture)); + harness.reset(1, 7); + harness.send_v2(4, Frame::Activate(&payload)); + assert_eq!(harness.next_v2(), Reply::Ready(4, ready)); + assert!(harness.stop_v2(5).diagnostics.is_empty()); +} + +#[test] +fn a_version_2_execution_takes_its_input_until_its_end_and_acknowledges_what_it_wrote() -> TestResult +{ + if super::super::super::tests::isolate_process_test( + "NVX_V2_INPUT_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::a_version_2_execution_takes_its_input_until_its_end_and_acknowledges_what_it_wrote", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (_temporary, mut harness) = direct_harness(); + harness.send_v2(42, Frame::Exec(&workload_payload(&["/bin/cat"], None)?)); + let input: Vec = (0..200_000u32).map(|index| (index % 251) as u8).collect(); + let (mut sent, mut acknowledged, mut ended) = (0, 0, false); + let mut output = Vec::new(); + let exit = loop { + // The host never has more than the window unacknowledged. + let room = STDIN_WINDOW as usize - (sent - acknowledged); + let chunk = room.min(16 * 1024).min(input.len() - sent); + if chunk != 0 { + harness.send_v2(42, Frame::Stdin(&input[sent..sent + chunk])); + sent += chunk; + continue; + } + if sent == input.len() && !ended { + harness.send_v2(42, Frame::StdinEnd); + ended = true; + } + match harness.next_v2() { + Reply::Ack(42, count) => { + acknowledged += count as usize; + assert!(acknowledged <= sent); + } + Reply::Stdout(42, bytes) => output.extend(bytes), + Reply::Exit(42, status, outcome) => break (status, outcome), + reply => panic!("unexpected {reply:?}"), + } + }; + // `cat` ends only once its input ends, after every byte that the agent wrote into its pipe. + assert_eq!(exit, (0, v2::Outcome::Exit)); + assert_eq!(output, input); + assert_eq!(acknowledged, input.len()); + // Input of an execution that ended takes no more bytes, and gets no answer. + harness.send_v2(42, Frame::Stdin(b"late")); + harness.send_v2(42, Frame::StdinEnd); + harness.send_v2(43, Frame::Ping); + assert_eq!(harness.next_v2(), Reply::Ready(43, Vec::new())); + assert!(harness.stop_v2(44).diagnostics.is_empty()); + Ok(()) +} + +#[test] +fn a_version_2_execution_never_has_more_than_the_window_unacknowledged() -> TestResult { + if super::super::super::tests::isolate_process_test( + "NVX_V2_WINDOW_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::a_version_2_execution_never_has_more_than_the_window_unacknowledged", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (_temporary, mut harness) = direct_harness(); + // The workload never reads its input, so its pipe takes what it holds, and no more. + harness.send_v2(42, Frame::Exec(&shell("echo $$; exec sleep 30"))); + let (process, _) = harness.shell_process(42); + let window = STDIN_WINDOW as usize; + let chunk = [0x5a; 16 * 1024]; + let (mut sent, mut acknowledged) = (0, 0); + loop { + while sent - acknowledged < window { + let length = (window - (sent - acknowledged)).min(chunk.len()); + harness.send_v2(42, Frame::Stdin(&chunk[..length])); + sent += length; + } + match harness.next_v2_within(Duration::from_millis(500)) { + Some(Reply::Ack(42, count)) => acknowledged += count as usize, + None => break, + Some(reply) => panic!("unexpected {reply:?}"), + } + } + // The agent acknowledged what the pipe took, and holds the rest of a full window. + assert!(acknowledged > 0); + assert_eq!(sent - acknowledged, window); + assert!(sys::process_exists(process)); + // A byte beyond the window ends the session, and the workload with it. + harness.send_v2(42, Frame::Stdin(b"x")); + let (outcome, system) = harness.thread.join().unwrap(); + assert_eq!(outcome.err(), Some(Errno::EPROTO)); + assert_eq!( + system.diagnostics, + format!( + "NVX-MANAGED-ERROR: stage=data status={}\n", + Errno::EPROTO as i32 + ) + .as_bytes() + ); + until_gone(process); + Ok(()) +} + +#[test] +fn a_version_2_execution_runs_alone_and_ends_on_cancel_or_stop() -> TestResult { + if super::super::super::tests::isolate_process_test( + "NVX_V2_BUSY_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::a_version_2_execution_runs_alone_and_ends_on_cancel_or_stop", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (_temporary, mut harness) = direct_harness(); + let sleeper = shell("echo $$; exec sleep 30"); + harness.send_v2(42, Frame::Exec(&sleeper)); + let (process, _) = harness.shell_process(42); + harness.send_v2(43, Frame::Exec(&sleeper)); + assert_eq!(harness.next_v2(), Reply::Busy(43)); + harness.send_v2(44, Frame::Ping); + assert_eq!(harness.next_v2(), Reply::Busy(44)); + // A cancellation of another request leaves the execution running. + harness.send_v2(41, Frame::Cancel); + harness.send_v2(42, Frame::Cancel); + assert_eq!( + harness.next_v2(), + Reply::Exit(42, 128 + 9, v2::Outcome::Cancelled) + ); + until_gone(process); + // `STOP` ends the running execution, which reports it, and then the session. + harness.send_v2(45, Frame::Exec(&sleeper)); + let (process, _) = harness.shell_process(45); + harness.send_v2(46, Frame::Stop); + assert_eq!( + harness.next_v2(), + Reply::Exit(45, 128 + 9, v2::Outcome::Cancelled) + ); + assert_eq!(harness.next_v2(), Reply::Stopped(46)); + let (outcome, system) = harness.thread.join().unwrap(); + assert!(matches!(outcome, Ok(Outcome::Stopped))); + assert_eq!(system.tty_drained, 1); + assert!(system.diagnostics.is_empty()); + until_gone(process); + Ok(()) +} + +#[test] +fn a_version_2_execution_forwards_output_past_the_old_cap() -> TestResult { + if super::super::super::tests::isolate_process_test( + "NVX_V2_OUTPUT_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::a_version_2_execution_forwards_output_past_the_old_cap", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (_temporary, mut harness) = direct_harness(); + let total = protocol::MAX_OUTPUT_BYTES * 3 / 2; + harness.send_v2( + 42, + Frame::Exec(&shell(&format!("head -c {total} /dev/zero"))), + ); + let mut forwarded = 0; + loop { + match harness.next_v2() { + Reply::Stdout(42, bytes) => { + assert!(bytes.iter().all(|&byte| byte == 0)); + forwarded += bytes.len(); + } + Reply::Exit(42, 0, v2::Outcome::Exit) => break, + reply => panic!("unexpected {reply:?}"), + } + } + assert_eq!(forwarded, total); + assert!(harness.stop_v2(43).diagnostics.is_empty()); + Ok(()) +} + +#[test] +fn the_agent_reads_no_output_while_its_send_queue_is_full() -> TestResult { + if super::super::super::tests::isolate_process_test( + "NVX_V2_BACKPRESSURE_TEST_CHILD", + "guest::managed_agent::dispatch::tests::streams::the_agent_reads_no_output_while_its_send_queue_is_full", + ) { + return Ok(()); + } + let _process = super::super::super::tests::PROCESS_TEST_LOCK.lock()?; + let (temporary, mut harness) = direct_harness(); + let marker = temporary.path().join("finished"); + let lines = 1_000_000; + let script = format!("echo $$; seq 1 {lines}; touch '{}'", marker.display()); + harness.send_v2(42, Frame::Exec(&shell(&script))); + let (process, mut output) = harness.shell_process(42); + // The host reads nothing: the console, then the send queue, then the workload's pipe fill, + // and the workload waits to write the rest of its output. + thread::sleep(Duration::from_secs(1)); + assert!(sys::process_exists(process)); + assert!(!marker.exists()); + // The agent still serves control requests, so the cancellation ends the workload. + harness.send_v2(42, Frame::Cancel); + until_gone(process); + // What it forwarded then arrives, in order, before the outcome. + let exit = loop { + match harness.next_v2() { + Reply::Stdout(42, bytes) => output.extend(bytes), + Reply::Exit(42, status, outcome) => break (status, outcome), + reply => panic!("unexpected {reply:?}"), + } + }; + assert_eq!(exit, (128 + 9, v2::Outcome::Cancelled)); + let expected: String = (1..=lines).map(|line| format!("{line}\n")).collect(); + assert!(output.len() < expected.len()); + assert!(expected.as_bytes().starts_with(&output)); + assert!(!marker.exists()); + assert!(harness.stop_v2(43).diagnostics.is_empty()); + Ok(()) +} diff --git a/nvx-rs/src/guest/managed_agent/exec.rs b/nvx-rs/src/guest/managed_agent/exec.rs index 4decd40e6..f047514de 100644 --- a/nvx-rs/src/guest/managed_agent/exec.rs +++ b/nvx-rs/src/guest/managed_agent/exec.rs @@ -1,18 +1,19 @@ +use std::os::fd::BorrowedFd; use std::os::unix::process::ExitStatusExt; use std::time::{Duration, Instant}; use crate::guest::init::{self, Stream}; -use crate::guest::sys::{self, Errno, FileDescriptor, c_string}; +use crate::guest::sys::{self, Errno, c_string}; use crate::protocol::{ - self, APP_ACTIVATE, APP_CANCEL, APP_CAPTURE, APP_EXIT, APP_STDERR, APP_STDOUT, BUSY, - CWD_FAILED, GuestExecPayload, LAUNCH_FAILED, OUTER_DATA, + self, APP_ACTIVATE, APP_CANCEL, APP_CAPTURE, APP_EXIT, APP_STDERR, APP_STDOUT, APP_STOP, BUSY, + CWD_FAILED, GuestExecPayload, LAUNCH_FAILED, OUTER_DATA, v2::STDIN_WINDOW, }; use super::AgentConfig; use super::diagnostics::with_description; use super::launch; use super::seam::System; -use super::session::{DataFailureKind, Event as SessionEvent, ReceiveFailure, Session}; +use super::session::{DataFailureKind, Event as SessionEvent, ReceiveFailure, Request, Session}; /// Maximum bytes read from one output descriptor at a time. const OUTPUT_CHUNK_BYTES: usize = 32 * 1024; @@ -28,12 +29,21 @@ const ENVIRONMENT_DIAGNOSTIC: &[u8] = b"nvx-managed-agent: cannot configure work /// collects the statuses of the agent's children, and checks the workload's timeout. The agent /// keeps the C agent's policy: the output cap, checked for each read of up to 32 KiB, and the /// precedence of a cancellation, a timeout, and the cap over the workload's own status. +/// +/// An execution of NVXC version 2 has no output cap. Its workload reads a pipe, into which each +/// turn writes what the host sent in `STDIN` frames, under [`STDIN_WINDOW`], and acknowledges +/// what it wrote, and which it closes after `STDIN_END`. The session streams while it runs: its +/// records go through the send queue, which never waits for the console, and the agent reads no +/// more output while the queue is full, so the workload waits on its pipes while the agent serves +/// control requests. `STOP` cancels it, and once its `EXIT` reports that, `stop` holds the +/// `STOP`'s request ID. pub(crate) fn run( system: &mut impl System, session: &mut Session, config: &AgentConfig, request_id: u64, execution: &GuestExecPayload, + stop: &mut Option, ) -> Result<(), Errno> { if let Err(failure) = validate_launch_environment(system, config, execution) { match failure { @@ -58,12 +68,17 @@ pub(crate) fn run( Err(LaunchError::Other) => return session.send_error(request_id, 125, LAUNCH_FAILED), }; let started = Instant::now(); - let mut state = RunState::default(); + let mut state = RunState { + v2: session.v2(), + ..RunState::default() + }; let mut buffer = [0u8; OUTPUT_CHUNK_BYTES]; + session.stream(state.v2); while state.running(&workload) || (workload.has_open_output() && !state.ending()) { let session_ready = match poll( &workload, - (!state.session_lost).then_some(session.descriptor()), + (!state.session_lost).then_some(&*session), + !state.input.pending.is_empty(), ) { Ok(ready) => ready, Err(Errno::EINTR) => false, @@ -73,7 +88,8 @@ pub(crate) fn run( } }; if session_ready { - match handle_control(system, session, request_id) { + let stop = state.v2.then_some(&mut *stop); + match handle_control(system, session, request_id, &mut state.input, stop) { Ok(ExecControl::None) => {} Ok(ExecControl::Cancel) if state.running(&workload) && !state.cancelled => { state.cancelled = true; @@ -104,6 +120,13 @@ pub(crate) fn run( return Err(error); } } + session.answer_in(state.v2); + } + if !state.session_lost { + // The space that the console freed takes the input's acknowledgement before more + // output, so that output cannot starve it. + session.send_queued()?; + state.input.write(&mut workload, session, request_id)?; } let output = if state.session_lost || state.output_limited { &mut [][..] @@ -130,11 +153,16 @@ pub(crate) fn run( } else { state.finish(system, session, request_id, &mut workload, &mut buffer)? }; + session.stream(false); let launch_report = state.launch.take().unwrap_or(Ok(None)); if state.session_lost { return Ok(()); } + // Input that the pipe took while the send queue was full is acknowledged before the end. + if state.input.written != 0 { + session.send_ack(request_id, state.input.written)?; + } match launch_report { Ok(Some(failure)) => match failure.kind() { LaunchKind::WorkingDirectory => { @@ -251,6 +279,27 @@ impl Workload { } } + /// Lends the writer of the workload's standard input while it is open. + fn input(&self) -> Option> { + match self { + Self::Direct(workload) => workload.input(), + } + } + + /// Writes as much of `bytes` as the workload's standard input takes now. + fn write_input(&mut self, bytes: &[u8]) -> std::io::Result { + match self { + Self::Direct(workload) => workload.write_input(bytes), + } + } + + /// Closes the workload's standard input. + fn close_input(&mut self) { + match self { + Self::Direct(workload) => workload.close_input(), + } + } + fn next_event<'a>(&mut self, buffer: &'a mut [u8]) -> std::io::Result>> { match self { Self::Direct(workload) => workload.next_event(buffer).map(|event| { @@ -313,7 +362,7 @@ impl Workload { if let Some(Event::DirectKillFailed(error)) = self.next_event(&mut [])? { return Err(error); } - match poll(self, None) { + match poll(self, None, false) { Ok(_) | Err(Errno::EINTR) => {} Err(error) => return Err(error.into()), } @@ -376,10 +425,81 @@ fn settle_start( Ok(()) } +/// The standard input of an execution of NVXC version 2: what the host sent and the agent has +/// yet to write into the workload's pipe, and what it wrote and has yet to acknowledge. +#[derive(Default)] +struct Input { + /// Bytes that the host sent and the agent has not written. + pending: Vec, + /// Bytes written into the pipe and not acknowledged yet. + written: u32, + /// Bytes that the host sent and the agent has not acknowledged, at most [`STDIN_WINDOW`]: + /// those that a closed pipe never took stay unacknowledged. + outstanding: u32, + /// Whether the host ended the input with `STDIN_END`. + ended: bool, +} + +impl Input { + /// Takes a `STDIN` frame's bytes, or, for `None`, its `STDIN_END`, refusing bytes beyond the + /// window and anything after the end as `EPROTO`. + fn receive(&mut self, bytes: Option<&[u8]>) -> Result<(), Errno> { + if self.ended { + return Err(Errno::EPROTO); + } + let Some(bytes) = bytes else { + self.ended = true; + return Ok(()); + }; + self.outstanding = u32::try_from(bytes.len()) + .ok() + .and_then(|length| self.outstanding.checked_add(length)) + .filter(|&outstanding| outstanding <= STDIN_WINDOW) + .ok_or(Errno::EPROTO)?; + self.pending.extend_from_slice(bytes); + Ok(()) + } + + /// Writes as much as the workload's pipe takes, closes the pipe once the input ended and the + /// agent wrote every byte, and, unless the send queue is full, acknowledges what it wrote. + fn write( + &mut self, + workload: &mut Workload, + session: &mut Session, + request_id: u64, + ) -> Result<(), Errno> { + if !self.pending.is_empty() { + match workload.write_input(&self.pending) { + Ok(written) => { + self.pending.drain(..written); + self.written += written as u32; + } + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {} + // A closed pipe takes nothing more, and the workload reads no more of it. + Err(_) => self.pending.clear(), + } + } + if self.ended && self.pending.is_empty() { + workload.close_input(); + } + if self.written != 0 && !session.queue_full() { + session.send_ack(request_id, self.written)?; + self.outstanding -= self.written; + self.written = 0; + } + Ok(()) + } +} + /// What the agent learned of one workload while it ran: the C agent's flags, which decide the /// outcome that it reports, and the workload's launch report and outcome. #[derive(Default)] struct RunState { + /// Whether the execution is of NVXC version 2, which streams the workload's input, and caps + /// no output. + v2: bool, + /// The workload's standard input, in version 2. + input: Input, /// Bytes of output forwarded so far, which the 1 MiB cap bounds. output_bytes: usize, /// Whether the host cancelled the workload while it ran. @@ -425,11 +545,19 @@ impl RunState { workload: &mut Workload, buffer: &mut [u8], ) -> Result<(), Errno> { + // An execution of version 2, whose output has no cap, reads one chunk of it a turn, so + // that its output cannot keep the agent from serving control requests or its timeout. + let mut read = false; loop { if let Some((id, status)) = session.take_status() { workload.observe(id, status); } - let event = match workload.next_event(buffer) { + let chunk = if session.queue_full() || (self.v2 && read) { + &mut [][..] + } else { + &mut *buffer + }; + let event = match workload.next_event(chunk) { Ok(Some(event)) => event, Ok(None) => return Ok(()), Err(error) => { @@ -439,7 +567,7 @@ impl RunState { }; match event { Event::Output(stream, bytes) => { - if self.output_bytes > protocol::MAX_OUTPUT_BYTES - bytes.len() { + if !self.v2 && self.output_bytes > protocol::MAX_OUTPUT_BYTES - bytes.len() { self.output_limited = true; // The cap ends forwarding: read and drop the rest of both streams, // as a lost session does, so the client receives a prefix of what the @@ -451,6 +579,7 @@ impl RunState { return Ok(()); } self.output_bytes += bytes.len(); + read = true; let kind = match stream { Stream::Stdout => APP_STDOUT, Stream::Stderr => APP_STDERR, @@ -499,7 +628,7 @@ impl RunState { if self.outcome.is_some() || self.status_lost { return Ok(self.outcome); } - match poll(workload, None) { + match poll(workload, None, false) { Ok(_) | Err(Errno::EINTR) => {} Err(error) => return Err(error), } @@ -507,15 +636,23 @@ impl RunState { } } -/// Waits at most one poll interval until one of the workload's descriptors or the session's, -/// if given, polls ready, and returns whether the session's did. -fn poll(workload: &Workload, session: Option<&FileDescriptor>) -> Result { - let mut descriptors = workload.descriptors(); +/// Waits at most one poll interval until one of the workload's descriptors polls ready, but its +/// output while the session's send queue is full, its input, if `input`, writable, or the +/// session's descriptor, if given, readable, or writable while it has queued bytes, and returns +/// whether the session's descriptor polled readable. +fn poll(workload: &Workload, session: Option<&Session>, input: bool) -> Result { + let mut descriptors = Vec::new(); if let Some(session) = session { - descriptors.push(session.borrowed()); + let descriptor = session.descriptor().borrowed(); + descriptors.push((descriptor, false)); + descriptors.extend(session.queued().then_some((descriptor, true))); } - let ready = sys::readable_or_hung_up(&descriptors, POLL_INTERVAL_MS)?; - Ok(session.is_some() && ready.last().copied().unwrap_or(false)) + if !session.is_some_and(Session::queue_full) { + descriptors.extend(workload.descriptors().into_iter().map(|fd| (fd, false))); + } + descriptors.extend(workload.input().filter(|_| input).map(|fd| (fd, true))); + let ready = sys::ready(&descriptors, POLL_INTERVAL_MS)?; + Ok(session.is_some() && ready[0]) } /// Reports the kill failures that the workload's execution has yet to report. @@ -553,9 +690,9 @@ fn launch( config: &AgentConfig, execution: &GuestExecPayload, ) -> Result { - let workload = system - .start_direct(launch::direct_request(config, execution)?) - .map_err(LaunchError::Start)?; + let mut request = launch::direct_request(config, execution)?; + request.input = session.v2(); + let workload = system.start_direct(request).map_err(LaunchError::Start)?; session.watch(workload.id()); Ok(Workload::Direct(workload)) } @@ -640,15 +777,26 @@ enum ExecControl { Capture, } -/// Processes one control-console record while a workload runs. +/// Processes one control-console record while a workload runs: in version 2, the workload's +/// standard input, of which bytes beyond the window or after its end end the session, and, for an +/// execution of version 2, whose `stop` is given, a `STOP`, which cancels the workload and sets +/// `stop`. fn handle_control( system: &mut impl System, session: &mut Session, exec_id: u64, + input: &mut Input, + stop: Option<&mut Option>, ) -> Result { let payload = match session.next_event() { Ok(SessionEvent::Reset) => { + // The new epoch's acknowledgement follows what the earlier one queued, and the + // earlier epoch's `STOP` is abandoned with its work. + session.stream(false); session.acknowledge()?; + if let Some(stop) = stop { + *stop = None; + } return Ok(ExecControl::SessionLost); } Ok(SessionEvent::Data(payload)) => payload, @@ -656,7 +804,14 @@ fn handle_control( Err(failure) => return Err(report_control_failure(system, failure)), }; let request = match session.accept_request(&payload) { - Ok(request) => request, + Ok(Request::App(request)) => request, + // The input of an execution that ended takes no more bytes. + Ok(Request::Stdin(id, bytes)) => { + if id == exec_id { + input.receive(bytes)?; + } + return Ok(ExecControl::None); + } Err(failure) => { if failure.kind != DataFailureKind::Credit { super::diagnostics::error(system, "exec-record", i32::from(OUTER_DATA)); @@ -664,6 +819,13 @@ fn handle_control( return Err(failure.into_error()); } }; + if request.kind == APP_STOP + && session.v2() + && let Some(stop) = stop + { + *stop = Some(request.request_id); + return Ok(ExecControl::Cancel); + } if request.kind == APP_CANCEL { return Ok( if request.payload.is_empty() && request.request_id == exec_id { diff --git a/nvx-rs/src/guest/managed_agent/mod.rs b/nvx-rs/src/guest/managed_agent/mod.rs index f27f53ec8..e6055dffc 100644 --- a/nvx-rs/src/guest/managed_agent/mod.rs +++ b/nvx-rs/src/guest/managed_agent/mod.rs @@ -9,6 +9,14 @@ //! status and error categories, 1 MiB output and receive windows, and timeout and cancellation //! behavior. //! +//! It also serves NVXC version 2, whose frames carry the same requests, and in which each answer +//! of a request takes its version. An execution of version 2 runs as one of version 1 does, alone, +//! and a second `EXEC` gets `BUSY`, but its workload reads a pipe of the standard input that the +//! host streams, which the agent acknowledges as it writes it, up to its explicit end; its output +//! has no cap; and `STOP` ends it. While it runs, the agent's records wait in a send queue rather +//! than for the console, and the agent reads none of the workload's output while the queue is +//! full. +//! //! The agent exits with 0, or with 125 when it fails. //! //! # Structure @@ -29,7 +37,7 @@ //! ends it, and reports its launch and outcome, and the //! agent forwards the output as NVXC frames, serves the control session while the workload runs, //! and reports the outcome. The agent keeps its own policy: the null device as the workload's -//! standard input, the identity, the output cap, +//! standard input and the output cap of version 1, the identity, //! the timeout, the precedence of a cancellation, a timeout, and the cap over the workload's own //! status, and the outcome that it reports for a failed launch stage. It builds the workload's //! environment in memory, as C's environment functions changed it, and passes it to `execve`, so diff --git a/nvx-rs/src/guest/managed_agent/session.rs b/nvx-rs/src/guest/managed_agent/session.rs index 15aa65004..aac1bbf22 100644 --- a/nvx-rs/src/guest/managed_agent/session.rs +++ b/nvx-rs/src/guest/managed_agent/session.rs @@ -9,10 +9,23 @@ use crate::guest::nvxs::{GuestConnection, GuestConnectionError}; use crate::guest::snapshot::{Captured, Disposition}; use crate::guest::sys::{self, Errno, FileDescriptor, Pollable}; use crate::protocol::nvxs::{GuestEvent, RecordError, RecordType, SessionError}; -use crate::protocol::{self, APP_ERROR, OUTER_MAX_PAYLOAD, OUTER_RECEIVE_WINDOW}; +use crate::protocol::{ + self, APP_ERROR, APP_EXIT, APP_READY, APP_STDERR, APP_STDOUT, APP_STOPPED, OUTER_MAX_PAYLOAD, + OUTER_RECEIVE_WINDOW, v2, +}; + +/// Bytes that the send queue holds before the agent stops reading an execution's output, until +/// the control console takes them. +const SEND_QUEUE_BYTES: usize = 64 * 1024; /// The agent's control session: the guest's role of the shared NVXS session on the control /// console, and the NVXC frames that its DATA records carry. +/// +/// It serves requests of NVXC versions 1 and 2, which keep the codes of the kinds that both have, +/// and answers each in the version of the request that it serves. While an execution of version +/// 2 runs, the session's records go through a send queue, which never waits for the console, so +/// that the agent keeps serving control requests while the host does not read; the record that +/// it sends next without streaming waits until the console has taken the queue and itself. pub(crate) struct Session<'r> { /// The guest's NVXS connection on the owned control-console descriptor. connection: GuestConnection>, @@ -21,6 +34,9 @@ pub(crate) struct Session<'r> { /// Error of the connection's latest failure, which a connection that a failure ended reports /// again. last_error: Option, + /// Whether the request that the session serves is of version 2, whose frames its answers + /// take. + v2: bool, } /// Byte transport only: the existing connection still owns all record framing. @@ -32,6 +48,10 @@ struct Control<'r> { outcome: RefCell>, /// Bounds partial I/O until the restored activation's Ready has been written. deadline: Cell>, + /// Records that the console has not taken yet, in order. + queue: RefCell>, + /// Whether writes add to the queue rather than wait for the console. + queueing: Cell, } impl Control<'_> { @@ -74,6 +94,21 @@ impl Control<'_> { .copied() .map_err(|_| Errno::ECHILD) } + + /// Writes as much of the queue as the console takes without waiting. + fn send_queued(&self) -> io::Result<()> { + let mut queue = self.queue.borrow_mut(); + while !queue.is_empty() { + match self.descriptor.write(&queue) { + Ok(0) => return Err(Errno::EIO.into()), + Ok(written) => drop(queue.drain(..written)), + Err(Errno::EAGAIN) => break, + Err(Errno::EINTR) => {} + Err(error) => return Err(error.into()), + } + } + Ok(()) + } } impl Read for Control<'_> { @@ -81,9 +116,6 @@ impl Read for Control<'_> { if bytes.is_empty() { return Read::read(&mut self.descriptor, bytes); } - if self.runtime.is_none() && self.deadline.get().is_none() { - return Read::read(&mut self.descriptor, bytes); - } loop { if self.runtime.is_some() { self.service_leader()?; @@ -114,12 +146,16 @@ impl Read for Control<'_> { impl Write for Control<'_> { fn write(&mut self, bytes: &[u8]) -> io::Result { - if self.runtime.is_none() && self.deadline.get().is_none() { - return Write::write(&mut self.descriptor, bytes); - } if bytes.is_empty() { return Write::write(&mut self.descriptor, bytes); } + if self.queueing.get() { + self.queue.get_mut().extend_from_slice(bytes); + self.send_queued()?; + return Ok(bytes.len()); + } + let queued = std::mem::take(self.queue.get_mut()); + self.write_all(&queued)?; loop { self.remaining()?; if self.runtime.is_some() { @@ -169,6 +205,16 @@ pub(crate) enum Event { Data(Vec), } +/// An NVXC request that the session accepted, whose answers take its version. +pub(crate) enum Request<'a> { + /// A request of version 1, or one of version 2 of a kind that version 1 has too, with the + /// code that both give it. + App(protocol::AppFrame<'a>), + /// Version 2's standard input of the execution of this ID: a `STDIN` frame's bytes, or `None` + /// for its `STDIN_END`. + Stdin(u64, Option<&'a [u8]>), +} + /// A record from the broker that ended the session, classified by the stage at which the C agent /// reported it. pub(crate) enum ReceiveFailure { @@ -242,9 +288,12 @@ impl<'r> Session<'r> { status: Cell::new(None), outcome: RefCell::new(Ok(false)), deadline: Cell::new(None), + queue: RefCell::new(Vec::new()), + queueing: Cell::new(false), }), capture: None, last_error: None, + v2: false, } } @@ -452,7 +501,8 @@ impl<'r> Session<'r> { )); } }; - let mut replies = vec![request.request_id]; + // Each joined request gets its Ready in its own version. + let mut replies = vec![(request.request_id, self.v2)]; while sys::readable_or_hung_up(&[self.descriptor().borrowed()], 0)?[0] { match self .next_event() @@ -464,17 +514,22 @@ impl<'r> Session<'r> { replies.clear(); } Event::Data(bytes) => { - let retry = self + let retry = match self .accept_request(&bytes) - .map_err(DataFailure::into_error)?; - if retry.kind != protocol::APP_ACTIVATE { - return Err(activation_failure( - restored - .activation - .input(Instant::now(), ActivationInput::Request) - .expect_err("an activation takes no request yet"), - )); - } + .map_err(DataFailure::into_error)? + { + Request::App(retry) if retry.kind == protocol::APP_ACTIVATE => { + retry + } + _ => { + return Err(activation_failure( + restored + .activation + .input(Instant::now(), ActivationInput::Request) + .expect_err("an activation takes no request yet"), + )); + } + }; if !matches!( restored .accept(retry.payload, Instant::now()) @@ -484,7 +539,7 @@ impl<'r> Session<'r> { ) { return Err(Errno::EPROTO); } - replies.push(retry.request_id); + replies.push((retry.request_id, self.v2)); } } } @@ -494,10 +549,13 @@ impl<'r> Session<'r> { .map_err(activation_failure)?; (Some(retained), ready, replies) } - ActivationAction::Ready(ready) => (None, ready, vec![request.request_id]), + ActivationAction::Ready(ready) => { + (None, ready, vec![(request.request_id, self.v2)]) + } _ => return Err(Errno::EPROTO), }; - for id in replies { + for (id, v2) in replies { + self.v2 = v2; self.send_app(protocol::APP_READY, id, 0, &ready.encode())?; } self.connection.get_ref().deadline.set(None); @@ -616,29 +674,63 @@ impl<'r> Session<'r> { acknowledged.map_err(|error| self.error(error)) } - /// Decodes the NVXC request that a DATA payload carries, then returns the payload's bytes to - /// the receive window before the agent serves the request, as the C agent did. + /// Decodes the NVXC request that a DATA payload carries, of version 1 or 2, then returns the + /// payload's bytes to the receive window before the agent serves the request, as the C agent + /// did. The session answers in the request's version until it accepts another. pub(crate) fn accept_request<'a>( &mut self, payload: &'a [u8], - ) -> Result, DataFailure> { + ) -> Result, DataFailure> { // A DATA payload has at most `OUTER_MAX_PAYLOAD` bytes, so its length fits either type. let status = payload.len() as i32; - let frame = protocol::decode_app(payload) - .map_err(|_| data_failure(DataFailureKind::Application, status))?; - if frame.request_id == 0 || frame.status != 0 { - return Err(data_failure(DataFailureKind::Application, status)); + let streams = payload.get(4) == Some(&v2::VERSION); + let request = if streams { + v2::Frame::decode(payload) + .ok() + .map(|(request_id, frame)| match frame { + v2::Frame::Stdin(bytes) => Request::Stdin(request_id, Some(bytes)), + v2::Frame::StdinEnd => Request::Stdin(request_id, None), + // Version 2 keeps version 1's header, and the codes of the kinds that both + // have, which its decoder checked. + _ => Request::App(protocol::AppFrame { + kind: payload[5], + request_id, + status: 0, + payload: &payload[protocol::APP_HEADER_LEN..], + }), + }) + } else { + protocol::decode_app(payload) + .ok() + .filter(|frame| frame.request_id != 0 && frame.status == 0) + .map(Request::App) } + .ok_or_else(|| data_failure(DataFailureKind::Application, status))?; let credited = self.connection.credit(payload.len() as u32); credited.map_err(|error| DataFailure { kind: DataFailureKind::Credit, status, error: self.error(error), })?; - Ok(frame) + self.v2 = streams; + Ok(request) } - /// Encodes and sends one application frame, in one DATA record. + /// Whether the session answers in version 2. + pub(crate) fn v2(&self) -> bool { + self.v2 + } + + /// Answers in version 2, or in version 1, such as an execution's frames after a control + /// request of its own version. + pub(crate) fn answer_in(&mut self, v2: bool) { + self.v2 = v2; + } + + /// Encodes and sends one application frame, in one DATA record, in the session's version. + /// + /// In version 2, `READY` starts with the protocol version, an `ERROR` of the `busy` category + /// is `BUSY`, and an `EXIT` names its outcome as version 1 does. pub(crate) fn send_app( &mut self, kind: u8, @@ -646,17 +738,74 @@ impl<'r> Session<'r> { status: i32, payload: &[u8], ) -> Result<(), Errno> { - if protocol::APP_HEADER_LEN - .checked_add(payload.len()) - .is_none_or(|length| length > OUTER_MAX_PAYLOAD) - { - return Err(Errno::EMSGSIZE); - } - let frame = protocol::encode_app(kind, request_id, status, payload); + let frame = if self.v2 { + let frame = match kind { + APP_READY => v2::Frame::Ready { + version: v2::VERSION, + activation: payload, + }, + APP_STDOUT => v2::Frame::Stdout(payload), + APP_STDERR => v2::Frame::Stderr(payload), + APP_EXIT => v2::Frame::Exit { + status, + outcome: outcome(payload), + }, + APP_STOPPED => v2::Frame::Stopped, + APP_ERROR if payload == protocol::BUSY.as_bytes() => v2::Frame::Busy, + APP_ERROR => v2::Frame::Error { + status, + category: payload, + }, + _ => return Err(Errno::EINVAL), + }; + frame.encode(request_id).map_err(|_| Errno::EMSGSIZE)? + } else { + if protocol::APP_HEADER_LEN + .checked_add(payload.len()) + .is_none_or(|length| length > OUTER_MAX_PAYLOAD) + { + return Err(Errno::EMSGSIZE); + } + protocol::encode_app(kind, request_id, status, payload) + }; + let sent = self.connection.send(&frame); + sent.map_err(|error| self.error(error)) + } + + /// Acknowledges `count` bytes written into the standard input of the execution of + /// `request_id`, in version 2. + pub(crate) fn send_ack(&mut self, request_id: u64, count: u32) -> Result<(), Errno> { + let frame = v2::Frame::StdinAck(count) + .encode(request_id) + .map_err(|_| Errno::EINVAL)?; let sent = self.connection.send(&frame); sent.map_err(|error| self.error(error)) } + /// Streams, while an execution of version 2 runs: every record goes through the send queue, + /// which never waits for the console, and which the next record sent without streaming + /// follows. + pub(crate) fn stream(&self, on: bool) { + self.connection.get_ref().queueing.set(on); + } + + /// Whether the send queue holds [`SEND_QUEUE_BYTES`], so that the agent reads no more of an + /// execution's output. + pub(crate) fn queue_full(&self) -> bool { + self.connection.get_ref().queue.borrow().len() >= SEND_QUEUE_BYTES + } + + /// Whether the send queue holds bytes, so that the agent polls the console writable. + pub(crate) fn queued(&self) -> bool { + !self.connection.get_ref().queue.borrow().is_empty() + } + + /// Writes as much of the send queue as the console takes without waiting. + pub(crate) fn send_queued(&mut self) -> Result<(), Errno> { + let sent = self.connection.get_ref().send_queued(); + sent.map_err(|error| self.error(GuestConnectionError::Io(error))) + } + /// Sends one application ERROR frame with an ASCII category. pub(crate) fn send_error( &mut self, @@ -705,6 +854,17 @@ fn activation_failure(error: impl std::fmt::Display) -> Errno { Errno::EPROTO } +/// Returns version 2's outcome of the execution whose version-1 `EXIT` category is `category`. +fn outcome(category: &[u8]) -> v2::Outcome { + match category { + b"exit" => v2::Outcome::Exit, + b"signal" => v2::Outcome::Signal, + b"timeout" => v2::Outcome::Timeout, + b"cancelled" => v2::Outcome::Cancelled, + _ => v2::Outcome::Failed, + } +} + /// Returns the stage at which the C agent reported a record that the session refused, or `None` /// for one that it reported as a failure to read. /// diff --git a/nvx-rs/src/guest/managed_agent/session/tests.rs b/nvx-rs/src/guest/managed_agent/session/tests.rs index 62232fd52..482e5340b 100644 --- a/nvx-rs/src/guest/managed_agent/session/tests.rs +++ b/nvx-rs/src/guest/managed_agent/session/tests.rs @@ -78,6 +78,19 @@ impl Peer { self.host_sequence += 1; } + fn send_v2(&mut self, id: u64, frame: v2::Frame<'_>) { + let bytes = frame.encode(id).unwrap(); + let record = encode_outer( + OUTER_DATA, + &INSTANCE, + self.epoch, + self.host_sequence, + &bytes, + ); + self.stream.write_all(&record.unwrap()).unwrap(); + self.host_sequence += 1; + } + fn credit(&mut self) { let (header, _) = self.read(); assert_eq!(header.record_type, OUTER_CREDIT); @@ -142,9 +155,9 @@ fn a_replacement_session_joins_one_activation_and_caches_ready_without_reapplyin let (record, _) = protocol::decode_activate_payload(&payload).unwrap(); let ready = crate::activation::ReadyActivation::of(&record).encode(); let bytes = activate_bytes(&mut session, &mut peer, 61, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); let retained = session .activate_with(request, |restored, _, accepted, workload| { @@ -167,9 +180,9 @@ fn a_replacement_session_joins_one_activation_and_caches_ready_without_reapplyin assert!(!session.rootless()); assert_eq!(session.connection.get_ref().deadline.get(), None); let bytes = activate_bytes(&mut session, &mut peer, 63, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid retry")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid retry"); + }; peer.credit(); assert!( session @@ -188,6 +201,37 @@ fn a_replacement_session_joins_one_activation_and_caches_ready_without_reapplyin ); } +#[test] +fn joined_activations_get_ready_in_their_own_versions() { + let (mut session, mut peer, payload) = restored(); + let (record, _) = protocol::decode_activate_payload(&payload).unwrap(); + let ready = crate::activation::ReadyActivation::of(&record).encode(); + let bytes = activate_bytes(&mut session, &mut peer, 61, &payload); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; + peer.credit(); + let retained = session.activate_with(request, |_, _, _, _| { + peer.send_v2(62, v2::Frame::Activate(&payload)); + Ok("one applied space") + }); + assert_eq!(retained.unwrap(), Some("one applied space")); + peer.credit(); + peer.app(APP_READY, 61, 0, &ready); + let (_, frame) = peer.read(); + let activation = ready.as_slice(); + assert_eq!( + v2::Frame::decode(&frame).unwrap(), + ( + 62, + v2::Frame::Ready { + version: v2::VERSION, + activation + } + ) + ); +} + #[test] fn postactivation_capture_stays_refused_without_pending_cleanup() { let (mut session, mut peer, payload) = restored(); @@ -229,9 +273,9 @@ fn changed_record_or_workload_bytes_during_application_fail_without_ready() { changed.extend(record.encode()); } let bytes = activate_bytes(&mut session, &mut peer, 64, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); let result = session.activate_with(request, |_, _, _, _| { peer.send(protocol::APP_ACTIVATE, 65, &changed); @@ -252,9 +296,9 @@ fn changed_record_or_workload_bytes_during_application_fail_without_ready() { fn failed_application_never_reports_ready_or_admits_root_dependent_work() { let (mut session, mut peer, payload) = restored(); let bytes = activate_bytes(&mut session, &mut peer, 66, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); assert_eq!( session @@ -283,9 +327,9 @@ fn a_failed_restored_activation_gives_init_the_status_of_its_class() { let (mut session, mut peer, payload) = restored(); assert_eq!(session.activation_status(), None); let bytes = activate_bytes(&mut session, &mut peer, 68, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); let failure = crate::Error::io("the network settings cannot be applied").with_source( crate::guest::activation::ApplyError::Network { @@ -306,9 +350,9 @@ fn a_failed_restored_activation_gives_init_the_status_of_its_class() { // A request while the record applies fails the activation as the protocol's failure. let (mut session, mut peer, payload) = restored(); let bytes = activate_bytes(&mut session, &mut peer, 69, &payload); - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid activate")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; peer.credit(); let result = session.activate_with(request, |_, _, _, _| { peer.send(protocol::APP_PING, 70, &[]); @@ -325,6 +369,25 @@ fn a_failed_restored_activation_gives_init_the_status_of_its_class() { ); } +#[test] +fn standard_input_while_the_record_applies_fails_the_activation_as_the_protocols_failure() { + let (mut session, mut peer, payload) = restored(); + let bytes = activate_bytes(&mut session, &mut peer, 71, &payload); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid activate"); + }; + peer.credit(); + let result = session.activate_with(request, |_, _, _, _| { + peer.send_v2(71, v2::Frame::Stdin(b"x")); + Ok(()) + }); + assert_eq!(result.err(), Some(Errno::EPROTO)); + assert_eq!( + session.activation_status(), + Some(protocol::status::ACTIVATION_PROTOCOL) + ); +} + fn short_deadline(session: &mut Session<'_>) -> Instant { let now = Instant::now(); let repaired = crate::guest::snapshot::repair::tests::repaired_fixture(); @@ -390,9 +453,9 @@ fn queue( else { panic!("the capture request is DATA"); }; - let request = session - .accept_request(&bytes) - .unwrap_or_else(|_| panic!("valid frame")); + let Ok(Request::App(request)) = session.accept_request(&bytes) else { + panic!("valid frame"); + }; let accepted = session.queue_capture(request, active).unwrap(); peer.credit(); accepted diff --git a/nvx-rs/src/guest/sys/mod.rs b/nvx-rs/src/guest/sys/mod.rs index 196b2fe1e..669326e24 100644 --- a/nvx-rs/src/guest/sys/mod.rs +++ b/nvx-rs/src/guest/sys/mod.rs @@ -111,7 +111,7 @@ pub(crate) use ioctl::{ pub(crate) use mapping::MemoryMapping; #[cfg(target_arch = "x86_64")] pub(crate) use msr::{MsrAccess, MsrFile}; -pub(crate) use poll::{Pollable, hung_up, poll, readable_or_hung_up}; +pub(crate) use poll::{Pollable, hung_up, poll, readable_or_hung_up, ready}; pub(crate) use port::{PORT_DEVICE, Port}; #[cfg(target_arch = "x86_64")] pub(crate) use port::{PortAccess, PortError, check_released, power_off, probe_port}; diff --git a/nvx-rs/src/guest/sys/poll.rs b/nvx-rs/src/guest/sys/poll.rs index a6f9cfc0e..a30346532 100644 --- a/nvx-rs/src/guest/sys/poll.rs +++ b/nvx-rs/src/guest/sys/poll.rs @@ -15,10 +15,30 @@ use super::{ChildSignalWait, Errno, FileDescriptor, SignalDescriptor, Timer}; pub(crate) fn readable_or_hung_up( descriptors: &[BorrowedFd<'_>], timeout_milliseconds: u16, +) -> Result, Errno> { + let descriptors: Vec<_> = descriptors + .iter() + .map(|&descriptor| (descriptor, false)) + .collect(); + ready(&descriptors, timeout_milliseconds) +} + +/// Polls each descriptor for readable data or hangup, as [`readable_or_hung_up`] does, or, where +/// its flag is set, for space to write, and returns whether each one woke, in the caller's order. +pub(crate) fn ready( + descriptors: &[(BorrowedFd<'_>, bool)], + timeout_milliseconds: u16, ) -> Result, Errno> { let mut polls = descriptors .iter() - .map(|descriptor| PollFd::new(*descriptor, PollFlags::POLLIN | PollFlags::POLLHUP)) + .map(|&(descriptor, write)| { + let interest = if write { + PollFlags::POLLOUT + } else { + PollFlags::POLLIN | PollFlags::POLLHUP + }; + PollFd::new(descriptor, interest) + }) .collect::>(); nix::poll::poll(&mut polls, timeout_milliseconds)?; Ok(polls diff --git a/nvx-rs/src/protocol/app.rs b/nvx-rs/src/protocol/app.rs index 5cf505fd6..6ca904452 100644 --- a/nvx-rs/src/protocol/app.rs +++ b/nvx-rs/src/protocol/app.rs @@ -44,17 +44,9 @@ pub(crate) mod workload { } #[cfg(any( - feature = "cli", feature = "testing", all(feature = "guest", target_os = "linux", target_arch = "x86_64") ))] -#[cfg_attr( - not(test), - expect( - dead_code, - reason = "SL4.4b's guest loop and fake OpenVMM serve NVXC v2, and SL4.4c's host speaks it" - ) -)] pub(crate) mod v2; /// Magic of every application frame. diff --git a/nvx-rs/src/protocol/app/v2.rs b/nvx-rs/src/protocol/app/v2.rs index d6ba1757a..a509a1e2f 100644 --- a/nvx-rs/src/protocol/app/v2.rs +++ b/nvx-rs/src/protocol/app/v2.rs @@ -46,6 +46,12 @@ //! has written them into the execution's standard input, and the host never has more than //! [`STDIN_WINDOW`] bytes that the guest agent has not acknowledged. `CANCEL` ends the execution, //! whose `EXIT` reports it. Version 1's `FEATURES` request, `0x05`, has no place in version 2. +//! +//! NVX's guest agents and the fake OpenVMM run one execution at a time. A `STOP` while it runs +//! ends it, whose `EXIT` comes before `STOPPED`. Bytes that its standard input never took, as +//! once the workload closed it, stay unacknowledged; `STDIN`, `STDIN_END` and `CANCEL` of an +//! execution that ended get no answer; and standard input beyond the window, or after its end, +//! ends the session. use crate::protocol::{OUTER_MAX_PAYLOAD, ProtocolError, array, violation}; diff --git a/nvx-rs/src/protocol/mod.rs b/nvx-rs/src/protocol/mod.rs index de401d1f1..e9ae7e0f7 100644 --- a/nvx-rs/src/protocol/mod.rs +++ b/nvx-rs/src/protocol/mod.rs @@ -74,6 +74,7 @@ pub(crate) use app::APP_ACTIVATE; pub(crate) use app::APP_CAPTURE; #[cfg(any( feature = "cli", + feature = "testing", all(feature = "guest", target_os = "linux", target_arch = "x86_64") ))] pub(crate) use app::APP_HEADER_LEN; @@ -85,6 +86,11 @@ pub(crate) use app::ExitCategory; pub(crate) use app::encode_activate_payload; #[cfg(feature = "cli")] pub(crate) use app::encode_capture_payload; +#[cfg(any( + feature = "testing", + all(feature = "guest", target_os = "linux", target_arch = "x86_64") +))] +pub(crate) use app::v2; #[cfg(any( feature = "cli", all(feature = "guest", target_os = "linux", target_arch = "x86_64") diff --git a/nvx-rs/src/testing/fake_openvmm.rs b/nvx-rs/src/testing/fake_openvmm.rs index a6da98ec8..b488b66c8 100644 --- a/nvx-rs/src/testing/fake_openvmm.rs +++ b/nvx-rs/src/testing/fake_openvmm.rs @@ -17,10 +17,12 @@ //! or `printenv` in any directory, which prints its environment. A script is a `;`-separated //! list of commands: `echo TEXT`, `echoerr TEXT`, `printerr TEXT`, which writes `TEXT` to //! standard error without a line end, `sleep MS`, `exit CODE`, `signal NUMBER`, `flood BYTES`, -//! `write KEY VALUE`, `read KEY`, `mkdir DIRECTORY`, `touch PATH`, which creates the host file -//! `PATH`, as a workload that writes a read-write share does, `env`, `printenv NAME`, `pwd`, -//! `timeout`, `fail`, and `launchfail`. Like `sh`, a script exits with the status of its -//! last command; `printenv NAME` prints nothing and fails with status 1 when `NAME` is unset. +//! `cat`, which copies the workload's standard input to its standard output until the input ends +//! or the workload times out, `write KEY VALUE`, `read KEY`, `mkdir DIRECTORY`, `touch PATH`, which +//! creates the host file `PATH`, as a workload that writes a read-write share does, `env`, +//! `printenv NAME`, `pwd`, `timeout`, `fail`, and `launchfail`. Like `sh`, a script exits with the +//! status of its last command; `printenv NAME` prints nothing and fails with status 1 when `NAME` +//! is unset. //! Values written with `write` and directories made with `mkdir` live in memory until the VM //! stops, like files in the guest's RAM root file system. A `CANCEL` request ends a sleeping //! workload with the cancelled outcome, and a client that disconnects during an exec abandons it, @@ -30,6 +32,18 @@ //! made with `mkdir`), or that is a mapped regular file, with a diagnostic and the `cwd-failed` //! category. //! +//! Like the guest agent, the fake serves NVXC versions 1 and 2, and answers each request in its +//! version. A workload of version 1 reads no standard input, so its `cat` copies nothing, and +//! ends with `output-limit` once its output would pass 1 MiB. A workload of version 2 runs alone, +//! and another `EXEC` gets `BUSY`, as other requests do while it runs; its output has no cap; its +//! standard input is what the host's `STDIN` frames sent, which the fake acknowledges as the +//! workload reads it, until `STDIN_END`, and of which more than the window unacknowledged, or +//! more after its end, ends the session; and a `STOP` while it sleeps or reads its input ends it +//! with the cancelled outcome before the fake acknowledges the `STOP`, unless `fake_ignore_stop=1` +//! ignores it, as between workloads; with `fake_drop_stop=1`, the fake closes the connection after +//! that `EXIT`, without acknowledging the `STOP`. Each refusal while a workload runs takes the +//! version of the request that it refuses. +//! //! Host paths are mapped as OpenVMM and the guest agent map them: the command line exports //! numbered `--mount-child` directories, and files with the `file` flag, through //! `--mount-aggregate`, the kernel command line announces the number of mappings with @@ -192,12 +206,13 @@ use std::time::{Duration, Instant}; use crate::protocol::nvxs::ErrorCode; use crate::protocol::{ - APP_CANCEL, APP_ERROR, APP_EXEC, APP_EXIT, APP_FEATURES, APP_MAPS, APP_PING, APP_READY, - APP_STDERR, APP_STDOUT, APP_STOP, APP_STOPPED, BUSY, CAPABILITY_LEN, CWD_FAILED, ExitCategory, - GuestFeatures, INVALID_REQUEST, LAUNCH_FAILED, MAPPING_FAILED, MAPPINGS_INCOMPLETE, MAPS_TOKEN, - MAX_HOST_MAPPINGS, MAX_OUTPUT_BYTES, MapsPayload, OUTER_DATA, OUTER_ERROR, OUTER_HEADER_LEN, - OUTER_HOST_ATTACH, OUTER_READY, OUTER_WAIT, OuterHeader, UNSUPPORTED_OPERATION, decode_app, - decode_exec_payload, decode_maps_payload, decode_outer_header, encode_app, encode_outer, + APP_CANCEL, APP_ERROR, APP_EXEC, APP_EXIT, APP_FEATURES, APP_HEADER_LEN, APP_MAPS, APP_PING, + APP_READY, APP_STDERR, APP_STDOUT, APP_STOP, APP_STOPPED, BUSY, CAPABILITY_LEN, CWD_FAILED, + ExitCategory, GuestFeatures, INVALID_REQUEST, LAUNCH_FAILED, MAPPING_FAILED, + MAPPINGS_INCOMPLETE, MAPS_TOKEN, MAX_HOST_MAPPINGS, MAX_OUTPUT_BYTES, MapsPayload, OUTER_DATA, + OUTER_ERROR, OUTER_HEADER_LEN, OUTER_HOST_ATTACH, OUTER_READY, OUTER_WAIT, OuterHeader, + UNSUPPORTED_OPERATION, decode_app, decode_exec_payload, decode_maps_payload, + decode_outer_header, encode_app, encode_outer, v2, }; /// Largest workload-output frame emitted by the fake guest. @@ -2555,12 +2570,22 @@ enum Flow { enum Interrupt { /// No control request is waiting. None, - /// The workload's matching cancellation request arrived. + /// The workload's matching cancellation request, or a `STOP` of version 2, arrived. Cancel, /// The host disconnected and abandoned the workload. Disconnected, } +/// A host request that a DATA record carries. +enum Request { + /// A request: whether it is of NVXC version 2, its kind, whose code version 2 shares with + /// version 1, its ID, and its payload. + App(bool, u8, u64, Vec), + /// Version 2's standard input of the workload with this ID: a `STDIN` frame's bytes, or + /// `None` for its `STDIN_END`. + Stdin(u64, Option>), +} + /// A control stream that can report whether input is waiting without consuming it. trait Pending { /// Returns whether a read would make progress, including at end of stream. @@ -2643,26 +2668,67 @@ struct Session<'a, S> { host_sequence: u64, /// Whether matching cancellation requests are ignored for fault injection. ignore_cancel: bool, + /// Whether stop requests are ignored for fault injection, also while a workload runs. + ignore_stop: bool, + /// Whether the request that the fake serves, or the workload that runs, is of NVXC version + /// 2, whose frames its answers take. + v2: bool, + /// What the host sent of the running workload's standard input in version 2 and the workload + /// has not read, which the fake acknowledges as the workload reads it. + stdin: Vec, + /// Whether the running workload's standard input ended. + stdin_ended: bool, + /// The request ID of a `STOP` of version 2 that ended the running workload. + stop: Option, } impl Session<'_, S> { - /// Encodes and sends one sequenced guest application frame. + /// Encodes and sends one sequenced guest application frame, in the session's version. fn send(&mut self, kind: u8, request_id: u64, status: i32, payload: &[u8]) -> io::Result<()> { - let frame = encode_app(kind, request_id, status, payload); + if !self.v2 { + return self.send_frame(&encode_app(kind, request_id, status, payload)); + } + let frame = match kind { + APP_READY => v2::Frame::Ready { + version: v2::VERSION, + activation: payload, + }, + // Version 2 carries no empty chunk of output. + APP_STDOUT | APP_STDERR if payload.is_empty() => return Ok(()), + APP_STDOUT => v2::Frame::Stdout(payload), + APP_STDERR => v2::Frame::Stderr(payload), + APP_STOPPED => v2::Frame::Stopped, + APP_ERROR if payload == BUSY.as_bytes() => v2::Frame::Busy, + _ => v2::Frame::Error { + status, + category: payload, + }, + }; + self.send_v2(request_id, frame) + } + + /// Encodes and sends one frame of version 2. + fn send_v2(&mut self, request_id: u64, frame: v2::Frame<'_>) -> io::Result<()> { + let frame = frame.encode(request_id).map_err(io::Error::other)?; + self.send_frame(&frame) + } + + /// Sends one sequenced DATA record that carries `frame`. + fn send_frame(&mut self, frame: &[u8]) -> io::Result<()> { let record = encode_outer( OUTER_DATA, &self.instance, self.epoch, self.guest_sequence, - &frame, + frame, ) .map_err(io::Error::other)?; self.guest_sequence += 1; self.stream.write_all(&record) } - /// Reads the next host request, returning `None` at end of stream. - fn request(&mut self) -> io::Result)>> { + /// Reads the next host request, of either version, returning `None` at end of stream. + fn request(&mut self) -> io::Result> { let Some((header, frame)) = read_record(self.stream)? else { return Ok(None); }; @@ -2670,20 +2736,39 @@ impl Session<'_, S> { && header.instance_id == self.instance && header.epoch == self.epoch && header.sequence == self.host_sequence; - let frame = decode_app(&frame) - .ok() + let request = if frame.get(4) == Some(&v2::VERSION) { + v2::Frame::decode(&frame) + .ok() + .map(|(request_id, decoded)| match decoded { + v2::Frame::Stdin(bytes) => Request::Stdin(request_id, Some(bytes.to_vec())), + v2::Frame::StdinEnd => Request::Stdin(request_id, None), + // Version 2 keeps version 1's header and the codes of the kinds that both + // have. + _ => Request::App(true, frame[5], request_id, frame[APP_HEADER_LEN..].to_vec()), + }) + } else { + decode_app(&frame).ok().map(|frame| { + Request::App(false, frame.kind, frame.request_id, frame.payload.to_vec()) + }) + }; + let request = request .filter(|_| valid) .ok_or_else(|| io::Error::other("invalid data record"))?; self.host_sequence += 1; - Ok(Some((frame.kind, frame.request_id, frame.payload.to_vec()))) + Ok(Some(request)) } /// Serves requests until the host disconnects or asks the VM to stop. fn run(&mut self, options: &Options, guest: &mut Guest) -> io::Result { loop { - let Some((kind, request_id, payload)) = self.request()? else { + let Some(request) = self.request()? else { return Ok(Flow::Continue); }; + // The standard input of a workload that ended takes no more bytes. + let Request::App(v2, kind, request_id, payload) = request else { + continue; + }; + self.v2 = v2; match kind { APP_PING => self.send(APP_READY, request_id, 0, &[])?, APP_FEATURES if !options.legacy_guest => { @@ -2696,6 +2781,13 @@ impl Session<'_, S> { if !self.exec(request_id, &payload, guest)? { return Ok(Flow::Continue); } + if let Some(stop) = self.stop.take() { + if options.drop_stop { + return Ok(Flow::Drop); + } + let _ = self.send(APP_STOPPED, stop, 0, &[]); + return Ok(Flow::Exit); + } } APP_MAPS => match guest.map(options, &payload) { Ok(()) => self.send(APP_READY, request_id, 0, &[])?, @@ -2716,16 +2808,41 @@ impl Session<'_, S> { } } - /// Handles control traffic that arrives while workload `request_id` runs. + /// Handles control traffic that arrives while workload `request_id` runs. The workload's + /// standard input of version 2 queues for it, under the window; bytes beyond it, or after its + /// end, fail the session. fn interrupt(&mut self, request_id: u64) -> io::Result { while self.stream.pending()? { match self.request()? { None => return Ok(Interrupt::Disconnected), - Some((APP_CANCEL, id, _)) if id == request_id && !self.ignore_cancel => { + Some(Request::Stdin(id, _)) if id != request_id => {} + Some(Request::Stdin(_, _)) if self.stdin_ended => { + return Err(io::Error::other("standard input after its end")); + } + Some(Request::Stdin(_, None)) => self.stdin_ended = true, + Some(Request::Stdin(_, Some(bytes))) => { + if self.stdin.len() + bytes.len() > v2::STDIN_WINDOW as usize { + return Err(io::Error::other("standard input beyond the window")); + } + self.stdin.extend(bytes); + } + Some(Request::App(true, APP_STOP, _, _)) if self.v2 && self.ignore_stop => {} + Some(Request::App(true, APP_STOP, id, _)) if self.v2 => { + self.stop = Some(id); + return Ok(Interrupt::Cancel); + } + Some(Request::App(_, APP_CANCEL, id, _)) + if id == request_id && !self.ignore_cancel => + { return Ok(Interrupt::Cancel); } - Some((APP_CANCEL, _, _)) => {} - Some((_, id, _)) => self.send(APP_ERROR, id, 16, BUSY.as_bytes())?, + Some(Request::App(_, APP_CANCEL, _, _)) => {} + // The refusal takes the request's version, and the workload's frames their own. + Some(Request::App(v2, _, id, _)) => { + let running = std::mem::replace(&mut self.v2, v2); + self.send(APP_ERROR, id, 16, BUSY.as_bytes())?; + self.v2 = running; + } } } Ok(Interrupt::None) @@ -2747,6 +2864,9 @@ impl Session<'_, S> { let Ok(exec) = decode_exec_payload(payload) else { return self.send_some(APP_ERROR, request_id, 22, INVALID_REQUEST.as_bytes()); }; + // A workload of version 1 has no standard input, as one that reads the null device. + self.stdin.clear(); + self.stdin_ended = !self.v2; let mut environment = if exec.inherit_default_env { guest.default_environment.clone() } else { @@ -2868,6 +2988,27 @@ impl Session<'_, S> { remaining -= chunk; } } + "cat" => loop { + let chunk = self.stdin.len().min(OUTPUT_CHUNK_BYTES); + if limit.is_some_and(|limit| Instant::now() >= limit) { + return self.exit(request_id, 124, ExitCategory::Timeout); + } else if chunk > 0 { + // The workload reads these bytes now, so the fake acknowledges them. + let bytes: Vec = self.stdin.drain(..chunk).collect(); + self.send_v2(request_id, v2::Frame::StdinAck(chunk as u32))?; + self.send(APP_STDOUT, request_id, 0, &bytes)?; + } else if self.stdin_ended { + break; + } else { + match self.interrupt(request_id)? { + Interrupt::None => thread::sleep(Duration::from_millis(5)), + Interrupt::Cancel => { + return self.exit(request_id, 137, ExitCategory::Cancelled); + } + Interrupt::Disconnected => return Ok(None), + } + } + }, "write" => { let (key, value) = argument.split_once(' ').unwrap_or((argument, "")); guest.values.insert(key.to_owned(), value.to_owned()); @@ -2947,11 +3088,24 @@ impl Session<'_, S> { status: i32, category: ExitCategory, ) -> io::Result> { + if self.v2 { + let outcome = match category { + ExitCategory::Exit => v2::Outcome::Exit, + ExitCategory::Signal => v2::Outcome::Signal, + ExitCategory::Timeout => v2::Outcome::Timeout, + ExitCategory::Cancelled => v2::Outcome::Cancelled, + ExitCategory::Failed | ExitCategory::OutputLimit => v2::Outcome::Failed, + }; + return self + .send_v2(request_id, v2::Frame::Exit { status, outcome }) + .map(Some); + } self.send_some(APP_EXIT, request_id, status, category.as_str().as_bytes()) } - /// Forwards workload output unless it would exceed the guest agent's output limit, and - /// returns whether it did. A workload whose output was refused ends with `output-limit`. + /// Forwards workload output unless it would exceed the guest agent's output limit of version + /// 1, and returns whether it did. A workload whose output was refused ends with + /// `output-limit`. fn forward( &mut self, kind: u8, @@ -2959,7 +3113,7 @@ impl Session<'_, S> { forwarded: &mut usize, output: &[u8], ) -> io::Result { - if *forwarded + output.len() > MAX_OUTPUT_BYTES { + if !self.v2 && *forwarded + output.len() > MAX_OUTPUT_BYTES { return Ok(false); } *forwarded += output.len(); @@ -3020,6 +3174,11 @@ fn serve_client( guest_sequence: 1, host_sequence: 0, ignore_cancel: options.ignore_cancel, + ignore_stop: options.ignore_stop, + v2: false, + stdin: Vec::new(), + stdin_ended: true, + stop: None, }; match session.run(options, guest) { Ok(Flow::Exit) => { diff --git a/nvx-rs/src/testing/fake_openvmm/tests.rs b/nvx-rs/src/testing/fake_openvmm/tests.rs index ea78459ad..d03852068 100644 --- a/nvx-rs/src/testing/fake_openvmm/tests.rs +++ b/nvx-rs/src/testing/fake_openvmm/tests.rs @@ -1108,3 +1108,221 @@ fn capture_scenarios_take_defaults_and_refuse_unknown_fields() { ); } } + +/// A frame of version 2 that the fake guest sent, with its request ID. +#[cfg(all(unix, feature = "cli"))] +#[derive(Debug, PartialEq, Eq)] +enum Reply { + Ready(u64, Vec), + Stdout(u64, Vec), + Exit(u64, i32, v2::Outcome), + Stopped(u64), + Ack(u64, u32), + Busy(u64), + Error(u64, i32, Vec), +} + +/// A host attached to the control session of a fake guest of a managed run, which a thread +/// serves over a socket pair, as the fake serves a client of its control console, and which +/// returns whether the session stopped the VM. +#[cfg(all(unix, feature = "cli"))] +struct Host { + stream: std::os::unix::net::UnixStream, + instance: [u8; 16], + epoch: u64, + sequence: u64, + fake: thread::JoinHandle, + _directory: tempfile::TempDir, +} + +#[cfg(all(unix, feature = "cli"))] +impl Host { + /// Attaches to a new fake guest. + fn attach() -> Self { + let managed = boot_with( + "nvx_mode=managed nvx_identity=1000:1001", + &[ + "--virtio-console", + "none", + "--microvm-control-console", + "listen={dir}/control.sock", + "--microvm-control-auth-stdin", + "--microvm-report", + "{dir}/outcome.json", + ], + ); + let (directory, options) = parse_run_in(&managed, private_directory); + let options = options.unwrap(); + let (mut stream, mut fake) = std::os::unix::net::UnixStream::pair().unwrap(); + let capability = [7; CAPABILITY_LEN]; + let fake = thread::spawn(move || { + let mut guest = Guest::new(&options); + let flow = serve_client(&mut fake, &options, &capability, 1, &mut guest); + matches!(flow, Flow::Exit) + }); + let attach = encode_outer(OUTER_HOST_ATTACH, &[0; 16], 0, 0, &capability).unwrap(); + stream.write_all(&attach).unwrap(); + let (ready, _) = read_record(&mut stream).unwrap().unwrap(); + assert_eq!(ready.record_type, OUTER_READY); + Self { + stream, + instance: ready.instance_id, + epoch: ready.epoch, + sequence: 0, + fake, + _directory: directory, + } + } + + /// Sends one frame in a DATA record. + fn send_frame(&mut self, frame: &[u8]) { + let record = encode_outer(OUTER_DATA, &self.instance, self.epoch, self.sequence, frame); + self.stream.write_all(&record.unwrap()).unwrap(); + self.sequence += 1; + } + + /// Sends a request of version 2. + fn send(&mut self, request_id: u64, frame: v2::Frame<'_>) { + self.send_frame(&frame.encode(request_id).unwrap()); + } + + /// Reads the payload of the fake's next DATA record. + fn read(&mut self) -> Vec { + let (header, payload) = read_record(&mut self.stream).unwrap().unwrap(); + assert_eq!(header.record_type, OUTER_DATA); + payload + } + + /// Reads the fake's next frame, of version 2. + fn next(&mut self) -> Reply { + let payload = self.read(); + let (id, frame) = v2::Frame::decode(&payload).unwrap(); + match frame { + v2::Frame::Ready { + version, + activation, + } => { + assert_eq!(version, v2::VERSION); + Reply::Ready(id, activation.to_vec()) + } + v2::Frame::Stdout(bytes) => Reply::Stdout(id, bytes.to_vec()), + v2::Frame::Exit { status, outcome } => Reply::Exit(id, status, outcome), + v2::Frame::Stopped => Reply::Stopped(id), + v2::Frame::StdinAck(count) => Reply::Ack(id, count), + v2::Frame::Busy => Reply::Busy(id), + v2::Frame::Error { status, category } => Reply::Error(id, status, category.to_vec()), + frame => panic!("unexpected {frame:?}"), + } + } + + /// Stops the VM over version 2, which ends the fake's session. + fn stop(mut self, request_id: u64) { + self.send(request_id, v2::Frame::Stop); + assert_eq!(self.next(), Reply::Stopped(request_id)); + assert!(self.fake.join().unwrap()); + } +} + +/// Returns the payload of an `EXEC` request of the fake's workload `script`. +#[cfg(all(unix, feature = "cli"))] +fn script(script: &str) -> Vec { + timed(script, 0) +} + +/// Returns the payload of an `EXEC` request of the fake's workload `script`, which times out +/// after `timeout_ms`, unless it is 0. +#[cfg(all(unix, feature = "cli"))] +fn timed(script: &str, timeout_ms: u32) -> Vec { + crate::protocol::encode_exec_payload(&crate::protocol::Workload { + argv: ["/bin/sh", "-c", script].map(str::to_owned).to_vec(), + timeout_ms, + cwd: None, + environment: crate::protocol::WorkloadEnvironment::Default, + }) + .unwrap() +} + +#[cfg(all(unix, feature = "cli"))] +#[test] +fn version_2_workloads_read_their_input_until_its_end_under_the_window() { + let mut host = Host::attach(); + host.send(1, v2::Frame::Ping); + assert_eq!(host.next(), Reply::Ready(1, Vec::new())); + // The fake acknowledges what the workload reads, and `cat` reads until the input ends. + host.send(2, v2::Frame::Exec(&script("cat; echo done"))); + for input in [b"hello ", b"world\n"] { + host.send(2, v2::Frame::Stdin(input)); + assert_eq!(host.next(), Reply::Ack(2, 6)); + assert_eq!(host.next(), Reply::Stdout(2, input.to_vec())); + } + host.send(2, v2::Frame::StdinEnd); + assert_eq!(host.next(), Reply::Stdout(2, b"done\n".to_vec())); + assert_eq!(host.next(), Reply::Exit(2, 0, v2::Outcome::Exit)); + // A workload that waits for its input still ends at its timeout. + host.send(3, v2::Frame::Exec(&timed("cat", 50))); + assert_eq!(host.next(), Reply::Exit(3, 124, v2::Outcome::Timeout)); + // Input that the workload has not read stays unacknowledged, up to the window, and a byte + // beyond it ends the session without an answer. + host.send(4, v2::Frame::Exec(&script("sleep 60000; cat"))); + let half = vec![b'x'; v2::STDIN_WINDOW as usize / 2]; + host.send(4, v2::Frame::Stdin(&half)); + host.send(4, v2::Frame::Stdin(&half)); + host.send(4, v2::Frame::Stdin(b"x")); + assert!(read_record(&mut host.stream).unwrap().is_none()); + assert!(!host.fake.join().unwrap()); +} + +#[cfg(all(unix, feature = "cli"))] +#[test] +fn version_2_workloads_send_output_past_the_cap_of_version_1() { + let mut host = Host::attach(); + let total = MAX_OUTPUT_BYTES * 3 / 2; + host.send(1, v2::Frame::Exec(&script(&format!("flood {total}")))); + let mut forwarded = 0; + loop { + match host.next() { + Reply::Stdout(1, bytes) => forwarded += bytes.len(), + Reply::Exit(1, 0, v2::Outcome::Exit) => break, + reply => panic!("unexpected {reply:?}"), + } + } + assert_eq!(forwarded, total); + // Version 1, which the same session still serves, caps it. + host.send_frame(&encode_app( + APP_EXEC, + 2, + 0, + &script(&format!("flood {total}")), + )); + let mut capped = 0; + let exit = loop { + let payload = host.read(); + let frame = decode_app(&payload).unwrap(); + match frame.kind { + APP_STDOUT => capped += frame.payload.len(), + _ => break (frame.kind, frame.status, frame.payload.to_vec()), + } + }; + assert!(capped <= MAX_OUTPUT_BYTES); + assert_eq!(exit, (APP_EXIT, 125, b"output-limit".to_vec())); + host.stop(3); +} + +#[cfg(all(unix, feature = "cli"))] +#[test] +fn version_2_workloads_run_alone_and_end_on_cancel_or_stop() { + let mut host = Host::attach(); + host.send(1, v2::Frame::Exec(&script("sleep 60000"))); + host.send(2, v2::Frame::Exec(&script("echo second"))); + assert_eq!(host.next(), Reply::Busy(2)); + // The cancellation of another request leaves the workload running. + host.send(3, v2::Frame::Cancel); + host.send(1, v2::Frame::Cancel); + assert_eq!(host.next(), Reply::Exit(1, 137, v2::Outcome::Cancelled)); + // `STOP` ends the running workload, which reports it, and then the VM. + host.send(4, v2::Frame::Exec(&script("sleep 60000"))); + host.send(5, v2::Frame::Stop); + assert_eq!(host.next(), Reply::Exit(4, 137, v2::Outcome::Cancelled)); + assert_eq!(host.next(), Reply::Stopped(5)); + assert!(host.fake.join().unwrap()); +} diff --git a/scripts/test-layout-allowlist.json b/scripts/test-layout-allowlist.json index 1178a893f..ddb768d07 100644 --- a/scripts/test-layout-allowlist.json +++ b/scripts/test-layout-allowlist.json @@ -17,7 +17,6 @@ "nvx-rs/src/host/platform/windows/process.rs": 2, "nvx-rs/src/host/platform/windows/spawn.rs": 1, "nvx-rs/src/host/session.rs": 5, - "nvx-rs/src/host/vm.rs": 2, - "nvx-rs/src/protocol/app.rs": 1 + "nvx-rs/src/host/vm.rs": 2 } }