vmm: Wire postcopy live migration from source VM

Wire up the source side of postcopy migration over TCP. When
`mode=postcopy` is requested on vm.send-migration, the source skips
the pre-copy dirty-tracking loop and lets the destination resume early,
then serves guest pages on demand over a dedicated connection.

Signed-off-by: Sebastien Boeuf <sboeuf@meta.com>
Assisted-by: Claude:claude-opus-4-7

vmm/src/lib.rs

@@ -37,8 +37,8 @@ use serde::{Deserialize, Serialize};
 use signal_hook::iterator::{Handle, Signals};
 use thiserror::Error;
 use tracer::trace_scoped;
-use vm_memory::GuestMemoryAtomic;
 use vm_memory::bitmap::AtomicBitmap;
+use vm_memory::{Bytes, GuestAddress, GuestAddressSpace, GuestMemoryAtomic};
 use vm_migration::protocol::*;
 use vm_migration::{
     MemoryMigrationContext, Migratable, MigratableError, OngoingMigrationContext, Pausable,
@@ -1613,7 +1613,9 @@ impl Vmm {
         // Let every Migratable object know about the migration being started.
         vm.start_migration()?;
 
-        if send_data_migration.local {
+        if send_data_migration.local
+            || matches!(send_data_migration.memory_mode, MigrationMode::Postcopy)
+        {
             // Now pause VM (skip if already paused, e.g. migrating a paused VM)
             let downtime_begin = Instant::now();
             if vm.get_state() != VmState::Paused {
@@ -1656,12 +1658,34 @@ impl Vmm {
         vm.release_disk_locks()
             .map_err(|e| MigratableError::UnlockError(anyhow!("{e}")))?;
 
+        // For postcopy, serve faults before sending State so the destination
+        // can fault pages in during restore.
+        let postcopy_handle = if matches!(send_data_migration.memory_mode, MigrationMode::Postcopy)
+        {
+            let fault_stream = transport::open_fault_connection(
+                &send_data_migration.destination_url,
+                send_data_migration.tls_dir.as_deref(),
+            )?;
+            let guest_memory = vm.guest_memory();
+            let handle = thread::Builder::new()
+                .name("migrate-send-postcopy".to_owned())
+                .spawn(move || Self::serve_postcopy(fault_stream, guest_memory))
+                .map_err(|e| {
+                    MigratableError::MigrateSend(anyhow!("spawning postcopy serve thread: {e}"))
+                })?;
+            Some(handle)
+        } else {
+            None
+        };
+
         let (vm_snapshot, snapshot_duration) = measure_ok(|| {
             // Capture snapshot. This may have side effects, e.g. vhost-user backend inflight drain
             let snapshot = vm.snapshot()?;
 
             // One final memory iteration to handle side effects from snapshot.
-            if !send_data_migration.local {
+            if !send_data_migration.local
+                && !matches!(send_data_migration.memory_mode, MigrationMode::Postcopy)
+            {
                 let memory_ranges = vm.dirty_log()?;
                 transport::send_memory_ranges(&vm.guest_memory(), &memory_ranges, &mut socket)?;
             }
@@ -1701,14 +1725,92 @@ impl Vmm {
         debug!("Downtime breakdown: {}", ctx.downtime_ctx);
 
         // Stop logging dirty pages
-        if !send_data_migration.local {
+        if !send_data_migration.local
+            && !matches!(send_data_migration.memory_mode, MigrationMode::Postcopy)
+        {
             vm.stop_dirty_log()?;
         }
 
+        // Wait for the serve thread to drain every page
+        if let Some(handle) = postcopy_handle {
+            handle.join().map_err(|e| {
+                MigratableError::MigrateSend(anyhow!("postcopy serve thread panicked: {e:?}"))
+            })??;
+            // Signal that postcopy has drained every page to the destination.
+            event!("vm", "postcopy-migration-completed");
+        }
+
         // Let every Migratable object know about the migration being complete
         vm.complete_migration()
     }
 
+    /// Serve `Command::PageFault` requests from local guest memory on the fault
+    /// connection until the destination closes it. Runs on its own thread.
+    #[expect(
+        clippy::needless_pass_by_value,
+        reason = "runs on a dedicated thread and must own its arguments"
+    )]
+    fn serve_postcopy(
+        mut socket: SocketStream,
+        guest_memory: GuestMemoryAtomic<GuestMemoryMmap>,
+    ) -> result::Result<(), MigratableError> {
+        let mut buf: Vec<u8> = Vec::new();
+        info!("Postcopy: source entering PageFault serve loop");
+
+        loop {
+            let req = match Request::read_from(&mut socket) {
+                Ok(r) => r,
+                Err(MigratableError::MigrateSocket(e))
+                    if matches!(
+                        e.kind(),
+                        io::ErrorKind::UnexpectedEof | io::ErrorKind::BrokenPipe
+                    ) =>
+                {
+                    info!("Postcopy: destination closed the fault connection — drain complete");
+                    return Ok(());
+                }
+                Err(e) => return Err(e),
+            };
+
+            match req.command() {
+                Command::PageFault => {
+                    let range = MemoryRange::read_from(&mut socket)?;
+                    let len = range.length as usize;
+                    const MAX_PAGE: usize = 1 << 30; // 1 GiB
+                    if len == 0 || len > MAX_PAGE {
+                        return Err(MigratableError::MigrateSend(anyhow!(
+                            "Postcopy: invalid page length {len}"
+                        )));
+                    }
+                    buf.resize(len, 0);
+                    let mem = guest_memory.memory();
+                    mem.read_slice(&mut buf[..len], GuestAddress(range.gpa))
+                        .map_err(|e| {
+                            MigratableError::MigrateSend(anyhow!(
+                                "Postcopy: reading guest memory gpa={:#x} len={}: {e}",
+                                range.gpa,
+                                range.length
+                            ))
+                        })?;
+                    Response::new(Status::Ok, range.length).write_to(&mut socket)?;
+                    socket
+                        .write_all(&buf[..len])
+                        .map_err(MigratableError::MigrateSocket)?;
+                }
+                Command::Abandon => {
+                    Response::ok().write_to(&mut socket)?;
+                    info!("Postcopy: received Abandon, exiting serve loop");
+                    return Ok(());
+                }
+                c => {
+                    return Err(MigratableError::MigrateSend(anyhow!(
+                        "Postcopy: unexpected command in serve loop: {c:?}",
+                    )));
+                }
+            }
+        }
+    }
+
     #[cfg(all(feature = "kvm", target_arch = "x86_64"))]
     fn vm_check_cpuid_compatibility(
         &self,

vmm/src/migration/transport.rs

@@ -958,6 +958,22 @@ pub(crate) fn send_migration_socket(
     }
 }
 
+/// Open a dedicated fault connection with the destination VM and announce
+/// the [`ConnectionRole::Fault`] role. Used by the source to serve `PageFault`
+/// requests asynchronously.
+pub(crate) fn open_fault_connection(
+    destination_url: &str,
+    tls_dir: Option<&Path>,
+) -> Result<SocketStream, MigratableError> {
+    let mut socket = send_migration_socket(destination_url, tls_dir)?;
+    ConnectionRole::Fault.write_to(&mut socket)?;
+    // Enable fault request/response round-trips.
+    socket
+        .set_nodelay(true)
+        .map_err(MigratableError::MigrateSocket)?;
+    Ok(socket)
+}
+
 /// Bind a migration listener for the receiver side.
 pub(crate) fn receive_migration_listener(
     receiver_url: &str,