underhill_core/
lib.rs

1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4//! This module implements the interactive control process and the entry point
5//! for the underhill environment.
6
7#![cfg(target_os = "linux")]
8#![expect(missing_docs)]
9#![forbid(unsafe_code)]
10
11mod diag;
12mod dispatch;
13mod emuplat;
14mod get_tracing;
15mod inspect_internal;
16mod inspect_proc;
17mod livedump;
18mod loader;
19mod nvme_manager;
20mod options;
21mod reference_time;
22mod servicing;
23mod threadpool_vm_task_backend;
24mod vmbus_relay_unit;
25mod vmgs_logger;
26mod vp;
27mod vpci;
28mod worker;
29mod wrapped_partition;
30
31// `pub` so that the missing_docs warning fires for options without
32// documentation.
33pub use options::Options;
34
35use crate::diag::DiagWorker;
36use crate::dispatch::UhVmRpc;
37use crate::worker::UnderhillEnvCfg;
38use crate::worker::UnderhillRemoteConsoleCfg;
39use crate::worker::UnderhillVmWorker;
40use crate::worker::UnderhillWorkerParameters;
41use anyhow::Context;
42use bootloader_fdt_parser::BootTimes;
43use cvm_tracing::CVM_ALLOWED;
44use framebuffer::FRAMEBUFFER_SIZE;
45use framebuffer::FramebufferAccess;
46use futures::StreamExt;
47use futures_concurrency::stream::Merge;
48use get_tracing::init_tracing;
49use get_tracing::init_tracing_backend;
50use inspect::Inspect;
51use inspect::SensitivityLevel;
52use mesh::CancelContext;
53use mesh::CancelReason;
54use mesh::MeshPayload;
55use mesh::error::RemoteError;
56use mesh::rpc::Rpc;
57use mesh::rpc::RpcSend;
58use mesh_process::Mesh;
59use mesh_process::ProcessConfig;
60use mesh_process::try_run_mesh_host;
61use mesh_tracing::RemoteTracer;
62use mesh_tracing::TracingBackend;
63use mesh_worker::RegisteredWorkers;
64use mesh_worker::WorkerEvent;
65use mesh_worker::WorkerHandle;
66use mesh_worker::WorkerHost;
67use mesh_worker::WorkerHostRunner;
68use mesh_worker::launch_local_worker;
69use mesh_worker::register_workers;
70use pal_async::DefaultDriver;
71use pal_async::DefaultPool;
72use pal_async::task::Spawn;
73#[cfg(feature = "profiler")]
74use profiler_worker::ProfilerWorker;
75#[cfg(feature = "profiler")]
76use profiler_worker::ProfilerWorkerParameters;
77use std::time::Duration;
78use vmsocket::VmAddress;
79use vmsocket::VmListener;
80use vnc_worker_defs::VncParameters;
81
82fn new_underhill_remote_console_cfg(
83    framebuffer_gpa_base: Option<u64>,
84) -> anyhow::Result<(UnderhillRemoteConsoleCfg, Option<FramebufferAccess>)> {
85    if let Some(framebuffer_gpa_base) = framebuffer_gpa_base {
86        // Underhill accesses the framebuffer by using /dev/mshv_vtl_low to read
87        // from a second mapping placed after the end of RAM at a static
88        // location specified by the host.
89        //
90        // Open the file directly rather than use the `hcl` crate to avoid
91        // leaking `hcl` stuff into this crate.
92        //
93        // FUTURE: use an approach that doesn't require double mapping the
94        // framebuffer from the host.
95        let gpa_fd = fs_err::OpenOptions::new()
96            .read(true)
97            .write(true)
98            .open("/dev/mshv_vtl_low")
99            .context("failed to open gpa device")?;
100
101        let vram = sparse_mmap::new_mappable_from_file(gpa_fd.file(), true, false)?;
102        let (fb, fba) = framebuffer::framebuffer(vram, FRAMEBUFFER_SIZE, framebuffer_gpa_base)
103            .context("allocating framebuffer")?;
104        tracing::debug!("framebuffer_gpa_base: {:#x}", framebuffer_gpa_base);
105
106        Ok((
107            UnderhillRemoteConsoleCfg {
108                synth_keyboard: true,
109                synth_mouse: true,
110                synth_video: true,
111                input: mesh::Receiver::new(),
112                framebuffer: Some(fb),
113            },
114            Some(fba),
115        ))
116    } else {
117        Ok((
118            UnderhillRemoteConsoleCfg {
119                synth_keyboard: false,
120                synth_mouse: false,
121                synth_video: false,
122                input: mesh::Receiver::new(),
123                framebuffer: None,
124            },
125            None,
126        ))
127    }
128}
129
130pub fn main() -> anyhow::Result<()> {
131    // Install a panic hook to prefix the current async task name before the
132    // standard panic output.
133    install_task_name_panic_hook();
134
135    if let Some(path) = std::env::var_os("OPENVMM_WRITE_SAVED_STATE_PROTO") {
136        if cfg!(debug_assertions) {
137            mesh::payload::protofile::DescriptorWriter::new(
138                vmcore::save_restore::saved_state_roots(),
139            )
140            .write_to_path(path)
141            .context("failed to write protobuf descriptors")?;
142            return Ok(());
143        } else {
144            // The generated code for this is too large for release builds.
145            anyhow::bail!(".proto output only supported in debug builds");
146        }
147    }
148
149    // FUTURE: create and use the affinitized threadpool here.
150    let (_, tracing_driver) = DefaultPool::spawn_on_thread("tracing");
151
152    // Try to run as a worker host, sending a remote tracer that will forward
153    // tracing events back to the initial process for logging to the host. See
154    // [`get_tracing`] doc comments for more details.
155    //
156    // On success the worker runs to completion and then exits the process (does
157    // not return). Any worker host setup errors are return and bubbled up.
158    try_run_mesh_host("underhill", {
159        let tracing_driver = tracing_driver.clone();
160        async |params: MeshHostParams| {
161            if let Some(remote_tracer) = params.tracer {
162                init_tracing(tracing_driver, remote_tracer).context("failed to init tracing")?;
163            }
164            params.runner.run(RegisteredWorkers).await;
165            Ok(())
166        }
167    })?;
168
169    // Initialize the tracing backend used by this and all subprocesses.
170    let mut tracing = init_tracing_backend(tracing_driver.clone())?;
171    // Initialize tracing from the backend.
172    init_tracing(tracing_driver, tracing.tracer()).context("failed to init tracing")?;
173    DefaultPool::run_with(|driver| do_main(driver, tracing))
174}
175
176fn install_task_name_panic_hook() {
177    use std::io::Write;
178
179    let panic_hook = std::panic::take_hook();
180    std::panic::set_hook(Box::new(move |info| {
181        pal_async::task::with_current_task_metadata(|metadata| {
182            if let Some(metadata) = metadata {
183                let _ = write!(std::io::stderr(), "task '{}', ", metadata.name());
184            }
185        });
186        // This will proceed with writing "thread ... panicked at ..."
187        panic_hook(info);
188    }));
189}
190
191async fn do_main(driver: DefaultDriver, mut tracing: TracingBackend) -> anyhow::Result<()> {
192    let opt = Options::parse(Vec::new(), Vec::new())?;
193
194    let crate_name = build_info::get().crate_name();
195    let crate_revision = build_info::get().scm_revision();
196    tracing::info!(CVM_ALLOWED, ?crate_name, ?crate_revision, "VMM process");
197    log_boot_times().context("failure logging boot times")?;
198
199    // Write the current pid to a file.
200    if let Some(pid_path) = &opt.pid {
201        std::fs::write(pid_path, std::process::id().to_string())
202            .with_context(|| format!("failed to write pid to {}", pid_path.display()))?;
203    }
204
205    let mesh = Mesh::new("underhill".to_string()).context("failed to create mesh")?;
206
207    let r = run_control(driver, &mesh, opt, &mut tracing).await;
208    if let Err(err) = &r {
209        tracing::error!(
210            CVM_ALLOWED,
211            error = err.as_ref() as &dyn std::error::Error,
212            "VM failure"
213        );
214    }
215
216    // Wait a few seconds for child processes to terminate and tracing to finish.
217    CancelContext::new()
218        .with_timeout(Duration::from_secs(10))
219        .until_cancelled(async {
220            mesh.shutdown().await;
221            tracing.shutdown().await;
222        })
223        .await
224        .ok();
225
226    r
227}
228
229fn log_boot_times() -> anyhow::Result<()> {
230    fn diff(start: Option<u64>, end: Option<u64>) -> Option<tracing::field::DebugValue<Duration>> {
231        use reference_time::ReferenceTime;
232        Some(tracing::field::debug(
233            ReferenceTime::new(end?).since(ReferenceTime::new(start?))?,
234        ))
235    }
236
237    // Read boot times provided by the bootloader.
238    let BootTimes {
239        start,
240        end,
241        sidecar_start,
242        sidecar_end,
243    } = BootTimes::new().context("failed to parse boot times")?;
244    tracing::info!(
245        CVM_ALLOWED,
246        start,
247        end,
248        sidecar_start,
249        sidecar_end,
250        elapsed = diff(start, end),
251        sidecar_elapsed = diff(sidecar_start, sidecar_end),
252        "boot loader times"
253    );
254    Ok(())
255}
256
257struct DiagState {
258    _worker: WorkerHandle,
259    request_recv: mesh::Receiver<diag_server::DiagRequest>,
260}
261
262impl DiagState {
263    async fn new() -> anyhow::Result<Self> {
264        // Start the diagnostics worker immediately.
265        let (request_send, request_recv) = mesh::channel();
266        let worker = launch_local_worker::<DiagWorker>(diag::DiagWorkerParameters { request_send })
267            .await
268            .context("failed to launch diagnostics worker")?;
269        Ok(Self {
270            _worker: worker,
271            request_recv,
272        })
273    }
274}
275
276#[derive(Inspect)]
277struct Workers {
278    vm: WorkerHandle,
279    #[inspect(skip)]
280    vm_rpc: mesh::Sender<UhVmRpc>,
281    vnc: Option<WorkerHandle>,
282    #[cfg(feature = "gdb")]
283    gdb: Option<WorkerHandle>,
284}
285
286#[derive(MeshPayload)]
287struct MeshHostParams {
288    tracer: Option<RemoteTracer>,
289    runner: WorkerHostRunner,
290}
291
292async fn launch_mesh_host(
293    mesh: &Mesh,
294    name: &str,
295    tracer: Option<RemoteTracer>,
296) -> anyhow::Result<WorkerHost> {
297    let (host, runner) = mesh_worker::worker_host();
298    mesh.launch_host(ProcessConfig::new(name), MeshHostParams { tracer, runner })
299        .await?;
300    Ok(host)
301}
302
303async fn launch_workers(
304    mesh: &Mesh,
305    tracing: &mut TracingBackend,
306    control_send: mesh::Sender<ControlRequest>,
307    opt: Options,
308) -> anyhow::Result<Workers> {
309    let env_cfg = UnderhillEnvCfg {
310        vmbus_max_version: opt.vmbus_max_version,
311        vmbus_enable_mnf: opt.vmbus_enable_mnf,
312        vmbus_force_confidential_external_memory: opt.vmbus_force_confidential_external_memory,
313        vmbus_channel_unstick_delay: (opt.vmbus_channel_unstick_delay_ms != 0)
314            .then(|| Duration::from_millis(opt.vmbus_channel_unstick_delay_ms)),
315        cmdline_append: opt.cmdline_append.clone(),
316        reformat_vmgs: opt.reformat_vmgs,
317        vtl0_starts_paused: opt.vtl0_starts_paused,
318        emulated_serial_wait_for_rts: opt.serial_wait_for_rts,
319        force_load_vtl0_image: opt.force_load_vtl0_image,
320        nvme_vfio: opt.nvme_vfio,
321        mcr: opt.mcr,
322        halt_on_guest_halt: opt.halt_on_guest_halt,
323        no_sidecar_hotplug: opt.no_sidecar_hotplug,
324        gdbstub: opt.gdbstub,
325        hide_isolation: opt.hide_isolation,
326        nvme_keep_alive: opt.nvme_keep_alive,
327        mana_keep_alive: opt.mana_keep_alive,
328        nvme_always_flr: opt.nvme_always_flr,
329        test_configuration: opt.test_configuration,
330        disable_uefi_frontpage: opt.disable_uefi_frontpage,
331        default_boot_always_attempt: opt.default_boot_always_attempt,
332        guest_state_lifetime: opt.guest_state_lifetime,
333        guest_state_encryption_policy: opt.guest_state_encryption_policy,
334        strict_encryption_policy: opt.strict_encryption_policy,
335        attempt_ak_cert_callback: opt.attempt_ak_cert_callback,
336        enable_vpci_relay: opt.enable_vpci_relay,
337        disable_proxy_redirect: opt.disable_proxy_redirect,
338        disable_lower_vtl_timer_virt: opt.disable_lower_vtl_timer_virt,
339        config_timeout_in_seconds: opt.config_timeout_in_seconds,
340        servicing_timeout_dump_collection_in_ms: opt.servicing_timeout_dump_collection_in_ms,
341    };
342
343    let (mut remote_console_cfg, framebuffer_access) =
344        new_underhill_remote_console_cfg(opt.framebuffer_gpa_base)?;
345
346    let mut vnc_worker = None;
347    if let Some(framebuffer) = framebuffer_access {
348        let listener = VmListener::bind(VmAddress::vsock_any(opt.vnc_port))
349            .context("failed to bind socket")?;
350
351        let input_send = remote_console_cfg.input.sender();
352
353        let vnc_host = launch_mesh_host(mesh, "vnc", Some(tracing.tracer()))
354            .await
355            .context("spawning vnc process failed")?;
356
357        vnc_worker = Some(
358            vnc_host
359                .launch_worker(
360                    vnc_worker_defs::VNC_WORKER_VMSOCKET,
361                    VncParameters {
362                        listener,
363                        framebuffer,
364                        input_send,
365                    },
366                )
367                .await?,
368        )
369    }
370
371    #[cfg(feature = "gdb")]
372    let mut gdbstub_worker = None;
373    #[cfg_attr(not(feature = "gdb"), expect(unused_mut))]
374    let mut debugger_rpc = None;
375    #[cfg(feature = "gdb")]
376    if opt.gdbstub {
377        let listener = VmListener::bind(VmAddress::vsock_any(opt.gdbstub_port))
378            .context("failed to bind socket")?;
379
380        let gdb_host = launch_mesh_host(mesh, "gdb", Some(tracing.tracer()))
381            .await
382            .context("failed to spawn gdb host process")?;
383
384        // Get the VP count of this machine. It's too early to read it directly
385        // from IGVM parameters, but the kernel already has the IGVM parsed VP
386        // count via the boot loader anyways.
387        let vp_count =
388            pal::unix::affinity::max_present_cpu().context("failed to get max present cpu")? + 1;
389
390        let (send, recv) = mesh::channel();
391        debugger_rpc = Some(recv);
392        gdbstub_worker = Some(
393            gdb_host
394                .launch_worker(
395                    debug_worker_defs::DEBUGGER_VSOCK_WORKER,
396                    debug_worker_defs::DebuggerParameters {
397                        listener,
398                        req_chan: send,
399                        vp_count,
400                        target_arch: if cfg!(guest_arch = "x86_64") {
401                            debug_worker_defs::TargetArch::X86_64
402                        } else {
403                            debug_worker_defs::TargetArch::Aarch64
404                        },
405                    },
406                )
407                .await?,
408        );
409    }
410    let (vm_rpc, vm_rpc_rx) = mesh::channel();
411
412    // Spawn the worker in a separate process in case the diagnostics server (in
413    // this process) is used to run gdbserver against it, or in case it needs to
414    // be restarted.
415    let host = launch_mesh_host(mesh, "vm", Some(tracing.tracer()))
416        .await
417        .context("failed to launch worker process")?;
418
419    let vm_worker = host
420        .start_worker(
421            worker::UNDERHILL_WORKER,
422            UnderhillWorkerParameters {
423                env_cfg,
424                remote_console_cfg,
425                debugger_rpc,
426                vm_rpc: vm_rpc_rx,
427                control_send,
428            },
429        )
430        .context("failed to launch worker")?;
431
432    Ok(Workers {
433        vm: vm_worker,
434        vm_rpc,
435        vnc: vnc_worker,
436        #[cfg(feature = "gdb")]
437        gdb: gdbstub_worker,
438    })
439}
440
441/// State for inspect only.
442#[derive(Inspect)]
443enum ControlState {
444    WaitingForStart,
445    Starting,
446    Started,
447    Restarting,
448}
449
450#[derive(MeshPayload)]
451pub enum ControlRequest {
452    FlushLogs(Rpc<CancelContext, Result<(), CancelReason>>),
453    MakeWorker(Rpc<String, Result<WorkerHost, RemoteError>>),
454}
455
456async fn run_control(
457    driver: DefaultDriver,
458    mesh: &Mesh,
459    opt: Options,
460    mut tracing: &mut TracingBackend,
461) -> anyhow::Result<()> {
462    let (control_send, mut control_recv) = mesh::channel();
463    let mut control_send = Some(control_send);
464
465    if opt.signal_vtl0_started {
466        signal_vtl0_started(&driver)
467            .await
468            .context("failed to signal vtl0 started")?;
469    }
470
471    let mut diag = DiagState::new().await?;
472
473    let (diag_reinspect_send, mut diag_reinspect_recv) = mesh::channel();
474    #[cfg(feature = "profiler")]
475    let mut profiler_host = None;
476    let mut state;
477    let mut workers = if opt.wait_for_start {
478        state = ControlState::WaitingForStart;
479        None
480    } else {
481        state = ControlState::Starting;
482        let workers = launch_workers(mesh, tracing, control_send.take().unwrap(), opt)
483            .await
484            .context("failed to launch workers")?;
485        Some(workers)
486    };
487
488    enum Event {
489        Diag(diag_server::DiagRequest),
490        Worker(WorkerEvent),
491        Control(ControlRequest),
492    }
493
494    let mut restart_rpc = None;
495    loop {
496        let event = {
497            let mut stream = (
498                (&mut diag.request_recv).map(Event::Diag),
499                (&mut diag_reinspect_recv)
500                    .map(|req| Event::Diag(diag_server::DiagRequest::Inspect(req))),
501                (&mut control_recv).map(Event::Control),
502                futures::stream::select_all(workers.as_mut().map(|w| &mut w.vm)).map(Event::Worker),
503            )
504                .merge();
505
506            let Some(event) = stream.next().await else {
507                break;
508            };
509            event
510        };
511
512        match event {
513            Event::Diag(request) => {
514                match request {
515                    diag_server::DiagRequest::Start(rpc) => {
516                        rpc.handle_failable(async |params| {
517                            if workers.is_some() {
518                                Err(anyhow::anyhow!("workers have already been started"))?;
519                            }
520                            let new_opt = Options::parse(params.args, params.env)
521                                .context("failed to parse new options")?;
522
523                            workers = Some(
524                                launch_workers(
525                                    mesh,
526                                    tracing,
527                                    control_send.take().unwrap(),
528                                    new_opt,
529                                )
530                                .await?,
531                            );
532                            state = ControlState::Starting;
533                            anyhow::Ok(())
534                        })
535                        .await
536                    }
537                    diag_server::DiagRequest::Inspect(deferred) => deferred.respond(|resp| {
538                        resp.sensitivity_field("mesh", SensitivityLevel::Safe, mesh)
539                            .sensitivity_field_mut("trace", SensitivityLevel::Safe, &mut tracing)
540                            .sensitivity_field(
541                                "build_info",
542                                SensitivityLevel::Safe,
543                                build_info::get(),
544                            )
545                            .sensitivity_child(
546                                "proc",
547                                SensitivityLevel::Safe,
548                                inspect_proc::inspect_proc,
549                            )
550                            .sensitivity_field("control_state", SensitivityLevel::Safe, &state)
551                            // This node can not be renamed due to stability guarantees.
552                            // See the comment at the top of inspect_internal for more details.
553                            .sensitivity_child("uhdiag", SensitivityLevel::Safe, |req| {
554                                inspect_internal::inspect_internal_diagnostics(
555                                    req,
556                                    &diag_reinspect_send,
557                                    &driver,
558                                )
559                            });
560
561                        resp.merge(&workers);
562                    }),
563                    diag_server::DiagRequest::Crash(pid) => {
564                        mesh.crash(pid);
565                    }
566                    diag_server::DiagRequest::Restart(rpc) => {
567                        let Some(workers) = &mut workers else {
568                            rpc.complete(Err(RemoteError::new(anyhow::anyhow!(
569                                "worker has not been started yet"
570                            ))));
571                            continue;
572                        };
573
574                        let r = async {
575                            if restart_rpc.is_some() {
576                                anyhow::bail!("previous restart still in progress");
577                            }
578
579                            let host = launch_mesh_host(mesh, "vm", Some(tracing.tracer()))
580                                .await
581                                .context("failed to launch worker process")?;
582
583                            workers.vm.restart(&host);
584                            Ok(())
585                        }
586                        .await;
587
588                        if r.is_err() {
589                            rpc.complete(r.map_err(RemoteError::new));
590                        } else {
591                            state = ControlState::Restarting;
592                            restart_rpc = Some(rpc);
593                        }
594                    }
595                    diag_server::DiagRequest::Pause(rpc) => {
596                        let Some(workers) = &mut workers else {
597                            rpc.complete(Err(RemoteError::new(anyhow::anyhow!(
598                                "worker has not been started yet"
599                            ))));
600                            continue;
601                        };
602
603                        // create the req future output the spawn, so that
604                        // we don't need to clone + move vm_rpc.
605                        let req = workers.vm_rpc.call(UhVmRpc::Pause, ());
606
607                        // FUTURE: consider supporting cancellation
608                        driver
609                            .spawn("diag-pause", async move {
610                                let was_paused = req.await.expect("failed to pause VM");
611                                rpc.handle_failable_sync(|_| {
612                                    if !was_paused {
613                                        Err(anyhow::anyhow!("VM is already paused"))
614                                    } else {
615                                        Ok(())
616                                    }
617                                });
618                            })
619                            .detach();
620                    }
621                    diag_server::DiagRequest::PacketCapture(rpc) => {
622                        let Some(workers) = &mut workers else {
623                            rpc.complete(Err(RemoteError::new(anyhow::anyhow!(
624                                "worker has not been started yet"
625                            ))));
626                            continue;
627                        };
628
629                        workers.vm_rpc.send(UhVmRpc::PacketCapture(rpc));
630                    }
631                    diag_server::DiagRequest::Resume(rpc) => {
632                        let Some(workers) = &mut workers else {
633                            rpc.complete(Err(RemoteError::new(anyhow::anyhow!(
634                                "worker has not been started yet"
635                            ))));
636                            continue;
637                        };
638
639                        let was_resumed = workers
640                            .vm_rpc
641                            .call(UhVmRpc::Resume, ())
642                            .await
643                            .context("failed to resumed VM")?;
644
645                        let was_halted = workers
646                            .vm_rpc
647                            .call(UhVmRpc::ClearHalt, ())
648                            .await
649                            .context("failed to clear halt from VPs")?;
650
651                        rpc.handle_sync(|_| {
652                            if was_resumed || was_halted {
653                                Ok(())
654                            } else {
655                                Err(RemoteError::new(anyhow::anyhow!("VM is currently running")))
656                            }
657                        });
658                    }
659                    diag_server::DiagRequest::Save(rpc) => {
660                        let Some(workers) = &mut workers else {
661                            rpc.complete(Err(RemoteError::new(anyhow::anyhow!(
662                                "worker has not been started yet"
663                            ))));
664                            continue;
665                        };
666
667                        workers.vm_rpc.send(UhVmRpc::Save(rpc));
668                    }
669                    #[cfg(feature = "profiler")]
670                    diag_server::DiagRequest::Profile(rpc) => {
671                        let (rpc_params, rpc_sender) = rpc.split();
672                        // Create profiler host if there is none created before
673                        if profiler_host.is_none() {
674                            match launch_mesh_host(mesh, "profiler", Some(tracing.tracer()))
675                                .await
676                                .context("failed to launch profiler host")
677                            {
678                                Ok(host) => {
679                                    profiler_host = Some(host);
680                                }
681                                Err(e) => {
682                                    rpc_sender.complete(Err(RemoteError::new(e)));
683                                    continue;
684                                }
685                            }
686                        }
687
688                        let profiling_duration = rpc_params.duration;
689                        let host = profiler_host.as_ref().unwrap();
690                        let mut profiler_worker;
691                        match host
692                            .launch_worker(
693                                profiler_worker::PROFILER_WORKER,
694                                ProfilerWorkerParameters {
695                                    profiler_request: rpc_params,
696                                },
697                            )
698                            .await
699                        {
700                            Ok(worker) => {
701                                profiler_worker = worker;
702                            }
703                            Err(e) => {
704                                rpc_sender.complete(Err(RemoteError::new(e)));
705                                continue;
706                            }
707                        }
708
709                        driver
710                            .spawn("profiler_worker", async move {
711                                let result = CancelContext::new()
712                                    .with_timeout(Duration::from_secs(profiling_duration + 30))
713                                    .until_cancelled(profiler_worker.join())
714                                    .await
715                                    .context("profiler worker cancelled")
716                                    .and_then(|result| result.context("profiler worker failed"))
717                                    .map_err(RemoteError::new);
718
719                                rpc_sender.complete(result);
720                            })
721                            .detach();
722                    }
723                }
724            }
725            Event::Worker(event) => match event {
726                WorkerEvent::Started => {
727                    if let Some(response) = restart_rpc.take() {
728                        tracing::info!(CVM_ALLOWED, "restart complete");
729                        response.complete(Ok(()));
730                    } else {
731                        tracing::info!(CVM_ALLOWED, "vm worker started");
732                    }
733                    state = ControlState::Started;
734                }
735                WorkerEvent::Stopped => {
736                    anyhow::bail!("worker unexpectedly stopped");
737                }
738                WorkerEvent::Failed(err) => {
739                    return Err(anyhow::Error::from(err)).context("vm worker failed");
740                }
741                WorkerEvent::RestartFailed(err) => {
742                    tracing::error!(
743                        CVM_ALLOWED,
744                        error = &err as &dyn std::error::Error,
745                        "restart failed"
746                    );
747                    restart_rpc.take().unwrap().complete(Err(err));
748                    state = ControlState::Started;
749                }
750            },
751            Event::Control(req) => match req {
752                ControlRequest::FlushLogs(rpc) => {
753                    rpc.handle(async |mut ctx| {
754                        tracing::info!(CVM_ALLOWED, "flushing logs");
755                        ctx.until_cancelled(tracing.flush()).await?;
756                        Ok(())
757                    })
758                    .await
759                }
760                ControlRequest::MakeWorker(rpc) => {
761                    rpc.handle_failable(async |name| {
762                        launch_mesh_host(mesh, &name, Some(tracing.tracer())).await
763                    })
764                    .await
765                }
766            },
767        }
768    }
769
770    Ok(())
771}
772
773async fn signal_vtl0_started(driver: &DefaultDriver) -> anyhow::Result<()> {
774    tracing::info!(CVM_ALLOWED, "signaling vtl0 started early");
775    let (client, task) = guest_emulation_transport::spawn_get_worker(driver.clone())
776        .await
777        .context("failed to spawn GET")?;
778    client.complete_start_vtl0(None).await;
779    // Disconnect the GET so that it can be reused.
780    drop(client);
781    task.await.unwrap();
782    tracing::info!(CVM_ALLOWED, "signaled vtl0 start");
783    Ok(())
784}
785
786// The "base" workers for Underhill. Other workers are defined in the
787// `underhill_resources` crate.
788//
789// FUTURE: split these workers into separate crates and move them to
790// `underhill_resources`, too.
791register_workers! {
792    UnderhillVmWorker,
793    DiagWorker,
794    #[cfg(feature = "profiler")]
795    ProfilerWorker,
796}