Backport Asynchronization Series from Upstream to Align Patchset

cloud-hypervisor/tests/integration.rs

@@ -6675,6 +6675,8 @@ mod common_parallel {
 
         let src_vm_path = clh_command("cloud-hypervisor");
         let src_api_socket = temp_api_path(&guest.tmp_dir);
+        let event_path = temp_event_monitor_path(&guest.tmp_dir);
+        let src_event_path = format!("{event_path}.src");
         let mut src_vm_cmd = GuestCommand::new_with_binary_path(&guest, &src_vm_path);
         src_vm_cmd
             .args(["--cpus", format!("boot={boot_vcpus}").as_str()])
@@ -6684,6 +6686,7 @@ mod common_parallel {
             .default_disks()
             .args(["--net", net_params.as_str()])
             .args(["--api-socket", &src_api_socket])
+            .args(["--event-monitor", format!("path={src_event_path}").as_str()])
             .capture_output();
         let mut src_child = src_vm_cmd.spawn().unwrap();
 
@@ -6731,6 +6734,7 @@ mod common_parallel {
                 .args([
                     &format!("--api-socket={src_api_socket}"),
                     "send-migration",
+                    "--dispatch",
                     &format!(
                         "destination_url=tcp:{host_ip}:{migration_port},downtime_ms=1,timeout_s=1,timeout_strategy={timeout_strategy:?}"
                     ),
@@ -6748,6 +6752,18 @@ mod common_parallel {
                 .wait_timeout(Duration::from_secs(60))
                 .unwrap();
 
+            let send_dispatched = match send_status {
+                Some(status) => status.success(),
+                None => {
+                    let _ = send_migration.kill();
+                    false
+                }
+            };
+            assert!(
+                send_dispatched,
+                "send-migration should have dispatched successfully"
+            );
+
             // Clean up receive-migration regardless of its outcome
             if receive_status.is_none() {
                 let _ = receive_migration.kill();
@@ -6759,19 +6775,21 @@ mod common_parallel {
 
             match timeout_strategy {
                 TimeoutStrategy::Cancel => {
-                    // With cancel strategy the send must fail and the source VM
-                    // must keep running.
-                    let send_failed = match send_status {
-                        Some(status) => !status.success(),
-                        None => {
-                            let _ = send_migration.kill();
-                            false
-                        }
-                    };
-                    assert!(
-                        send_failed,
-                        "send-migration should have failed due to 1s timeout with cancel strategy"
-                    );
+                    let expected_events = [
+                        &MetaEvent {
+                            event: "migration-started".to_string(),
+                            device_id: None,
+                        },
+                        &MetaEvent {
+                            event: "migration-failed".to_string(),
+                            device_id: None,
+                        },
+                    ];
+                    assert!(wait_for_sequential_events(
+                        Duration::from_secs(30),
+                        &expected_events,
+                        &src_event_path
+                    ));
 
                     thread::sleep(Duration::from_secs(2));
                     assert!(
@@ -6783,19 +6801,21 @@ mod common_parallel {
                     assert_eq!(guest.get_cpu_count().unwrap_or_default(), boot_vcpus);
                 }
                 TimeoutStrategy::Ignore => {
-                    // With Ignore strategy the send must succeed despite the timeout
-                    // being reached, and the source VM must have terminated.
-                    let send_succeeded = match send_status {
-                        Some(status) => status.success(),
-                        None => {
-                            let _ = send_migration.kill();
-                            false
-                        }
-                    };
-                    assert!(
-                        send_succeeded,
-                        "send-migration should have succeeded with timeout_strategy=ignore"
-                    );
+                    let expected_events = [
+                        &MetaEvent {
+                            event: "migration-started".to_string(),
+                            device_id: None,
+                        },
+                        &MetaEvent {
+                            event: "migration-finished".to_string(),
+                            device_id: None,
+                        },
+                    ];
+                    assert!(wait_for_sequential_events(
+                        Duration::from_secs(30),
+                        &expected_events,
+                        &src_event_path
+                    ));
 
                     thread::sleep(Duration::from_secs(3));
                     assert!(

fuzz/fuzz_targets/http_api.rs

@@ -306,17 +306,17 @@ impl RequestHandler for StubApiRequestHandler {
         Ok(())
     }
 
+    fn vm_post_migration_announce(&mut self) -> Result<(), VmError> {
+        Ok(())
+    }
+
     fn vm_migration_progress(&mut self) -> Option<MigrationProgress> {
         None
     }
 
     fn vm_cancel_migration(&mut self) -> Result<(), MigratableError> {
         Ok(())
     }
-
-    fn vm_post_migration_announce(&mut self) -> Result<(), VmError> {
-        Ok(())
-    }
 }
 
 fn http_receiver_stub(exit_evt: EventFd, api_evt: EventFd, api_receiver: Receiver<ApiRequest>) {

vmm/src/api/http/http_endpoint.rs

@@ -37,7 +37,6 @@
 use std::fs::File;
 use std::sync::mpsc::Sender;
 
-use log::info;
 use micro_http::{Body, Method, Request, Response, StatusCode, Version};
 use vmm_sys_util::eventfd::EventFd;
 
@@ -433,6 +432,7 @@ vm_action_put_handler_body!(VmRemoveDevice);
 vm_action_put_handler_body!(VmResizeDisk);
 vm_action_put_handler_body!(VmResizeZone);
 vm_action_put_handler_body!(VmSnapshot);
+vm_action_put_handler_body!(VmSendMigration);
 
 #[cfg(all(target_arch = "x86_64", feature = "guest_debug"))]
 vm_action_put_handler_body!(VmCoredump);
@@ -489,35 +489,6 @@ impl PutHandler for VmReceiveMigration {
 
 impl GetHandler for VmReceiveMigration {}
 
-// Special Handling for virtio-net Devices Backed by Network File Descriptors
-//
-// See above.
-impl PutHandler for VmSendMigration {
-    fn handle_request(
-        &'static self,
-        api_notifier: EventFd,
-        api_sender: Sender<ApiRequest>,
-        body: &Option<Body>,
-        _files: Vec<File>,
-    ) -> std::result::Result<Option<Body>, HttpError> {
-        if let Some(body) = body {
-            self.send(
-                api_notifier,
-                api_sender,
-                serde_json::from_slice(body.raw())?,
-            )
-            .inspect(|_| {
-                info!("live migration started (in background)");
-            })
-            .map_err(HttpError::ApiError)
-        } else {
-            Err(HttpError::BadRequest)
-        }
-    }
-}
-
-impl GetHandler for VmSendMigration {}
-
 impl PutHandler for VmResize {
     fn handle_request(
         &'static self,

vmm/src/api/mod.rs

@@ -55,7 +55,7 @@ pub use self::http::{start_http_fd_thread, start_http_path_thread};
 use crate::Error as VmmError;
 use crate::config::{RestoreConfig, RestoredNetConfig};
 use crate::device_tree::DeviceTree;
-use crate::migration_transport::MAX_MIGRATION_CONNECTIONS;
+use crate::migration::transport::MAX_MIGRATION_CONNECTIONS;
 use crate::vm::{Error as VmError, VmState};
 use crate::vm_config::{
     DeviceConfig, DiskConfig, FsConfig, GenericVhostUserConfig, MemoryZoneConfig, NetConfig,