petri/vm/openvmm/
runtime.rs

1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4//! Methods to interact with a running [`PetriVmOpenVmm`].
5
6use super::PetriVmResourcesOpenVmm;
7use crate::OpenHclServicingFlags;
8use crate::PetriHaltReason;
9use crate::PetriHaltReasonDetail;
10use crate::PetriVmFramebufferAccess;
11use crate::PetriVmInspector;
12use crate::PetriVmRuntime;
13use crate::ShutdownKind;
14use crate::VmScreenshotMeta;
15use crate::openhcl_diag::OpenHclDiagHandler;
16use crate::worker::Worker;
17use anyhow::Context;
18use async_trait::async_trait;
19use framebuffer::View;
20use futures::FutureExt;
21use futures_concurrency::future::Race;
22use get_resources::ged::FirmwareEvent;
23use hyperv_ic_resources::shutdown::ShutdownRpc;
24use mesh::CancelContext;
25use mesh::Receiver;
26use mesh::RecvError;
27use mesh::rpc::RpcError;
28use mesh::rpc::RpcSend;
29use mesh_process::Mesh;
30use openvmm_defs::rpc::PulseSaveRestoreError;
31use pal_async::socket::PolledSocket;
32use petri_artifacts_core::ResolvedArtifact;
33#[cfg(target_os = "linux")]
34use pipette_client::PIPETTE_PORT;
35use pipette_client::PipetteClient;
36use std::future::Future;
37use std::path::Path;
38use std::sync::Arc;
39use std::time::Duration;
40use vmm_core_defs::HaltReason;
41#[cfg(target_os = "linux")]
42use vmsocket::VmAddress;
43#[cfg(target_os = "linux")]
44use vmsocket::VmSocket;
45use vtl2_settings_proto::Vtl2Settings;
46
47/// A running VM that tests can interact with.
48// DEVNOTE: Really the PetriVmInner is the actual VM and channels that we interact
49// with. This struct exists as a wrapper to provide error handling, such as not
50// hanging indefinitely when waiting on certain channels if the VM crashes.
51pub struct PetriVmOpenVmm {
52    inner: PetriVmInner,
53    halt: PetriVmHaltReceiver,
54}
55
56#[async_trait]
57impl PetriVmRuntime for PetriVmOpenVmm {
58    type VmInspector = OpenVmmInspector;
59    type VmFramebufferAccess = OpenVmmFramebufferAccess;
60
61    async fn teardown(self) -> anyhow::Result<()> {
62        tracing::info!("waiting for worker");
63        let worker = Arc::into_inner(self.inner.worker)
64            .context("all references to the OpenVMM worker have not been closed")?;
65        worker.shutdown().await?;
66
67        tracing::info!("Worker quit, waiting for mesh");
68        self.inner.mesh.shutdown().await;
69
70        tracing::info!("Mesh shutdown, waiting for logging tasks");
71        for t in self.inner.resources.log_stream_tasks {
72            t.await?;
73        }
74
75        Ok(())
76    }
77
78    async fn wait_for_halt(&mut self, allow_reset: bool) -> anyhow::Result<PetriHaltReasonDetail> {
79        let halt_reason = if let Some(already) = self.halt.already_received.take() {
80            already.map_err(anyhow::Error::from)
81        } else {
82            self.halt
83                .halt_notif
84                .recv()
85                .await
86                .context("Failed to get halt reason")
87        }?;
88
89        tracing::info!(?halt_reason, "Got halt reason");
90
91        let reason = match halt_reason {
92            HaltReason::PowerOff => PetriHaltReason::PowerOff,
93            HaltReason::Reset => PetriHaltReason::Reset,
94            HaltReason::Hibernate => PetriHaltReason::Hibernate,
95            HaltReason::TripleFault { .. } => PetriHaltReason::TripleFault,
96            _ => PetriHaltReason::Other,
97        };
98
99        if allow_reset && reason == PetriHaltReason::Reset {
100            self.reset().await?
101        }
102
103        Ok(PetriHaltReasonDetail {
104            reason,
105            detail: format!("{halt_reason:?}"),
106        })
107    }
108
109    async fn wait_for_agent(&mut self, set_high_vtl: bool) -> anyhow::Result<PipetteClient> {
110        Self::wait_for_agent(self, set_high_vtl).await
111    }
112
113    fn openhcl_diag(&self) -> Option<OpenHclDiagHandler> {
114        self.inner.resources.vtl2_vsock_path.as_ref().map(|path| {
115            OpenHclDiagHandler::new(diag_client::DiagClient::from_hybrid_vsock(
116                self.inner.resources.driver.clone(),
117                path,
118            ))
119        })
120    }
121
122    async fn wait_for_boot_event(
123        &mut self,
124        timeout: Option<Duration>,
125    ) -> anyhow::Result<Option<FirmwareEvent>> {
126        // The event arrives on a channel, so cancelling the wait can't discard
127        // one that has already been delivered.
128        CancelContext::new()
129            .with_timeout(timeout.unwrap_or(Duration::MAX))
130            .until_cancelled(Self::wait_for_boot_event(self))
131            .await
132            .ok()
133            .transpose()
134    }
135
136    async fn wait_for_enlightened_shutdown_ready(&mut self) -> anyhow::Result<()> {
137        Self::wait_for_enlightened_shutdown_ready(self)
138            .await
139            .map(|_| ())
140    }
141
142    async fn send_enlightened_shutdown(&mut self, kind: ShutdownKind) -> anyhow::Result<()> {
143        Self::send_enlightened_shutdown(self, kind).await
144    }
145
146    async fn restart_openhcl(
147        &mut self,
148        new_openhcl: &ResolvedArtifact,
149        flags: OpenHclServicingFlags,
150    ) -> anyhow::Result<()> {
151        Self::save_openhcl(self, new_openhcl, flags).await?;
152        Self::restore_openhcl(self).await
153    }
154
155    async fn save_openhcl(
156        &mut self,
157        new_openhcl: &ResolvedArtifact,
158        flags: OpenHclServicingFlags,
159    ) -> anyhow::Result<()> {
160        Self::save_openhcl(self, new_openhcl, flags).await
161    }
162
163    async fn restore_openhcl(&mut self) -> anyhow::Result<()> {
164        Self::restore_openhcl(self).await
165    }
166
167    async fn update_command_line(&mut self, command_line: &str) -> anyhow::Result<()> {
168        Self::update_command_line(self, command_line).await
169    }
170
171    fn inspector(&self) -> Option<OpenVmmInspector> {
172        Some(OpenVmmInspector {
173            worker: self.inner.worker.clone(),
174        })
175    }
176
177    fn take_framebuffer_access(&mut self) -> Option<OpenVmmFramebufferAccess> {
178        self.inner
179            .framebuffer_view
180            .take()
181            .map(|view| OpenVmmFramebufferAccess { view })
182    }
183
184    async fn reset(&mut self) -> anyhow::Result<()> {
185        Self::reset(self).await
186    }
187
188    async fn set_vtl2_settings(&mut self, settings: &Vtl2Settings) -> anyhow::Result<()> {
189        Self::set_vtl2_settings(self, settings).await
190    }
191
192    async fn set_vmbus_drive(
193        &mut self,
194        _disk: &crate::Drive,
195        _controller_id: &guid::Guid,
196        _controller_location: u32,
197    ) -> anyhow::Result<()> {
198        todo!("openvmm set vmbus drive")
199    }
200
201    async fn add_pcie_device(
202        &mut self,
203        port_name: String,
204        resource: vm_resource::Resource<vm_resource::kind::PciDeviceHandleKind>,
205    ) -> anyhow::Result<()> {
206        Self::add_pcie_device(self, port_name, resource).await
207    }
208
209    async fn remove_pcie_device(&mut self, port_name: String) -> anyhow::Result<()> {
210        Self::remove_pcie_device(self, port_name).await
211    }
212}
213
214pub(super) struct PetriVmInner {
215    pub(super) resources: PetriVmResourcesOpenVmm,
216    pub(super) mesh: Mesh,
217    pub(super) worker: Arc<Worker>,
218    pub(super) framebuffer_view: Option<View>,
219    /// Whether CIDATA has already been mounted inside the guest.
220    /// Used to skip re-mounting after save/restore (where guest state is
221    /// preserved) while still mounting after a full reset/reboot.
222    pub(super) cidata_mounted: bool,
223    /// Resolved TCP pipette port for no-vmbus Windows guests. Set once
224    /// during startup and reused across reconnections (e.g. after reset).
225    pub(super) tcp_pipette_port: Option<u16>,
226    pub(super) pid: i32,
227}
228
229struct PetriVmHaltReceiver {
230    halt_notif: Receiver<HaltReason>,
231    already_received: Option<Result<HaltReason, RecvError>>,
232}
233
234// Wrap a PetriVmInner function in [`PetriVmOpenVmm::wait_for_halt_or_internal`] to
235// provide better error handling.
236macro_rules! petri_vm_fn {
237    ($(#[$($attrss:tt)*])* $vis:vis async fn $fn_name:ident (&mut self $(,$arg:ident: $ty:ty)*) $(-> $ret:ty)?) => {
238        $(#[$($attrss)*])*
239        $vis async fn $fn_name(&mut self, $($arg:$ty,)*) $(-> $ret)? {
240            Self::wait_for_halt_or_internal(&mut self.halt, self.inner.$fn_name($($arg,)*)).await
241        }
242    };
243}
244
245// TODO: Add all runtime functions that are not backend specific
246// to the `PetriVmRuntime` trait
247impl PetriVmOpenVmm {
248    pub(super) fn new(inner: PetriVmInner, halt_notif: Receiver<HaltReason>) -> Self {
249        Self {
250            inner,
251            halt: PetriVmHaltReceiver {
252                halt_notif,
253                already_received: None,
254            },
255        }
256    }
257
258    /// Get the path to the VTL 2 vsock socket, if the VM is configured with OpenHCL.
259    pub fn vtl2_vsock_path(&self) -> anyhow::Result<&Path> {
260        self.inner
261            .resources
262            .vtl2_vsock_path
263            .as_deref()
264            .context("VM is not configured with OpenHCL")
265    }
266
267    /// Get the PID of the openvmm child process.
268    pub fn pid(&self) -> i32 {
269        self.inner.pid
270    }
271
272    petri_vm_fn!(
273        /// Waits for an event emitted by the firmware about its boot status, and
274        /// returns that status.
275        pub async fn wait_for_boot_event(&mut self) -> anyhow::Result<FirmwareEvent>
276    );
277    petri_vm_fn!(
278        /// Waits for the Hyper-V shutdown IC to be ready, returning a receiver
279        /// that will be closed when it is no longer ready. Returns `None` if
280        /// the shutdown IC is not configured.
281        pub async fn wait_for_enlightened_shutdown_ready(&mut self) -> anyhow::Result<Option<mesh::OneshotReceiver<()>>>
282    );
283    petri_vm_fn!(
284        /// Instruct the guest to shutdown via the Hyper-V shutdown IC.
285        pub async fn send_enlightened_shutdown(&mut self, kind: ShutdownKind) -> anyhow::Result<()>
286    );
287    petri_vm_fn!(
288        /// Waits for the KVP IC to be ready, returning a sender that can be used
289        /// to send requests to it.
290        pub async fn wait_for_kvp(&mut self) -> anyhow::Result<mesh::Sender<hyperv_ic_resources::kvp::KvpRpc>>
291    );
292    petri_vm_fn!(
293        /// Stages the new OpenHCL file and saves the existing state.
294        pub async fn save_openhcl(
295            &mut self,
296            new_openhcl: &ResolvedArtifact,
297            flags: OpenHclServicingFlags
298        ) -> anyhow::Result<()>
299    );
300    petri_vm_fn!(
301        /// Restores OpenHCL from a previously saved state.
302        pub async fn restore_openhcl(
303            &mut self
304        ) -> anyhow::Result<()>
305    );
306    petri_vm_fn!(
307        /// Updates the command line parameters of the running VM.
308        pub async fn update_command_line(
309            &mut self,
310            command_line: &str
311        ) -> anyhow::Result<()>
312    );
313
314    petri_vm_fn!(
315        /// Hot-add a PCIe device to a named port at runtime.
316        pub async fn add_pcie_device(
317            &mut self,
318            port_name: String,
319            resource: vm_resource::Resource<vm_resource::kind::PciDeviceHandleKind>
320        ) -> anyhow::Result<()>
321    );
322    petri_vm_fn!(
323        /// Hot-remove a PCIe device from a named port at runtime.
324        pub async fn remove_pcie_device(
325            &mut self,
326            port_name: String
327        ) -> anyhow::Result<()>
328    );
329    petri_vm_fn!(
330        /// Resets the hardware state of the VM, simulating a power cycle.
331        pub async fn reset(&mut self) -> anyhow::Result<()>
332    );
333    petri_vm_fn!(
334        /// Dumps the VM's processor and memory state to a `.vmrs` file at `path`.
335        pub async fn dump_state(&mut self, path: &Path) -> anyhow::Result<()>
336    );
337    petri_vm_fn!(
338        /// Wait for a connection from a pipette agent
339        pub async fn wait_for_agent(&mut self, set_high_vtl: bool) -> anyhow::Result<PipetteClient>
340    );
341    petri_vm_fn!(
342        /// Set the OpenHCL VTL2 settings.
343        pub async fn set_vtl2_settings(&mut self, settings: &Vtl2Settings) -> anyhow::Result<()>
344    );
345
346    petri_vm_fn!(
347        /// Pause the VM. Call [`resume`](Self::resume) to continue execution.
348        pub async fn pause(&mut self) -> anyhow::Result<()>
349    );
350    petri_vm_fn!(
351        /// Save the VM's device and processor state, returning the serialized
352        /// bytes. The VM should be paused before calling this.
353        pub async fn save_state(&mut self) -> anyhow::Result<Vec<u8>>
354    );
355    petri_vm_fn!(
356        /// Resume a paused VM.
357        pub async fn resume(&mut self) -> anyhow::Result<()>
358    );
359    petri_vm_fn!(
360        /// Perform a pulse save/restore cycle: pause the VM, save all state,
361        /// reset, restore, and resume. Useful for verifying that device state
362        /// survives a save/restore round-trip.
363        pub async fn verify_save_restore(&mut self) -> anyhow::Result<()>
364    );
365    petri_vm_fn!(pub(crate) async fn launch_linux_direct_pipette(&mut self) -> anyhow::Result<()>);
366
367    /// Wrap the provided future in a race with the worker process's halt
368    /// notification channel. This is useful for preventing a future from
369    /// waiting indefinitely if the VM dies for any reason. If the worker
370    /// process crashes the halt notification channel will return an error, and
371    /// if the VM halts for any other reason the future will complete with that
372    /// reason.
373    pub async fn wait_for_halt_or<T, F: Future<Output = anyhow::Result<T>>>(
374        &mut self,
375        future: F,
376    ) -> anyhow::Result<T> {
377        Self::wait_for_halt_or_internal(&mut self.halt, future).await
378    }
379
380    async fn wait_for_halt_or_internal<T, F: Future<Output = anyhow::Result<T>>>(
381        halt: &mut PetriVmHaltReceiver,
382        future: F,
383    ) -> anyhow::Result<T> {
384        let future = &mut std::pin::pin!(future);
385        enum Either<T> {
386            Future(anyhow::Result<T>),
387            Halt(Result<HaltReason, RecvError>),
388        }
389        let res = (
390            future.map(Either::Future),
391            halt.halt_notif.recv().map(Either::Halt),
392        )
393            .race()
394            .await;
395
396        match res {
397            Either::Future(Ok(success)) => Ok(success),
398            Either::Future(Err(e)) => {
399                tracing::warn!(
400                    ?e,
401                    "Future returned with an error, sleeping for 5 seconds to let outstanding work finish"
402                );
403                let mut c = CancelContext::new().with_timeout(Duration::from_secs(5));
404                c.cancelled().await;
405                Err(e)
406            }
407            Either::Halt(halt_result) => {
408                tracing::warn!(
409                    halt_result = format_args!("{:x?}", halt_result),
410                    "Halt channel returned while waiting for other future, sleeping for 5 seconds to let outstanding work finish"
411                );
412                let mut c = CancelContext::new().with_timeout(Duration::from_secs(5));
413                let try_again = c.until_cancelled(future).await;
414
415                match try_again {
416                    Ok(fut_result) => {
417                        halt.already_received = Some(halt_result);
418                        if let Err(e) = &fut_result {
419                            tracing::warn!(
420                                ?e,
421                                "Future returned with an error, sleeping for 5 seconds to let outstanding work finish"
422                            );
423                            let mut c = CancelContext::new().with_timeout(Duration::from_secs(5));
424                            c.cancelled().await;
425                        }
426                        fut_result
427                    }
428                    Err(_cancel) => match halt_result {
429                        Ok(halt_reason) => Err(anyhow::anyhow!("VM halted: {:x?}", halt_reason)),
430                        Err(e) => Err(e).context("VM disappeared"),
431                    },
432                }
433            }
434        }
435    }
436}
437
438impl PetriVmInner {
439    async fn wait_for_boot_event(&mut self) -> anyhow::Result<FirmwareEvent> {
440        self.resources
441            .firmware_event_recv
442            .recv()
443            .await
444            .context("Failed to get firmware boot event")
445    }
446
447    async fn wait_for_enlightened_shutdown_ready(
448        &mut self,
449    ) -> anyhow::Result<Option<mesh::OneshotReceiver<()>>> {
450        let Some(send) = self.resources.shutdown_ic_send.as_ref() else {
451            return Ok(None);
452        };
453        let recv = send
454            .call(ShutdownRpc::WaitReady, ())
455            .await
456            .context("waiting for shutdown IC to be ready")?;
457        Ok(Some(recv))
458    }
459
460    async fn send_enlightened_shutdown(&mut self, kind: ShutdownKind) -> anyhow::Result<()> {
461        let send = self
462            .resources
463            .shutdown_ic_send
464            .as_ref()
465            .context("shutdown IC not configured")?;
466        let shutdown_result = send
467            .call(
468                ShutdownRpc::Shutdown,
469                hyperv_ic_resources::shutdown::ShutdownParams {
470                    shutdown_type: match kind {
471                        ShutdownKind::Shutdown => {
472                            hyperv_ic_resources::shutdown::ShutdownType::PowerOff
473                        }
474                        ShutdownKind::Reboot => hyperv_ic_resources::shutdown::ShutdownType::Reboot,
475                    },
476                    force: false,
477                },
478            )
479            .await?;
480
481        tracing::info!(?shutdown_result, "Shutdown sent");
482        anyhow::ensure!(
483            shutdown_result == hyperv_ic_resources::shutdown::ShutdownResult::Ok,
484            "Got non-Ok shutdown response"
485        );
486
487        Ok(())
488    }
489
490    async fn wait_for_kvp(
491        &mut self,
492    ) -> anyhow::Result<mesh::Sender<hyperv_ic_resources::kvp::KvpRpc>> {
493        tracing::info!("Waiting for KVP IC");
494        let send = self
495            .resources
496            .kvp_ic_send
497            .as_ref()
498            .context("KVP IC not configured")?;
499        let (send, _) = send
500            .call_failable(hyperv_ic_resources::kvp::KvpConnectRpc::WaitForGuest, ())
501            .await
502            .context("failed to connect to KVP IC")?;
503
504        Ok(send)
505    }
506
507    async fn save_openhcl(
508        &self,
509        new_openhcl: &ResolvedArtifact,
510        flags: OpenHclServicingFlags,
511    ) -> anyhow::Result<()> {
512        let ged_send = self
513            .resources
514            .ged_send
515            .as_ref()
516            .context("openhcl not configured")?;
517
518        let igvm_file = fs_err::File::open(new_openhcl).context("failed to open igvm file")?;
519        self.worker
520            .save_openhcl(ged_send, flags, igvm_file.into())
521            .await
522    }
523
524    async fn update_command_line(&mut self, command_line: &str) -> anyhow::Result<()> {
525        self.worker.update_command_line(command_line).await
526    }
527
528    async fn add_pcie_device(
529        &mut self,
530        port_name: String,
531        resource: vm_resource::Resource<vm_resource::kind::PciDeviceHandleKind>,
532    ) -> anyhow::Result<()> {
533        self.worker.add_pcie_device(port_name, resource).await
534    }
535
536    async fn remove_pcie_device(&mut self, port_name: String) -> anyhow::Result<()> {
537        self.worker.remove_pcie_device(port_name).await
538    }
539    async fn dump_state(&mut self, path: &Path) -> anyhow::Result<()> {
540        self.worker.dump_state(path).await
541    }
542    async fn restore_openhcl(&self) -> anyhow::Result<()> {
543        let ged_send = self
544            .resources
545            .ged_send
546            .as_ref()
547            .context("openhcl not configured")?;
548
549        self.worker.restore_openhcl(ged_send).await
550    }
551
552    async fn set_vtl2_settings(&self, settings: &Vtl2Settings) -> anyhow::Result<()> {
553        let ged_send = self
554            .resources
555            .ged_send
556            .as_ref()
557            .context("openhcl not configured")?;
558
559        ged_send
560            .call_failable(
561                get_resources::ged::GuestEmulationRequest::ModifyVtl2Settings,
562                prost::Message::encode_to_vec(settings),
563            )
564            .await?;
565
566        Ok(())
567    }
568
569    async fn reset(&mut self) -> anyhow::Result<()> {
570        tracing::info!("Resetting VM");
571        self.worker.reset().await?;
572        // Discard any firmware events from the boot that was abandoned, so
573        // that they aren't mistaken for the results of the new boot.
574        while self.resources.firmware_event_recv.try_recv().is_ok() {}
575        // Guest state is lost on reset, so CIDATA needs to be remounted.
576        self.cidata_mounted = false;
577        // On linux direct, pipette won't auto-start unless it is the init
578        // process. When it isn't, restart it over serial. (When pipette runs
579        // as PID 1 via rdinit=/pipette, linux_direct_serial_agent is None, so
580        // this block is skipped and pipette restarts automatically on reboot.)
581        if let Some(agent) = self.resources.linux_direct_serial_agent.as_mut() {
582            agent.reset();
583
584            if self.resources.properties.using_vtl0_pipette {
585                self.launch_linux_direct_pipette().await?;
586            }
587        }
588        Ok(())
589    }
590
591    async fn wait_for_agent(&mut self, set_high_vtl: bool) -> anyhow::Result<PipetteClient> {
592        #[cfg(target_os = "linux")]
593        if let Some(guest_cid) = self.resources.properties.vhost_vsock_guest_cid {
594            assert!(
595                !set_high_vtl,
596                "kernel vhost-vsock pipette transport does not support VTL2"
597            );
598            return self.wait_for_agent_vhost_vsock(guest_cid).await;
599        }
600
601        // Use TCP transport if configured (Windows no-vmbus guests).
602        if let Some(port) = self.tcp_pipette_port {
603            assert!(!set_high_vtl, "TCP pipette transport does not support VTL2");
604            return self.wait_for_agent_tcp(port).await;
605        }
606
607        let listener = if set_high_vtl {
608            self.resources
609                .vtl2_pipette_listener
610                .as_mut()
611                .context("VM is not configured with VTL 2")?
612        } else {
613            &mut self.resources.pipette_listener
614        };
615
616        tracing::info!(set_high_vtl, "listening for pipette connection");
617        let client = loop {
618            let (conn, _) = listener
619                .accept()
620                .await
621                .context("failed to accept pipette connection")?;
622            tracing::info!(set_high_vtl, "handshaking with pipette");
623            let socket = PolledSocket::new(&self.resources.driver, conn)?;
624            match PipetteClient::new(&self.resources.driver, socket, &self.resources.output_dir)
625                .await
626            {
627                Ok(client) => break client,
628                Err(e) => {
629                    // During save/restore cycles, stale connections from
630                    // previous hvsock relay sessions can accumulate in the
631                    // listener backlog. These are already-closed sockets
632                    // that fail during the mesh handshake. Drain them and
633                    // retry until we get a live connection.
634                    tracing::warn!(
635                        error = e.as_ref() as &dyn std::error::Error,
636                        "pipette connection not live, retrying"
637                    );
638                }
639            }
640        };
641        tracing::info!(set_high_vtl, "completed pipette handshake");
642
643        // When pipette runs as PID 1 init and a CIDATA agent disk is
644        // attached, mount it so test files are available at /cidata.
645        // Skip if already mounted (e.g. reconnecting after save/restore
646        // where guest state is preserved).
647        if !set_high_vtl
648            && self.resources.properties.uses_pipette_as_init
649            && self.resources.properties.has_agent_disk
650            && !self.cidata_mounted
651        {
652            tracing::info!("mounting CIDATA agent disk via pipette");
653            client
654                .unix_shell()
655                .cmd("mkdir")
656                .arg("-p")
657                .arg("/cidata")
658                .run()
659                .await
660                .context("failed to create /cidata mount point")?;
661            client
662                .unix_shell()
663                .cmd("mount")
664                .arg("LABEL=cidata")
665                .arg("/cidata")
666                .run()
667                .await
668                .context("failed to mount CIDATA disk")?;
669            self.cidata_mounted = true;
670        }
671
672        Ok(client)
673    }
674
675    /// Connect to pipette directly through the host's AF_VSOCK namespace.
676    #[cfg(target_os = "linux")]
677    async fn wait_for_agent_vhost_vsock(
678        &mut self,
679        guest_cid: u32,
680    ) -> anyhow::Result<PipetteClient> {
681        tracing::info!(
682            guest_cid,
683            port = PIPETTE_PORT,
684            "connecting to pipette via kernel vhost-vsock"
685        );
686        let socket = loop {
687            let connect = async {
688                let socket = VmSocket::new().context("failed to create AF_VSOCK socket")?;
689                socket
690                    .set_connect_timeout(Duration::from_secs(5))
691                    .context("failed to set AF_VSOCK connect timeout")?;
692                let mut socket = PolledSocket::new(&self.resources.driver, socket)
693                    .context("failed to create polled AF_VSOCK socket")?
694                    .convert();
695                socket
696                    .connect(&VmAddress::vsock(guest_cid, PIPETTE_PORT).into())
697                    .await
698                    .context("failed to connect to guest AF_VSOCK listener")?;
699                Ok::<_, anyhow::Error>(socket)
700            };
701
702            match connect.await {
703                Ok(socket) => break socket,
704                Err(error) => {
705                    tracing::trace!(
706                        error = error.as_ref() as &dyn std::error::Error,
707                        "AF_VSOCK connect failed, guest not ready yet"
708                    );
709                }
710            }
711
712            pal_async::timer::PolledTimer::new(&self.resources.driver)
713                .sleep(Duration::from_secs(1))
714                .await;
715        };
716        tracing::info!("AF_VSOCK connected, handshaking with pipette");
717        let client = PipetteClient::new(&self.resources.driver, socket, &self.resources.output_dir)
718            .await
719            .context("pipette AF_VSOCK handshake failed")?;
720        tracing::info!("completed pipette AF_VSOCK handshake");
721        Ok(client)
722    }
723
724    /// Connect to pipette via TCP through consomme port forwarding.
725    ///
726    /// The guest pipette agent listens on `0.0.0.0:{port}` and consomme
727    /// forwards connections from `localhost:{port}` on the host into the
728    /// guest. We retry until the guest's network stack and pipette are up.
729    async fn wait_for_agent_tcp(&mut self, port: u16) -> anyhow::Result<PipetteClient> {
730        tracing::info!(port, "connecting to pipette via TCP");
731        let addr = std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port));
732        let client = loop {
733            match PolledSocket::connect_tcp(&self.resources.driver, addr).await {
734                Ok(socket) => {
735                    socket
736                        .get()
737                        .set_nodelay(true)
738                        .context("failed to set TCP_NODELAY")?;
739                    tracing::info!("TCP connected, handshaking with pipette");
740                    match PipetteClient::new(
741                        &self.resources.driver,
742                        socket,
743                        &self.resources.output_dir,
744                    )
745                    .await
746                    {
747                        Ok(client) => break client,
748                        Err(e) => {
749                            tracing::warn!(
750                                error = e.as_ref() as &dyn std::error::Error,
751                                "pipette TCP connection failed, retrying"
752                            );
753                        }
754                    }
755                }
756                Err(e) => {
757                    tracing::trace!(
758                        error = &e as &dyn std::error::Error,
759                        "TCP connect failed, guest not ready yet"
760                    );
761                }
762            }
763            // Wait before retrying — guest network stack may not be up yet.
764            pal_async::timer::PolledTimer::new(&self.resources.driver)
765                .sleep(Duration::from_secs(1))
766                .await;
767        };
768        tracing::info!("completed pipette TCP handshake");
769        Ok(client)
770    }
771
772    async fn pause(&self) -> anyhow::Result<()> {
773        self.worker.pause().await?;
774        Ok(())
775    }
776
777    async fn save_state(&self) -> anyhow::Result<Vec<u8>> {
778        let state_msg = self.worker.save().await?;
779        Ok(mesh::payload::encode(state_msg))
780    }
781
782    async fn resume(&self) -> anyhow::Result<()> {
783        self.worker.resume().await?;
784        Ok(())
785    }
786
787    async fn verify_save_restore(&self) -> anyhow::Result<()> {
788        for i in 0..2 {
789            let result = self.worker.pulse_save_restore().await;
790            match result {
791                Ok(()) => {}
792                Err(RpcError::Channel(err)) => return Err(err.into()),
793                Err(RpcError::Call(PulseSaveRestoreError::ResetNotSupported)) => {
794                    tracing::warn!("Reset not supported, could not test save + restore.");
795                    break;
796                }
797                Err(RpcError::Call(PulseSaveRestoreError::Other(err))) => {
798                    return Err(anyhow::Error::from(err))
799                        .context(format!("Save + restore {i} failed."));
800                }
801            }
802        }
803
804        Ok(())
805    }
806
807    async fn launch_linux_direct_pipette(&mut self) -> anyhow::Result<()> {
808        // Start pipette through serial on linux direct.
809        self.resources
810            .linux_direct_serial_agent
811            .as_mut()
812            .unwrap()
813            .run_command("mkdir /cidata && mount LABEL=cidata /cidata && sh -c '/cidata/pipette &'")
814            .await?;
815        Ok(())
816    }
817}
818
819/// Interface for inspecting OpenVMM
820pub struct OpenVmmInspector {
821    worker: Arc<Worker>,
822}
823
824#[async_trait]
825impl PetriVmInspector for OpenVmmInspector {
826    async fn inspect(&self, path: &str) -> anyhow::Result<inspect::Node> {
827        Ok(self.worker.inspect(path).await)
828    }
829}
830
831/// Interface to the OpenVMM framebuffer
832pub struct OpenVmmFramebufferAccess {
833    view: View,
834}
835
836#[async_trait]
837impl PetriVmFramebufferAccess for OpenVmmFramebufferAccess {
838    async fn screenshot(
839        &mut self,
840        image: &mut Vec<u8>,
841    ) -> anyhow::Result<Option<VmScreenshotMeta>> {
842        // Our framebuffer uses 4 bytes per pixel, approximating an
843        // BGRA image, however it only actually contains BGR data.
844        // The fourth byte is effectively noise. We can set the 'alpha'
845        // value to 0xFF to make the image opaque.
846        const BYTES_PER_PIXEL: usize = 4;
847        let (width, height) = self.view.resolution();
848        let (widthsize, heightsize) = (width as usize, height as usize);
849        let len = widthsize * heightsize * BYTES_PER_PIXEL;
850
851        image.resize(len, 0);
852        for (i, line) in (0..height).zip(image.chunks_exact_mut(widthsize * BYTES_PER_PIXEL)) {
853            self.view.read_line(i, line);
854            for pixel in line.chunks_exact_mut(BYTES_PER_PIXEL) {
855                pixel.swap(0, 2);
856                pixel[3] = 0xFF;
857            }
858        }
859
860        Ok(Some(VmScreenshotMeta {
861            color: image::ExtendedColorType::Rgba8,
862            width,
863            height,
864        }))
865    }
866}
petri/vm/openvmm/runtime.rs

petri/vm/openvmm/
runtime.rs
