[PATCH v7 5/5] migration: add MULTIFD_RECV_SYNC migration command

[Prasad Pandit <ppandit@redhat.com>]

From: Prasad Pandit <pjp@fedoraproject.org>

When Multifd and Postcopy migration is enabled together,
before starting Postcopy phase, Multifd threads are shutdown.

But before this shutdown, we need to flush the Multifd channels
on the source side and notify the destination migration thread
to synchronise with the Multifd 'recv_x' threads, so that all
the Multifd data is received and processed on the destination
side, before Multifd threads are shutdown.

This synchronisation happens when the main migration thread
waits for all the Multifd threads to receive their data.

Add 'MULTIFD_RECV_SYNC' migration command to notify the
destination side to synchronise with Multifd threads.

Suggested-by: Peter Xu <peterx@redhat.com>
Signed-off-by: Prasad Pandit <pjp@fedoraproject.org>
---
 migration/migration.c      |  3 +++
 migration/multifd-nocomp.c | 21 +++++++++++++++------
 migration/multifd.c        |  1 +
 migration/multifd.h        |  1 +
 migration/savevm.c         | 13 +++++++++++++
 migration/savevm.h         |  1 +
 6 files changed, 34 insertions(+), 6 deletions(-)

v6: New patch, not from earlier versions.
- https://lore.kernel.org/qemu-devel/20250215123119.814345-1-ppandit@redhat.com/T/#t

diff --git a/migration/migration.c b/migration/migration.c
index 5db9e18272..65fc4f5eed 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -3401,6 +3401,9 @@ static MigIterateState migration_iteration_run(MigrationState *s)
     if (!in_postcopy && must_precopy <= s->threshold_size
         && can_switchover && qatomic_read(&s->start_postcopy)) {
         if (migrate_multifd()) {
+            multifd_send_flush();
+            multifd_send_sync_main(MULTIFD_SYNC_LOCAL);
+            qemu_savevm_send_multifd_recv_sync(s->to_dst_file);
             multifd_send_shutdown();
         }
         if (postcopy_start(s, &local_err)) {
diff --git a/migration/multifd-nocomp.c b/migration/multifd-nocomp.c
index d0edec7cd1..bbe07e4f7e 100644
--- a/migration/multifd-nocomp.c
+++ b/migration/multifd-nocomp.c
@@ -334,7 +334,7 @@ retry:
      * After flush, always retry.
      */
     if (pages->block != block || multifd_queue_full(pages)) {
-        if (!multifd_send(&multifd_ram_send)) {
+        if (multifd_send_flush() < 0) {
             return false;
         }
         goto retry;
@@ -387,6 +387,18 @@ bool multifd_ram_sync_per_round(void)
     return !migrate_multifd_flush_after_each_section();
 }
 
+int multifd_send_flush(void)
+{
+    if (!multifd_payload_empty(multifd_ram_send)) {
+        if (!multifd_send(&multifd_ram_send)) {
+            error_report("%s: multifd_send fail", __func__);
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
 int multifd_ram_flush_and_sync(QEMUFile *f)
 {
     MultiFDSyncReq req;
@@ -396,11 +408,8 @@ int multifd_ram_flush_and_sync(QEMUFile *f)
         return 0;
     }
 
-    if (!multifd_payload_empty(multifd_ram_send)) {
-        if (!multifd_send(&multifd_ram_send)) {
-            error_report("%s: multifd_send fail", __func__);
-            return -1;
-        }
+    if ((ret = multifd_send_flush()) < 0) {
+        return ret;
     }
 
     /* File migrations only need to sync with threads */
diff --git a/migration/multifd.c b/migration/multifd.c
index 2bd8604ca1..8928ca2611 100644
--- a/migration/multifd.c
+++ b/migration/multifd.c
@@ -1265,6 +1265,7 @@ static void *multifd_recv_thread(void *opaque)
 
     rcu_unregister_thread();
     trace_multifd_recv_thread_end(p->id, p->packets_recved);
+    qemu_sem_post(&multifd_recv_state->sem_sync);
 
     return NULL;
 }
diff --git a/migration/multifd.h b/migration/multifd.h
index bff867ca6b..242b923633 100644
--- a/migration/multifd.h
+++ b/migration/multifd.h
@@ -361,6 +361,7 @@ static inline uint32_t multifd_ram_page_count(void)
 
 void multifd_ram_save_setup(void);
 void multifd_ram_save_cleanup(void);
+int multifd_send_flush(void);
 int multifd_ram_flush_and_sync(QEMUFile *f);
 bool multifd_ram_sync_per_round(void);
 bool multifd_ram_sync_per_section(void);
diff --git a/migration/savevm.c b/migration/savevm.c
index 4046faf009..0b71e988ba 100644
--- a/migration/savevm.c
+++ b/migration/savevm.c
@@ -37,6 +37,7 @@
 #include "migration/register.h"
 #include "migration/global_state.h"
 #include "migration/channel-block.h"
+#include "multifd.h"
 #include "ram.h"
 #include "qemu-file.h"
 #include "savevm.h"
@@ -90,6 +91,7 @@ enum qemu_vm_cmd {
     MIG_CMD_ENABLE_COLO,       /* Enable COLO */
     MIG_CMD_POSTCOPY_RESUME,   /* resume postcopy on dest */
     MIG_CMD_RECV_BITMAP,       /* Request for recved bitmap on dst */
+    MIG_CMD_MULTIFD_RECV_SYNC, /* Sync multifd recv_x and main threads */
     MIG_CMD_MAX
 };
 
@@ -109,6 +111,7 @@ static struct mig_cmd_args {
     [MIG_CMD_POSTCOPY_RESUME]  = { .len =  0, .name = "POSTCOPY_RESUME" },
     [MIG_CMD_PACKAGED]         = { .len =  4, .name = "PACKAGED" },
     [MIG_CMD_RECV_BITMAP]      = { .len = -1, .name = "RECV_BITMAP" },
+    [MIG_CMD_MULTIFD_RECV_SYNC] = { .len = 0, .name = "MULTIFD_RECV_SYNC" },
     [MIG_CMD_MAX]              = { .len = -1, .name = "MAX" },
 };
 
@@ -1201,6 +1204,12 @@ void qemu_savevm_send_recv_bitmap(QEMUFile *f, char *block_name)
     qemu_savevm_command_send(f, MIG_CMD_RECV_BITMAP, len + 1, (uint8_t *)buf);
 }
 
+void qemu_savevm_send_multifd_recv_sync(QEMUFile *f)
+{
+    /* TBD: trace_savevm_send_multifd_recv_sync(); */
+    qemu_savevm_command_send(f, MIG_CMD_MULTIFD_RECV_SYNC, 0, NULL);
+}
+
 bool qemu_savevm_state_blocked(Error **errp)
 {
     SaveStateEntry *se;
@@ -2479,6 +2488,10 @@ static int loadvm_process_command(QEMUFile *f)
     case MIG_CMD_RECV_BITMAP:
         return loadvm_handle_recv_bitmap(mis, len);
 
+    case MIG_CMD_MULTIFD_RECV_SYNC:
+        multifd_recv_sync_main();
+        break;
+
     case MIG_CMD_ENABLE_COLO:
         return loadvm_process_enable_colo(mis);
     }
diff --git a/migration/savevm.h b/migration/savevm.h
index 7957460062..91ae703925 100644
--- a/migration/savevm.h
+++ b/migration/savevm.h
@@ -53,6 +53,7 @@ void qemu_savevm_send_postcopy_listen(QEMUFile *f);
 void qemu_savevm_send_postcopy_run(QEMUFile *f);
 void qemu_savevm_send_postcopy_resume(QEMUFile *f);
 void qemu_savevm_send_recv_bitmap(QEMUFile *f, char *block_name);
+void qemu_savevm_send_multifd_recv_sync(QEMUFile *f);
 
 void qemu_savevm_send_postcopy_ram_discard(QEMUFile *f, const char *name,
                                            uint16_t len,
-- 
2.48.1