vmm: live migration asynchronization (dedicated thread)

This puts the send-migration action into a dedicated thread.

1. The send-migration call will exit sooner (just dispatch the migration)
2. The VMM can respond to VmInfo calls as the migration is ongoing

Currently missing functionality:
- query migration progress
- prevent VM change events during the life migration (device hot plug etc.)

The commit was inspired by @ljcore [0] but adapted by me.

[0] cloud-hypervisor#7038

Signed-off-by: Philipp Schuster <philipp.schuster@cyberus-technology.de>
On-behalf-of: SAP philipp.schuster@sap.com

Author: phip1611

vmm/src/lib.rs

@@ -26,7 +26,7 @@ use std::os::unix::net::{UnixListener, UnixStream};
 use std::panic::AssertUnwindSafe;
 use std::path::PathBuf;
 use std::rc::Rc;
-use std::sync::mpsc::{Receiver, RecvError, SendError, Sender};
+use std::sync::mpsc::{Receiver, RecvError, SendError, Sender, TryRecvError};
 use std::sync::{Arc, Mutex};
 #[cfg(not(target_arch = "riscv64"))]
 use std::time::{Duration, Instant};
@@ -238,6 +238,7 @@ pub enum EpollDispatch {
     Api = 2,
     ActivateVirtioDevices = 3,
     Debug = 4,
+    CheckMigration = 5,
     Unknown,
 }
 
@@ -250,6 +251,7 @@ impl From<u64> for EpollDispatch {
             2 => Api,
             3 => ActivateVirtioDevices,
             4 => Debug,
+            5 => CheckMigration,
             _ => Unknown,
         }
     }
@@ -698,6 +700,49 @@ impl MigrationState {
     }
 }
 
+struct VmGuard {
+    vm: Option<Arc<Mutex<Vm>>>,
+    sender: std::sync::mpsc::Sender<Option<Vm>>,
+    is_local: bool,
+    check_migration_evt: Option<EventFd>,
+}
+
+impl Drop for VmGuard {
+    fn drop(&mut self) {
+        // Return the VM only if the migration fails
+        if let Some(vm) = self.vm.take()
+            && let Ok(mutex) = Arc::try_unwrap(vm)
+            && let Ok(mut inner_vm) = mutex.into_inner()
+        {
+            // Stop logging dirty pages only for non-local migrations
+            if !self.is_local
+                && let Err(e) = inner_vm.stop_dirty_log()
+            {
+                error!("Failed to stop dirty log in guard: {:?}", e);
+            }
+
+            // If VM is paused, attempt to resume
+            if inner_vm.get_state().unwrap_or(VmState::Paused) == VmState::Paused
+                && let Err(e) = inner_vm.resume()
+            {
+                error!("Failed to resume VM in guard: {:?}", e);
+            }
+
+            info!("VmGuard returning VM to parent thread");
+            if let Err(e) = self.sender.send(Some(inner_vm)) {
+                error!("Failed to return VM to parent thread: {:?}", e);
+            }
+
+            // Trigger the event of checking migration results
+            if let Some(evt) = &self.check_migration_evt
+                && let Err(e) = evt.write(1)
+            {
+                error!("Failed to trigger check migration event: {:?}", e);
+            }
+        }
+    }
+}
+
 pub struct VmmThreadHandle {
     pub thread_handle: thread::JoinHandle<Result<()>>,
     #[cfg(feature = "dbus_api")]
@@ -725,6 +770,8 @@ pub struct Vmm {
     original_termios_opt: Arc<Mutex<Option<termios>>>,
     console_resize_pipe: Option<Arc<File>>,
     console_info: Option<ConsoleInfo>,
+    migration_receiver: Option<Receiver<Option<Vm>>>,
+    check_migration_evt: EventFd,
 }
 
 impl Vmm {
@@ -782,14 +829,14 @@ impl Vmm {
                         .name("vmm_signal_handler".to_string())
                         .spawn(move || {
                             if !signal_handler_seccomp_filter.is_empty() && let Err(e) = apply_filter(&signal_handler_seccomp_filter)
-                                    .map_err(Error::ApplySeccompFilter)
-                                {
-                                    error!("Error applying seccomp filter: {:?}", e);
-                                    exit_evt.write(1).ok();
-                                    return;
-                                }
+                                .map_err(Error::ApplySeccompFilter)
+                            {
+                                error!("Error applying seccomp filter: {:?}", e);
+                                exit_evt.write(1).ok();
+                                return;
+                            }
 
-                            if landlock_enable{
+                            if landlock_enable {
                                 match Landlock::new() {
                                     Ok(landlock) => {
                                         let _ = landlock.restrict_self().map_err(Error::ApplyLandlock).map_err(|e| {
@@ -807,11 +854,11 @@ impl Vmm {
                             std::panic::catch_unwind(AssertUnwindSafe(|| {
                                 Vmm::signal_handler(signals, original_termios_opt, &exit_evt);
                             }))
-                            .map_err(|_| {
-                                error!("vmm signal_handler thread panicked");
-                                exit_evt.write(1).ok()
-                            })
-                            .ok();
+                                .map_err(|_| {
+                                    error!("vmm signal_handler thread panicked");
+                                    exit_evt.write(1).ok()
+                                })
+                                .ok();
                         })
                         .map_err(Error::SignalHandlerSpawn)?,
                 );
@@ -834,6 +881,7 @@ impl Vmm {
         let mut epoll = EpollContext::new().map_err(Error::Epoll)?;
         let reset_evt = EventFd::new(EFD_NONBLOCK).map_err(Error::EventFdCreate)?;
         let activate_evt = EventFd::new(EFD_NONBLOCK).map_err(Error::EventFdCreate)?;
+        let check_migration_evt = EventFd::new(EFD_NONBLOCK).map_err(Error::EventFdCreate)?;
 
         epoll
             .add_event(&exit_evt, EpollDispatch::Exit)
@@ -856,6 +904,10 @@ impl Vmm {
             .add_event(&debug_evt, EpollDispatch::Debug)
             .map_err(Error::Epoll)?;
 
+        epoll
+            .add_event(&check_migration_evt, EpollDispatch::CheckMigration)
+            .map_err(Error::Epoll)?;
+
         Ok(Vmm {
             epoll,
             exit_evt,
@@ -876,6 +928,8 @@ impl Vmm {
             original_termios_opt: Arc::new(Mutex::new(None)),
             console_resize_pipe: None,
             console_info: None,
+            migration_receiver: None,
+            check_migration_evt,
         })
     }
 
@@ -1580,6 +1634,31 @@ impl Vmm {
         }
     }
 
+    fn check_migration_result(&mut self) {
+        if let Some(receiver) = &self.migration_receiver {
+            match receiver.try_recv() {
+                Ok(Some(vm)) => {
+                    info!("Migration failed, restoring VM");
+                    // Restore VM on migration failure
+                    self.vm = Some(vm);
+                    self.migration_receiver = None;
+                }
+                Ok(None) => {
+                    info!("Migration completed successfully, VM transferred");
+                    // VM successfully transferred, clean up receiver
+                    self.migration_receiver = None;
+                }
+                Err(TryRecvError::Empty) => {
+                    // Migration still in progress, continue waiting
+                }
+                Err(TryRecvError::Disconnected) => {
+                    warn!("Migration channel disconnected unexpectedly");
+                    self.migration_receiver = None;
+                }
+            }
+        }
+    }
+
     fn control_loop(
         &mut self,
         api_receiver: Rc<Receiver<ApiRequest>>,
@@ -1673,8 +1752,17 @@ impl Vmm {
                     }
                     #[cfg(not(feature = "guest_debug"))]
                     EpollDispatch::Debug => {}
+                    EpollDispatch::CheckMigration => {
+                        info!("VM migration check event");
+                        // Consume the event.
+                        self.check_migration_evt
+                            .read()
+                            .map_err(Error::EventFdRead)?;
+                        self.check_migration_result();
+                    }
                 }
             }
+            self.check_migration_result();
         }
 
         // Trigger the termination of the signal_handler thread
@@ -2523,6 +2611,13 @@ impl RequestHandler for Vmm {
             send_data_migration.destination_url, send_data_migration.local
         );
 
+        // Check if there is already a migration in progress
+        if self.migration_receiver.is_some() {
+            return Err(MigratableError::MigrateSend(anyhow!(
+                "Another migration is already in progress"
+            )));
+        }
+
         if !self
             .vm_config
             .as_ref()
@@ -2537,42 +2632,98 @@ impl RequestHandler for Vmm {
             )));
         }
 
-        if let Some(vm) = self.vm.as_mut() {
-            Self::send_migration(
-                vm,
-                #[cfg(all(feature = "kvm", target_arch = "x86_64"))]
-                self.hypervisor.clone(),
-                send_data_migration.clone(),
-            )
-            .map_err(|migration_err| {
-                error!("Migration failed: {:?}", migration_err);
+        // Create a channel for VM ownership transfer
+        let (sender, receiver) = std::sync::mpsc::channel();
+        self.migration_receiver = Some(receiver);
 
-                // Stop logging dirty pages only for non-local migrations
-                if !send_data_migration.local
-                    && let Err(e) = vm.stop_dirty_log()
-                {
-                    return e;
-                }
-
-                if vm.get_state().unwrap() == VmState::Paused
-                    && let Err(e) = vm.resume()
-                {
-                    return e;
-                }
+        // Take VM ownership
+        let vm = self
+            .vm
+            .take()
+            .ok_or(MigratableError::MigrateSend(anyhow!("VM is not running")))?;
 
-                migration_err
-            })?;
+        #[cfg(all(feature = "kvm", target_arch = "x86_64"))]
+        let hypervisor = self.hypervisor.clone();
+        let send_data = send_data_migration.clone();
+
+        // Clone required event descriptors
+        let exit_evt = match self.exit_evt.try_clone() {
+            Ok(evt) => evt,
+            Err(e) => {
+                // Return VM immediately if clone fails
+                self.vm = Some(vm);
+                return Err(MigratableError::MigrateSend(anyhow!(
+                    "Failed to clone exit eventfd: {}",
+                    e
+                )));
+            }
+        };
 
-            // Shutdown the VM after the migration succeeded
-            self.exit_evt.write(1).map_err(|e| {
-                MigratableError::MigrateSend(anyhow!(
-                    "Failed shutting down the VM after migration: {:?}",
+        let check_migration_evt = match self.check_migration_evt.try_clone() {
+            Ok(evt) => evt,
+            Err(e) => {
+                // Return VM immediately if clone fails
+                self.vm = Some(vm);
+                return Err(MigratableError::MigrateSend(anyhow!(
+                    "Failed to clone exit eventfd: {}",
                     e
-                ))
+                )));
+            }
+        };
+
+        let vm = Arc::new(Mutex::new(vm));
+        let vm_clone = Arc::clone(&vm);
+
+        // Start migration thread
+        if let Err(e) = thread::Builder::new()
+            .name("migration".into())
+            .spawn(move || {
+                // Create a VM guard to return the VM in case of an exception
+                let vm_guard = VmGuard {
+                    vm: Some(vm),
+                    sender: sender.clone(),
+                    is_local: send_data.local,
+                    check_migration_evt: Some(check_migration_evt),
+                };
+
+                let result = {
+                    let vm_ref = vm_guard.vm.as_ref().unwrap();
+                    let mut vm_lock = vm_ref.lock().unwrap();
+
+                    Self::send_migration(
+                        &mut vm_lock,
+                        #[cfg(all(feature = "kvm", target_arch = "x86_64"))]
+                        hypervisor,
+                        send_data.clone(),
+                    )
+                };
+
+                match result {
+                    Ok(_) => {
+                        // Shutdown the VM after the migration succeeded
+                        if let Err(e) = exit_evt.write(1) {
+                            error!("Failed shutting down the VM after migration: {}", e);
+                        }
+                    }
+                    Err(e) => {
+                        error!("Migration failed: {:?}", e);
+                        // VmGuard's drop will handle the VM return
+                    }
+                }
             })
-        } else {
-            Err(MigratableError::MigrateSend(anyhow!("VM is not running")))
+        {
+            // Return VM if thread creation fails
+            if let Ok(mutex) = Arc::try_unwrap(vm_clone) {
+                if let Ok(inner_vm) = mutex.into_inner() {
+                    self.vm = Some(inner_vm);
+                }
+            }
+            return Err(MigratableError::MigrateSend(anyhow!(
+                "Failed to spawn migration thread: {}",
+                e
+            )));
         }
+        Ok(())
     }
 }