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

[Peter Xu <peterx@redhat.com>]

On Fri, Feb 28, 2025 at 05:47:49PM +0530, Prasad Pandit wrote:
> 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>

Thanks for trying to grant me the credit, but.. I suggest not having a new
message at all.  We should be able to do multifd's flush and sync before VM
stopped in postcopy_start()..  So we can drop this line.

And whatever come up with, it needs to be with patch 2 because patch 2
currently is broken itself.

> 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);

And.. this is not what I meant..

So again, there're more than one way to make sure no multifd pages will
arrive during postcopy.

What I actually think the easiest is to do flush and sync once in
postcopy_start() as mentioned above.  I think that'll serialize with
postcopy messages later on the main channel, making sure all channels are
flushed before moving to postcopy work.

If you want to stick with shutdown channels, I'm OK.  But then we at least
need one message to say "the recv side finished joining all recv threads".
That needs to be an ACK sent from dest to src, not src to dest.

>              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
> 

-- 
Peter Xu