Peter Xu <pet...@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 <faro...@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
>>
