[DO NOT REVIEW] vmm_tests: drive the microVM guests without portb input

petri/src/vm/openvmm/microvm.rs

@@ -14,7 +14,6 @@ use chipset_resources::microvm::MicrovmShutdownHandle;
 use chipset_resources::microvm::MicrovmSnapshotRequestHandle;
 use chipset_resources::pit::PitDeviceHandle;
 use cpu_profile::HostCpuSignature;
-use futures::AsyncWriteExt;
 use openvmm_defs::config::Config;
 use openvmm_defs::config::LinuxDirectBootMode;
 use openvmm_defs::config::LinuxIsolationConfig;
@@ -24,7 +23,6 @@ use openvmm_defs::microvm::MicrovmConfig;
 use openvmm_defs::time_abi::TimeAbiParameters;
 use pal_async::DefaultDriver;
 use pal_async::socket::PolledSocket;
-use pal_async::socket::WriteHalf;
 use pal_async::task::Spawn;
 use petri_artifacts_common::tags::MachineArch;
 use serial_core::resources::DisconnectedSerialBackendHandle;
@@ -38,8 +36,9 @@ use vm_resource::kind::SerialBackendHandle;
 use vmotherboard::ChipsetDeviceHandle;
 
 pub(super) struct RuntimeResources {
-    input: WriteHalf<UnixStream>,
     output: mesh::Receiver<Vec<u8>>,
+    /// The output that followed the marker of the latest wait.
+    unread: Vec<u8>,
 }
 
 pub(super) struct PortbSerial {
@@ -356,19 +355,21 @@ pub(super) fn configure_serial(
     host: PolledSocket<UnixStream>,
     io: Option<Resource<SerialBackendHandle>>,
 ) -> anyhow::Result<SerialData> {
-    let (read, input) = host.split();
     let (output_send, output) = mesh::channel();
     let task = driver.spawn(
         "microvm-portb-console",
-        crate::log_task_with_output(log_file, read, "microvm-portb-console", output_send),
+        crate::log_task_with_output(log_file, host, "microvm-portb-console", output_send),
     );
     Ok(SerialData {
         emulated_serial_config: [None, None, None, None],
         serial_tasks: vec![task],
         linux_direct_serial_agent: None,
         microvm: Some(PortbSerial {
             io: io.unwrap_or_else(|| DisconnectedSerialBackendHandle.into_resource()),
-            runtime: RuntimeResources { input, output },
+            runtime: RuntimeResources {
+                output,
+                unread: Vec::new(),
+            },
         }),
     })
 }
@@ -469,46 +470,37 @@ impl PetriVmInner {
         self.wait_for_microvm_portb_bytes(marker.as_bytes()).await
     }
 
+    /// Waits for `marker` in the microVM portb output, consuming the output
+    /// through it. The output that follows it is kept for the next wait, so
+    /// that consecutive waits find markers that arrived together.
     pub(super) async fn wait_for_microvm_portb_bytes(
         &mut self,
         marker: &[u8],
     ) -> anyhow::Result<()> {
         anyhow::ensure!(!marker.is_empty(), "microVM portb marker cannot be empty");
-        let output = &mut self
+        let RuntimeResources { output, unread } = self
             .resources
             .microvm
             .as_mut()
-            .context("microVM portb output is not configured")?
-            .output;
-        let mut buffered = Vec::new();
+            .context("microVM portb output is not configured")?;
         loop {
+            if let Some(start) = unread
+                .windows(marker.len())
+                .position(|window| window == marker)
+            {
+                unread.drain(..start + marker.len());
+                return Ok(());
+            }
             let chunk = output.recv().await.map_err(|error| {
                 anyhow::anyhow!(
                     "microVM portb output disconnected ({error}) after bytes {:?}",
-                    String::from_utf8_lossy(&buffered[..buffered.len().min(256)])
+                    String::from_utf8_lossy(&unread[..unread.len().min(256)])
                 )
             })?;
-            buffered.extend_from_slice(&chunk);
-            if buffered
-                .windows(marker.len())
-                .any(|window| window == marker)
-            {
-                return Ok(());
-            }
+            unread.extend_from_slice(&chunk);
         }
     }
 
-    pub(super) async fn write_microvm_portb_input(&mut self, input: &[u8]) -> anyhow::Result<()> {
-        self.resources
-            .microvm
-            .as_mut()
-            .context("microVM portb input is not configured")?
-            .input
-            .write_all(input)
-            .await
-            .context("writing microVM portb input")
-    }
-
     pub(super) async fn pulse_save_restore(&self) -> anyhow::Result<()> {
         self.worker
             .pulse_save_restore()

petri/src/vm/openvmm/runtime.rs

@@ -375,13 +375,10 @@ impl PetriVmOpenVmm {
         pub async fn wait_for_microvm_portb_output(&mut self, marker: &str) -> anyhow::Result<()>
     );
     petri_vm_fn!(
-        /// Wait for an exact byte sequence from the microVM portb console.
+        /// Wait for an exact byte sequence from the microVM portb console,
+        /// consuming its output through the sequence.
         pub async fn wait_for_microvm_portb_bytes(&mut self, marker: &[u8]) -> anyhow::Result<()>
     );
-    petri_vm_fn!(
-        /// Write raw bytes to the microVM portb console input stream.
-        pub async fn write_microvm_portb_input(&mut self, input: &[u8]) -> anyhow::Result<()>
-    );
     petri_vm_fn!(
         /// Perform one pulse save/restore operation.
         pub async fn pulse_save_restore(&mut self) -> anyhow::Result<()>

vmm_tests/vmm_tests/tests/tests/ttrpc/microvm.rs

@@ -7,7 +7,6 @@ use super::launch_openvmm;
 use anyhow::Context;
 use futures::AsyncBufReadExt;
 use futures::AsyncReadExt;
-use futures::AsyncWriteExt;
 #[cfg(windows)]
 use guid::Guid;
 use mesh::CancelContext;
@@ -180,8 +179,6 @@ pub(super) async fn test_ttrpc_microvm_linux_direct_lifecycle_and_snapshot(
     const READY_MARKER: &[u8] = b"OPENVMM-LINUX-MPTABLE-SNAPSHOT-READY";
     const SNAPSHOT_CONTINUED_MARKER: &[u8] = b"SNAPSHOT-CONTINUED=1";
     const STATE_MARKER: &[u8] = b"STATE=1";
-    const COMMAND_STATE: u8 = 5;
-    const COMMAND_SHUTDOWN: u8 = 6;
     const KERNEL_OVERRIDE: &str = "OPENVMM_MICROVM_TEST_KERNEL";
     const INITRD_OVERRIDE: &str = "OPENVMM_MICROVM_TEST_INITRD";
 
@@ -361,25 +358,6 @@ write_port() {{
     octal=$(printf '%03o' "$2")
     printf "\\$octal" | dd of=/dev/port bs=1 seek="$1" count=1 conv=notrunc 2>/dev/null
 }}
-# Only kernels with the portb hvc driver expose /dev/hvc0, which then owns the
-# portb input. Other kernels, such as the packaged test kernel, poll portb.
-if [ -c /dev/hvc0 ]; then
-    stty -F /dev/hvc0 raw -echo
-    next_byte() {{
-        dd if=/dev/hvc0 bs=1 count=1 2>/dev/null | od -An -tu1 | tr -d ' '
-    }}
-else
-    next_byte() {{
-        while :; do
-            input_status=$(read_port 234)
-            [ -n "$input_status" ] || input_status=0
-            if [ $((input_status & 1)) -ne 0 ]; then
-                read_port 233
-                return
-            fi
-        done
-    }}
-fi
 generation=0
 {}
 write_port 1541 0
@@ -426,26 +404,15 @@ case "$range_count" in
         while :; do sleep 1; done
         ;;
 esac
-while :; do
-    command=$(next_byte)
-    case "$command" in
-        {COMMAND_STATE})
-            if [ "$generation" = 1 ]; then
+# portb takes no host input, so the restored guest reports its state and
+# shuts down without commands.
+if [ "$generation" = 1 ]; then
 {}
-            else
-{}
-            fi
-            ;;
-        {COMMAND_SHUTDOWN})
-            exit_status=$(next_byte)
-            write_port 1540 "$exit_status"
-            while :; do sleep 1; done
-            ;;
-        *)
+else
 {}
-            ;;
-    esac
-done
+fi
+write_port 1540 0
+while :; do sleep 1; done
 "#,
         portb_output(READY_MARKER)?,
         portb_output(SNAPSHOT_CONTINUED_MARKER)?,
@@ -457,7 +424,6 @@ done
         portb_output(b"RESTORE-RANGE-COUNT-INVALID")?,
         portb_output(STATE_MARKER)?,
         portb_output(b"STATE-INVALID")?,
-        portb_output(b"UNKNOWN-COMMAND")?,
     );
 
     let tempdir = if cfg!(target_os = "linux") {
@@ -609,7 +575,7 @@ done
                 )
             })?;
         let portb = PolledSocket::new(&driver, UnixStream::connect(&portb_path)?)?;
-        let (mut portb_read, mut portb_write) = portb.split();
+        let (mut portb_read, _portb_write) = portb.split();
         let mut resume_cancel = CancelContext::new().with_timeout(Duration::from_secs(15));
         let (resume, readiness) = futures::join!(
             resume_cancel.until_cancelled(client.call().start(vmservice::Vm::ResumeVm, ())),
@@ -643,15 +609,11 @@ done
             .with_context(|| {
                 format!("Linux direct restore {restore_index} did not report restore targets")
             })?;
-        portb_write.write_all(&[COMMAND_STATE]).await?;
-        portb_write.flush().await?;
         wait_for_bytes(&mut portb_read, &mut restore_output, STATE_MARKER)
             .await
             .with_context(|| {
-                format!("Linux direct restore {restore_index} did not answer state query")
+                format!("Linux direct restore {restore_index} did not report its state")
             })?;
-        portb_write.write_all(&[COMMAND_SHUTDOWN, 0]).await?;
-        portb_write.flush().await?;
         CancelContext::new()
             .with_timeout(Duration::from_secs(15))
             .until_cancelled(drain_until_closed(&mut portb_read, &mut restore_output))