From 130b936adcf35fad4b78a403cb1be9211fc274bc Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Mon, 5 Oct 2026 22:29:40 -0700 Subject: [PATCH 1/2] openvmm: add an authenticated microVM host pause endpoint A microVM that has a live authenticated control console cannot use the management RPC, the console relay, or --paused, and its REPL is headless, so its host had no way to pause or resume it. Add --microvm-state-control listen=PATH, a separate local endpoint that is bound and peer-checked like the control console. A host authenticates with the control-console capability within the control-console authentication timeout, then sends QUERY, PAUSE, and RESUME requests. The endpoint serves one host at a time, answers while the VM is paused, and is never recorded in a snapshot. The Guide documents protocol version 1. The VM worker gains MicrovmPause, MicrovmResume, and MicrovmRunState. A host pause stops the state units and holds guest monotonic time: once the vCPUs stop, it takes VP 0's TSC and every LAPIC, and resume sets them back with the time ABI's synchronized TSC set before any vCPU runs. Wall-clock time is not held; the CMOS RTC keeps following host UTC. A pause is busy during a snapshot boundary or post-restore gate, and is rejected for periodic or TSC-deadline LAPIC timers, as capture is. A guest reset while paused keeps the pause but discards the held time; a failed reset ends the pause. Also make the Windows named-pipe serial backend listen again when an accepted client's identity cannot be resolved, instead of dropping the pipe and panicking on the next poll. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- Guide/src/SUMMARY.md | 1 + Guide/src/reference/openvmm/management/cli.md | 19 + .../management/state_control_protocol.md | 137 +++ openvmm/openvmm_core/src/worker/dispatch.rs | 47 +- .../src/worker/dispatch/host_pause.rs | 347 ++++++++ .../src/worker/dispatch/microvm.rs | 5 + .../src/worker/dispatch/snapshot_rpc.rs | 43 +- .../src/worker/dispatch/time_abi.rs | 35 +- openvmm/openvmm_defs/src/rpc.rs | 54 ++ openvmm/openvmm_entry/src/cli_args/microvm.rs | 111 ++- openvmm/openvmm_entry/src/lib.rs | 9 + openvmm/openvmm_entry/src/microvm/config.rs | 30 + openvmm/openvmm_entry/src/microvm/launch.rs | 6 + openvmm/openvmm_entry/src/microvm/mod.rs | 6 + .../src/microvm/state_control.rs | 825 ++++++++++++++++++ .../src/microvm/state_control/protocol.rs | 402 +++++++++ .../openvmm_entry/src/serial_io/microvm.rs | 18 +- .../src/vm_controller/microvm.rs | 18 + .../serial/serial_socket/src/windows.rs | 33 +- .../serial/serial_socket/src/windows_tests.rs | 74 ++ vmm_core/src/partition_unit/snapshot.rs | 44 +- vmm_core/src/partition_unit/vp_set.rs | 26 + .../src/partition_unit/vp_set/boundary.rs | 5 +- .../src/partition_unit/vp_set/time_abi.rs | 138 ++- vmm_core/virt/src/time_abi/backend.rs | 4 +- 25 files changed, 2411 insertions(+), 26 deletions(-) create mode 100644 Guide/src/reference/openvmm/management/state_control_protocol.md create mode 100644 openvmm/openvmm_core/src/worker/dispatch/host_pause.rs create mode 100644 openvmm/openvmm_entry/src/microvm/state_control.rs create mode 100644 openvmm/openvmm_entry/src/microvm/state_control/protocol.rs diff --git a/Guide/src/SUMMARY.md b/Guide/src/SUMMARY.md index d4f602bc8c6..26df25ca1e7 100644 --- a/Guide/src/SUMMARY.md +++ b/Guide/src/SUMMARY.md @@ -89,6 +89,7 @@ - [Configuration and Management](./reference/openvmm/management.md) - [CLI](./reference/openvmm/management/cli.md) - [Control-session Protocol](./reference/openvmm/management/control_session_protocol.md) + - [MicroVM State-control Protocol](./reference/openvmm/management/state_control_protocol.md) - [Interactive Console](./reference/openvmm/management/interactive_console.md) - [gRPC / ttrpc](./reference/openvmm/management/grpc.md) - [Graphical Console](./reference/openvmm/graphical_console.md) diff --git a/Guide/src/reference/openvmm/management/cli.md b/Guide/src/reference/openvmm/management/cli.md index b4676a5be52..4a392e08011 100644 --- a/Guide/src/reference/openvmm/management/cli.md +++ b/Guide/src/reference/openvmm/management/cli.md @@ -855,6 +855,25 @@ Serial devices can be configured to appear as different devices inside the guest management RPC does not expose control-console restore attachments and rejects these snapshots explicitly. +* `--microvm-state-control `: With a live authenticated + `--microvm-control-console`, expose a separate host endpoint to pause, + resume, and query the microVM. The only backend is `listen=PATH`. It is bound + and peer-checked like the control console endpoint and must differ from the + boot console and control console endpoints. Hosts authenticate with the + control-console capability within `--microvm-control-auth-timeout-ms`. + + The endpoint serves one host at a time, and it answers while the VM is + paused. A pause holds guest monotonic time, so the guest observes no elapsed + time across it. Wall-clock time is not held: the CMOS RTC follows host UTC, + and the guest's system clock is behind host UTC after a resume until the + guest steps it. A pause is busy while a snapshot boundary or post-restore + gate is active, and is rejected when a vCPU has an armed periodic or + TSC-deadline LAPIC timer. + + The endpoint is never recorded in a snapshot. A restore caller may supply a + fresh one. See + [MicroVM State-control Protocol](./state_control_protocol.md). + The `BACKEND` argument is the same for all serial devices: * `none`: Serial output is dropped. diff --git a/Guide/src/reference/openvmm/management/state_control_protocol.md b/Guide/src/reference/openvmm/management/state_control_protocol.md new file mode 100644 index 00000000000..aa40fa1c32c --- /dev/null +++ b/Guide/src/reference/openvmm/management/state_control_protocol.md @@ -0,0 +1,137 @@ +# MicroVM state-control protocol + +`--microvm-state-control listen=` gives the host a local endpoint to +pause, resume, and query a microVM that has a live authenticated control +console. OpenVMM serves the endpoint outside the VM's device state units, so it +answers while the VM is paused. The protocol is version 1. + +## Endpoint and authentication + +The endpoint is bound like the control console endpoint and must differ from +the boot console and control console endpoints. On Linux it is an AF_UNIX +socket with mode `0600` in an owned, non-symlink directory with mode `0700`, +and the peer UID must match the OpenVMM process. On Windows it is a +`//./pipe/...` named pipe with a protected DACL, and the token user SID of the +connecting process must match the OpenVMM process user. + +The endpoint serves one host connection at a time: + +1. OpenVMM checks the peer identity before reading from the connection. +2. The first request must be `AUTHENTICATE`, carrying the 32-byte + control-console capability, within `--microvm-control-auth-timeout-ms`. +3. The host then sends requests. OpenVMM handles them one at a time and + answers them in request order. + +A failed or stalled authentication, a malformed request, or a second +`AUTHENTICATE` closes the connection without a response. OpenVMM also closes a +connection that sends no request for 60 seconds. The host may then reconnect. + +The endpoint is never recorded in a snapshot. A restore caller may supply a +fresh endpoint with the restore command. When a snapshot commits, the source +removes its endpoint along with the control console endpoint. + +## Header + +Every request and response starts with a 24-byte little-endian header: + +| Offset | Size | Field | +| --- | ---: | --- | +| 0 | 4 | ASCII magic `NVXV` | +| 4 | 2 | protocol version, `1` | +| 6 | 1 | operation; a response sets bit 7 (`0x80`) | +| 7 | 1 | flags, zero | +| 8 | 8 | sequence, chosen by the host and echoed in the response | +| 16 | 4 | status, zero in a request | +| 20 | 4 | payload length | + +| Value | Operation | Request payload | +| ---: | --- | --- | +| 1 | `AUTHENTICATE` | exactly 32 capability bytes | +| 2 | `QUERY` | empty | +| 3 | `PAUSE` | empty | +| 4 | `RESUME` | empty | + +A response payload is a 32-byte state record followed by an optional UTF-8 +detail of at most 512 bytes. The detail explains a `REJECTED` or `FAILED` +status for logs; hosts must not parse it. + +## State record + +| Offset | Size | Field | +| --- | ---: | --- | +| 0 | 1 | run state | +| 1 | 7 | reserved, zero | +| 8 | 8 | transitions | +| 16 | 16 | VMM instance ID of the control-console broker | + +| Value | Run state | Meaning | +| ---: | --- | --- | +| 0 | `UNKNOWN` | the VM worker did not answer | +| 1 | `RUNNING` | the guest is running | +| 2 | `PAUSED` | the host paused the VM and holds its guest time | +| 3 | `STOPPED` | the VM is stopped for another reason | +| 4 | `BUSY` | a snapshot boundary or post-restore gate is active | + +`transitions` counts how many times the VM entered or left the host-paused +state since the VMM process started, so it is odd exactly while the host holds +a pause. A host can compare it across responses to detect a pause or resume it +did not make. The instance ID changes when a snapshot is restored into a new +VMM process. + +## Status + +| Value | Status | Meaning | +| ---: | --- | --- | +| 0 | `OK` | the operation completed; the record is the resulting state | +| 1 | `BUSY` | a snapshot boundary or post-restore gate is active; nothing changed, so retry later | +| 2 | `REJECTED` | the operation was refused; the VM keeps the recorded state | +| 3 | `FAILED` | the VM worker did not answer, or the VM could not start again; its state is uncertain and it must be torn down | + +A successful `AUTHENTICATE` and every `QUERY` return `OK` with the current +state record. + +## Pause and resume + +`PAUSE` stops a running VM's vCPUs and devices and holds its guest time. Once +the vCPUs stop, OpenVMM takes VP 0's TSC and every vCPU's LAPIC state. Pausing +a VM that the host already paused returns `OK` and changes nothing. Pausing a +VM that is stopped for another reason is `REJECTED`. + +A pause is `REJECTED` with state `RUNNING` when guest time cannot be held, for +example because a vCPU has an armed periodic or TSC-deadline LAPIC timer. This +is the same restriction as snapshot capture. The vCPUs start again, so the +guest only observes a brief stall. If the VM cannot start again, the status is +`FAILED`. + +`RESUME` sets every vCPU's TSC back to the held value and sets the held LAPIC +state again before any vCPU runs, and then starts the VM. Resuming a running +VM returns `OK` and changes nothing. Resuming a VM that is stopped for another +reason is `REJECTED`. If OpenVMM cannot restore the held guest time, the status +is `REJECTED`, the VM stays `PAUSED` with its held time, and the host may +retry. If the VM cannot start, the status is `FAILED`. + +## Guest time across a pause + +The guest observes no elapsed monotonic time across a host pause. The TSC, the +LAPIC timers, and the VM time that drives emulated timers, such as the PIT and +the RTC's periodic and alarm interrupts, all stop with the vCPUs and continue +from the same values. + +Wall-clock time is not held. The CMOS RTC's date and time follow host UTC, so +after a resume they read the current host time. The guest's system clock is +behind host UTC by the paused duration until the guest steps it, for example +from the RTC. The NVX guest's wall-clock discipline does this at its next +poll. + +## Guest activity while paused + +While the VM is paused, nothing that needs the guest makes progress. Guest +timers, guest-requested snapshots, and control-console traffic wait for the +resume. A guest snapshot request that is still pending when a pause stops the +vCPUs is cancelled, as when the VM stops for any other reason. A request that +already reached its snapshot boundary makes the pause `BUSY` instead. + +If OpenVMM processes a guest-requested reset while the host holds a pause, the +reset discards the held guest time but keeps the pause. The reset VM starts at +the next `RESUME`. If the reset fails, the pause ends and the VM stays +`STOPPED`, so a `RESUME` is `REJECTED`. diff --git a/openvmm/openvmm_core/src/worker/dispatch.rs b/openvmm/openvmm_core/src/worker/dispatch.rs index 6f974b6b657..347b62ad7bd 100644 --- a/openvmm/openvmm_core/src/worker/dispatch.rs +++ b/openvmm/openvmm_core/src/worker/dispatch.rs @@ -4,6 +4,7 @@ mod amd_iommu_wiring; mod dump; mod ecam_config_access; +mod host_pause; mod intel_vtd_wiring; mod ioapic_iommu_wiring; mod microvm; @@ -926,6 +927,7 @@ pub(crate) struct LoadedVm { running: bool, snapshot_restore: restore::SnapshotRestore, snapshot_boundary: microvm::SnapshotBoundary, + host_pause: host_pause::HostPause, } struct DynamicVpciDeviceEntry { @@ -3418,6 +3420,7 @@ impl InitializedVm { running: false, snapshot_restore, snapshot_boundary: Default::default(), + host_pause: Default::default(), inner: LoadedVmInner { driver_source, resolver, @@ -3949,8 +3952,12 @@ impl LoadedVm { if self.running { return Ok(false); } + self.restore_held_guest_time() + .await + .context("failed to restore the guest time held by the host pause")?; self.start_state_units().await?; self.running = true; + self.complete_host_resume(); Ok(true) } @@ -4129,6 +4136,15 @@ impl LoadedVm { rpc.handle_failable(async |()| self.resume().await).await } VmRpc::Pause(rpc) => rpc.handle(async |()| self.pause().await).await, + VmRpc::MicrovmPause(rpc) => { + rpc.handle(async |()| self.microvm_pause().await).await + } + VmRpc::MicrovmResume(rpc) => { + rpc.handle(async |()| self.microvm_resume().await).await + } + VmRpc::MicrovmRunState(rpc) => { + rpc.handle_sync(|()| self.microvm_run_status()) + } VmRpc::Save(rpc) => { rpc.handle_failable(async |()| self.save().await.map(ProtobufMessage::new)) .await @@ -4701,16 +4717,27 @@ impl LoadedVm { async fn reset(&mut self, reload_firmware: bool) -> anyhow::Result<()> { let resume = self.pause().await; - self.state_units.reset().await?; - // TODO: _vmnic - // TODO: gdb? - - // Load again - if reload_firmware { - // Assign PCI resources before rebuilding firmware so the ACPI - // tables reflect the freshly assigned bus numbers. - self.assign_pci_resources().await?; - self.inner.load_firmware(false).await?; + // A reset reinitializes guest time, so it discards the time that a + // host pause held. The host pause survives only a complete reset. + self.host_pause.discard_held_time(); + let reset = async { + self.state_units.reset().await?; + // TODO: _vmnic + // TODO: gdb? + + // Load again + if reload_firmware { + // Assign PCI resources before rebuilding firmware so the ACPI + // tables reflect the freshly assigned bus numbers. + self.assign_pci_resources().await?; + self.inner.load_firmware(false).await?; + } + anyhow::Ok(()) + } + .await; + if let Err(error) = reset { + self.host_pause.abandon(); + return Err(error); } if resume { diff --git a/openvmm/openvmm_core/src/worker/dispatch/host_pause.rs b/openvmm/openvmm_core/src/worker/dispatch/host_pause.rs new file mode 100644 index 00000000000..9a6066a9936 --- /dev/null +++ b/openvmm/openvmm_core/src/worker/dispatch/host_pause.rs @@ -0,0 +1,347 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! Host pause of a microVM: stop the vCPUs and devices, and hold guest time so +//! the guest observes no elapsed time across the pause. +//! +//! A pause takes VP 0's TSC and every VP's LAPIC state once the vCPUs have +//! stopped. Resume sets every VP's TSC back to the held value with the +//! backend's synchronized TSC set, and sets the held LAPIC state again, before +//! any vCPU runs. VM time stops and starts with the state units. Host UTC, +//! which the CMOS RTC follows, keeps running, so the guest's wall-clock +//! discipline steps `CLOCK_REALTIME` at its next poll. +//! +//! A guest reset processed while the host holds the pause reinitializes guest +//! time, so it discards the held time but keeps the pause: the reset VM starts +//! when the host resumes it. A failed reset ends the pause and leaves the VM +//! stopped. + +use super::LoadedVm; +use mesh::error::RemoteError; +use openvmm_defs::microvm::MachineProfile; +use openvmm_defs::rpc::MicrovmHostPauseError; +use openvmm_defs::rpc::MicrovmRunState; +use openvmm_defs::rpc::MicrovmRunStatus; + +/// Host pause state of a [`LoadedVm`]. +#[derive(Default)] +pub(super) struct HostPause { + /// Whether the host holds the VM paused. + paused: bool, + /// Guest time held by the current host pause, unless a reset discarded it. + held: Option, + /// How many times the VM entered or left the host-paused state. + transitions: u64, +} + +/// Guest time held while the host has paused the VM. +struct HeldGuestTime { + /// VP 0's TSC when the vCPUs stopped. + #[cfg(guest_arch = "x86_64")] + tsc: u64, + /// Every VP's LAPIC state when the vCPUs stopped, in VP order. + #[cfg(guest_arch = "x86_64")] + lapics: Vec, +} + +impl HostPause { + /// Records a completed host pause that holds `held`. + fn hold(&mut self, held: HeldGuestTime) { + self.paused = true; + self.held = Some(held); + self.transitions += 1; + } + + /// Discards the held guest time before a reset reinitializes the + /// partition. The host pause stays in place. + pub(super) fn discard_held_time(&mut self) { + self.held = None; + } + + /// Ends the host pause after a failed reset, so that a host resume cannot + /// start the partially reset VM. + pub(super) fn abandon(&mut self) { + self.held = None; + if std::mem::take(&mut self.paused) { + self.transitions += 1; + } + } + + /// Records that the VM runs again. Returns whether this released a host + /// pause. + fn release(&mut self) -> bool { + if !std::mem::take(&mut self.paused) { + return false; + } + self.held = None; + self.transitions += 1; + true + } + + /// The run status of a VM that is `running`, or that a snapshot boundary + /// or restore gate `blocked`. + fn status(&self, running: bool, blocked: bool) -> MicrovmRunStatus { + let state = if blocked { + MicrovmRunState::Busy + } else if running { + MicrovmRunState::Running + } else if self.paused { + MicrovmRunState::Paused + } else { + MicrovmRunState::Stopped + }; + MicrovmRunStatus { + state, + transitions: self.transitions, + } + } +} + +impl LoadedVm { + /// Handles [`openvmm_defs::rpc::VmRpc::MicrovmPause`]. Returns `false` if + /// the host already paused the VM. + pub(super) async fn microvm_pause(&mut self) -> Result { + if self.inner.machine_profile != MachineProfile::Microvm { + return Err(rejected(anyhow::anyhow!( + "host pause requires the microVM profile" + ))); + } + if self.host_pause_blocked() { + return Err(MicrovmHostPauseError::Busy); + } + if !self.running { + return if self.host_pause.paused { + Ok(false) + } else { + Err(rejected(anyhow::anyhow!("the VM is not running"))) + }; + } + + self.state_units.stop().await; + self.running = false; + match self.hold_guest_time().await { + Ok(held) => { + self.host_pause.hold(held); + tracing::info!( + transitions = self.host_pause.transitions, + "microVM paused by the host" + ); + Ok(true) + } + Err(error) => { + // Nothing is held, so the guest only observes a brief stall. + if let Err(start_error) = self.start_state_units().await { + tracing::error!( + error = format!("{error:#}"), + start_error = format!("{start_error:#}"), + "microVM host pause failed and the VM could not start again" + ); + return Err(MicrovmHostPauseError::Uncertain(RemoteError::new( + start_error.context(format!("host pause failed: {error:#}")), + ))); + } + self.running = true; + tracing::warn!( + error = format!("{error:#}"), + "microVM host pause rejected; the guest continues" + ); + Err(rejected(error)) + } + } + } + + /// Handles [`openvmm_defs::rpc::VmRpc::MicrovmResume`]: resumes only a VM + /// that the host paused. Returns `false` if the VM is already running. + pub(super) async fn microvm_resume(&mut self) -> Result { + if self.host_pause_blocked() { + return Err(MicrovmHostPauseError::Busy); + } + if self.running { + return Ok(false); + } + if !self.host_pause.paused { + return Err(rejected(anyhow::anyhow!( + "the VM is stopped and was not paused by the host" + ))); + } + // The held time stays in place on failure, so the host can retry. + self.restore_held_guest_time().await.map_err(|error| { + rejected(error.context("failed to restore the guest time held by the host pause")) + })?; + if let Err(error) = self.start_state_units().await { + tracing::error!( + error = format!("{error:#}"), + "microVM host resume could not start the VM" + ); + return Err(MicrovmHostPauseError::Uncertain(RemoteError::new(error))); + } + self.running = true; + self.complete_host_resume(); + Ok(true) + } + + /// Restores the guest time held by a host pause, before the vCPUs run + /// again. The held time stays in place until the VM is running, so a + /// failed resume can be retried. + pub(super) async fn restore_held_guest_time(&mut self) -> anyhow::Result<()> { + let Some(held) = &self.host_pause.held else { + return Ok(()); + }; + #[cfg(guest_arch = "x86_64")] + { + let state = self + .inner + .time_abi + .as_ref() + .ok_or_else(|| anyhow::anyhow!("host pause requires the time ABI"))?; + super::time_abi::set_held_tsc(self.inner.partition.as_ref(), state, held.tsc)?; + self.inner + .partition_unit + .set_lapics(held.lapics.clone()) + .await?; + } + #[cfg(not(guest_arch = "x86_64"))] + let _ = held; + Ok(()) + } + + /// Records that the VM is running again, releasing any host pause. + pub(super) fn complete_host_resume(&mut self) { + if self.host_pause.release() { + tracing::info!( + transitions = self.host_pause.transitions, + "microVM resumed by the host" + ); + } + } + + /// Handles [`openvmm_defs::rpc::VmRpc::MicrovmRunState`]. + pub(super) fn microvm_run_status(&self) -> MicrovmRunStatus { + self.host_pause + .status(self.running, self.host_pause_blocked()) + } + + /// Returns whether a snapshot boundary or post-restore gate is active. + fn host_pause_blocked(&self) -> bool { + self.snapshot_boundary.is_active() || self.snapshot_restore.gate_timeout.is_some() + } + + /// Takes the guest time to hold with every vCPU stopped: checks that the + /// LAPIC timers are one-shot, takes VP 0's TSC, and reads every LAPIC. + #[cfg(guest_arch = "x86_64")] + async fn hold_guest_time(&mut self) -> anyhow::Result { + let state = self + .inner + .time_abi + .as_ref() + .ok_or_else(|| anyhow::anyhow!("host pause requires the time ABI"))?; + let tsc = super::time_abi::held_tsc(self.inner.partition.as_ref(), state)?; + self.inner.partition_unit.check_one_shot_timers().await?; + let lapics = self.inner.partition_unit.get_lapics().await?; + Ok(HeldGuestTime { tsc, lapics }) + } + + #[cfg(not(guest_arch = "x86_64"))] + async fn hold_guest_time(&mut self) -> anyhow::Result { + anyhow::bail!("host pause requires an x86-64 guest") + } +} + +fn rejected(error: anyhow::Error) -> MicrovmHostPauseError { + MicrovmHostPauseError::Rejected(RemoteError::new(error)) +} + +#[cfg(test)] +mod tests { + use super::*; + use test_with_tracing::test; + + fn held() -> HeldGuestTime { + HeldGuestTime { + #[cfg(guest_arch = "x86_64")] + tsc: 1, + #[cfg(guest_arch = "x86_64")] + lapics: Vec::new(), + } + } + + fn status(state: MicrovmRunState, transitions: u64) -> MicrovmRunStatus { + MicrovmRunStatus { state, transitions } + } + + #[test] + fn pause_and_resume_count_transitions() { + let mut pause = HostPause::default(); + assert_eq!( + pause.status(true, false), + status(MicrovmRunState::Running, 0) + ); + assert_eq!( + pause.status(false, false), + status(MicrovmRunState::Stopped, 0) + ); + // A start that did not release a host pause is not a transition. + assert!(!pause.release()); + + pause.hold(held()); + assert_eq!( + pause.status(false, false), + status(MicrovmRunState::Paused, 1) + ); + assert!(pause.release()); + assert!(pause.held.is_none()); + assert_eq!( + pause.status(true, false), + status(MicrovmRunState::Running, 2) + ); + assert!(!pause.release()); + assert_eq!(pause.transitions, 2); + } + + #[test] + fn reset_discards_held_time_but_keeps_the_pause() { + let mut pause = HostPause::default(); + pause.hold(held()); + pause.discard_held_time(); + assert!(pause.held.is_none()); + assert_eq!( + pause.status(false, false), + status(MicrovmRunState::Paused, 1) + ); + assert!(pause.release()); + assert_eq!(pause.transitions, 2); + + // Without a host pause, a reset leaves a stopped VM stopped. + pause.discard_held_time(); + assert_eq!( + pause.status(false, false), + status(MicrovmRunState::Stopped, 2) + ); + } + + #[test] + fn failed_reset_ends_the_pause() { + let mut pause = HostPause::default(); + pause.hold(held()); + pause.abandon(); + assert!(pause.held.is_none()); + assert_eq!( + pause.status(false, false), + status(MicrovmRunState::Stopped, 2) + ); + // A later resume releases nothing. + assert!(!pause.release()); + + // Without a host pause, a failed reset is not a transition. + pause.abandon(); + assert_eq!(pause.transitions, 2); + } + + #[test] + fn snapshot_boundary_or_restore_gate_reports_busy() { + let mut pause = HostPause::default(); + assert_eq!(pause.status(true, true), status(MicrovmRunState::Busy, 0)); + pause.hold(held()); + assert_eq!(pause.status(false, true), status(MicrovmRunState::Busy, 1)); + } +} diff --git a/openvmm/openvmm_core/src/worker/dispatch/microvm.rs b/openvmm/openvmm_core/src/worker/dispatch/microvm.rs index 420daafb8b8..6ca5f51450c 100644 --- a/openvmm/openvmm_core/src/worker/dispatch/microvm.rs +++ b/openvmm/openvmm_core/src/worker/dispatch/microvm.rs @@ -145,6 +145,11 @@ impl SnapshotBoundary { None => std::future::pending().await, } } + + /// Returns whether a boundary holds the vCPUs stopped. + pub(super) fn is_active(&self) -> bool { + self.stop_guard.is_some() + } } #[cfg(guest_arch = "x86_64")] diff --git a/openvmm/openvmm_core/src/worker/dispatch/snapshot_rpc.rs b/openvmm/openvmm_core/src/worker/dispatch/snapshot_rpc.rs index 8bf211c05c0..db14cb526ec 100644 --- a/openvmm/openvmm_core/src/worker/dispatch/snapshot_rpc.rs +++ b/openvmm/openvmm_core/src/worker/dispatch/snapshot_rpc.rs @@ -5,6 +5,7 @@ use chipset_resources::microvm::MicrovmSnapshotBoundaryRequest; use mesh::error::RemoteError; +use openvmm_defs::rpc::MicrovmHostPauseError; use openvmm_defs::rpc::PulseSaveRestoreError; use openvmm_defs::rpc::VmRpc; @@ -35,7 +36,11 @@ pub(super) fn filter(message: VmRpc, boundary_active: bool) -> Option { message @ (VmRpc::QuiesceForSnapshot(_) | VmRpc::ResumeAfterFailedSnapshot(_) | VmRpc::ReleaseSnapshotBoundary(_) - | VmRpc::ReadMemory(_)) => return Some(message), + | VmRpc::ReadMemory(_) + | VmRpc::MicrovmRunState(_)) => return Some(message), + VmRpc::MicrovmPause(rpc) | VmRpc::MicrovmResume(rpc) => { + rpc.complete(Err(MicrovmHostPauseError::Busy)) + } VmRpc::Save(rpc) => rpc.fail(SnapshotBoundaryActive), VmRpc::Resume(rpc) => rpc.fail(SnapshotBoundaryActive), VmRpc::Reset(rpc) => rpc.fail(SnapshotBoundaryActive), @@ -166,11 +171,47 @@ mod tests { VmRpc::Reset(Rpc::detached(())), VmRpc::Resume(Rpc::detached(())), VmRpc::Pause(Rpc::detached(())), + VmRpc::MicrovmPause(Rpc::detached(())), + VmRpc::MicrovmResume(Rpc::detached(())), + VmRpc::MicrovmRunState(Rpc::detached(())), ] { assert!(filter(message, false).is_some()); } } + #[test] + fn host_pause_is_busy_during_boundary_but_run_state_reaches_dispatch() { + use openvmm_defs::rpc::MicrovmHostPauseError; + use openvmm_defs::rpc::MicrovmRunState; + use openvmm_defs::rpc::MicrovmRunStatus; + + let (send, mut recv) = mesh::channel(); + let pause = send.call(VmRpc::MicrovmPause, ()); + let resume = send.call(VmRpc::MicrovmResume, ()); + for _ in 0..2 { + assert!(filter(recv.try_recv().unwrap(), true).is_none()); + } + for result in [block_on(pause), block_on(resume)] { + assert!(matches!(result.unwrap(), Err(MicrovmHostPauseError::Busy))); + } + + let status = send.call(VmRpc::MicrovmRunState, ()); + let Some(VmRpc::MicrovmRunState(rpc)) = filter(recv.try_recv().unwrap(), true) else { + panic!("run state was blocked during the boundary"); + }; + rpc.complete(MicrovmRunStatus { + state: MicrovmRunState::Busy, + transitions: 3, + }); + assert_eq!( + block_on(status).unwrap(), + MicrovmRunStatus { + state: MicrovmRunState::Busy, + transitions: 3, + } + ); + } + #[test] fn snapshot_control_and_reads_reach_dispatch() { for message in [ diff --git a/openvmm/openvmm_core/src/worker/dispatch/time_abi.rs b/openvmm/openvmm_core/src/worker/dispatch/time_abi.rs index e03a61cc301..4799fb31d0c 100644 --- a/openvmm/openvmm_core/src/worker/dispatch/time_abi.rs +++ b/openvmm/openvmm_core/src/worker/dispatch/time_abi.rs @@ -3,7 +3,8 @@ //! NVX time ABI v1 in the VM worker, which every microVM uses: the partition //! configuration, the backend preflight and rate policy, the `time-abi` state -//! unit, the capture records, and the restore clock. +//! unit, the capture records, the restore clock, and the TSC held by a host +//! pause. #![cfg(guest_arch = "x86_64")] @@ -617,6 +618,38 @@ pub(super) fn capture_records( }) } +/// Takes VP 0's TSC for a host pause, with every VP stopped. +pub(super) fn held_tsc( + partition: &dyn HvlitePartition, + state: &TimeAbiState, +) -> anyhow::Result { + let anchor = backend(partition, &state.report.hypervisor)?.capture_anchor()?; + tracing::info!( + tsc = anchor.tsc, + pairing_ns = anchor.pairing_ns, + "time ABI host pause anchor" + ); + Ok(anchor.tsc) +} + +/// Sets every VP's TSC to `tsc`, the value held by a host pause, before any +/// VP runs again, so the guest observes no elapsed time across the pause. +pub(super) fn set_held_tsc( + partition: &dyn HvlitePartition, + state: &TimeAbiState, + tsc: u64, +) -> anyhow::Result<()> { + let set = + backend(partition, &state.report.hypervisor)?.set_synchronized_tsc(&mut |_| Ok(tsc))?; + tracing::info!( + target = set.target, + method = ?set.method, + vps = set.readback.len(), + "time ABI host resume TSC set" + ); + Ok(()) +} + impl LoadedVm { /// Restores the guest clocks: restore steps 12 to 16 of the /// specification, after the saved state is restored and before any diff --git a/openvmm/openvmm_defs/src/rpc.rs b/openvmm/openvmm_defs/src/rpc.rs index 79080ef23c0..c4a8244508e 100644 --- a/openvmm/openvmm_defs/src/rpc.rs +++ b/openvmm/openvmm_defs/src/rpc.rs @@ -29,6 +29,15 @@ pub enum VmRpc { ReleaseSnapshotBoundary(FailableRpc<(), ()>), Resume(FailableRpc<(), bool>), Pause(Rpc<(), bool>), + /// Pause a running microVM for its host while holding guest time, so the + /// guest observes no elapsed time across the pause. Returns `false` if the + /// host already paused it. + MicrovmPause(Rpc<(), Result>), + /// Resume a microVM that its host paused, restoring the held guest time. + /// Returns `false` if the VM is already running. + MicrovmResume(Rpc<(), Result>), + /// Report the host-visible run state of the VM. + MicrovmRunState(Rpc<(), MicrovmRunStatus>), ClearHalt(Rpc<(), bool>), Reset(FailableRpc<(), ()>), Nmi(Rpc), @@ -88,6 +97,48 @@ pub enum SnapshotQuiesceError { Uncertain(#[source] RemoteError), } +/// Host-visible run state of a microVM. +#[derive(Debug, Copy, Clone, PartialEq, Eq, MeshPayload)] +pub enum MicrovmRunState { + /// The guest is running. + Running, + /// The host paused the guest with [`VmRpc::MicrovmPause`] and holds its + /// time until [`VmRpc::MicrovmResume`]. + Paused, + /// The VM is stopped for another reason, such as before its first start. + Stopped, + /// A snapshot boundary or post-restore gate is active. Host pause is + /// unavailable until it ends. + Busy, +} + +/// The run state of a VM and the number of times it entered or left the +/// host-paused state since it was launched. +#[derive(Debug, Copy, Clone, PartialEq, Eq, MeshPayload)] +pub struct MicrovmRunStatus { + /// The run state. + pub state: MicrovmRunState, + /// How many times the VM entered or left the host-paused state. It is odd + /// exactly while the host holds a pause. + pub transitions: u64, +} + +/// Failure of a [`VmRpc::MicrovmPause`] or [`VmRpc::MicrovmResume`] request. +#[derive(Debug, MeshPayload, thiserror::Error)] +pub enum MicrovmHostPauseError { + /// A snapshot boundary or post-restore gate is active. The VM is unchanged. + #[error("host pause is unavailable while a snapshot boundary or restore gate is active")] + Busy, + /// The request was rejected and the VM keeps its run state, apart from a + /// brief vCPU stop when a pause is rejected after the vCPUs stopped. + #[error("the request was rejected and the VM keeps its run state")] + Rejected(#[source] RemoteError), + /// The VM could not start again. Its state is uncertain and it must be + /// torn down. + #[error("the VM could not start again and its state is uncertain")] + Uncertain(#[source] RemoteError), +} + #[derive(Debug, MeshPayload, thiserror::Error)] pub enum PulseSaveRestoreError { #[error("reset not supported")] @@ -114,6 +165,9 @@ impl fmt::Debug for VmRpc { VmRpc::ReleaseSnapshotBoundary(_) => "ReleaseSnapshotBoundary", VmRpc::Resume(_) => "Resume", VmRpc::Pause(_) => "Pause", + VmRpc::MicrovmPause(_) => "MicrovmPause", + VmRpc::MicrovmResume(_) => "MicrovmResume", + VmRpc::MicrovmRunState(_) => "MicrovmRunState", VmRpc::ClearHalt(_) => "ClearHalt", VmRpc::Nmi(_) => "Nmi", VmRpc::AddVmbusDevice(_) => "AddVmbusDevice", diff --git a/openvmm/openvmm_entry/src/cli_args/microvm.rs b/openvmm/openvmm_entry/src/cli_args/microvm.rs index 8c435890943..4074a007ee4 100644 --- a/openvmm/openvmm_entry/src/cli_args/microvm.rs +++ b/openvmm/openvmm_entry/src/cli_args/microvm.rs @@ -406,6 +406,13 @@ pub struct MicrovmCli { )] pub microvm_control_auth_timeout_ms: u64, + /// authenticated host endpoint for microVM pause, resume, and run state + /// + /// Accepts listen=\. Requires a live --microvm-control-console; + /// hosts authenticate with the control-console capability. + #[clap(long, value_name = "SERIAL", requires("microvm_control_auth_stdin"))] + pub microvm_state_control: Option, + /// microVM CPU profile: a pinned profile ID, `auto` (the default) to /// select the host's profile, or `host` to derive a development profile /// from this host; a restore must name the snapshot's profile @@ -543,8 +550,9 @@ impl Options { && self.microvm.restore_memory.is_none() && self.microvm.memory_capacity.is_none() && self.microvm.microvm_control_console.is_none() - && !self.microvm.microvm_control_auth_stdin, - "--network-profile, --net-tap, --mount, --microvm-sandbox-block, --microvm-workload-identity, --microvm-lifecycle, --microvm-control-console, --microvm-control-auth-stdin, --restore-processors, --restore-memory, --memory-capacity, and microVM network policy require a microVM machine" + && !self.microvm.microvm_control_auth_stdin + && self.microvm.microvm_state_control.is_none(), + "--network-profile, --net-tap, --mount, --microvm-sandbox-block, --microvm-workload-identity, --microvm-lifecycle, --microvm-control-console, --microvm-control-auth-stdin, --microvm-state-control, --restore-processors, --restore-memory, --memory-capacity, and microVM network policy require a microVM machine" ); return Ok(()); } @@ -741,6 +749,20 @@ impl Options { "--microvm-control-auth-stdin requires --microvm-control-console" ); } + if let Some(state_control) = &self.microvm.microvm_state_control { + anyhow::ensure!( + matches!(state_control, SerialConfigCli::Pipe(_)), + "microVM state control requires listen=..." + ); + anyhow::ensure!( + self.microvm + .microvm_control_console + .as_ref() + .is_some_and(|console| matches!(console, SerialConfigCli::Pipe(_))) + && self.microvm.microvm_control_auth_stdin, + "--microvm-state-control requires a live --microvm-control-console with --microvm-control-auth-stdin" + ); + } if self.microvm.microvm_lifecycle == Some(MicrovmLifecycleCli::Managed) { anyhow::ensure!( self.microvm.microvm_workload_identity.is_some(), @@ -1757,6 +1779,91 @@ mod tests { ); } + #[test] + fn state_control_requires_a_live_authenticated_control_console() { + let (control, state) = if cfg!(windows) { + ( + "listen=//./pipe/openvmm-microvm-control", + "listen=//./pipe/openvmm-microvm-state", + ) + } else { + ("listen=control.sock", "listen=state.sock") + }; + let base = [ + "openvmm", + "--machine", + "microvm", + "--virtio-console", + "none", + "--microvm-control-console", + control, + "--microvm-control-auth-stdin", + ]; + let options = + Options::try_parse_from(base.into_iter().chain(["--microvm-state-control", state])) + .unwrap(); + assert!(matches!( + options.microvm.microvm_state_control, + Some(SerialConfigCli::Pipe(_)) + )); + assert_eq!( + options.validate_microvm_options().is_ok(), + cfg!(any(target_os = "linux", windows)) + ); + + // The endpoint must listen. + let options = + Options::try_parse_from(base.into_iter().chain(["--microvm-state-control", "none"])) + .unwrap(); + assert!(options.validate_microvm_options().is_err()); + + // It authenticates with the live control console's capability. + assert!( + Options::try_parse_from([ + "openvmm", + "--machine", + "microvm", + "--virtio-console", + "none", + "--microvm-state-control", + state, + ]) + .is_err() + ); + let options = Options::try_parse_from([ + "openvmm", + "--machine", + "microvm", + "--virtio-console", + "none", + "--microvm-control-console", + "none", + "--microvm-control-auth-stdin", + "--microvm-state-control", + state, + ]) + .unwrap(); + assert!(options.validate_microvm_options().is_err()); + + // A standard VM has no state-control endpoint. + let options = Options::try_parse_from([ + "openvmm", + "--microvm-control-console", + control, + "--microvm-control-auth-stdin", + "--microvm-state-control", + state, + ]) + .unwrap(); + assert!( + options + .validate_microvm_options() + .unwrap_err() + .to_string() + .contains("--microvm-state-control") + ); + } + #[test] fn test_microvm_host_loopback_generic_allow_is_rejected() { let common = [ diff --git a/openvmm/openvmm_entry/src/lib.rs b/openvmm/openvmm_entry/src/lib.rs index 30f74668b37..a8961a60be8 100644 --- a/openvmm/openvmm_entry/src/lib.rs +++ b/openvmm/openvmm_entry/src/lib.rs @@ -3026,6 +3026,14 @@ async fn run_control_inner( tracing::info!("restoring VM from snapshot"); } + // Serve host pause, resume, and run-state requests. The worker rejects a + // host pause until the VM runs. The task stops before the serial driver + // that backs its endpoint. + let state_control = microvm + .take_state_control() + .map(|endpoint| endpoint.spawn(vm_rpc.clone())) + .transpose()?; + if !opt.paused { anyhow::ensure!( vm_rpc.call_failable(VmRpc::Resume, ()).await?, @@ -3112,6 +3120,7 @@ async fn run_control_inner( // Wait for the controller task to finish (it stops the VM worker and // shuts down the mesh). let teardown = controller_task.await; + drop(state_control); drop(serial_driver); // run_repl returns the exit status: the code the guest drove via an opt-in diff --git a/openvmm/openvmm_entry/src/microvm/config.rs b/openvmm/openvmm_entry/src/microvm/config.rs index 3ede267782c..a4a881719db 100644 --- a/openvmm/openvmm_entry/src/microvm/config.rs +++ b/openvmm/openvmm_entry/src/microvm/config.rs @@ -22,6 +22,8 @@ use super::output::MicrovmOutputDrain; use super::output::OutputCompletion; use super::output::spawn_output; use super::restore::fresh_microvm_generation_id; +use super::state_control::StateControlConfig; +use super::state_control::StateControlEndpoint; use crate::ConsoleState; use crate::Options; use crate::VmResources; @@ -75,6 +77,7 @@ pub(crate) struct MicrovmConfigBuilder<'a> { control_console: Option, control_broker_config: Option, control_console_backend: Option>, + state_control: Option, portb: Option<(Resource, OutputCompletion)>, resources: MicrovmResources, } @@ -158,6 +161,27 @@ impl<'a> MicrovmConfigBuilder<'a> { .as_ref() .map(|(endpoint, _, _)| microvm_control_broker_config(opt, endpoint)) .transpose()?; + let state_control = match (&opt.microvm.microvm_state_control, &control_console) { + (None, _) => None, + (Some(SerialConfigCli::Pipe(path)), Some((SerialConfigCli::Pipe(control), _, _))) => { + anyhow::ensure!( + path != control + && !matches!( + console.as_ref(), + Some((SerialConfigCli::Pipe(boot), _, _)) if boot == path + ), + "microVM state control requires its own endpoint" + ); + let broker = control_broker_config + .as_ref() + .context("microVM state control requires the control-console broker")?; + Some(StateControlConfig::new(path.clone(), broker)) + } + (Some(SerialConfigCli::Pipe(_)), _) => { + bail!("microVM state control requires a live --microvm-control-console") + } + (Some(_), _) => bail!("microVM state control requires listen=..."), + }; anyhow::ensure!( control_console.is_none() || console.is_some(), "microVM control console requires the boot virtio-console" @@ -219,6 +243,7 @@ impl<'a> MicrovmConfigBuilder<'a> { control_console, control_broker_config, control_console_backend: None, + state_control, portb: None, resources, }) @@ -353,6 +378,11 @@ impl<'a> MicrovmConfigBuilder<'a> { _ => unreachable!("microVM control console backend was validated"), }; } + if let Some(config) = self.state_control.take() { + let path = config.path().to_owned(); + self.resources.state_control = Some(StateControlEndpoint::bind(config, serial_driver)?); + self.resources.state_control_socket_cleanup = microvm_console_socket_cleanup(path)?; + } Ok(virtio_console_backend) } diff --git a/openvmm/openvmm_entry/src/microvm/launch.rs b/openvmm/openvmm_entry/src/microvm/launch.rs index 1e26f528a23..6162908355e 100644 --- a/openvmm/openvmm_entry/src/microvm/launch.rs +++ b/openvmm/openvmm_entry/src/microvm/launch.rs @@ -179,6 +179,12 @@ impl MicrovmLaunch { self.resources.restore_packet_selected.take() } + /// Takes the bound host state-control endpoint, to serve it once the VM + /// worker runs. + pub(crate) fn take_state_control(&mut self) -> Option { + self.resources.state_control.take() + } + /// Creates the channels that carry guest-requested snapshot boundaries from /// the VM worker to the VM controller. pub(crate) fn snapshot_channels( diff --git a/openvmm/openvmm_entry/src/microvm/mod.rs b/openvmm/openvmm_entry/src/microvm/mod.rs index 85a93092610..499aeb2acf5 100644 --- a/openvmm/openvmm_entry/src/microvm/mod.rs +++ b/openvmm/openvmm_entry/src/microvm/mod.rs @@ -11,6 +11,7 @@ mod network; pub(crate) mod output; pub(crate) mod report; mod restore; +mod state_control; mod verify; pub(crate) use config::MicrovmConfigBuilder; @@ -49,6 +50,7 @@ pub(crate) use restore::prepare_restore; #[cfg(any(feature = "ttrpc", feature = "grpc"))] pub(crate) use restore::restore_packet_base; pub(crate) use restore::validate_restore_contract; +pub(crate) use state_control::StateControlEndpoint; pub(crate) use verify::fatal_error_message; pub(crate) use verify::report_time_abi_verification; @@ -77,6 +79,10 @@ pub(crate) struct MicrovmResources { pub(crate) control_console_attachment: Option, /// Removes the control console Unix socket on teardown. pub(crate) control_console_socket_cleanup: Option, + /// The bound host state-control endpoint, until it is served. + pub(crate) state_control: Option, + /// Removes the state-control Unix socket on teardown. + pub(crate) state_control_socket_cleanup: Option, /// Snapshot identity of the portable network attachment. pub(crate) network_attachment: Option, /// The bound run-scoped egress policy. diff --git a/openvmm/openvmm_entry/src/microvm/state_control.rs b/openvmm/openvmm_entry/src/microvm/state_control.rs new file mode 100644 index 00000000000..27332ba70af --- /dev/null +++ b/openvmm/openvmm_entry/src/microvm/state_control.rs @@ -0,0 +1,825 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! Authenticated host endpoint for microVM pause, resume, and run-state +//! queries. +//! +//! The endpoint is bound like the live control console: a private local +//! endpoint that serves one host at a time and checks the local peer identity +//! before the protocol starts. The first request must present the +//! control-console capability within the control-console authentication +//! timeout. A failed or stalled authentication, and any malformed request, +//! closes the connection without a response. The endpoint is served by the +//! VMM process outside the VM's device state units, so it answers while the +//! VM is paused. + +mod protocol; + +use self::protocol::Operation; +use self::protocol::StateRecord; +use self::protocol::Status; +use self::protocol::WireState; +use anyhow::Context as _; +use futures::AsyncRead; +use futures::AsyncReadExt as _; +use futures::AsyncWrite; +use futures::AsyncWriteExt as _; +use mesh::CancelContext; +use mesh::rpc::Rpc; +use mesh::rpc::RpcSend as _; +use openvmm_defs::rpc::MicrovmHostPauseError; +use openvmm_defs::rpc::MicrovmRunState; +use openvmm_defs::rpc::MicrovmRunStatus; +use openvmm_defs::rpc::VmRpc; +use pal_async::DefaultDriver; +use pal_async::task::Spawn as _; +use pal_async::task::Task; +use serial_core::LocalPeerIdentity; +use serial_core::SerialIo; +use std::path::Path; +use std::path::PathBuf; +use std::time::Duration; + +/// How long an authenticated host may stay idle between requests. +const IDLE_TIMEOUT: Duration = Duration::from_secs(60); + +/// How long to wait before accepting again after the endpoint failed to +/// accept a host. +const ACCEPT_RETRY_DELAY: Duration = Duration::from_millis(100); + +/// The configuration of a state-control endpoint. +pub(crate) struct StateControlConfig { + path: PathBuf, + capability: [u8; protocol::CAPABILITY_LEN], + instance_id: [u8; 16], + expected_peer_identity: LocalPeerIdentity, + auth_timeout: Duration, +} + +impl StateControlConfig { + /// Takes the capability, broker instance ID, expected peer identity, and + /// authentication timeout of the live control console. + pub(crate) fn new( + path: PathBuf, + broker: &virtio_resources::console::control::VirtioControlConsoleBrokerConfig, + ) -> Self { + Self { + path, + capability: broker.capability, + instance_id: broker.instance_id, + expected_peer_identity: broker.expected_peer_identity.clone(), + auth_timeout: Duration::from_millis(broker.auth_timeout_ms), + } + } + + /// The endpoint path. + pub(crate) fn path(&self) -> &Path { + &self.path + } +} + +/// A bound state-control endpoint, ready to serve once the VM worker runs. +pub(crate) struct StateControlEndpoint { + listener: Listener, + driver: DefaultDriver, + config: StateControlConfig, +} + +#[cfg(target_os = "linux")] +type Listener = serial_socket::net::OpenSocketSerialConfig; +#[cfg(windows)] +type Listener = serial_socket::windows::OpenWindowsPipeSerialConfig; +#[cfg(not(any(target_os = "linux", windows)))] +type Listener = std::convert::Infallible; + +impl StateControlEndpoint { + /// Binds the endpoint exclusively, with the private permissions of the + /// control console. + pub(crate) fn bind(config: StateControlConfig, driver: &DefaultDriver) -> anyhow::Result { + let listener = bind(&config.path).with_context(|| { + format!( + "failed to bind microVM state-control listener {}", + config.path.display() + ) + })?; + Ok(Self { + listener, + driver: driver.clone(), + config, + }) + } + + /// Serves hosts until the returned task is dropped, controlling the VM + /// worker behind `vm_rpc`. + pub(crate) fn spawn(self, vm_rpc: mesh::Sender) -> anyhow::Result> { + let Self { + listener, + driver, + config, + } = self; + let io = backend(listener, &driver) + .context("failed to start the microVM state-control endpoint")?; + Ok(driver.spawn("microvm-state-control", serve(io, config, vm_rpc))) + } +} + +#[cfg(any(target_os = "linux", windows))] +fn bind(path: &Path) -> std::io::Result { + crate::serial_io::microvm::bind_control_endpoint(path) +} + +#[cfg(not(any(target_os = "linux", windows)))] +fn bind(_path: &Path) -> std::io::Result { + Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "microVM state control requires Linux Unix sockets or Windows named pipes", + )) +} + +#[cfg(target_os = "linux")] +fn backend(listener: Listener, driver: &DefaultDriver) -> std::io::Result> { + Ok(Box::new(serial_socket::net::SocketSerialBackend::new( + Box::new(driver.clone()), + listener, + )?)) +} + +#[cfg(windows)] +fn backend(listener: Listener, driver: &DefaultDriver) -> std::io::Result> { + Ok(Box::new( + serial_socket::windows::WindowsPipeSerialBackend::new(Box::new(driver.clone()), listener)?, + )) +} + +#[cfg(not(any(target_os = "linux", windows)))] +fn backend(listener: Listener, _driver: &DefaultDriver) -> std::io::Result> { + match listener {} +} + +/// Accepts hosts one at a time and serves each authenticated session. +async fn serve(mut io: Box, config: StateControlConfig, vm_rpc: mesh::Sender) { + loop { + if let Err(error) = futures::future::poll_fn(|cx| io.poll_connect(cx)).await { + tracelimit::error_ratelimited!( + error = &error as &dyn std::error::Error, + "microVM state-control endpoint failed to accept a host" + ); + CancelContext::new() + .with_timeout(ACCEPT_RETRY_DELAY) + .cancelled() + .await; + continue; + } + match io.local_peer_identity() { + Ok(Some(identity)) if identity == config.expected_peer_identity => { + if let Err(error) = serve_session(&mut *io, &config, &vm_rpc).await { + tracelimit::warn_ratelimited!( + error = error.as_ref() as &dyn std::error::Error, + "microVM state-control session closed" + ); + } + } + identity => { + tracelimit::warn_ratelimited!( + identity_available = matches!(identity, Ok(Some(_))), + "microVM state-control host rejected because its local peer identity is unavailable or unexpected" + ); + } + } + if let Err(error) = io.disconnect_current() { + tracing::error!( + error = &error as &dyn std::error::Error, + "microVM state-control endpoint cannot accept another host" + ); + return; + } + } +} + +/// Serves one connected host: authentication, then requests until the host +/// disconnects, stays idle for [`IDLE_TIMEOUT`], or sends a malformed request. +async fn serve_session( + stream: &mut (impl AsyncRead + AsyncWrite + Unpin + ?Sized), + config: &StateControlConfig, + vm_rpc: &mesh::Sender, +) -> anyhow::Result<()> { + let (header, capability) = CancelContext::new() + .with_timeout(config.auth_timeout) + .until_cancelled(read_request(&mut *stream)) + .await + .context("authentication timed out")?? + .context("host disconnected before authenticating")?; + anyhow::ensure!( + header.operation == Operation::Authenticate, + "the first request did not authenticate" + ); + anyhow::ensure!( + capability_matches(&config.capability, &capability), + "authentication failed" + ); + let run = run_status(vm_rpc).await; + respond(&mut *stream, header, Status::Ok, config, run, "").await?; + + loop { + let Ok(request) = CancelContext::new() + .with_timeout(IDLE_TIMEOUT) + .until_cancelled(read_request(&mut *stream)) + .await + else { + return Ok(()); + }; + let Some((header, _)) = request? else { + return Ok(()); + }; + let (status, run, detail) = match header.operation { + Operation::Authenticate => anyhow::bail!("the host authenticated twice"), + Operation::Query => (Status::Ok, run_status(vm_rpc).await, String::new()), + Operation::Pause => transition(vm_rpc, VmRpc::MicrovmPause).await, + Operation::Resume => transition(vm_rpc, VmRpc::MicrovmResume).await, + }; + respond(&mut *stream, header, status, config, run, &detail).await?; + } +} + +/// Reads one request. Returns `None` if the host disconnected cleanly before +/// a header. +async fn read_request( + stream: &mut (impl AsyncRead + Unpin + ?Sized), +) -> anyhow::Result> { + let mut header = [0; protocol::HEADER_LEN]; + let first = stream.read(&mut header).await?; + if first == 0 { + return Ok(None); + } + stream.read_exact(&mut header[first..]).await?; + let header = protocol::decode_request_header(&header)?; + let mut capability = [0; protocol::CAPABILITY_LEN]; + if header.payload_len() != 0 { + stream.read_exact(&mut capability).await?; + } + Ok(Some((header, capability))) +} + +async fn respond( + stream: &mut (impl AsyncWrite + Unpin + ?Sized), + request: protocol::RequestHeader, + status: Status, + config: &StateControlConfig, + run: Option, + detail: &str, +) -> anyhow::Result<()> { + let record = StateRecord { + state: run.map_or(WireState::Unknown, |run| wire_state(run.state)), + transitions: run.map_or(0, |run| run.transitions), + instance_id: config.instance_id, + }; + let response = + protocol::encode_response(request.operation, request.sequence, status, &record, detail); + stream.write_all(&response).await?; + stream.flush().await?; + Ok(()) +} + +fn wire_state(state: MicrovmRunState) -> WireState { + match state { + MicrovmRunState::Running => WireState::Running, + MicrovmRunState::Paused => WireState::Paused, + MicrovmRunState::Stopped => WireState::Stopped, + MicrovmRunState::Busy => WireState::Busy, + } +} + +/// Returns the VM worker's run status, or `None` if it did not answer. +async fn run_status(vm_rpc: &mesh::Sender) -> Option { + vm_rpc.call(VmRpc::MicrovmRunState, ()).await.ok() +} + +/// Asks the VM worker to pause or resume, then reports its run state. +async fn transition( + vm_rpc: &mesh::Sender, + request: fn(Rpc<(), Result>) -> VmRpc, +) -> (Status, Option, String) { + let (status, detail) = match vm_rpc.call(request, ()).await { + Ok(Ok(_)) => (Status::Ok, String::new()), + Ok(Err(MicrovmHostPauseError::Busy)) => (Status::Busy, String::new()), + Ok(Err(error @ MicrovmHostPauseError::Rejected(_))) => { + (Status::Rejected, error_detail(&error)) + } + Ok(Err(error @ MicrovmHostPauseError::Uncertain(_))) => { + (Status::Failed, error_detail(&error)) + } + Err(error) => (Status::Failed, error_detail(&error)), + }; + (status, run_status(vm_rpc).await, detail) +} + +fn error_detail(error: &(dyn std::error::Error + 'static)) -> String { + let mut detail = error.to_string(); + let mut source = error.source(); + while let Some(error) = source { + detail.push_str(": "); + detail.push_str(&error.to_string()); + source = error.source(); + } + detail +} + +/// Compares the capability in constant time. +fn capability_matches(expected: &[u8; 32], actual: &[u8; 32]) -> bool { + std::hint::black_box( + expected + .iter() + .zip(actual) + .fold(0, |difference, (expected, actual)| { + difference | (expected ^ actual) + }), + ) == 0 +} + +#[cfg(test)] +mod tests { + use super::protocol::client::Response; + use super::protocol::client::decode_responses; + use super::protocol::client::encode_request; + use super::*; + use futures::FutureExt as _; + use mesh::error::RemoteError; + use std::pin::Pin; + use std::task::Context; + use std::task::Poll; + use test_with_tracing::test; + + const CAPABILITY: [u8; 32] = [0x5a; 32]; + const INSTANCE: [u8; 16] = [3; 16]; + + fn config() -> StateControlConfig { + StateControlConfig { + path: PathBuf::from("state.sock"), + capability: CAPABILITY, + instance_id: INSTANCE, + expected_peer_identity: LocalPeerIdentity::UnixUid(1), + auth_timeout: Duration::from_secs(5), + } + } + + /// A host connection with scripted input that records the responses. + struct Scripted { + input: futures::io::Cursor>, + output: Vec, + } + + impl AsyncRead for Scripted { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + Pin::new(&mut self.input).poll_read(cx, buf) + } + } + + impl AsyncWrite for Scripted { + fn poll_write( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + self.output.extend_from_slice(buf); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + + /// A VM worker that answers run-state, pause, and resume requests. + #[derive(Default)] + struct FakeWorker { + state: Option, + transitions: u64, + pause_error: Option MicrovmHostPauseError>, + resume_error: Option MicrovmHostPauseError>, + requests: Vec<&'static str>, + } + + impl FakeWorker { + fn running() -> Self { + Self { + state: Some(MicrovmRunState::Running), + ..Default::default() + } + } + + fn status(&self) -> MicrovmRunStatus { + MicrovmRunStatus { + state: self.state.unwrap(), + transitions: self.transitions, + } + } + + async fn run(&mut self, mut recv: mesh::Receiver) { + while let Ok(message) = recv.recv().await { + match message { + VmRpc::MicrovmRunState(rpc) => rpc.complete(self.status()), + VmRpc::MicrovmPause(rpc) => { + self.requests.push("pause"); + let result = self.transition( + self.pause_error, + MicrovmRunState::Running, + MicrovmRunState::Paused, + ); + rpc.complete(result); + } + VmRpc::MicrovmResume(rpc) => { + self.requests.push("resume"); + let result = self.transition( + self.resume_error, + MicrovmRunState::Paused, + MicrovmRunState::Running, + ); + rpc.complete(result); + } + message => panic!("unexpected VM RPC {message:?}"), + } + } + } + + /// Moves from `from` to `to`; already being in `to` is idempotent. + fn transition( + &mut self, + error: Option MicrovmHostPauseError>, + from: MicrovmRunState, + to: MicrovmRunState, + ) -> Result { + if let Some(error) = error { + return Err(error()); + } + match self.state.unwrap() { + state if state == from => { + self.state = Some(to); + self.transitions += 1; + Ok(true) + } + state if state == to => Ok(false), + state => Err(MicrovmHostPauseError::Rejected(RemoteError::new( + anyhow::anyhow!("cannot leave {state:?}"), + ))), + } + } + } + + fn requests(sequence: &[(Operation, &[u8])]) -> Vec { + sequence + .iter() + .enumerate() + .flat_map(|(index, (operation, payload))| { + encode_request(*operation, index as u64 + 1, payload) + }) + .collect() + } + + /// Runs one scripted session against `worker` and returns the session + /// result and responses. + fn run_session(input: Vec, worker: &mut FakeWorker) -> (anyhow::Result<()>, Vec) { + let (send, recv) = mesh::channel(); + let mut stream = Scripted { + input: futures::io::Cursor::new(input), + output: Vec::new(), + }; + let config = config(); + let session = async { + let result = serve_session(&mut stream, &config, &send).await; + drop(send); + result + }; + let (result, ()) = + futures::executor::block_on(futures::future::join(session, worker.run(recv))); + (result, decode_responses(&stream.output).unwrap()) + } + + fn record(state: WireState, transitions: u64) -> StateRecord { + StateRecord { + state, + transitions, + instance_id: INSTANCE, + } + } + + #[test] + fn pause_query_and_resume_report_observed_state() { + let mut worker = FakeWorker::running(); + let (result, responses) = run_session( + requests(&[ + (Operation::Authenticate, &CAPABILITY), + (Operation::Pause, &[]), + (Operation::Pause, &[]), + (Operation::Query, &[]), + (Operation::Resume, &[]), + (Operation::Resume, &[]), + ]), + &mut worker, + ); + result.unwrap(); + let expected = [ + (Operation::Authenticate, WireState::Running, 0), + (Operation::Pause, WireState::Paused, 1), + (Operation::Pause, WireState::Paused, 1), + (Operation::Query, WireState::Paused, 1), + (Operation::Resume, WireState::Running, 2), + (Operation::Resume, WireState::Running, 2), + ]; + assert_eq!(responses.len(), expected.len()); + for (sequence, (response, (operation, state, transitions))) in + (1..).zip(responses.iter().zip(expected)) + { + assert_eq!(response.operation, operation); + assert_eq!(response.sequence, sequence); + assert_eq!(response.status, Status::Ok); + assert_eq!(response.record, record(state, transitions)); + assert!(response.detail.is_empty()); + } + // Repeated pause and resume are idempotent in the worker. + assert_eq!(worker.requests, ["pause", "pause", "resume", "resume"]); + } + + #[test] + fn transition_failures_map_to_their_statuses() { + type Inject = fn() -> MicrovmHostPauseError; + let busy: Inject = || MicrovmHostPauseError::Busy; + let rejected: Inject = + || MicrovmHostPauseError::Rejected(RemoteError::new(anyhow::anyhow!("periodic timer"))); + let uncertain: Inject = + || MicrovmHostPauseError::Uncertain(RemoteError::new(anyhow::anyhow!("start failed"))); + for operation in [Operation::Pause, Operation::Resume] { + for (error, status, state) in [ + (busy, Status::Busy, MicrovmRunState::Busy), + (rejected, Status::Rejected, MicrovmRunState::Paused), + (uncertain, Status::Failed, MicrovmRunState::Stopped), + ] { + let mut worker = FakeWorker { + state: Some(state), + pause_error: Some(error), + resume_error: Some(error), + ..Default::default() + }; + let (result, responses) = run_session( + requests(&[(Operation::Authenticate, &CAPABILITY), (operation, &[])]), + &mut worker, + ); + result.unwrap(); + assert_eq!(responses[1].operation, operation); + assert_eq!(responses[1].status, status); + assert_eq!(responses[1].record.state, wire_state(state)); + assert_eq!(responses[1].detail.is_empty(), status == Status::Busy); + } + } + } + + #[test] + fn resume_reports_the_state_the_worker_kept() { + let mut worker = FakeWorker { + state: Some(MicrovmRunState::Stopped), + ..Default::default() + }; + let (result, responses) = run_session( + requests(&[ + (Operation::Authenticate, &CAPABILITY), + (Operation::Resume, &[]), + ]), + &mut worker, + ); + result.unwrap(); + assert_eq!(responses[1].status, Status::Rejected); + assert_eq!(responses[1].record.state, WireState::Stopped); + assert_eq!(worker.requests, ["resume"]); + + let mut worker = FakeWorker { + state: Some(MicrovmRunState::Paused), + transitions: 1, + resume_error: Some(|| { + MicrovmHostPauseError::Rejected(RemoteError::new(anyhow::anyhow!( + "TSC read-back mismatch" + ))) + }), + ..Default::default() + }; + let (result, responses) = run_session( + requests(&[ + (Operation::Authenticate, &CAPABILITY), + (Operation::Resume, &[]), + ]), + &mut worker, + ); + result.unwrap(); + // The held time is still in place, so the host can retry. + assert_eq!(responses[1].status, Status::Rejected); + assert_eq!(responses[1].record, record(WireState::Paused, 1)); + assert!(responses[1].detail.contains("TSC read-back mismatch")); + } + + #[test] + fn authentication_failures_close_without_a_response() { + let mut wrong = CAPABILITY; + wrong[31] ^= 1; + for input in [ + requests(&[(Operation::Authenticate, &wrong)]), + requests(&[(Operation::Query, &[])]), + Vec::new(), + // Truncated capability. + requests(&[(Operation::Authenticate, &CAPABILITY)])[..40].to_vec(), + ] { + let mut worker = FakeWorker::running(); + let (result, responses) = run_session(input, &mut worker); + assert!(result.is_err()); + assert!(responses.is_empty()); + assert!(worker.requests.is_empty()); + } + } + + #[test] + fn malformed_and_repeated_authentication_requests_close_the_session() { + let mut bad = encode_request(Operation::Query, 2, &[]); + bad[7] = 1; + let mut input = requests(&[(Operation::Authenticate, &CAPABILITY)]); + input.extend(bad); + input.extend(encode_request(Operation::Pause, 3, &[])); + let mut worker = FakeWorker::running(); + let (result, responses) = run_session(input, &mut worker); + assert!(result.is_err()); + assert_eq!(responses.len(), 1); + assert!(worker.requests.is_empty()); + + let mut worker = FakeWorker::running(); + let (result, responses) = run_session( + requests(&[ + (Operation::Authenticate, &CAPABILITY), + (Operation::Authenticate, &CAPABILITY), + ]), + &mut worker, + ); + assert!(result.is_err()); + assert_eq!(responses.len(), 1); + } + + #[test] + fn unavailable_worker_reports_unknown_state() { + let (send, recv) = mesh::channel::(); + drop(recv); + let mut stream = Scripted { + input: futures::io::Cursor::new(requests(&[ + (Operation::Authenticate, &CAPABILITY), + (Operation::Pause, &[]), + (Operation::Resume, &[]), + ])), + output: Vec::new(), + }; + let config = config(); + serve_session(&mut stream, &config, &send) + .now_or_never() + .unwrap() + .unwrap(); + let responses = decode_responses(&stream.output).unwrap(); + assert_eq!(responses[0].record.state, WireState::Unknown); + assert_eq!(responses[1].status, Status::Failed); + assert_eq!(responses[2].status, Status::Failed); + assert_eq!(responses[2].record.state, WireState::Unknown); + } + + #[test] + fn capability_comparison_checks_every_byte() { + assert!(capability_matches(&CAPABILITY, &CAPABILITY)); + for index in 0..32 { + let mut other = CAPABILITY; + other[index] ^= 0x80; + assert!(!capability_matches(&CAPABILITY, &other)); + } + } + + #[cfg(any(target_os = "linux", windows))] + const TEST_TIMEOUT: Duration = Duration::from_secs(10); + + #[cfg(target_os = "linux")] + fn endpoint_path(directory: &tempfile::TempDir) -> PathBuf { + use std::os::unix::fs::PermissionsExt; + + std::fs::set_permissions(directory.path(), std::fs::Permissions::from_mode(0o700)).unwrap(); + directory.path().join("state.sock") + } + + #[cfg(windows)] + fn endpoint_path(_directory: &tempfile::TempDir) -> PathBuf { + PathBuf::from(format!( + r"\\.\pipe\openvmm-state-control-test-{}", + guid::Guid::new_random() + )) + } + + #[cfg(target_os = "linux")] + fn current_peer_identity() -> LocalPeerIdentity { + LocalPeerIdentity::UnixUid(pal::unix::effective_user_id()) + } + + #[cfg(windows)] + fn current_peer_identity() -> LocalPeerIdentity { + let sid = pal::windows::security::user_sid::current_process_user_sid().unwrap(); + let (bytes, length) = sid.to_fixed_bytes(); + LocalPeerIdentity::windows_sid(bytes, length).unwrap() + } + + #[cfg(target_os = "linux")] + fn connect(path: &Path) -> std::os::unix::net::UnixStream { + std::os::unix::net::UnixStream::connect(path).unwrap() + } + + #[cfg(windows)] + fn connect(path: &Path) -> std::fs::File { + let deadline = std::time::Instant::now() + TEST_TIMEOUT; + loop { + match std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(path) + { + // The endpoint listens again shortly after closing a host. + Err(error) + if error.raw_os_error() + == Some(windows_sys::Win32::Foundation::ERROR_PIPE_BUSY as i32) + && std::time::Instant::now() < deadline => + { + std::thread::sleep(Duration::from_millis(10)); + } + result => return result.unwrap(), + } + } + } + + /// Serves a real endpoint: a host with the wrong capability is closed + /// without a response, and the next host is served. + #[cfg(any(target_os = "linux", windows))] + #[test] + fn endpoint_closes_a_failed_host_and_serves_the_next() { + use std::io::Read as _; + use std::io::Write as _; + + let directory = tempfile::tempdir().unwrap(); + let path = endpoint_path(&directory); + let config = StateControlConfig { + path: path.clone(), + expected_peer_identity: current_peer_identity(), + ..config() + }; + let (bound_send, bound_recv) = std::sync::mpsc::channel(); + let (done_send, done_recv) = mesh::oneshot::<()>(); + let server = std::thread::spawn(move || { + pal_async::DefaultPool::run_with(async |driver| { + let endpoint = StateControlEndpoint::bind(config, &driver).unwrap(); + bound_send.send(()).unwrap(); + let (send, recv) = mesh::channel(); + let _serving = endpoint.spawn(send).unwrap(); + let mut worker = FakeWorker::running(); + futures::future::select(std::pin::pin!(worker.run(recv)), done_recv).await; + worker.requests + }) + }); + bound_recv.recv_timeout(TEST_TIMEOUT).unwrap(); + + let (host_send, host_recv) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let mut wrong = CAPABILITY; + wrong[0] ^= 1; + let mut host = connect(&path); + host.write_all(&encode_request(Operation::Authenticate, 1, &wrong)) + .unwrap(); + let mut rejected = Vec::new(); + // Windows reports the closed pipe as an error rather than EOF. + let _ = host.read_to_end(&mut rejected); + drop(host); + + let mut host = connect(&path); + host.write_all(&requests(&[ + (Operation::Authenticate, &CAPABILITY), + (Operation::Pause, &[]), + ])) + .unwrap(); + let mut served = vec![0; 2 * (protocol::HEADER_LEN + protocol::STATE_RECORD_LEN)]; + host.read_exact(&mut served).unwrap(); + host_send + .send((rejected, decode_responses(&served).unwrap())) + .unwrap(); + }); + let (rejected, responses) = host_recv + .recv_timeout(TEST_TIMEOUT) + .expect("the endpoint did not serve the host"); + assert!(rejected.is_empty()); + assert_eq!(responses[0].status, Status::Ok); + assert_eq!(responses[1].status, Status::Ok); + assert_eq!(responses[1].record, record(WireState::Paused, 1)); + done_send.send(()); + assert_eq!(server.join().unwrap(), ["pause"]); + } +} diff --git a/openvmm/openvmm_entry/src/microvm/state_control/protocol.rs b/openvmm/openvmm_entry/src/microvm/state_control/protocol.rs new file mode 100644 index 00000000000..7b54090ec1f --- /dev/null +++ b/openvmm/openvmm_entry/src/microvm/state_control/protocol.rs @@ -0,0 +1,402 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! Wire format of the microVM state-control protocol, version 1. +//! +//! Every message starts with a 24-byte little-endian header. Requests carry no +//! payload except the authentication capability. Responses carry a 32-byte +//! state record and an optional UTF-8 detail. + +/// ASCII magic of every message. +pub(crate) const MAGIC: [u8; 4] = *b"NVXV"; +/// The protocol version. +pub(crate) const VERSION: u16 = 1; +/// The header length. +pub(crate) const HEADER_LEN: usize = 24; +/// The length of the authentication capability. +pub(crate) const CAPABILITY_LEN: usize = 32; +/// The length of the state record. +pub(crate) const STATE_RECORD_LEN: usize = 32; +/// The longest detail a response carries. +pub(crate) const MAX_DETAIL_LEN: usize = 512; +/// The bit that marks a response operation. +pub(crate) const RESPONSE: u8 = 0x80; + +/// A request operation. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub(crate) enum Operation { + /// Present the capability; must be the first request. + Authenticate = 1, + /// Report the run state. + Query = 2, + /// Pause the VM and hold guest time. + Pause = 3, + /// Resume a VM the host paused. + Resume = 4, +} + +impl Operation { + fn from_wire(value: u8) -> Option { + Some(match value { + 1 => Self::Authenticate, + 2 => Self::Query, + 3 => Self::Pause, + 4 => Self::Resume, + _ => return None, + }) + } + + /// The payload length a request of this operation carries. + fn request_payload_len(self) -> usize { + match self { + Self::Authenticate => CAPABILITY_LEN, + Self::Query | Self::Pause | Self::Resume => 0, + } + } +} + +/// The status of a response. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub(crate) enum Status { + /// The operation completed; the record is the resulting state. + Ok = 0, + /// A snapshot boundary or post-restore gate is active. Nothing changed; + /// retry later. + Busy = 1, + /// The operation was refused. The VM is in the recorded state. + Rejected = 2, + /// The operation failed and the VM state may be uncertain. + Failed = 3, +} + +/// The run state carried by a state record. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub(crate) enum WireState { + /// The VM worker did not answer. + Unknown = 0, + /// The guest is running. + Running = 1, + /// The host paused the guest and holds its time. + Paused = 2, + /// The VM is stopped for another reason. + Stopped = 3, + /// A snapshot boundary or post-restore gate is active. + Busy = 4, +} + +/// A malformed request. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub(crate) enum ProtocolError { + #[error("invalid state-control magic")] + Magic, + #[error("unsupported state-control protocol version {0}")] + Version(u16), + #[error("unknown state-control operation {0:#x}")] + Operation(u8), + #[error("state-control flags must be zero")] + Flags, + #[error("state-control request status must be zero")] + RequestStatus, + #[error("invalid state-control payload length {0}")] + Length(u32), +} + +/// A decoded request header. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub(crate) struct RequestHeader { + /// The operation. + pub operation: Operation, + /// The client-chosen sequence the response echoes. + pub sequence: u64, +} + +impl RequestHeader { + /// The payload length that follows the header. + pub fn payload_len(&self) -> usize { + self.operation.request_payload_len() + } +} + +/// The state record of a response. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub(crate) struct StateRecord { + /// The run state. + pub state: WireState, + /// How many times the VM entered or left the host-paused state. + pub transitions: u64, + /// The control-console broker instance ID of this VMM process. + pub instance_id: [u8; 16], +} + +impl StateRecord { + fn encode(&self) -> [u8; STATE_RECORD_LEN] { + let mut bytes = [0; STATE_RECORD_LEN]; + bytes[0] = self.state as u8; + bytes[8..16].copy_from_slice(&self.transitions.to_le_bytes()); + bytes[16..32].copy_from_slice(&self.instance_id); + bytes + } +} + +fn header(kind: u8, sequence: u64, status: u32, length: usize) -> [u8; HEADER_LEN] { + let mut bytes = [0; HEADER_LEN]; + bytes[0..4].copy_from_slice(&MAGIC); + bytes[4..6].copy_from_slice(&VERSION.to_le_bytes()); + bytes[6] = kind; + bytes[8..16].copy_from_slice(&sequence.to_le_bytes()); + bytes[16..20].copy_from_slice(&status.to_le_bytes()); + bytes[20..24].copy_from_slice(&(length as u32).to_le_bytes()); + bytes +} + +/// Decodes and validates a request header. +pub(crate) fn decode_request_header( + bytes: &[u8; HEADER_LEN], +) -> Result { + if bytes[0..4] != MAGIC { + return Err(ProtocolError::Magic); + } + let version = u16::from_le_bytes([bytes[4], bytes[5]]); + if version != VERSION { + return Err(ProtocolError::Version(version)); + } + let operation = Operation::from_wire(bytes[6]).ok_or(ProtocolError::Operation(bytes[6]))?; + if bytes[7] != 0 { + return Err(ProtocolError::Flags); + } + if bytes[16..20] != [0; 4] { + return Err(ProtocolError::RequestStatus); + } + let length = u32::from_le_bytes(bytes[20..24].try_into().unwrap()); + if length as usize != operation.request_payload_len() { + return Err(ProtocolError::Length(length)); + } + Ok(RequestHeader { + operation, + sequence: u64::from_le_bytes(bytes[8..16].try_into().unwrap()), + }) +} + +/// Truncates `detail` to at most [`MAX_DETAIL_LEN`] bytes on a character +/// boundary. +fn bounded_detail(detail: &str) -> &str { + if detail.len() <= MAX_DETAIL_LEN { + return detail; + } + let mut end = MAX_DETAIL_LEN; + while !detail.is_char_boundary(end) { + end -= 1; + } + &detail[..end] +} + +/// Encodes a response. +pub(crate) fn encode_response( + operation: Operation, + sequence: u64, + status: Status, + record: &StateRecord, + detail: &str, +) -> Vec { + let detail = bounded_detail(detail).as_bytes(); + let mut bytes = header( + operation as u8 | RESPONSE, + sequence, + status as u32, + STATE_RECORD_LEN + detail.len(), + ) + .to_vec(); + bytes.extend_from_slice(&record.encode()); + bytes.extend_from_slice(detail); + bytes +} + +/// A reference client codec for tests. +#[cfg(test)] +pub(crate) mod client { + use super::*; + + /// A decoded response. + #[derive(Debug, Clone, PartialEq, Eq)] + pub(crate) struct Response { + pub operation: Operation, + pub sequence: u64, + pub status: Status, + pub record: StateRecord, + pub detail: String, + } + + /// Encodes a request. `payload` is the capability of + /// [`Operation::Authenticate`] and empty otherwise. + pub(crate) fn encode_request(operation: Operation, sequence: u64, payload: &[u8]) -> Vec { + let mut bytes = header(operation as u8, sequence, 0, payload.len()).to_vec(); + bytes.extend_from_slice(payload); + bytes + } + + /// Decodes consecutive responses, failing on any malformed one. + pub(crate) fn decode_responses(mut bytes: &[u8]) -> Result, String> { + let mut responses = Vec::new(); + while !bytes.is_empty() { + let header = bytes.get(..HEADER_LEN).ok_or("truncated header")?; + if header[0..4] != MAGIC || header[4..6] != VERSION.to_le_bytes() || header[7] != 0 { + return Err("invalid header".into()); + } + let kind = header[6]; + if kind & RESPONSE == 0 { + return Err(format!("not a response: {kind:#x}")); + } + let operation = Operation::from_wire(kind & !RESPONSE).ok_or("unknown operation")?; + let status = match u32::from_le_bytes(header[16..20].try_into().unwrap()) { + 0 => Status::Ok, + 1 => Status::Busy, + 2 => Status::Rejected, + 3 => Status::Failed, + status => return Err(format!("unknown status {status}")), + }; + let length = u32::from_le_bytes(header[20..24].try_into().unwrap()) as usize; + if !(STATE_RECORD_LEN..=STATE_RECORD_LEN + MAX_DETAIL_LEN).contains(&length) { + return Err(format!("invalid length {length}")); + } + let payload = bytes + .get(HEADER_LEN..HEADER_LEN + length) + .ok_or("truncated payload")?; + let state = match payload[0] { + 0 => WireState::Unknown, + 1 => WireState::Running, + 2 => WireState::Paused, + 3 => WireState::Stopped, + 4 => WireState::Busy, + state => return Err(format!("unknown state {state}")), + }; + if payload[1..8] != [0; 7] { + return Err("nonzero reserved bytes".into()); + } + responses.push(Response { + operation, + sequence: u64::from_le_bytes(header[8..16].try_into().unwrap()), + status, + record: StateRecord { + state, + transitions: u64::from_le_bytes(payload[8..16].try_into().unwrap()), + instance_id: payload[16..32].try_into().unwrap(), + }, + detail: String::from_utf8(payload[STATE_RECORD_LEN..].to_vec()) + .map_err(|_| "detail is not UTF-8")?, + }); + bytes = &bytes[HEADER_LEN + length..]; + } + Ok(responses) + } +} + +#[cfg(test)] +mod tests { + use super::client::*; + use super::*; + + const INSTANCE: [u8; 16] = [7; 16]; + + fn record(state: WireState, transitions: u64) -> StateRecord { + StateRecord { + state, + transitions, + instance_id: INSTANCE, + } + } + + #[test] + fn request_header_has_the_documented_layout() { + let bytes = encode_request(Operation::Pause, 0x0102_0304_0506_0708, &[]); + assert_eq!( + bytes, + [ + b'N', b'V', b'X', b'V', 1, 0, 3, 0, 8, 7, 6, 5, 4, 3, 2, 1, 0, 0, 0, 0, 0, 0, 0, 0 + ] + ); + let header = decode_request_header(bytes.as_slice().try_into().unwrap()).unwrap(); + assert_eq!( + header, + RequestHeader { + operation: Operation::Pause, + sequence: 0x0102_0304_0506_0708, + } + ); + assert_eq!(header.payload_len(), 0); + + let auth = encode_request(Operation::Authenticate, 1, &[0x5a; CAPABILITY_LEN]); + assert_eq!(auth.len(), HEADER_LEN + CAPABILITY_LEN); + assert_eq!(&auth[20..24], &32u32.to_le_bytes()); + let header = decode_request_header(auth[..HEADER_LEN].try_into().unwrap()).unwrap(); + assert_eq!(header.payload_len(), CAPABILITY_LEN); + } + + #[test] + fn malformed_request_headers_are_rejected() { + let valid = encode_request(Operation::Query, 9, &[]); + let decode = |mutate: &dyn Fn(&mut Vec)| { + let mut bytes = valid.clone(); + mutate(&mut bytes); + decode_request_header(bytes.as_slice().try_into().unwrap()) + }; + assert_eq!(decode(&|b| b[0] = b'X'), Err(ProtocolError::Magic)); + assert_eq!(decode(&|b| b[4] = 2), Err(ProtocolError::Version(2))); + assert_eq!(decode(&|b| b[6] = 0), Err(ProtocolError::Operation(0))); + assert_eq!(decode(&|b| b[6] = 5), Err(ProtocolError::Operation(5))); + assert_eq!( + decode(&|b| b[6] = 2 | RESPONSE), + Err(ProtocolError::Operation(0x82)) + ); + assert_eq!(decode(&|b| b[7] = 1), Err(ProtocolError::Flags)); + assert_eq!(decode(&|b| b[16] = 1), Err(ProtocolError::RequestStatus)); + assert_eq!(decode(&|b| b[20] = 1), Err(ProtocolError::Length(1))); + // Authenticate must carry exactly the capability. + assert_eq!( + decode(&|b| b[6] = Operation::Authenticate as u8), + Err(ProtocolError::Length(0)) + ); + } + + #[test] + fn responses_have_the_documented_layout_and_bounded_detail() { + let bytes = encode_response( + Operation::Resume, + 42, + Status::Ok, + &record(WireState::Running, 2), + "", + ); + assert_eq!(bytes.len(), HEADER_LEN + STATE_RECORD_LEN); + assert_eq!(bytes[6], Operation::Resume as u8 | RESPONSE); + assert_eq!( + &bytes[HEADER_LEN..], + &[ + 1, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, + 7, 7, 7, 7 + ] + ); + assert_eq!( + decode_responses(&bytes).unwrap(), + [Response { + operation: Operation::Resume, + sequence: 42, + status: Status::Ok, + record: record(WireState::Running, 2), + detail: String::new(), + }] + ); + + let long = "é".repeat(MAX_DETAIL_LEN); + let bytes = encode_response( + Operation::Pause, + 1, + Status::Rejected, + &record(WireState::Running, 0), + &long, + ); + let response = decode_responses(&bytes).unwrap().remove(0); + assert_eq!(response.status, Status::Rejected); + assert_eq!(response.detail.len(), MAX_DETAIL_LEN); + assert!(long.starts_with(&response.detail)); + } +} diff --git a/openvmm/openvmm_entry/src/serial_io/microvm.rs b/openvmm/openvmm_entry/src/serial_io/microvm.rs index 833d1c609c9..d0022ea9700 100644 --- a/openvmm/openvmm_entry/src/serial_io/microvm.rs +++ b/openvmm/openvmm_entry/src/serial_io/microvm.rs @@ -18,6 +18,12 @@ use vm_resource::kind::SerialBackendHandle; #[cfg(target_os = "linux")] pub fn bind_control_serial(path: &Path) -> io::Result> { + Ok(bind_control_endpoint(path)?.into_resource()) +} + +/// Binds an owned, private Unix socket for a control endpoint. +#[cfg(target_os = "linux")] +pub fn bind_control_endpoint(path: &Path) -> io::Result { use std::os::unix::fs::FileTypeExt; use std::os::unix::fs::MetadataExt; use std::os::unix::fs::PermissionsExt; @@ -69,11 +75,19 @@ pub fn bind_control_serial(path: &Path) -> io::Result io::Result> { + Ok(bind_control_endpoint(path)?.into_resource()) +} + +/// Creates a protected named pipe for a control endpoint. +#[cfg(windows)] +pub fn bind_control_endpoint( + path: &Path, +) -> io::Result { use pal::windows::security::LocalSecurityDescriptor; use serial_socket::windows::OpenWindowsPipeSerialConfig; @@ -104,7 +118,7 @@ pub fn bind_control_serial(path: &Path) -> io::Result io::Result<()> { + // A pipe without a client fails to disconnect but can still listen. + let _ = pipe.disconnect_pipe(); + self.state = PipeState::Listening(ListeningPipe::new(&self.driver, pipe)?); + Ok(()) + } } impl From for Resource { @@ -162,10 +171,26 @@ impl SerialIo for WindowsPipeSerialBackend { match &mut self.state { PipeState::Done => Poll::Pending, PipeState::Listening(accept) => { - let file = ready!(accept.poll_unpin(cx)); - let file = file?; - self.peer_identity = Some(named_pipe_peer_identity(&file)?); - self.state = PipeState::Done; + let accepted = ready!(accept.poll_unpin(cx)); + let PipeState::Listening(accept) = + std::mem::replace(&mut self.state, PipeState::Done) + else { + unreachable!() + }; + let file = match accepted { + Ok(file) => file, + // A failed accept leaves the pipe with the listener. + Err(error) => { + return Poll::Ready(self.listen_again(accept.into_inner()).and(Err(error))); + } + }; + // Reject a client whose identity cannot be resolved, such as + // one that already exited, and listen for the next client. + let peer_identity = match named_pipe_peer_identity(&file) { + Ok(peer_identity) => peer_identity, + Err(error) => return Poll::Ready(self.listen_again(file).and(Err(error))), + }; + self.peer_identity = Some(peer_identity); self.state = PipeState::Connected(PolledPipe::new(&self.driver, file)?); Poll::Ready(Ok(())) } diff --git a/vm/devices/serial/serial_socket/src/windows_tests.rs b/vm/devices/serial/serial_socket/src/windows_tests.rs index 509ca3083b9..ad736cfb9a3 100644 --- a/vm/devices/serial/serial_socket/src/windows_tests.rs +++ b/vm/devices/serial/serial_socket/src/windows_tests.rs @@ -52,3 +52,77 @@ async fn reconnects_after_client_close(driver: DefaultDriver) { .unwrap(); poll_fn(|cx| backend.poll_connect(cx)).await.unwrap(); } + +/// Connects to the pipe that `SERIAL_SOCKET_TEST_PIPE` names and exits. +#[test] +fn exiting_client_child() { + let Some(path) = std::env::var_os("SERIAL_SOCKET_TEST_PIPE") else { + return; + }; + OpenOptions::new() + .read(true) + .write(true) + .open(path) + .unwrap(); +} + +#[async_test] +async fn accepts_the_next_client_after_an_unresolved_identity(driver: DefaultDriver) { + use std::os::windows::process::CommandExt; + use std::process::Command; + use std::process::Stdio; + use std::time::Duration; + use std::time::Instant; + + // Without a console, no console host keeps the exited client open. + const DETACHED_PROCESS: u32 = 0x0000_0008; + + let mut id = [0; 16]; + getrandom::fill(&mut id).unwrap(); + let path = format!(r#"\\.\pipe\{:0x}"#, u128::from_ne_bytes(id)); + let server = new_named_pipe( + &path, + GENERIC_READ | GENERIC_WRITE, + Disposition::Create, + PipeMode::Byte, + ) + .unwrap(); + let mut backend = WindowsPipeSerialBackend::new( + Box::new(driver.clone()), + OpenWindowsPipeSerialConfig::from(server), + ) + .unwrap(); + + // The client connects and exits before the backend resolves its identity. + let mut child = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "windows_tests::exiting_client_child"]) + .env("SERIAL_SOCKET_TEST_PIPE", &path) + .creation_flags(DETACHED_PROCESS) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let process_id = child.id(); + assert!(child.wait().unwrap().success()); + drop(child); + let deadline = Instant::now() + Duration::from_secs(10); + while pal::windows::security::user_sid::process_user_sid(process_id).is_ok() { + assert!( + Instant::now() < deadline, + "the exited client's identity stayed resolvable" + ); + std::thread::sleep(Duration::from_millis(10)); + } + poll_fn(|cx| backend.poll_connect(cx)).await.unwrap_err(); + assert!(!backend.is_connected()); + + // The backend rejected that client and listens for the next one. + let _next = OpenOptions::new() + .read(true) + .write(true) + .open(&path) + .unwrap(); + poll_fn(|cx| backend.poll_connect(cx)).await.unwrap(); + assert!(backend.local_peer_identity().unwrap().is_some()); +} diff --git a/vmm_core/src/partition_unit/snapshot.rs b/vmm_core/src/partition_unit/snapshot.rs index 9a5a0658758..80cdd74059b 100644 --- a/vmm_core/src/partition_unit/snapshot.rs +++ b/vmm_core/src/partition_unit/snapshot.rs @@ -2,8 +2,9 @@ // Licensed under the MIT License. //! Partition unit support for snapshots: validating the instantiated VP prefix, -//! stopping VPs at a deferred I/O boundary for capture, and the NVX time ABI's -//! LAPIC timer checks and restore advance. +//! stopping VPs at a deferred I/O boundary for capture, the NVX time ABI's +//! LAPIC timer checks and restore advance, and the LAPIC state that a host +//! pause holds while the vCPUs are stopped. use super::Error; use super::PartitionRequest; @@ -22,6 +23,10 @@ pub(super) enum SnapshotRequest { #[cfg(guest_arch = "x86_64")] AdvanceLapic(FailableRpc<(u64, u64), ()>), #[cfg(guest_arch = "x86_64")] + GetLapics(FailableRpc<(), Vec>), + #[cfg(guest_arch = "x86_64")] + SetLapics(FailableRpc, ()>), + #[cfg(guest_arch = "x86_64")] OmitSavedTsc(mesh::rpc::Rpc<(), ()>), } @@ -112,6 +117,31 @@ impl PartitionUnit { .await .unwrap(); } + + /// Reads the LAPIC state of every stopped vCPU, in VP order, so that a + /// host pause can set the same timer state again when it resumes. + #[cfg(guest_arch = "x86_64")] + pub async fn get_lapics(&mut self) -> anyhow::Result> { + Ok(self + .req_send + .call_failable( + |rpc| PartitionRequest::Snapshot(SnapshotRequest::GetLapics(rpc)), + (), + ) + .await?) + } + + /// Sets the LAPIC state of every stopped vCPU to `lapics`, in VP order. + #[cfg(guest_arch = "x86_64")] + pub async fn set_lapics(&mut self, lapics: Vec) -> anyhow::Result<()> { + self.req_send + .call_failable( + |rpc| PartitionRequest::Snapshot(SnapshotRequest::SetLapics(rpc)), + lapics, + ) + .await?; + Ok(()) + } } impl PartitionUnitRunner { @@ -142,6 +172,16 @@ impl PartitionUnitRunner { .await } #[cfg(guest_arch = "x86_64")] + SnapshotRequest::GetLapics(rpc) => { + rpc.handle_failable(async |()| self.vp_set.get_lapics().await) + .await + } + #[cfg(guest_arch = "x86_64")] + SnapshotRequest::SetLapics(rpc) => { + rpc.handle_failable(async |lapics| self.vp_set.set_lapics(lapics).await) + .await + } + #[cfg(guest_arch = "x86_64")] SnapshotRequest::OmitSavedTsc(rpc) => rpc.handle_sync(|()| { self.omit_saved_tsc = true; }), diff --git a/vmm_core/src/partition_unit/vp_set.rs b/vmm_core/src/partition_unit/vp_set.rs index 67935fe3ea5..a5dfe8a13a9 100644 --- a/vmm_core/src/partition_unit/vp_set.rs +++ b/vmm_core/src/partition_unit/vp_set.rs @@ -91,6 +91,14 @@ trait ControlVp: ProtobufSaveRestore { #[cfg(guest_arch = "x86_64")] fn advance_lapic_timer(&mut self, advance: time_abi::LapicAdvance) -> anyhow::Result<()>; + /// Reads the stopped vCPU's LAPIC state. + #[cfg(guest_arch = "x86_64")] + fn get_lapic(&mut self) -> anyhow::Result; + + /// Sets the stopped vCPU's LAPIC state. + #[cfg(guest_arch = "x86_64")] + fn set_lapic(&mut self, apic: &virt::x86::vp::Apic) -> anyhow::Result<()>; + #[cfg(feature = "gdb")] fn debug(&mut self) -> &mut dyn DebugVp; @@ -198,6 +206,16 @@ where advance.apply(self.vp) } + #[cfg(guest_arch = "x86_64")] + fn get_lapic(&mut self) -> anyhow::Result { + time_abi::get_lapic(self.vp) + } + + #[cfg(guest_arch = "x86_64")] + fn set_lapic(&mut self, apic: &virt::x86::vp::Apic) -> anyhow::Result<()> { + time_abi::set_lapic(self.vp, apic) + } + fn inspect_vp( &mut self, vtl_guest_memory: &[Option; NUM_VTLS], @@ -1137,6 +1155,10 @@ enum StateEvent { CheckTimers(mesh::rpc::FailableRpc<(), ()>), #[cfg(guest_arch = "x86_64")] AdvanceLapic(mesh::rpc::FailableRpc), + #[cfg(guest_arch = "x86_64")] + GetLapic(mesh::rpc::FailableRpc<(), virt::x86::vp::Apic>), + #[cfg(guest_arch = "x86_64")] + SetLapic(mesh::rpc::FailableRpc, ()>), #[cfg(feature = "dump")] GetDumpVpState(Rpc>), #[cfg(feature = "gdb")] @@ -1391,6 +1413,10 @@ impl RunnerInner { StateEvent::AdvanceLapic(rpc) => { rpc.handle_failable_sync(|advance| vp.advance_lapic_timer(advance)) } + #[cfg(guest_arch = "x86_64")] + StateEvent::GetLapic(rpc) => rpc.handle_failable_sync(|()| vp.get_lapic()), + #[cfg(guest_arch = "x86_64")] + StateEvent::SetLapic(rpc) => rpc.handle_failable_sync(|apic| vp.set_lapic(&apic)), #[cfg(feature = "dump")] StateEvent::GetDumpVpState(rpc) => rpc.handle_sync(|vtl| vp.get_dump_vp_state(vtl)), #[cfg(feature = "gdb")] diff --git a/vmm_core/src/partition_unit/vp_set/boundary.rs b/vmm_core/src/partition_unit/vp_set/boundary.rs index aecba8b75ae..9b95035282b 100644 --- a/vmm_core/src/partition_unit/vp_set/boundary.rs +++ b/vmm_core/src/partition_unit/vp_set/boundary.rs @@ -47,12 +47,13 @@ impl VpSet { } #[cfg(test)] -mod tests { +pub(super) mod tests { use crate::partition_unit::vp_set::*; use test_with_tracing::test; use vm_topology::processor::VpInfo; - fn vp_set(count: u32) -> (VpSet, Vec) { + /// Creates a stopped set of `count` VPs and their runners. + pub(in crate::partition_unit::vp_set) fn vp_set(count: u32) -> (VpSet, Vec) { let (halt, _halt_recv) = Halt::new(); let mut vps = VpSet::new([None, None, None], Arc::new(halt), count as usize); let runners = (0..count) diff --git a/vmm_core/src/partition_unit/vp_set/time_abi.rs b/vmm_core/src/partition_unit/vp_set/time_abi.rs index 19989453748..9f05be8ff9e 100644 --- a/vmm_core/src/partition_unit/vp_set/time_abi.rs +++ b/vmm_core/src/partition_unit/vp_set/time_abi.rs @@ -1,7 +1,8 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -//! LAPIC timer checks and advancement for an NVX time ABI snapshot. +//! LAPIC timer checks and advancement for an NVX time ABI snapshot, and the +//! LAPIC state that a host pause holds while the vCPUs are stopped. use super::StateEvent; use super::VpEvent; @@ -50,6 +51,25 @@ impl LapicAdvance { } } +/// Reads the LAPIC state of the stopped VP `vp`. +pub(super) fn get_lapic(vp: &mut impl Processor) -> anyhow::Result { + vp.access_state(Vtl::Vtl0) + .apic() + .context("failed to read stopped LAPIC") +} + +/// Sets the LAPIC state of the stopped VP `vp` to `apic`. Like +/// [`LapicAdvance::apply`], this makes KVM derive its timer deadline from the +/// current guest TSC. +pub(super) fn set_lapic(vp: &mut impl Processor, apic: &virt::x86::vp::Apic) -> anyhow::Result<()> { + let mut access = vp.access_state(Vtl::Vtl0); + access + .set_apic(apic) + .context("failed to set stopped LAPIC")?; + access.commit().context("failed to commit stopped LAPIC")?; + Ok(()) +} + impl VpSet { /// Checks the LAPIC timer of every stopped VP: capture step 1 of the time /// ABI. Restore needs no separate check, because the LAPIC advance @@ -106,4 +126,120 @@ impl VpSet { .await?; Ok(()) } + + /// Reads the LAPIC state of every stopped VP, in VP order. + pub async fn get_lapics(&mut self) -> anyhow::Result> { + anyhow::ensure!( + !self.started, + "vCPUs must be stopped before reading LAPIC state" + ); + self.vps + .iter() + .enumerate() + .map(|(index, vp)| async move { + vp.send + .call_failable(|rpc| VpEvent::State(StateEvent::GetLapic(rpc)), ()) + .await + .with_context(|| format!("vp{index} LAPIC read")) + }) + .collect::>() + .await + } + + /// Sets the LAPIC state of every stopped VP to `lapics`, in VP order. + pub async fn set_lapics(&mut self, lapics: Vec) -> anyhow::Result<()> { + anyhow::ensure!( + !self.started, + "vCPUs must be stopped before setting LAPIC state" + ); + anyhow::ensure!( + lapics.len() == self.vps.len(), + "{} LAPIC states were supplied for {} vCPUs", + lapics.len(), + self.vps.len() + ); + self.vps + .iter() + .zip(lapics) + .enumerate() + .map(|(index, (vp, apic))| async move { + vp.send + .call_failable( + |rpc| VpEvent::State(StateEvent::SetLapic(rpc)), + Box::new(apic), + ) + .await + .with_context(|| format!("vp{index} LAPIC set")) + }) + .collect::>() + .await?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::super::boundary::tests::vp_set; + use super::*; + use futures::FutureExt as _; + use test_with_tracing::test; + use virt::x86::vp::Apic; + + fn apic(index: u32) -> Apic { + Apic { + apic_base: 0xfee0_0900, + registers: [index; 64], + auto_eoi: [0; 8], + } + } + + #[test] + fn lapic_state_is_read_and_set_in_vp_order() { + let (mut vps, mut runners) = vp_set(2); + + let mut get = Box::pin(vps.get_lapics()); + assert!(get.as_mut().now_or_never().is_none()); + for (index, runner) in runners.iter_mut().enumerate().rev() { + let Ok(VpEvent::State(StateEvent::GetLapic(rpc))) = runner.recv.try_recv() else { + panic!("expected a LAPIC read"); + }; + rpc.complete(Ok(apic(index as u32))); + } + assert_eq!(get.now_or_never().unwrap().unwrap(), [apic(0), apic(1)]); + + let mut set = Box::pin(vps.set_lapics(vec![apic(0), apic(1)])); + assert!(set.as_mut().now_or_never().is_none()); + for (index, runner) in runners.iter_mut().enumerate() { + let Ok(VpEvent::State(StateEvent::SetLapic(rpc))) = runner.recv.try_recv() else { + panic!("expected a LAPIC set"); + }; + let (state, response) = rpc.split(); + assert_eq!(*state, apic(index as u32)); + response.complete(Ok(())); + } + set.now_or_never().unwrap().unwrap(); + } + + #[test] + fn lapic_state_requires_stopped_vps_and_one_state_per_vp() { + let (mut vps, mut runners) = vp_set(2); + assert!( + vps.set_lapics(vec![apic(0)]) + .now_or_never() + .unwrap() + .is_err() + ); + vps.start(); + assert!(vps.get_lapics().now_or_never().unwrap().is_err()); + assert!( + vps.set_lapics(vec![apic(0), apic(1)]) + .now_or_never() + .unwrap() + .is_err() + ); + for runner in &mut runners { + assert!(matches!(runner.recv.try_recv(), Ok(VpEvent::Start))); + assert!(runner.recv.try_recv().is_err()); + } + } } diff --git a/vmm_core/virt/src/time_abi/backend.rs b/vmm_core/virt/src/time_abi/backend.rs index 2721eb20c90..a2f7af379e2 100644 --- a/vmm_core/virt/src/time_abi/backend.rs +++ b/vmm_core/virt/src/time_abi/backend.rs @@ -90,7 +90,9 @@ pub trait TimeAbiBackend: Send + Sync { /// /// Core calls it once per restore, after every instantiated VP is bound /// (which creates it on MSHV) and its saved state is restored, and before - /// any restored VP runs. A cold boot never calls it. + /// any restored VP runs. A cold boot never calls it before the guest runs. + /// Core also calls it at every resume from a host pause, with every VP + /// stopped, to set back the TSC that the pause held. fn set_synchronized_tsc( &self, target: &mut dyn FnMut(&HostTimeSample) -> Result, From 647e635c308b2ca463843153116582c4192d0e1c Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Mon, 5 Oct 2026 23:18:03 -0700 Subject: [PATCH 2/2] Guide: describe guest wall-clock repair after a host pause precisely The NVX guest's wall-clock discipline reads the time ABI's time sample, not the CMOS RTC. Both follow host UTC across a host pause; say so, and that the guest steps its clock from such a source at its next poll. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- Guide/src/reference/openvmm/management/cli.md | 6 +++--- .../openvmm/management/state_control_protocol.md | 11 ++++++----- .../openvmm_core/src/worker/dispatch/host_pause.rs | 4 ++-- 3 files changed, 11 insertions(+), 10 deletions(-) diff --git a/Guide/src/reference/openvmm/management/cli.md b/Guide/src/reference/openvmm/management/cli.md index 4a392e08011..9d62d1ffceb 100644 --- a/Guide/src/reference/openvmm/management/cli.md +++ b/Guide/src/reference/openvmm/management/cli.md @@ -864,9 +864,9 @@ Serial devices can be configured to appear as different devices inside the guest The endpoint serves one host at a time, and it answers while the VM is paused. A pause holds guest monotonic time, so the guest observes no elapsed - time across it. Wall-clock time is not held: the CMOS RTC follows host UTC, - and the guest's system clock is behind host UTC after a resume until the - guest steps it. A pause is busy while a snapshot boundary or post-restore + time across it. Wall-clock time is not held: the CMOS RTC and the time + sample follow host UTC, and the guest's system clock is behind host UTC + after a resume until the guest steps it. A pause is busy while a snapshot boundary or post-restore gate is active, and is rejected when a vCPU has an armed periodic or TSC-deadline LAPIC timer. diff --git a/Guide/src/reference/openvmm/management/state_control_protocol.md b/Guide/src/reference/openvmm/management/state_control_protocol.md index aa40fa1c32c..f9f91d5b0bf 100644 --- a/Guide/src/reference/openvmm/management/state_control_protocol.md +++ b/Guide/src/reference/openvmm/management/state_control_protocol.md @@ -117,11 +117,12 @@ LAPIC timers, and the VM time that drives emulated timers, such as the PIT and the RTC's periodic and alarm interrupts, all stop with the vCPUs and continue from the same values. -Wall-clock time is not held. The CMOS RTC's date and time follow host UTC, so -after a resume they read the current host time. The guest's system clock is -behind host UTC by the paused duration until the guest steps it, for example -from the RTC. The NVX guest's wall-clock discipline does this at its next -poll. +Wall-clock time is not held. Host UTC keeps running, and so do the guest-visible +sources that follow it: the CMOS RTC's date and time, and the time sample on the +portb status port. After a resume, the guest's system clock is behind host UTC +by the paused duration until the guest steps it from such a source. The NVX +guest's wall-clock discipline, which reads the time sample, steps it at its +next poll. ## Guest activity while paused diff --git a/openvmm/openvmm_core/src/worker/dispatch/host_pause.rs b/openvmm/openvmm_core/src/worker/dispatch/host_pause.rs index 9a6066a9936..60a00bd5862 100644 --- a/openvmm/openvmm_core/src/worker/dispatch/host_pause.rs +++ b/openvmm/openvmm_core/src/worker/dispatch/host_pause.rs @@ -8,8 +8,8 @@ //! stopped. Resume sets every VP's TSC back to the held value with the //! backend's synchronized TSC set, and sets the held LAPIC state again, before //! any vCPU runs. VM time stops and starts with the state units. Host UTC, -//! which the CMOS RTC follows, keeps running, so the guest's wall-clock -//! discipline steps `CLOCK_REALTIME` at its next poll. +//! which the CMOS RTC and the time sample follow, keeps running, so the +//! guest's wall-clock discipline steps `CLOCK_REALTIME` at its next poll. //! //! A guest reset processed while the host holds the pause reinitializes guest //! time, so it discards the held time but keeps the pause: the reset VM starts