Re: [RFC PATCH v2 9/9] migration/multifd: Replace multifd_send_state->pages with client data

[Peter Xu]

On Mon, Jul 22, 2024 at 02:59:14PM -0300, Fabiano Rosas wrote:
> Multifd currently has a simple scheduling mechanism that distributes
> work to the various channels by keeping storage space within each
> channel and an extra space that is given to the client. Each time the
> client fills the space with data and calls into multifd, that space is
> given to the next idle channel and a free storage space is taken from
> the channel and given to client for the next iteration.
> 
> This means we always need (#multifd_channels + 1) memory slots to
> operate multifd.
> 
> This is fine, except that the presence of this one extra memory slot
> doesn't allow different types of payloads to be processed at the same
> time in different channels, i.e. the data type of
> multifd_send_state->pages needs to be the same as p->pages.
> 
> For each new data type different from MultiFDPage_t that is to be
> handled, this logic would need to be duplicated by adding new fields
> to multifd_send_state, to the channels and to multifd_send_pages().
> 
> Fix this situation by moving the extra slot into the client and using
> only the generic type MultiFDSendData in the multifd core.
> 
> Signed-off-by: Fabiano Rosas <farosas@suse.de>
> ---
>  migration/multifd.c | 58 ++++++++++++++++++++++++++++-----------------
>  migration/multifd.h |  2 ++
>  migration/ram.c     |  1 +
>  3 files changed, 39 insertions(+), 22 deletions(-)
> 
> diff --git a/migration/multifd.c b/migration/multifd.c
> index 4394ca6ade..0a85951d58 100644
> --- a/migration/multifd.c
> +++ b/migration/multifd.c
> @@ -49,7 +49,6 @@ typedef struct {
>  
>  struct {
>      MultiFDSendParams *params;
> -    MultiFDSendData *data;
>      /*
>       * Global number of generated multifd packets.
>       *
> @@ -97,6 +96,8 @@ struct {
>      MultiFDMethods *ops;
>  } *multifd_recv_state;
>  
> +/* TODO: move these to multifd-ram.c */
> +static MultiFDSendData *multifd_ram_send;
>  static size_t multifd_ram_payload_size(void)
>  {
>      uint32_t n = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> @@ -118,6 +119,14 @@ static MultiFDSendData *multifd_send_data_alloc(void)
>      return g_malloc0(sizeof(MultiFDPayloadType) + max_payload_size);
>  }
>  
> +void multifd_ram_save_setup(void)
> +{
> +    uint32_t n = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> +
> +    multifd_ram_send = multifd_send_data_alloc();
> +    multifd_ram_send->u.ram.allocated = n;

IIUC this line won't help, as the type is still NONE.. We may need to reset
this in multifd_pages_reset() even if it's a constant to RAM code.

Side note: looks like multifd_ram_send is leaked across the patch.

> +}
> +
>  static bool multifd_use_packets(void)
>  {
>      return !migrate_mapped_ram();
> @@ -620,7 +629,7 @@ static void multifd_send_kick_main(MultiFDSendParams *p)
>   *
>   * Returns true if succeed, false otherwise.
>   */
> -static bool multifd_send_pages(void)
> +static bool multifd_send(MultiFDSendData **send_data)
>  {
>      int i;
>      static int next_channel;
> @@ -661,11 +670,16 @@ static bool multifd_send_pages(void)
>       */
>      smp_mb_acquire();
>  
> -    assert(!p->data->u.ram.num);
> +    assert(multifd_payload_empty(p->data));
>  
> -    tmp = multifd_send_state->data;
> -    multifd_send_state->data = p->data;
> +    /*
> +     * Swap the pointers. The channel gets the client data for
> +     * transferring and the client gets back an unused data slot.
> +     */
> +    tmp = *send_data;
> +    *send_data = p->data;
>      p->data = tmp;
> +
>      /*
>       * Making sure p->data is setup before marking pending_job=true. Pairs
>       * with the qatomic_load_acquire() in multifd_send_thread().
> @@ -697,7 +711,12 @@ bool multifd_queue_page(RAMBlock *block, ram_addr_t 
> offset)
>      MultiFDPages_t *pages;
>  
>  retry:
> -    pages = &multifd_send_state->data->u.ram;
> +    pages = &multifd_ram_send->u.ram;
> +
> +    if (multifd_payload_empty(multifd_ram_send)) {
> +        multifd_pages_reset(pages);
> +        multifd_set_payload_type(multifd_ram_send, MULTIFD_PAYLOAD_RAM);
> +    }
>  
>      /* If the queue is empty, we can already enqueue now */
>      if (multifd_queue_empty(pages)) {
> @@ -715,7 +734,7 @@ retry:
>       * After flush, always retry.
>       */
>      if (pages->block != block || multifd_queue_full(pages)) {
> -        if (!multifd_send_pages()) {
> +        if (!multifd_send(&multifd_ram_send)) {
>              return false;
>          }
>          goto retry;
> @@ -833,6 +852,8 @@ static bool 
> multifd_send_cleanup_channel(MultiFDSendParams *p, Error **errp)
>      g_free(p->packet);
>      p->packet = NULL;
>      multifd_send_state->ops->send_cleanup(p, errp);
> +    g_free(p->data);
> +    p->data = NULL;

These two lines look superfluous.

>  
>      return *errp == NULL;
>  }
> @@ -845,8 +866,6 @@ static void multifd_send_cleanup_state(void)
>      qemu_sem_destroy(&multifd_send_state->channels_ready);
>      g_free(multifd_send_state->params);
>      multifd_send_state->params = NULL;
> -    g_free(multifd_send_state->data);
> -    multifd_send_state->data = NULL;
>      g_free(multifd_send_state);
>      multifd_send_state = NULL;
>  }
> @@ -895,15 +914,14 @@ int multifd_send_sync_main(void)
>  {
>      int i;
>      bool flush_zero_copy;
> -    MultiFDPages_t *pages;
>  
>      if (!migrate_multifd()) {
>          return 0;
>      }
> -    pages = &multifd_send_state->data->u.ram;
> -    if (pages->num) {
> -        if (!multifd_send_pages()) {
> -            error_report("%s: multifd_send_pages fail", __func__);
> +
> +    if (!multifd_payload_empty(multifd_ram_send)) {
> +        if (!multifd_send(&multifd_ram_send)) {
> +            error_report("%s: multifd_send fail", __func__);
>              return -1;
>          }
>      }
> @@ -977,13 +995,11 @@ static void *multifd_send_thread(void *opaque)
>  
>          /*
>           * Read pending_job flag before p->data.  Pairs with the
> -         * qatomic_store_release() in multifd_send_pages().
> +         * qatomic_store_release() in multifd_send().
>           */
>          if (qatomic_load_acquire(&p->pending_job)) {
> -            MultiFDPages_t *pages = &p->data->u.ram;
> -
>              p->iovs_num = 0;
> -            assert(pages->num);
> +            assert(!multifd_payload_empty(p->data));
>  
>              ret = multifd_send_state->ops->send_prepare(p, &local_err);
>              if (ret != 0) {
> @@ -1006,13 +1022,13 @@ static void *multifd_send_thread(void *opaque)
>              stat64_add(&mig_stats.multifd_bytes,
>                         p->next_packet_size + p->packet_len);
>  
> -            multifd_pages_reset(pages);
>              p->next_packet_size = 0;
> +            multifd_set_payload_type(p->data, MULTIFD_PAYLOAD_NONE);
>  
>              /*
>               * Making sure p->data is published before saying "we're
>               * free".  Pairs with the smp_mb_acquire() in
> -             * multifd_send_pages().
> +             * multifd_send().
>               */
>              qatomic_store_release(&p->pending_job, false);
>          } else {
> @@ -1206,8 +1222,6 @@ bool multifd_send_setup(void)
>      thread_count = migrate_multifd_channels();
>      multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
>      multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
> -    multifd_send_state->data = multifd_send_data_alloc();
> -    multifd_send_state->data->u.ram.allocated = page_count;
>      qemu_sem_init(&multifd_send_state->channels_created, 0);
>      qemu_sem_init(&multifd_send_state->channels_ready, 0);
>      qatomic_set(&multifd_send_state->exiting, 0);
> diff --git a/migration/multifd.h b/migration/multifd.h
> index c9c01579a0..04c000f435 100644
> --- a/migration/multifd.h
> +++ b/migration/multifd.h
> @@ -113,6 +113,8 @@ static inline void 
> multifd_set_payload_type(MultiFDSendData *data,
>      data->type = type;
>  }
>  
> +void multifd_ram_save_setup(void);
> +
>  typedef struct {
>      /* Fields are only written at creating/deletion time */
>      /* No lock required for them, they are read only */
> diff --git a/migration/ram.c b/migration/ram.c
> index edec1a2d07..2b90396b3c 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -3058,6 +3058,7 @@ static int ram_save_setup(QEMUFile *f, void *opaque, 
> Error **errp)
>      migration_ops = g_malloc0(sizeof(MigrationOps));
>  
>      if (migrate_multifd()) {
> +        multifd_ram_save_setup();
>          migration_ops->ram_save_target_page = ram_save_target_page_multifd;
>      } else {
>          migration_ops->ram_save_target_page = ram_save_target_page_legacy;
> -- 
> 2.35.3
> 

-- 
Peter Xu