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

[Fabiano Rosas]

Peter Xu <peterx@redhat.com> writes:

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

I could maybe just hardcode it in the packet. No point setting this
every time.

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

Hm I'll look into it, thanks.

>> +}
>> +
>>  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.
>

Rebase mistake, sorry.

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