nvx-rs: serve NVXC v2 in the guest loop and the fake OpenVMM (SL4.4b)

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`

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<Option<Outcome>, 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<Option<Outcome>, 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;

nvx-rs/src/guest/managed_agent/dispatch/tests.rs

@@ -225,6 +225,7 @@ fn read_outer(host: &mut UnixStream) -> (OuterHeader, Vec<u8>) {
 }
 
 mod runtime;
+mod streams;
 
 fn response(kind: u8, request_id: u64, status: i32, payload: &[u8]) -> Response {
     Response {

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);