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

petri/vm/openvmm/
runtime.rs
