[v3,17/21] multifd: Split multifd code into its own file
diff mbox series

Message ID 20200123115831.36842-18-quintela@redhat.com
State New
Headers show
Series
  • Multifd Migration Compression
Related show

Commit Message

Juan Quintela Jan. 23, 2020, 11:58 a.m. UTC
Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/Makefile.objs |   1 +
 migration/migration.c   |   1 +
 migration/multifd.c     | 891 ++++++++++++++++++++++++++++++++++++
 migration/multifd.h     | 139 ++++++
 migration/ram.c         | 980 +---------------------------------------
 migration/ram.h         |   7 -
 6 files changed, 1033 insertions(+), 986 deletions(-)
 create mode 100644 migration/multifd.c
 create mode 100644 migration/multifd.h

Comments

Dr. David Alan Gilbert Jan. 24, 2020, 12:10 p.m. UTC | #1
* Juan Quintela (quintela@redhat.com) wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>

Good, ram.c was getting WAY too big.


Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>

> ---
>  migration/Makefile.objs |   1 +
>  migration/migration.c   |   1 +
>  migration/multifd.c     | 891 ++++++++++++++++++++++++++++++++++++
>  migration/multifd.h     | 139 ++++++
>  migration/ram.c         | 980 +---------------------------------------
>  migration/ram.h         |   7 -
>  6 files changed, 1033 insertions(+), 986 deletions(-)
>  create mode 100644 migration/multifd.c
>  create mode 100644 migration/multifd.h
> 
> diff --git a/migration/Makefile.objs b/migration/Makefile.objs
> index a4f3bafd86..d3623d5f9b 100644
> --- a/migration/Makefile.objs
> +++ b/migration/Makefile.objs
> @@ -7,6 +7,7 @@ common-obj-y += qemu-file-channel.o
>  common-obj-y += xbzrle.o postcopy-ram.o
>  common-obj-y += qjson.o
>  common-obj-y += block-dirty-bitmap.o
> +common-obj-y += multifd.o
>  
>  common-obj-$(CONFIG_RDMA) += rdma.o
>  
> diff --git a/migration/migration.c b/migration/migration.c
> index ecb56afd50..3501bc3353 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -53,6 +53,7 @@
>  #include "monitor/monitor.h"
>  #include "net/announce.h"
>  #include "qemu/queue.h"
> +#include "multifd.h"
>  
>  #define MAX_THROTTLE  (32 << 20)      /* Migration transfer speed throttling */
>  
> diff --git a/migration/multifd.c b/migration/multifd.c
> new file mode 100644
> index 0000000000..1875bb3aaa
> --- /dev/null
> +++ b/migration/multifd.c
> @@ -0,0 +1,891 @@
> +/*
> + * Multifd common code
> + *
> + * Copyright (c) 2019-2020 Red Hat Inc
> + *
> + * Authors:
> + *  Juan Quintela <quintela@redhat.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or later.
> + * See the COPYING file in the top-level directory.
> + */
> +
> +#include "qemu/osdep.h"
> +#include "qemu/rcu.h"
> +#include "exec/target_page.h"
> +#include "sysemu/sysemu.h"
> +#include "exec/ramblock.h"
> +#include "qemu/error-report.h"
> +#include "qapi/error.h"
> +#include "ram.h"
> +#include "migration.h"
> +#include "socket.h"
> +#include "qemu-file.h"
> +#include "trace.h"
> +#include "multifd.h"
> +
> +/* Multiple fd's */
> +
> +#define MULTIFD_MAGIC 0x11223344U
> +#define MULTIFD_VERSION 1
> +
> +typedef struct {
> +    uint32_t magic;
> +    uint32_t version;
> +    unsigned char uuid[16]; /* QemuUUID */
> +    uint8_t id;
> +    uint8_t unused1[7];     /* Reserved for future use */
> +    uint64_t unused2[4];    /* Reserved for future use */
> +} __attribute__((packed)) MultiFDInit_t;
> +
> +static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
> +{
> +    MultiFDInit_t msg = {};
> +    int ret;
> +
> +    msg.magic = cpu_to_be32(MULTIFD_MAGIC);
> +    msg.version = cpu_to_be32(MULTIFD_VERSION);
> +    msg.id = p->id;
> +    memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
> +
> +    ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), errp);
> +    if (ret != 0) {
> +        return -1;
> +    }
> +    return 0;
> +}
> +
> +static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
> +{
> +    MultiFDInit_t msg;
> +    int ret;
> +
> +    ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
> +    if (ret != 0) {
> +        return -1;
> +    }
> +
> +    msg.magic = be32_to_cpu(msg.magic);
> +    msg.version = be32_to_cpu(msg.version);
> +
> +    if (msg.magic != MULTIFD_MAGIC) {
> +        error_setg(errp, "multifd: received packet magic %x "
> +                   "expected %x", msg.magic, MULTIFD_MAGIC);
> +        return -1;
> +    }
> +
> +    if (msg.version != MULTIFD_VERSION) {
> +        error_setg(errp, "multifd: received packet version %d "
> +                   "expected %d", msg.version, MULTIFD_VERSION);
> +        return -1;
> +    }
> +
> +    if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
> +        char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
> +        char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
> +
> +        error_setg(errp, "multifd: received uuid '%s' and expected "
> +                   "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
> +        g_free(uuid);
> +        g_free(msg_uuid);
> +        return -1;
> +    }
> +
> +    if (msg.id > migrate_multifd_channels()) {
> +        error_setg(errp, "multifd: received channel version %d "
> +                   "expected %d", msg.version, MULTIFD_VERSION);
> +        return -1;
> +    }
> +
> +    return msg.id;
> +}
> +
> +static MultiFDPages_t *multifd_pages_init(size_t size)
> +{
> +    MultiFDPages_t *pages = g_new0(MultiFDPages_t, 1);
> +
> +    pages->allocated = size;
> +    pages->iov = g_new0(struct iovec, size);
> +    pages->offset = g_new0(ram_addr_t, size);
> +
> +    return pages;
> +}
> +
> +static void multifd_pages_clear(MultiFDPages_t *pages)
> +{
> +    pages->used = 0;
> +    pages->allocated = 0;
> +    pages->packet_num = 0;
> +    pages->block = NULL;
> +    g_free(pages->iov);
> +    pages->iov = NULL;
> +    g_free(pages->offset);
> +    pages->offset = NULL;
> +    g_free(pages);
> +}
> +
> +static void multifd_send_fill_packet(MultiFDSendParams *p)
> +{
> +    MultiFDPacket_t *packet = p->packet;
> +    int i;
> +
> +    packet->flags = cpu_to_be32(p->flags);
> +    packet->pages_alloc = cpu_to_be32(p->pages->allocated);
> +    packet->pages_used = cpu_to_be32(p->pages->used);
> +    packet->next_packet_size = cpu_to_be32(p->next_packet_size);
> +    packet->packet_num = cpu_to_be64(p->packet_num);
> +
> +    if (p->pages->block) {
> +        strncpy(packet->ramblock, p->pages->block->idstr, 256);
> +    }
> +
> +    for (i = 0; i < p->pages->used; i++) {
> +        /* there are architectures where ram_addr_t is 32 bit */
> +        uint64_t temp = p->pages->offset[i];
> +
> +        packet->offset[i] = cpu_to_be64(temp);
> +    }
> +}
> +
> +static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
> +{
> +    MultiFDPacket_t *packet = p->packet;
> +    uint32_t pages_max = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> +    RAMBlock *block;
> +    int i;
> +
> +    packet->magic = be32_to_cpu(packet->magic);
> +    if (packet->magic != MULTIFD_MAGIC) {
> +        error_setg(errp, "multifd: received packet "
> +                   "magic %x and expected magic %x",
> +                   packet->magic, MULTIFD_MAGIC);
> +        return -1;
> +    }
> +
> +    packet->version = be32_to_cpu(packet->version);
> +    if (packet->version != MULTIFD_VERSION) {
> +        error_setg(errp, "multifd: received packet "
> +                   "version %d and expected version %d",
> +                   packet->version, MULTIFD_VERSION);
> +        return -1;
> +    }
> +
> +    p->flags = be32_to_cpu(packet->flags);
> +
> +    packet->pages_alloc = be32_to_cpu(packet->pages_alloc);
> +    /*
> +     * If we received a packet that is 100 times bigger than expected
> +     * just stop migration.  It is a magic number.
> +     */
> +    if (packet->pages_alloc > pages_max * 100) {
> +        error_setg(errp, "multifd: received packet "
> +                   "with size %d and expected a maximum size of %d",
> +                   packet->pages_alloc, pages_max * 100) ;
> +        return -1;
> +    }
> +    /*
> +     * We received a packet that is bigger than expected but inside
> +     * reasonable limits (see previous comment).  Just reallocate.
> +     */
> +    if (packet->pages_alloc > p->pages->allocated) {
> +        multifd_pages_clear(p->pages);
> +        p->pages = multifd_pages_init(packet->pages_alloc);
> +    }
> +
> +    p->pages->used = be32_to_cpu(packet->pages_used);
> +    if (p->pages->used > packet->pages_alloc) {
> +        error_setg(errp, "multifd: received packet "
> +                   "with %d pages and expected maximum pages are %d",
> +                   p->pages->used, packet->pages_alloc) ;
> +        return -1;
> +    }
> +
> +    p->next_packet_size = be32_to_cpu(packet->next_packet_size);
> +    p->packet_num = be64_to_cpu(packet->packet_num);
> +
> +    if (p->pages->used == 0) {
> +        return 0;
> +    }
> +
> +    /* make sure that ramblock is 0 terminated */
> +    packet->ramblock[255] = 0;
> +    block = qemu_ram_block_by_name(packet->ramblock);
> +    if (!block) {
> +        error_setg(errp, "multifd: unknown ram block %s",
> +                   packet->ramblock);
> +        return -1;
> +    }
> +
> +    for (i = 0; i < p->pages->used; i++) {
> +        uint64_t offset = be64_to_cpu(packet->offset[i]);
> +
> +        if (offset > (block->used_length - qemu_target_page_size())) {
> +            error_setg(errp, "multifd: offset too long %" PRIu64
> +                       " (max " RAM_ADDR_FMT ")",
> +                       offset, block->max_length);
> +            return -1;
> +        }
> +        p->pages->iov[i].iov_base = block->host + offset;
> +        p->pages->iov[i].iov_len = qemu_target_page_size();
> +    }
> +
> +    return 0;
> +}
> +
> +struct {
> +    MultiFDSendParams *params;
> +    /* array of pages to sent */
> +    MultiFDPages_t *pages;
> +    /* global number of generated multifd packets */
> +    uint64_t packet_num;
> +    /* send channels ready */
> +    QemuSemaphore channels_ready;
> +    /*
> +     * Have we already run terminate threads.  There is a race when it
> +     * happens that we got one error while we are exiting.
> +     * We will use atomic operations.  Only valid values are 0 and 1.
> +     */
> +    int exiting;
> +} *multifd_send_state;
> +
> +/*
> + * How we use multifd_send_state->pages and channel->pages?
> + *
> + * We create a pages for each channel, and a main one.  Each time that
> + * we need to send a batch of pages we interchange the ones between
> + * multifd_send_state and the channel that is sending it.  There are
> + * two reasons for that:
> + *    - to not have to do so many mallocs during migration
> + *    - to make easier to know what to free at the end of migration
> + *
> + * This way we always know who is the owner of each "pages" struct,
> + * and we don't need any locking.  It belongs to the migration thread
> + * or to the channel thread.  Switching is safe because the migration
> + * thread is using the channel mutex when changing it, and the channel
> + * have to had finish with its own, otherwise pending_job can't be
> + * false.
> + */
> +
> +static int multifd_send_pages(QEMUFile *f)
> +{
> +    int i;
> +    static int next_channel;
> +    MultiFDSendParams *p = NULL; /* make happy gcc */
> +    MultiFDPages_t *pages = multifd_send_state->pages;
> +    uint64_t transferred;
> +
> +    if (atomic_read(&multifd_send_state->exiting)) {
> +        return -1;
> +    }
> +
> +    qemu_sem_wait(&multifd_send_state->channels_ready);
> +    for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
> +        p = &multifd_send_state->params[i];
> +
> +        qemu_mutex_lock(&p->mutex);
> +        if (p->quit) {
> +            error_report("%s: channel %d has already quit!", __func__, i);
> +            qemu_mutex_unlock(&p->mutex);
> +            return -1;
> +        }
> +        if (!p->pending_job) {
> +            p->pending_job++;
> +            next_channel = (i + 1) % migrate_multifd_channels();
> +            break;
> +        }
> +        qemu_mutex_unlock(&p->mutex);
> +    }
> +    assert(!p->pages->used);
> +    assert(!p->pages->block);
> +
> +    p->packet_num = multifd_send_state->packet_num++;
> +    multifd_send_state->pages = p->pages;
> +    p->pages = pages;
> +    transferred = ((uint64_t) pages->used) * qemu_target_page_size()
> +                + p->packet_len;
> +    qemu_file_update_transfer(f, transferred);
> +    ram_counters.multifd_bytes += transferred;
> +    ram_counters.transferred += transferred;;
> +    qemu_mutex_unlock(&p->mutex);
> +    qemu_sem_post(&p->sem);
> +
> +    return 1;
> +}
> +
> +int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
> +{
> +    MultiFDPages_t *pages = multifd_send_state->pages;
> +
> +    if (!pages->block) {
> +        pages->block = block;
> +    }
> +
> +    if (pages->block == block) {
> +        pages->offset[pages->used] = offset;
> +        pages->iov[pages->used].iov_base = block->host + offset;
> +        pages->iov[pages->used].iov_len = qemu_target_page_size();
> +        pages->used++;
> +
> +        if (pages->used < pages->allocated) {
> +            return 1;
> +        }
> +    }
> +
> +    if (multifd_send_pages(f) < 0) {
> +        return -1;
> +    }
> +
> +    if (pages->block != block) {
> +        return  multifd_queue_page(f, block, offset);
> +    }
> +
> +    return 1;
> +}
> +
> +static void multifd_send_terminate_threads(Error *err)
> +{
> +    int i;
> +
> +    trace_multifd_send_terminate_threads(err != NULL);
> +
> +    if (err) {
> +        MigrationState *s = migrate_get_current();
> +        migrate_set_error(s, err);
> +        if (s->state == MIGRATION_STATUS_SETUP ||
> +            s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
> +            s->state == MIGRATION_STATUS_DEVICE ||
> +            s->state == MIGRATION_STATUS_ACTIVE) {
> +            migrate_set_state(&s->state, s->state,
> +                              MIGRATION_STATUS_FAILED);
> +        }
> +    }
> +
> +    /*
> +     * We don't want to exit each threads twice.  Depending on where
> +     * we get the error, or if there are two independent errors in two
> +     * threads at the same time, we can end calling this function
> +     * twice.
> +     */
> +    if (atomic_xchg(&multifd_send_state->exiting, 1)) {
> +        return;
> +    }
> +
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDSendParams *p = &multifd_send_state->params[i];
> +
> +        qemu_mutex_lock(&p->mutex);
> +        p->quit = true;
> +        qemu_sem_post(&p->sem);
> +        qemu_mutex_unlock(&p->mutex);
> +    }
> +}
> +
> +void multifd_save_cleanup(void)
> +{
> +    int i;
> +
> +    if (!migrate_use_multifd()) {
> +        return;
> +    }
> +    multifd_send_terminate_threads(NULL);
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDSendParams *p = &multifd_send_state->params[i];
> +
> +        if (p->running) {
> +            qemu_thread_join(&p->thread);
> +        }
> +    }
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDSendParams *p = &multifd_send_state->params[i];
> +
> +        socket_send_channel_destroy(p->c);
> +        p->c = NULL;
> +        qemu_mutex_destroy(&p->mutex);
> +        qemu_sem_destroy(&p->sem);
> +        qemu_sem_destroy(&p->sem_sync);
> +        g_free(p->name);
> +        p->name = NULL;
> +        multifd_pages_clear(p->pages);
> +        p->pages = NULL;
> +        p->packet_len = 0;
> +        g_free(p->packet);
> +        p->packet = NULL;
> +    }
> +    qemu_sem_destroy(&multifd_send_state->channels_ready);
> +    g_free(multifd_send_state->params);
> +    multifd_send_state->params = NULL;
> +    multifd_pages_clear(multifd_send_state->pages);
> +    multifd_send_state->pages = NULL;
> +    g_free(multifd_send_state);
> +    multifd_send_state = NULL;
> +}
> +
> +void multifd_send_sync_main(QEMUFile *f)
> +{
> +    int i;
> +
> +    if (!migrate_use_multifd()) {
> +        return;
> +    }
> +    if (multifd_send_state->pages->used) {
> +        if (multifd_send_pages(f) < 0) {
> +            error_report("%s: multifd_send_pages fail", __func__);
> +            return;
> +        }
> +    }
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDSendParams *p = &multifd_send_state->params[i];
> +
> +        trace_multifd_send_sync_main_signal(p->id);
> +
> +        qemu_mutex_lock(&p->mutex);
> +
> +        if (p->quit) {
> +            error_report("%s: channel %d has already quit", __func__, i);
> +            qemu_mutex_unlock(&p->mutex);
> +            return;
> +        }
> +
> +        p->packet_num = multifd_send_state->packet_num++;
> +        p->flags |= MULTIFD_FLAG_SYNC;
> +        p->pending_job++;
> +        qemu_file_update_transfer(f, p->packet_len);
> +        ram_counters.multifd_bytes += p->packet_len;
> +        ram_counters.transferred += p->packet_len;
> +        qemu_mutex_unlock(&p->mutex);
> +        qemu_sem_post(&p->sem);
> +    }
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDSendParams *p = &multifd_send_state->params[i];
> +
> +        trace_multifd_send_sync_main_wait(p->id);
> +        qemu_sem_wait(&p->sem_sync);
> +    }
> +    trace_multifd_send_sync_main(multifd_send_state->packet_num);
> +}
> +
> +static void *multifd_send_thread(void *opaque)
> +{
> +    MultiFDSendParams *p = opaque;
> +    Error *local_err = NULL;
> +    int ret = 0;
> +    uint32_t flags = 0;
> +
> +    trace_multifd_send_thread_start(p->id);
> +    rcu_register_thread();
> +
> +    if (multifd_send_initial_packet(p, &local_err) < 0) {
> +        ret = -1;
> +        goto out;
> +    }
> +    /* initial packet */
> +    p->num_packets = 1;
> +
> +    while (true) {
> +        qemu_sem_wait(&p->sem);
> +
> +        if (atomic_read(&multifd_send_state->exiting)) {
> +            break;
> +        }
> +        qemu_mutex_lock(&p->mutex);
> +
> +        if (p->pending_job) {
> +            uint32_t used = p->pages->used;
> +            uint64_t packet_num = p->packet_num;
> +            flags = p->flags;
> +
> +            p->next_packet_size = used * qemu_target_page_size();
> +            multifd_send_fill_packet(p);
> +            p->flags = 0;
> +            p->num_packets++;
> +            p->num_pages += used;
> +            p->pages->used = 0;
> +            p->pages->block = NULL;
> +            qemu_mutex_unlock(&p->mutex);
> +
> +            trace_multifd_send(p->id, packet_num, used, flags,
> +                               p->next_packet_size);
> +
> +            ret = qio_channel_write_all(p->c, (void *)p->packet,
> +                                        p->packet_len, &local_err);
> +            if (ret != 0) {
> +                break;
> +            }
> +
> +            if (used) {
> +                ret = qio_channel_writev_all(p->c, p->pages->iov,
> +                                             used, &local_err);
> +                if (ret != 0) {
> +                    break;
> +                }
> +            }
> +
> +            qemu_mutex_lock(&p->mutex);
> +            p->pending_job--;
> +            qemu_mutex_unlock(&p->mutex);
> +
> +            if (flags & MULTIFD_FLAG_SYNC) {
> +                qemu_sem_post(&p->sem_sync);
> +            }
> +            qemu_sem_post(&multifd_send_state->channels_ready);
> +        } else if (p->quit) {
> +            qemu_mutex_unlock(&p->mutex);
> +            break;
> +        } else {
> +            qemu_mutex_unlock(&p->mutex);
> +            /* sometimes there are spurious wakeups */
> +        }
> +    }
> +
> +out:
> +    if (local_err) {
> +        trace_multifd_send_error(p->id);
> +        multifd_send_terminate_threads(local_err);
> +    }
> +
> +    /*
> +     * Error happen, I will exit, but I can't just leave, tell
> +     * who pay attention to me.
> +     */
> +    if (ret != 0) {
> +        qemu_sem_post(&p->sem_sync);
> +        qemu_sem_post(&multifd_send_state->channels_ready);
> +    }
> +
> +    qemu_mutex_lock(&p->mutex);
> +    p->running = false;
> +    qemu_mutex_unlock(&p->mutex);
> +
> +    rcu_unregister_thread();
> +    trace_multifd_send_thread_end(p->id, p->num_packets, p->num_pages);
> +
> +    return NULL;
> +}
> +
> +static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
> +{
> +    MultiFDSendParams *p = opaque;
> +    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
> +    Error *local_err = NULL;
> +
> +    trace_multifd_new_send_channel_async(p->id);
> +    if (qio_task_propagate_error(task, &local_err)) {
> +        migrate_set_error(migrate_get_current(), local_err);
> +        multifd_save_cleanup();
> +    } else {
> +        p->c = QIO_CHANNEL(sioc);
> +        qio_channel_set_delay(p->c, false);
> +        p->running = true;
> +        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
> +                           QEMU_THREAD_JOINABLE);
> +    }
> +}
> +
> +int multifd_save_setup(Error **errp)
> +{
> +    int thread_count;
> +    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> +    uint8_t i;
> +
> +    if (!migrate_use_multifd()) {
> +        return 0;
> +    }
> +    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->pages = multifd_pages_init(page_count);
> +    qemu_sem_init(&multifd_send_state->channels_ready, 0);
> +    atomic_set(&multifd_send_state->exiting, 0);
> +
> +    for (i = 0; i < thread_count; i++) {
> +        MultiFDSendParams *p = &multifd_send_state->params[i];
> +
> +        qemu_mutex_init(&p->mutex);
> +        qemu_sem_init(&p->sem, 0);
> +        qemu_sem_init(&p->sem_sync, 0);
> +        p->quit = false;
> +        p->pending_job = 0;
> +        p->id = i;
> +        p->pages = multifd_pages_init(page_count);
> +        p->packet_len = sizeof(MultiFDPacket_t)
> +                      + sizeof(uint64_t) * page_count;
> +        p->packet = g_malloc0(p->packet_len);
> +        p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
> +        p->packet->version = cpu_to_be32(MULTIFD_VERSION);
> +        p->name = g_strdup_printf("multifdsend_%d", i);
> +        socket_send_channel_create(multifd_new_send_channel_async, p);
> +    }
> +    return 0;
> +}
> +
> +struct {
> +    MultiFDRecvParams *params;
> +    /* number of created threads */
> +    int count;
> +    /* syncs main thread and channels */
> +    QemuSemaphore sem_sync;
> +    /* global number of generated multifd packets */
> +    uint64_t packet_num;
> +} *multifd_recv_state;
> +
> +static void multifd_recv_terminate_threads(Error *err)
> +{
> +    int i;
> +
> +    trace_multifd_recv_terminate_threads(err != NULL);
> +
> +    if (err) {
> +        MigrationState *s = migrate_get_current();
> +        migrate_set_error(s, err);
> +        if (s->state == MIGRATION_STATUS_SETUP ||
> +            s->state == MIGRATION_STATUS_ACTIVE) {
> +            migrate_set_state(&s->state, s->state,
> +                              MIGRATION_STATUS_FAILED);
> +        }
> +    }
> +
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> +
> +        qemu_mutex_lock(&p->mutex);
> +        p->quit = true;
> +        /*
> +         * We could arrive here for two reasons:
> +         *  - normal quit, i.e. everything went fine, just finished
> +         *  - error quit: We close the channels so the channel threads
> +         *    finish the qio_channel_read_all_eof()
> +         */
> +        if (p->c) {
> +            qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
> +        }
> +        qemu_mutex_unlock(&p->mutex);
> +    }
> +}
> +
> +int multifd_load_cleanup(Error **errp)
> +{
> +    int i;
> +    int ret = 0;
> +
> +    if (!migrate_use_multifd()) {
> +        return 0;
> +    }
> +    multifd_recv_terminate_threads(NULL);
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> +
> +        if (p->running) {
> +            p->quit = true;
> +            /*
> +             * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
> +             * however try to wakeup it without harm in cleanup phase.
> +             */
> +            qemu_sem_post(&p->sem_sync);
> +            qemu_thread_join(&p->thread);
> +        }
> +    }
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> +
> +        object_unref(OBJECT(p->c));
> +        p->c = NULL;
> +        qemu_mutex_destroy(&p->mutex);
> +        qemu_sem_destroy(&p->sem_sync);
> +        g_free(p->name);
> +        p->name = NULL;
> +        multifd_pages_clear(p->pages);
> +        p->pages = NULL;
> +        p->packet_len = 0;
> +        g_free(p->packet);
> +        p->packet = NULL;
> +    }
> +    qemu_sem_destroy(&multifd_recv_state->sem_sync);
> +    g_free(multifd_recv_state->params);
> +    multifd_recv_state->params = NULL;
> +    g_free(multifd_recv_state);
> +    multifd_recv_state = NULL;
> +
> +    return ret;
> +}
> +
> +void multifd_recv_sync_main(void)
> +{
> +    int i;
> +
> +    if (!migrate_use_multifd()) {
> +        return;
> +    }
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> +
> +        trace_multifd_recv_sync_main_wait(p->id);
> +        qemu_sem_wait(&multifd_recv_state->sem_sync);
> +    }
> +    for (i = 0; i < migrate_multifd_channels(); i++) {
> +        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> +
> +        qemu_mutex_lock(&p->mutex);
> +        if (multifd_recv_state->packet_num < p->packet_num) {
> +            multifd_recv_state->packet_num = p->packet_num;
> +        }
> +        qemu_mutex_unlock(&p->mutex);
> +        trace_multifd_recv_sync_main_signal(p->id);
> +        qemu_sem_post(&p->sem_sync);
> +    }
> +    trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
> +}
> +
> +static void *multifd_recv_thread(void *opaque)
> +{
> +    MultiFDRecvParams *p = opaque;
> +    Error *local_err = NULL;
> +    int ret;
> +
> +    trace_multifd_recv_thread_start(p->id);
> +    rcu_register_thread();
> +
> +    while (true) {
> +        uint32_t used;
> +        uint32_t flags;
> +
> +        if (p->quit) {
> +            break;
> +        }
> +
> +        ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
> +                                       p->packet_len, &local_err);
> +        if (ret == 0) {   /* EOF */
> +            break;
> +        }
> +        if (ret == -1) {   /* Error */
> +            break;
> +        }
> +
> +        qemu_mutex_lock(&p->mutex);
> +        ret = multifd_recv_unfill_packet(p, &local_err);
> +        if (ret) {
> +            qemu_mutex_unlock(&p->mutex);
> +            break;
> +        }
> +
> +        used = p->pages->used;
> +        flags = p->flags;
> +        trace_multifd_recv(p->id, p->packet_num, used, flags,
> +                           p->next_packet_size);
> +        p->num_packets++;
> +        p->num_pages += used;
> +        qemu_mutex_unlock(&p->mutex);
> +
> +        if (used) {
> +            ret = qio_channel_readv_all(p->c, p->pages->iov,
> +                                        used, &local_err);
> +            if (ret != 0) {
> +                break;
> +            }
> +        }
> +
> +        if (flags & MULTIFD_FLAG_SYNC) {
> +            qemu_sem_post(&multifd_recv_state->sem_sync);
> +            qemu_sem_wait(&p->sem_sync);
> +        }
> +    }
> +
> +    if (local_err) {
> +        multifd_recv_terminate_threads(local_err);
> +    }
> +    qemu_mutex_lock(&p->mutex);
> +    p->running = false;
> +    qemu_mutex_unlock(&p->mutex);
> +
> +    rcu_unregister_thread();
> +    trace_multifd_recv_thread_end(p->id, p->num_packets, p->num_pages);
> +
> +    return NULL;
> +}
> +
> +int multifd_load_setup(Error **errp)
> +{
> +    int thread_count;
> +    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> +    uint8_t i;
> +
> +    if (!migrate_use_multifd()) {
> +        return 0;
> +    }
> +    thread_count = migrate_multifd_channels();
> +    multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
> +    multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
> +    atomic_set(&multifd_recv_state->count, 0);
> +    qemu_sem_init(&multifd_recv_state->sem_sync, 0);
> +
> +    for (i = 0; i < thread_count; i++) {
> +        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> +
> +        qemu_mutex_init(&p->mutex);
> +        qemu_sem_init(&p->sem_sync, 0);
> +        p->quit = false;
> +        p->id = i;
> +        p->pages = multifd_pages_init(page_count);
> +        p->packet_len = sizeof(MultiFDPacket_t)
> +                      + sizeof(uint64_t) * page_count;
> +        p->packet = g_malloc0(p->packet_len);
> +        p->name = g_strdup_printf("multifdrecv_%d", i);
> +    }
> +    return 0;
> +}
> +
> +bool multifd_recv_all_channels_created(void)
> +{
> +    int thread_count = migrate_multifd_channels();
> +
> +    if (!migrate_use_multifd()) {
> +        return true;
> +    }
> +
> +    return thread_count == atomic_read(&multifd_recv_state->count);
> +}
> +
> +/*
> + * Try to receive all multifd channels to get ready for the migration.
> + * - Return true and do not set @errp when correctly receving all channels;
> + * - Return false and do not set @errp when correctly receiving the current one;
> + * - Return false and set @errp when failing to receive the current channel.
> + */
> +bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
> +{
> +    MultiFDRecvParams *p;
> +    Error *local_err = NULL;
> +    int id;
> +
> +    id = multifd_recv_initial_packet(ioc, &local_err);
> +    if (id < 0) {
> +        multifd_recv_terminate_threads(local_err);
> +        error_propagate_prepend(errp, local_err,
> +                                "failed to receive packet"
> +                                " via multifd channel %d: ",
> +                                atomic_read(&multifd_recv_state->count));
> +        return false;
> +    }
> +    trace_multifd_recv_new_channel(id);
> +
> +    p = &multifd_recv_state->params[id];
> +    if (p->c != NULL) {
> +        error_setg(&local_err, "multifd: received id '%d' already setup'",
> +                   id);
> +        multifd_recv_terminate_threads(local_err);
> +        error_propagate(errp, local_err);
> +        return false;
> +    }
> +    p->c = ioc;
> +    object_ref(OBJECT(ioc));
> +    /* initial packet */
> +    p->num_packets = 1;
> +
> +    p->running = true;
> +    qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
> +                       QEMU_THREAD_JOINABLE);
> +    atomic_inc(&multifd_recv_state->count);
> +    return atomic_read(&multifd_recv_state->count) ==
> +           migrate_multifd_channels();
> +}
> +
> diff --git a/migration/multifd.h b/migration/multifd.h
> new file mode 100644
> index 0000000000..d8b0205977
> --- /dev/null
> +++ b/migration/multifd.h
> @@ -0,0 +1,139 @@
> +/*
> + * Multifd common functions
> + *
> + * Copyright (c) 2019-2020 Red Hat Inc
> + *
> + * Authors:
> + *  Juan Quintela <quintela@redhat.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or later.
> + * See the COPYING file in the top-level directory.
> + */
> +
> +#ifndef QEMU_MIGRATION_MULTIFD_H
> +#define QEMU_MIGRATION_MULTIFD_H
> +
> +int multifd_save_setup(Error **errp);
> +void multifd_save_cleanup(void);
> +int multifd_load_setup(Error **errp);
> +int multifd_load_cleanup(Error **errp);
> +bool multifd_recv_all_channels_created(void);
> +bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp);
> +void multifd_recv_sync_main(void);
> +void multifd_send_sync_main(QEMUFile *f);
> +int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset);
> +
> +#define MULTIFD_FLAG_SYNC (1 << 0)
> +
> +/* This value needs to be a multiple of qemu_target_page_size() */
> +#define MULTIFD_PACKET_SIZE (512 * 1024)
> +
> +typedef struct {
> +    uint32_t magic;
> +    uint32_t version;
> +    uint32_t flags;
> +    /* maximum number of allocated pages */
> +    uint32_t pages_alloc;
> +    uint32_t pages_used;
> +    /* size of the next packet that contains pages */
> +    uint32_t next_packet_size;
> +    uint64_t packet_num;
> +    uint64_t unused[4];    /* Reserved for future use */
> +    char ramblock[256];
> +    uint64_t offset[];
> +} __attribute__((packed)) MultiFDPacket_t;
> +
> +typedef struct {
> +    /* number of used pages */
> +    uint32_t used;
> +    /* number of allocated pages */
> +    uint32_t allocated;
> +    /* global number of generated multifd packets */
> +    uint64_t packet_num;
> +    /* offset of each page */
> +    ram_addr_t *offset;
> +    /* pointer to each page */
> +    struct iovec *iov;
> +    RAMBlock *block;
> +} MultiFDPages_t;
> +
> +typedef struct {
> +    /* this fields are not changed once the thread is created */
> +    /* channel number */
> +    uint8_t id;
> +    /* channel thread name */
> +    char *name;
> +    /* channel thread id */
> +    QemuThread thread;
> +    /* communication channel */
> +    QIOChannel *c;
> +    /* sem where to wait for more work */
> +    QemuSemaphore sem;
> +    /* this mutex protects the following parameters */
> +    QemuMutex mutex;
> +    /* is this channel thread running */
> +    bool running;
> +    /* should this thread finish */
> +    bool quit;
> +    /* thread has work to do */
> +    int pending_job;
> +    /* array of pages to sent */
> +    MultiFDPages_t *pages;
> +    /* packet allocated len */
> +    uint32_t packet_len;
> +    /* pointer to the packet */
> +    MultiFDPacket_t *packet;
> +    /* multifd flags for each packet */
> +    uint32_t flags;
> +    /* size of the next packet that contains pages */
> +    uint32_t next_packet_size;
> +    /* global number of generated multifd packets */
> +    uint64_t packet_num;
> +    /* thread local variables */
> +    /* packets sent through this channel */
> +    uint64_t num_packets;
> +    /* pages sent through this channel */
> +    uint64_t num_pages;
> +    /* syncs main thread and channels */
> +    QemuSemaphore sem_sync;
> +}  MultiFDSendParams;
> +
> +typedef struct {
> +    /* this fields are not changed once the thread is created */
> +    /* channel number */
> +    uint8_t id;
> +    /* channel thread name */
> +    char *name;
> +    /* channel thread id */
> +    QemuThread thread;
> +    /* communication channel */
> +    QIOChannel *c;
> +    /* this mutex protects the following parameters */
> +    QemuMutex mutex;
> +    /* is this channel thread running */
> +    bool running;
> +    /* should this thread finish */
> +    bool quit;
> +    /* array of pages to receive */
> +    MultiFDPages_t *pages;
> +    /* packet allocated len */
> +    uint32_t packet_len;
> +    /* pointer to the packet */
> +    MultiFDPacket_t *packet;
> +    /* multifd flags for each packet */
> +    uint32_t flags;
> +    /* global number of generated multifd packets */
> +    uint64_t packet_num;
> +    /* thread local variables */
> +    /* size of the next packet that contains pages */
> +    uint32_t next_packet_size;
> +    /* packets sent through this channel */
> +    uint64_t num_packets;
> +    /* pages sent through this channel */
> +    uint64_t num_pages;
> +    /* syncs main thread and channels */
> +    QemuSemaphore sem_sync;
> +} MultiFDRecvParams;
> +
> +#endif
> +
> diff --git a/migration/ram.c b/migration/ram.c
> index c24b4cc771..ed23ed1c7c 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -36,7 +36,6 @@
>  #include "xbzrle.h"
>  #include "ram.h"
>  #include "migration.h"
> -#include "socket.h"
>  #include "migration/register.h"
>  #include "migration/misc.h"
>  #include "qemu-file.h"
> @@ -53,9 +52,9 @@
>  #include "migration/colo.h"
>  #include "block.h"
>  #include "sysemu/sysemu.h"
> -#include "qemu/uuid.h"
>  #include "savevm.h"
>  #include "qemu/iov.h"
> +#include "multifd.h"
>  
>  /***********************************************************/
>  /* ram save/restore */
> @@ -575,983 +574,6 @@ exit:
>      return -1;
>  }
>  
> -/* Multiple fd's */
> -
> -#define MULTIFD_MAGIC 0x11223344U
> -#define MULTIFD_VERSION 1
> -
> -#define MULTIFD_FLAG_SYNC (1 << 0)
> -
> -/* This value needs to be a multiple of qemu_target_page_size() */
> -#define MULTIFD_PACKET_SIZE (512 * 1024)
> -
> -typedef struct {
> -    uint32_t magic;
> -    uint32_t version;
> -    unsigned char uuid[16]; /* QemuUUID */
> -    uint8_t id;
> -    uint8_t unused1[7];     /* Reserved for future use */
> -    uint64_t unused2[4];    /* Reserved for future use */
> -} __attribute__((packed)) MultiFDInit_t;
> -
> -typedef struct {
> -    uint32_t magic;
> -    uint32_t version;
> -    uint32_t flags;
> -    /* maximum number of allocated pages */
> -    uint32_t pages_alloc;
> -    uint32_t pages_used;
> -    /* size of the next packet that contains pages */
> -    uint32_t next_packet_size;
> -    uint64_t packet_num;
> -    uint64_t unused[4];    /* Reserved for future use */
> -    char ramblock[256];
> -    uint64_t offset[];
> -} __attribute__((packed)) MultiFDPacket_t;
> -
> -typedef struct {
> -    /* number of used pages */
> -    uint32_t used;
> -    /* number of allocated pages */
> -    uint32_t allocated;
> -    /* global number of generated multifd packets */
> -    uint64_t packet_num;
> -    /* offset of each page */
> -    ram_addr_t *offset;
> -    /* pointer to each page */
> -    struct iovec *iov;
> -    RAMBlock *block;
> -} MultiFDPages_t;
> -
> -typedef struct {
> -    /* this fields are not changed once the thread is created */
> -    /* channel number */
> -    uint8_t id;
> -    /* channel thread name */
> -    char *name;
> -    /* channel thread id */
> -    QemuThread thread;
> -    /* communication channel */
> -    QIOChannel *c;
> -    /* sem where to wait for more work */
> -    QemuSemaphore sem;
> -    /* this mutex protects the following parameters */
> -    QemuMutex mutex;
> -    /* is this channel thread running */
> -    bool running;
> -    /* should this thread finish */
> -    bool quit;
> -    /* thread has work to do */
> -    int pending_job;
> -    /* array of pages to sent */
> -    MultiFDPages_t *pages;
> -    /* packet allocated len */
> -    uint32_t packet_len;
> -    /* pointer to the packet */
> -    MultiFDPacket_t *packet;
> -    /* multifd flags for each packet */
> -    uint32_t flags;
> -    /* size of the next packet that contains pages */
> -    uint32_t next_packet_size;
> -    /* global number of generated multifd packets */
> -    uint64_t packet_num;
> -    /* thread local variables */
> -    /* packets sent through this channel */
> -    uint64_t num_packets;
> -    /* pages sent through this channel */
> -    uint64_t num_pages;
> -    /* syncs main thread and channels */
> -    QemuSemaphore sem_sync;
> -}  MultiFDSendParams;
> -
> -typedef struct {
> -    /* this fields are not changed once the thread is created */
> -    /* channel number */
> -    uint8_t id;
> -    /* channel thread name */
> -    char *name;
> -    /* channel thread id */
> -    QemuThread thread;
> -    /* communication channel */
> -    QIOChannel *c;
> -    /* this mutex protects the following parameters */
> -    QemuMutex mutex;
> -    /* is this channel thread running */
> -    bool running;
> -    /* should this thread finish */
> -    bool quit;
> -    /* array of pages to receive */
> -    MultiFDPages_t *pages;
> -    /* packet allocated len */
> -    uint32_t packet_len;
> -    /* pointer to the packet */
> -    MultiFDPacket_t *packet;
> -    /* multifd flags for each packet */
> -    uint32_t flags;
> -    /* global number of generated multifd packets */
> -    uint64_t packet_num;
> -    /* thread local variables */
> -    /* size of the next packet that contains pages */
> -    uint32_t next_packet_size;
> -    /* packets sent through this channel */
> -    uint64_t num_packets;
> -    /* pages sent through this channel */
> -    uint64_t num_pages;
> -    /* syncs main thread and channels */
> -    QemuSemaphore sem_sync;
> -} MultiFDRecvParams;
> -
> -static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
> -{
> -    MultiFDInit_t msg = {};
> -    int ret;
> -
> -    msg.magic = cpu_to_be32(MULTIFD_MAGIC);
> -    msg.version = cpu_to_be32(MULTIFD_VERSION);
> -    msg.id = p->id;
> -    memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
> -
> -    ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), errp);
> -    if (ret != 0) {
> -        return -1;
> -    }
> -    return 0;
> -}
> -
> -static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
> -{
> -    MultiFDInit_t msg;
> -    int ret;
> -
> -    ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
> -    if (ret != 0) {
> -        return -1;
> -    }
> -
> -    msg.magic = be32_to_cpu(msg.magic);
> -    msg.version = be32_to_cpu(msg.version);
> -
> -    if (msg.magic != MULTIFD_MAGIC) {
> -        error_setg(errp, "multifd: received packet magic %x "
> -                   "expected %x", msg.magic, MULTIFD_MAGIC);
> -        return -1;
> -    }
> -
> -    if (msg.version != MULTIFD_VERSION) {
> -        error_setg(errp, "multifd: received packet version %d "
> -                   "expected %d", msg.version, MULTIFD_VERSION);
> -        return -1;
> -    }
> -
> -    if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
> -        char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
> -        char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
> -
> -        error_setg(errp, "multifd: received uuid '%s' and expected "
> -                   "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
> -        g_free(uuid);
> -        g_free(msg_uuid);
> -        return -1;
> -    }
> -
> -    if (msg.id > migrate_multifd_channels()) {
> -        error_setg(errp, "multifd: received channel version %d "
> -                   "expected %d", msg.version, MULTIFD_VERSION);
> -        return -1;
> -    }
> -
> -    return msg.id;
> -}
> -
> -static MultiFDPages_t *multifd_pages_init(size_t size)
> -{
> -    MultiFDPages_t *pages = g_new0(MultiFDPages_t, 1);
> -
> -    pages->allocated = size;
> -    pages->iov = g_new0(struct iovec, size);
> -    pages->offset = g_new0(ram_addr_t, size);
> -
> -    return pages;
> -}
> -
> -static void multifd_pages_clear(MultiFDPages_t *pages)
> -{
> -    pages->used = 0;
> -    pages->allocated = 0;
> -    pages->packet_num = 0;
> -    pages->block = NULL;
> -    g_free(pages->iov);
> -    pages->iov = NULL;
> -    g_free(pages->offset);
> -    pages->offset = NULL;
> -    g_free(pages);
> -}
> -
> -static void multifd_send_fill_packet(MultiFDSendParams *p)
> -{
> -    MultiFDPacket_t *packet = p->packet;
> -    int i;
> -
> -    packet->flags = cpu_to_be32(p->flags);
> -    packet->pages_alloc = cpu_to_be32(p->pages->allocated);
> -    packet->pages_used = cpu_to_be32(p->pages->used);
> -    packet->next_packet_size = cpu_to_be32(p->next_packet_size);
> -    packet->packet_num = cpu_to_be64(p->packet_num);
> -
> -    if (p->pages->block) {
> -        strncpy(packet->ramblock, p->pages->block->idstr, 256);
> -    }
> -
> -    for (i = 0; i < p->pages->used; i++) {
> -        /* there are architectures where ram_addr_t is 32 bit */
> -        uint64_t temp = p->pages->offset[i];
> -
> -        packet->offset[i] = cpu_to_be64(temp);
> -    }
> -}
> -
> -static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
> -{
> -    MultiFDPacket_t *packet = p->packet;
> -    uint32_t pages_max = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> -    RAMBlock *block;
> -    int i;
> -
> -    packet->magic = be32_to_cpu(packet->magic);
> -    if (packet->magic != MULTIFD_MAGIC) {
> -        error_setg(errp, "multifd: received packet "
> -                   "magic %x and expected magic %x",
> -                   packet->magic, MULTIFD_MAGIC);
> -        return -1;
> -    }
> -
> -    packet->version = be32_to_cpu(packet->version);
> -    if (packet->version != MULTIFD_VERSION) {
> -        error_setg(errp, "multifd: received packet "
> -                   "version %d and expected version %d",
> -                   packet->version, MULTIFD_VERSION);
> -        return -1;
> -    }
> -
> -    p->flags = be32_to_cpu(packet->flags);
> -
> -    packet->pages_alloc = be32_to_cpu(packet->pages_alloc);
> -    /*
> -     * If we received a packet that is 100 times bigger than expected
> -     * just stop migration.  It is a magic number.
> -     */
> -    if (packet->pages_alloc > pages_max * 100) {
> -        error_setg(errp, "multifd: received packet "
> -                   "with size %d and expected a maximum size of %d",
> -                   packet->pages_alloc, pages_max * 100) ;
> -        return -1;
> -    }
> -    /*
> -     * We received a packet that is bigger than expected but inside
> -     * reasonable limits (see previous comment).  Just reallocate.
> -     */
> -    if (packet->pages_alloc > p->pages->allocated) {
> -        multifd_pages_clear(p->pages);
> -        p->pages = multifd_pages_init(packet->pages_alloc);
> -    }
> -
> -    p->pages->used = be32_to_cpu(packet->pages_used);
> -    if (p->pages->used > packet->pages_alloc) {
> -        error_setg(errp, "multifd: received packet "
> -                   "with %d pages and expected maximum pages are %d",
> -                   p->pages->used, packet->pages_alloc) ;
> -        return -1;
> -    }
> -
> -    p->next_packet_size = be32_to_cpu(packet->next_packet_size);
> -    p->packet_num = be64_to_cpu(packet->packet_num);
> -
> -    if (p->pages->used == 0) {
> -        return 0;
> -    }
> -
> -    /* make sure that ramblock is 0 terminated */
> -    packet->ramblock[255] = 0;
> -    block = qemu_ram_block_by_name(packet->ramblock);
> -    if (!block) {
> -        error_setg(errp, "multifd: unknown ram block %s",
> -                   packet->ramblock);
> -        return -1;
> -    }
> -
> -    for (i = 0; i < p->pages->used; i++) {
> -        uint64_t offset = be64_to_cpu(packet->offset[i]);
> -
> -        if (offset > (block->used_length - qemu_target_page_size())) {
> -            error_setg(errp, "multifd: offset too long %" PRIu64
> -                       " (max " RAM_ADDR_FMT ")",
> -                       offset, block->max_length);
> -            return -1;
> -        }
> -        p->pages->iov[i].iov_base = block->host + offset;
> -        p->pages->iov[i].iov_len = qemu_target_page_size();
> -    }
> -
> -    return 0;
> -}
> -
> -struct {
> -    MultiFDSendParams *params;
> -    /* array of pages to sent */
> -    MultiFDPages_t *pages;
> -    /* global number of generated multifd packets */
> -    uint64_t packet_num;
> -    /* send channels ready */
> -    QemuSemaphore channels_ready;
> -    /*
> -     * Have we already run terminate threads.  There is a race when it
> -     * happens that we got one error while we are exiting.
> -     * We will use atomic operations.  Only valid values are 0 and 1.
> -     */
> -    int exiting;
> -} *multifd_send_state;
> -
> -/*
> - * How we use multifd_send_state->pages and channel->pages?
> - *
> - * We create a pages for each channel, and a main one.  Each time that
> - * we need to send a batch of pages we interchange the ones between
> - * multifd_send_state and the channel that is sending it.  There are
> - * two reasons for that:
> - *    - to not have to do so many mallocs during migration
> - *    - to make easier to know what to free at the end of migration
> - *
> - * This way we always know who is the owner of each "pages" struct,
> - * and we don't need any locking.  It belongs to the migration thread
> - * or to the channel thread.  Switching is safe because the migration
> - * thread is using the channel mutex when changing it, and the channel
> - * have to had finish with its own, otherwise pending_job can't be
> - * false.
> - */
> -
> -static int multifd_send_pages(QEMUFile *f)
> -{
> -    int i;
> -    static int next_channel;
> -    MultiFDSendParams *p = NULL; /* make happy gcc */
> -    MultiFDPages_t *pages = multifd_send_state->pages;
> -    uint64_t transferred;
> -
> -    if (atomic_read(&multifd_send_state->exiting)) {
> -        return -1;
> -    }
> -
> -    qemu_sem_wait(&multifd_send_state->channels_ready);
> -    for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
> -        p = &multifd_send_state->params[i];
> -
> -        qemu_mutex_lock(&p->mutex);
> -        if (p->quit) {
> -            error_report("%s: channel %d has already quit!", __func__, i);
> -            qemu_mutex_unlock(&p->mutex);
> -            return -1;
> -        }
> -        if (!p->pending_job) {
> -            p->pending_job++;
> -            next_channel = (i + 1) % migrate_multifd_channels();
> -            break;
> -        }
> -        qemu_mutex_unlock(&p->mutex);
> -    }
> -    assert(!p->pages->used);
> -    assert(!p->pages->block);
> -
> -    p->packet_num = multifd_send_state->packet_num++;
> -    multifd_send_state->pages = p->pages;
> -    p->pages = pages;
> -    transferred = ((uint64_t) pages->used) * qemu_target_page_size()
> -                + p->packet_len;
> -    qemu_file_update_transfer(f, transferred);
> -    ram_counters.multifd_bytes += transferred;
> -    ram_counters.transferred += transferred;;
> -    qemu_mutex_unlock(&p->mutex);
> -    qemu_sem_post(&p->sem);
> -
> -    return 1;
> -}
> -
> -static int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
> -{
> -    MultiFDPages_t *pages = multifd_send_state->pages;
> -
> -    if (!pages->block) {
> -        pages->block = block;
> -    }
> -
> -    if (pages->block == block) {
> -        pages->offset[pages->used] = offset;
> -        pages->iov[pages->used].iov_base = block->host + offset;
> -        pages->iov[pages->used].iov_len = qemu_target_page_size();
> -        pages->used++;
> -
> -        if (pages->used < pages->allocated) {
> -            return 1;
> -        }
> -    }
> -
> -    if (multifd_send_pages(f) < 0) {
> -        return -1;
> -    }
> -
> -    if (pages->block != block) {
> -        return  multifd_queue_page(f, block, offset);
> -    }
> -
> -    return 1;
> -}
> -
> -static void multifd_send_terminate_threads(Error *err)
> -{
> -    int i;
> -
> -    trace_multifd_send_terminate_threads(err != NULL);
> -
> -    if (err) {
> -        MigrationState *s = migrate_get_current();
> -        migrate_set_error(s, err);
> -        if (s->state == MIGRATION_STATUS_SETUP ||
> -            s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
> -            s->state == MIGRATION_STATUS_DEVICE ||
> -            s->state == MIGRATION_STATUS_ACTIVE) {
> -            migrate_set_state(&s->state, s->state,
> -                              MIGRATION_STATUS_FAILED);
> -        }
> -    }
> -
> -    /*
> -     * We don't want to exit each threads twice.  Depending on where
> -     * we get the error, or if there are two independent errors in two
> -     * threads at the same time, we can end calling this function
> -     * twice.
> -     */
> -    if (atomic_xchg(&multifd_send_state->exiting, 1)) {
> -        return;
> -    }
> -
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDSendParams *p = &multifd_send_state->params[i];
> -
> -        qemu_mutex_lock(&p->mutex);
> -        p->quit = true;
> -        qemu_sem_post(&p->sem);
> -        qemu_mutex_unlock(&p->mutex);
> -    }
> -}
> -
> -void multifd_save_cleanup(void)
> -{
> -    int i;
> -
> -    if (!migrate_use_multifd()) {
> -        return;
> -    }
> -    multifd_send_terminate_threads(NULL);
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDSendParams *p = &multifd_send_state->params[i];
> -
> -        if (p->running) {
> -            qemu_thread_join(&p->thread);
> -        }
> -    }
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDSendParams *p = &multifd_send_state->params[i];
> -
> -        socket_send_channel_destroy(p->c);
> -        p->c = NULL;
> -        qemu_mutex_destroy(&p->mutex);
> -        qemu_sem_destroy(&p->sem);
> -        qemu_sem_destroy(&p->sem_sync);
> -        g_free(p->name);
> -        p->name = NULL;
> -        multifd_pages_clear(p->pages);
> -        p->pages = NULL;
> -        p->packet_len = 0;
> -        g_free(p->packet);
> -        p->packet = NULL;
> -    }
> -    qemu_sem_destroy(&multifd_send_state->channels_ready);
> -    g_free(multifd_send_state->params);
> -    multifd_send_state->params = NULL;
> -    multifd_pages_clear(multifd_send_state->pages);
> -    multifd_send_state->pages = NULL;
> -    g_free(multifd_send_state);
> -    multifd_send_state = NULL;
> -}
> -
> -static void multifd_send_sync_main(QEMUFile *f)
> -{
> -    int i;
> -
> -    if (!migrate_use_multifd()) {
> -        return;
> -    }
> -    if (multifd_send_state->pages->used) {
> -        if (multifd_send_pages(f) < 0) {
> -            error_report("%s: multifd_send_pages fail", __func__);
> -            return;
> -        }
> -    }
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDSendParams *p = &multifd_send_state->params[i];
> -
> -        trace_multifd_send_sync_main_signal(p->id);
> -
> -        qemu_mutex_lock(&p->mutex);
> -
> -        if (p->quit) {
> -            error_report("%s: channel %d has already quit", __func__, i);
> -            qemu_mutex_unlock(&p->mutex);
> -            return;
> -        }
> -
> -        p->packet_num = multifd_send_state->packet_num++;
> -        p->flags |= MULTIFD_FLAG_SYNC;
> -        p->pending_job++;
> -        qemu_file_update_transfer(f, p->packet_len);
> -        ram_counters.multifd_bytes += p->packet_len;
> -        ram_counters.transferred += p->packet_len;
> -        qemu_mutex_unlock(&p->mutex);
> -        qemu_sem_post(&p->sem);
> -    }
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDSendParams *p = &multifd_send_state->params[i];
> -
> -        trace_multifd_send_sync_main_wait(p->id);
> -        qemu_sem_wait(&p->sem_sync);
> -    }
> -    trace_multifd_send_sync_main(multifd_send_state->packet_num);
> -}
> -
> -static void *multifd_send_thread(void *opaque)
> -{
> -    MultiFDSendParams *p = opaque;
> -    Error *local_err = NULL;
> -    int ret = 0;
> -    uint32_t flags = 0;
> -
> -    trace_multifd_send_thread_start(p->id);
> -    rcu_register_thread();
> -
> -    if (multifd_send_initial_packet(p, &local_err) < 0) {
> -        ret = -1;
> -        goto out;
> -    }
> -    /* initial packet */
> -    p->num_packets = 1;
> -
> -    while (true) {
> -        qemu_sem_wait(&p->sem);
> -
> -        if (atomic_read(&multifd_send_state->exiting)) {
> -            break;
> -        }
> -        qemu_mutex_lock(&p->mutex);
> -
> -        if (p->pending_job) {
> -            uint32_t used = p->pages->used;
> -            uint64_t packet_num = p->packet_num;
> -            flags = p->flags;
> -
> -            p->next_packet_size = used * qemu_target_page_size();
> -            multifd_send_fill_packet(p);
> -            p->flags = 0;
> -            p->num_packets++;
> -            p->num_pages += used;
> -            p->pages->used = 0;
> -            p->pages->block = NULL;
> -            qemu_mutex_unlock(&p->mutex);
> -
> -            trace_multifd_send(p->id, packet_num, used, flags,
> -                               p->next_packet_size);
> -
> -            ret = qio_channel_write_all(p->c, (void *)p->packet,
> -                                        p->packet_len, &local_err);
> -            if (ret != 0) {
> -                break;
> -            }
> -
> -            if (used) {
> -                ret = qio_channel_writev_all(p->c, p->pages->iov,
> -                                             used, &local_err);
> -                if (ret != 0) {
> -                    break;
> -                }
> -            }
> -
> -            qemu_mutex_lock(&p->mutex);
> -            p->pending_job--;
> -            qemu_mutex_unlock(&p->mutex);
> -
> -            if (flags & MULTIFD_FLAG_SYNC) {
> -                qemu_sem_post(&p->sem_sync);
> -            }
> -            qemu_sem_post(&multifd_send_state->channels_ready);
> -        } else if (p->quit) {
> -            qemu_mutex_unlock(&p->mutex);
> -            break;
> -        } else {
> -            qemu_mutex_unlock(&p->mutex);
> -            /* sometimes there are spurious wakeups */
> -        }
> -    }
> -
> -out:
> -    if (local_err) {
> -        trace_multifd_send_error(p->id);
> -        multifd_send_terminate_threads(local_err);
> -    }
> -
> -    /*
> -     * Error happen, I will exit, but I can't just leave, tell
> -     * who pay attention to me.
> -     */
> -    if (ret != 0) {
> -        qemu_sem_post(&p->sem_sync);
> -        qemu_sem_post(&multifd_send_state->channels_ready);
> -    }
> -
> -    qemu_mutex_lock(&p->mutex);
> -    p->running = false;
> -    qemu_mutex_unlock(&p->mutex);
> -
> -    rcu_unregister_thread();
> -    trace_multifd_send_thread_end(p->id, p->num_packets, p->num_pages);
> -
> -    return NULL;
> -}
> -
> -static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
> -{
> -    MultiFDSendParams *p = opaque;
> -    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
> -    Error *local_err = NULL;
> -
> -    trace_multifd_new_send_channel_async(p->id);
> -    if (qio_task_propagate_error(task, &local_err)) {
> -        migrate_set_error(migrate_get_current(), local_err);
> -        multifd_save_cleanup();
> -    } else {
> -        p->c = QIO_CHANNEL(sioc);
> -        qio_channel_set_delay(p->c, false);
> -        p->running = true;
> -        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
> -                           QEMU_THREAD_JOINABLE);
> -    }
> -}
> -
> -int multifd_save_setup(Error **errp)
> -{
> -    int thread_count;
> -    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> -    uint8_t i;
> -
> -    if (!migrate_use_multifd()) {
> -        return 0;
> -    }
> -    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->pages = multifd_pages_init(page_count);
> -    qemu_sem_init(&multifd_send_state->channels_ready, 0);
> -    atomic_set(&multifd_send_state->exiting, 0);
> -
> -    for (i = 0; i < thread_count; i++) {
> -        MultiFDSendParams *p = &multifd_send_state->params[i];
> -
> -        qemu_mutex_init(&p->mutex);
> -        qemu_sem_init(&p->sem, 0);
> -        qemu_sem_init(&p->sem_sync, 0);
> -        p->quit = false;
> -        p->pending_job = 0;
> -        p->id = i;
> -        p->pages = multifd_pages_init(page_count);
> -        p->packet_len = sizeof(MultiFDPacket_t)
> -                      + sizeof(uint64_t) * page_count;
> -        p->packet = g_malloc0(p->packet_len);
> -        p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
> -        p->packet->version = cpu_to_be32(MULTIFD_VERSION);
> -        p->name = g_strdup_printf("multifdsend_%d", i);
> -        socket_send_channel_create(multifd_new_send_channel_async, p);
> -    }
> -    return 0;
> -}
> -
> -struct {
> -    MultiFDRecvParams *params;
> -    /* number of created threads */
> -    int count;
> -    /* syncs main thread and channels */
> -    QemuSemaphore sem_sync;
> -    /* global number of generated multifd packets */
> -    uint64_t packet_num;
> -} *multifd_recv_state;
> -
> -static void multifd_recv_terminate_threads(Error *err)
> -{
> -    int i;
> -
> -    trace_multifd_recv_terminate_threads(err != NULL);
> -
> -    if (err) {
> -        MigrationState *s = migrate_get_current();
> -        migrate_set_error(s, err);
> -        if (s->state == MIGRATION_STATUS_SETUP ||
> -            s->state == MIGRATION_STATUS_ACTIVE) {
> -            migrate_set_state(&s->state, s->state,
> -                              MIGRATION_STATUS_FAILED);
> -        }
> -    }
> -
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> -
> -        qemu_mutex_lock(&p->mutex);
> -        p->quit = true;
> -        /*
> -         * We could arrive here for two reasons:
> -         *  - normal quit, i.e. everything went fine, just finished
> -         *  - error quit: We close the channels so the channel threads
> -         *    finish the qio_channel_read_all_eof()
> -         */
> -        if (p->c) {
> -            qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
> -        }
> -        qemu_mutex_unlock(&p->mutex);
> -    }
> -}
> -
> -int multifd_load_cleanup(Error **errp)
> -{
> -    int i;
> -    int ret = 0;
> -
> -    if (!migrate_use_multifd()) {
> -        return 0;
> -    }
> -    multifd_recv_terminate_threads(NULL);
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> -
> -        if (p->running) {
> -            p->quit = true;
> -            /*
> -             * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
> -             * however try to wakeup it without harm in cleanup phase.
> -             */
> -            qemu_sem_post(&p->sem_sync);
> -            qemu_thread_join(&p->thread);
> -        }
> -    }
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> -
> -        object_unref(OBJECT(p->c));
> -        p->c = NULL;
> -        qemu_mutex_destroy(&p->mutex);
> -        qemu_sem_destroy(&p->sem_sync);
> -        g_free(p->name);
> -        p->name = NULL;
> -        multifd_pages_clear(p->pages);
> -        p->pages = NULL;
> -        p->packet_len = 0;
> -        g_free(p->packet);
> -        p->packet = NULL;
> -    }
> -    qemu_sem_destroy(&multifd_recv_state->sem_sync);
> -    g_free(multifd_recv_state->params);
> -    multifd_recv_state->params = NULL;
> -    g_free(multifd_recv_state);
> -    multifd_recv_state = NULL;
> -
> -    return ret;
> -}
> -
> -static void multifd_recv_sync_main(void)
> -{
> -    int i;
> -
> -    if (!migrate_use_multifd()) {
> -        return;
> -    }
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> -
> -        trace_multifd_recv_sync_main_wait(p->id);
> -        qemu_sem_wait(&multifd_recv_state->sem_sync);
> -    }
> -    for (i = 0; i < migrate_multifd_channels(); i++) {
> -        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> -
> -        qemu_mutex_lock(&p->mutex);
> -        if (multifd_recv_state->packet_num < p->packet_num) {
> -            multifd_recv_state->packet_num = p->packet_num;
> -        }
> -        qemu_mutex_unlock(&p->mutex);
> -        trace_multifd_recv_sync_main_signal(p->id);
> -        qemu_sem_post(&p->sem_sync);
> -    }
> -    trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
> -}
> -
> -static void *multifd_recv_thread(void *opaque)
> -{
> -    MultiFDRecvParams *p = opaque;
> -    Error *local_err = NULL;
> -    int ret;
> -
> -    trace_multifd_recv_thread_start(p->id);
> -    rcu_register_thread();
> -
> -    while (true) {
> -        uint32_t used;
> -        uint32_t flags;
> -
> -        if (p->quit) {
> -            break;
> -        }
> -
> -        ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
> -                                       p->packet_len, &local_err);
> -        if (ret == 0) {   /* EOF */
> -            break;
> -        }
> -        if (ret == -1) {   /* Error */
> -            break;
> -        }
> -
> -        qemu_mutex_lock(&p->mutex);
> -        ret = multifd_recv_unfill_packet(p, &local_err);
> -        if (ret) {
> -            qemu_mutex_unlock(&p->mutex);
> -            break;
> -        }
> -
> -        used = p->pages->used;
> -        flags = p->flags;
> -        trace_multifd_recv(p->id, p->packet_num, used, flags,
> -                           p->next_packet_size);
> -        p->num_packets++;
> -        p->num_pages += used;
> -        qemu_mutex_unlock(&p->mutex);
> -
> -        if (used) {
> -            ret = qio_channel_readv_all(p->c, p->pages->iov,
> -                                        used, &local_err);
> -            if (ret != 0) {
> -                break;
> -            }
> -        }
> -
> -        if (flags & MULTIFD_FLAG_SYNC) {
> -            qemu_sem_post(&multifd_recv_state->sem_sync);
> -            qemu_sem_wait(&p->sem_sync);
> -        }
> -    }
> -
> -    if (local_err) {
> -        multifd_recv_terminate_threads(local_err);
> -    }
> -    qemu_mutex_lock(&p->mutex);
> -    p->running = false;
> -    qemu_mutex_unlock(&p->mutex);
> -
> -    rcu_unregister_thread();
> -    trace_multifd_recv_thread_end(p->id, p->num_packets, p->num_pages);
> -
> -    return NULL;
> -}
> -
> -int multifd_load_setup(Error **errp)
> -{
> -    int thread_count;
> -    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> -    uint8_t i;
> -
> -    if (!migrate_use_multifd()) {
> -        return 0;
> -    }
> -    thread_count = migrate_multifd_channels();
> -    multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
> -    multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
> -    atomic_set(&multifd_recv_state->count, 0);
> -    qemu_sem_init(&multifd_recv_state->sem_sync, 0);
> -
> -    for (i = 0; i < thread_count; i++) {
> -        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> -
> -        qemu_mutex_init(&p->mutex);
> -        qemu_sem_init(&p->sem_sync, 0);
> -        p->quit = false;
> -        p->id = i;
> -        p->pages = multifd_pages_init(page_count);
> -        p->packet_len = sizeof(MultiFDPacket_t)
> -                      + sizeof(uint64_t) * page_count;
> -        p->packet = g_malloc0(p->packet_len);
> -        p->name = g_strdup_printf("multifdrecv_%d", i);
> -    }
> -    return 0;
> -}
> -
> -bool multifd_recv_all_channels_created(void)
> -{
> -    int thread_count = migrate_multifd_channels();
> -
> -    if (!migrate_use_multifd()) {
> -        return true;
> -    }
> -
> -    return thread_count == atomic_read(&multifd_recv_state->count);
> -}
> -
> -/*
> - * Try to receive all multifd channels to get ready for the migration.
> - * - Return true and do not set @errp when correctly receving all channels;
> - * - Return false and do not set @errp when correctly receiving the current one;
> - * - Return false and set @errp when failing to receive the current channel.
> - */
> -bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
> -{
> -    MultiFDRecvParams *p;
> -    Error *local_err = NULL;
> -    int id;
> -
> -    id = multifd_recv_initial_packet(ioc, &local_err);
> -    if (id < 0) {
> -        multifd_recv_terminate_threads(local_err);
> -        error_propagate_prepend(errp, local_err,
> -                                "failed to receive packet"
> -                                " via multifd channel %d: ",
> -                                atomic_read(&multifd_recv_state->count));
> -        return false;
> -    }
> -    trace_multifd_recv_new_channel(id);
> -
> -    p = &multifd_recv_state->params[id];
> -    if (p->c != NULL) {
> -        error_setg(&local_err, "multifd: received id '%d' already setup'",
> -                   id);
> -        multifd_recv_terminate_threads(local_err);
> -        error_propagate(errp, local_err);
> -        return false;
> -    }
> -    p->c = ioc;
> -    object_ref(OBJECT(ioc));
> -    /* initial packet */
> -    p->num_packets = 1;
> -
> -    p->running = true;
> -    qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
> -                       QEMU_THREAD_JOINABLE);
> -    atomic_inc(&multifd_recv_state->count);
> -    return atomic_read(&multifd_recv_state->count) ==
> -           migrate_multifd_channels();
> -}
> -
>  /**
>   * save_page_header: write page header to wire
>   *
> diff --git a/migration/ram.h b/migration/ram.h
> index 42be471d52..a553d40751 100644
> --- a/migration/ram.h
> +++ b/migration/ram.h
> @@ -41,13 +41,6 @@ int xbzrle_cache_resize(int64_t new_size, Error **errp);
>  uint64_t ram_bytes_remaining(void);
>  uint64_t ram_bytes_total(void);
>  
> -int multifd_save_setup(Error **errp);
> -void multifd_save_cleanup(void);
> -int multifd_load_setup(Error **errp);
> -int multifd_load_cleanup(Error **errp);
> -bool multifd_recv_all_channels_created(void);
> -bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp);
> -
>  uint64_t ram_pagesize_summary(void);
>  int ram_save_queue_pages(const char *rbname, ram_addr_t start, ram_addr_t len);
>  void acct_update_position(QEMUFile *f, size_t size, bool zero);
> -- 
> 2.24.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Patch
diff mbox series

diff --git a/migration/Makefile.objs b/migration/Makefile.objs
index a4f3bafd86..d3623d5f9b 100644
--- a/migration/Makefile.objs
+++ b/migration/Makefile.objs
@@ -7,6 +7,7 @@  common-obj-y += qemu-file-channel.o
 common-obj-y += xbzrle.o postcopy-ram.o
 common-obj-y += qjson.o
 common-obj-y += block-dirty-bitmap.o
+common-obj-y += multifd.o
 
 common-obj-$(CONFIG_RDMA) += rdma.o
 
diff --git a/migration/migration.c b/migration/migration.c
index ecb56afd50..3501bc3353 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -53,6 +53,7 @@ 
 #include "monitor/monitor.h"
 #include "net/announce.h"
 #include "qemu/queue.h"
+#include "multifd.h"
 
 #define MAX_THROTTLE  (32 << 20)      /* Migration transfer speed throttling */
 
diff --git a/migration/multifd.c b/migration/multifd.c
new file mode 100644
index 0000000000..1875bb3aaa
--- /dev/null
+++ b/migration/multifd.c
@@ -0,0 +1,891 @@ 
+/*
+ * Multifd common code
+ *
+ * Copyright (c) 2019-2020 Red Hat Inc
+ *
+ * Authors:
+ *  Juan Quintela <quintela@redhat.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or later.
+ * See the COPYING file in the top-level directory.
+ */
+
+#include "qemu/osdep.h"
+#include "qemu/rcu.h"
+#include "exec/target_page.h"
+#include "sysemu/sysemu.h"
+#include "exec/ramblock.h"
+#include "qemu/error-report.h"
+#include "qapi/error.h"
+#include "ram.h"
+#include "migration.h"
+#include "socket.h"
+#include "qemu-file.h"
+#include "trace.h"
+#include "multifd.h"
+
+/* Multiple fd's */
+
+#define MULTIFD_MAGIC 0x11223344U
+#define MULTIFD_VERSION 1
+
+typedef struct {
+    uint32_t magic;
+    uint32_t version;
+    unsigned char uuid[16]; /* QemuUUID */
+    uint8_t id;
+    uint8_t unused1[7];     /* Reserved for future use */
+    uint64_t unused2[4];    /* Reserved for future use */
+} __attribute__((packed)) MultiFDInit_t;
+
+static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
+{
+    MultiFDInit_t msg = {};
+    int ret;
+
+    msg.magic = cpu_to_be32(MULTIFD_MAGIC);
+    msg.version = cpu_to_be32(MULTIFD_VERSION);
+    msg.id = p->id;
+    memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
+
+    ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), errp);
+    if (ret != 0) {
+        return -1;
+    }
+    return 0;
+}
+
+static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
+{
+    MultiFDInit_t msg;
+    int ret;
+
+    ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
+    if (ret != 0) {
+        return -1;
+    }
+
+    msg.magic = be32_to_cpu(msg.magic);
+    msg.version = be32_to_cpu(msg.version);
+
+    if (msg.magic != MULTIFD_MAGIC) {
+        error_setg(errp, "multifd: received packet magic %x "
+                   "expected %x", msg.magic, MULTIFD_MAGIC);
+        return -1;
+    }
+
+    if (msg.version != MULTIFD_VERSION) {
+        error_setg(errp, "multifd: received packet version %d "
+                   "expected %d", msg.version, MULTIFD_VERSION);
+        return -1;
+    }
+
+    if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
+        char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
+        char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
+
+        error_setg(errp, "multifd: received uuid '%s' and expected "
+                   "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
+        g_free(uuid);
+        g_free(msg_uuid);
+        return -1;
+    }
+
+    if (msg.id > migrate_multifd_channels()) {
+        error_setg(errp, "multifd: received channel version %d "
+                   "expected %d", msg.version, MULTIFD_VERSION);
+        return -1;
+    }
+
+    return msg.id;
+}
+
+static MultiFDPages_t *multifd_pages_init(size_t size)
+{
+    MultiFDPages_t *pages = g_new0(MultiFDPages_t, 1);
+
+    pages->allocated = size;
+    pages->iov = g_new0(struct iovec, size);
+    pages->offset = g_new0(ram_addr_t, size);
+
+    return pages;
+}
+
+static void multifd_pages_clear(MultiFDPages_t *pages)
+{
+    pages->used = 0;
+    pages->allocated = 0;
+    pages->packet_num = 0;
+    pages->block = NULL;
+    g_free(pages->iov);
+    pages->iov = NULL;
+    g_free(pages->offset);
+    pages->offset = NULL;
+    g_free(pages);
+}
+
+static void multifd_send_fill_packet(MultiFDSendParams *p)
+{
+    MultiFDPacket_t *packet = p->packet;
+    int i;
+
+    packet->flags = cpu_to_be32(p->flags);
+    packet->pages_alloc = cpu_to_be32(p->pages->allocated);
+    packet->pages_used = cpu_to_be32(p->pages->used);
+    packet->next_packet_size = cpu_to_be32(p->next_packet_size);
+    packet->packet_num = cpu_to_be64(p->packet_num);
+
+    if (p->pages->block) {
+        strncpy(packet->ramblock, p->pages->block->idstr, 256);
+    }
+
+    for (i = 0; i < p->pages->used; i++) {
+        /* there are architectures where ram_addr_t is 32 bit */
+        uint64_t temp = p->pages->offset[i];
+
+        packet->offset[i] = cpu_to_be64(temp);
+    }
+}
+
+static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
+{
+    MultiFDPacket_t *packet = p->packet;
+    uint32_t pages_max = MULTIFD_PACKET_SIZE / qemu_target_page_size();
+    RAMBlock *block;
+    int i;
+
+    packet->magic = be32_to_cpu(packet->magic);
+    if (packet->magic != MULTIFD_MAGIC) {
+        error_setg(errp, "multifd: received packet "
+                   "magic %x and expected magic %x",
+                   packet->magic, MULTIFD_MAGIC);
+        return -1;
+    }
+
+    packet->version = be32_to_cpu(packet->version);
+    if (packet->version != MULTIFD_VERSION) {
+        error_setg(errp, "multifd: received packet "
+                   "version %d and expected version %d",
+                   packet->version, MULTIFD_VERSION);
+        return -1;
+    }
+
+    p->flags = be32_to_cpu(packet->flags);
+
+    packet->pages_alloc = be32_to_cpu(packet->pages_alloc);
+    /*
+     * If we received a packet that is 100 times bigger than expected
+     * just stop migration.  It is a magic number.
+     */
+    if (packet->pages_alloc > pages_max * 100) {
+        error_setg(errp, "multifd: received packet "
+                   "with size %d and expected a maximum size of %d",
+                   packet->pages_alloc, pages_max * 100) ;
+        return -1;
+    }
+    /*
+     * We received a packet that is bigger than expected but inside
+     * reasonable limits (see previous comment).  Just reallocate.
+     */
+    if (packet->pages_alloc > p->pages->allocated) {
+        multifd_pages_clear(p->pages);
+        p->pages = multifd_pages_init(packet->pages_alloc);
+    }
+
+    p->pages->used = be32_to_cpu(packet->pages_used);
+    if (p->pages->used > packet->pages_alloc) {
+        error_setg(errp, "multifd: received packet "
+                   "with %d pages and expected maximum pages are %d",
+                   p->pages->used, packet->pages_alloc) ;
+        return -1;
+    }
+
+    p->next_packet_size = be32_to_cpu(packet->next_packet_size);
+    p->packet_num = be64_to_cpu(packet->packet_num);
+
+    if (p->pages->used == 0) {
+        return 0;
+    }
+
+    /* make sure that ramblock is 0 terminated */
+    packet->ramblock[255] = 0;
+    block = qemu_ram_block_by_name(packet->ramblock);
+    if (!block) {
+        error_setg(errp, "multifd: unknown ram block %s",
+                   packet->ramblock);
+        return -1;
+    }
+
+    for (i = 0; i < p->pages->used; i++) {
+        uint64_t offset = be64_to_cpu(packet->offset[i]);
+
+        if (offset > (block->used_length - qemu_target_page_size())) {
+            error_setg(errp, "multifd: offset too long %" PRIu64
+                       " (max " RAM_ADDR_FMT ")",
+                       offset, block->max_length);
+            return -1;
+        }
+        p->pages->iov[i].iov_base = block->host + offset;
+        p->pages->iov[i].iov_len = qemu_target_page_size();
+    }
+
+    return 0;
+}
+
+struct {
+    MultiFDSendParams *params;
+    /* array of pages to sent */
+    MultiFDPages_t *pages;
+    /* global number of generated multifd packets */
+    uint64_t packet_num;
+    /* send channels ready */
+    QemuSemaphore channels_ready;
+    /*
+     * Have we already run terminate threads.  There is a race when it
+     * happens that we got one error while we are exiting.
+     * We will use atomic operations.  Only valid values are 0 and 1.
+     */
+    int exiting;
+} *multifd_send_state;
+
+/*
+ * How we use multifd_send_state->pages and channel->pages?
+ *
+ * We create a pages for each channel, and a main one.  Each time that
+ * we need to send a batch of pages we interchange the ones between
+ * multifd_send_state and the channel that is sending it.  There are
+ * two reasons for that:
+ *    - to not have to do so many mallocs during migration
+ *    - to make easier to know what to free at the end of migration
+ *
+ * This way we always know who is the owner of each "pages" struct,
+ * and we don't need any locking.  It belongs to the migration thread
+ * or to the channel thread.  Switching is safe because the migration
+ * thread is using the channel mutex when changing it, and the channel
+ * have to had finish with its own, otherwise pending_job can't be
+ * false.
+ */
+
+static int multifd_send_pages(QEMUFile *f)
+{
+    int i;
+    static int next_channel;
+    MultiFDSendParams *p = NULL; /* make happy gcc */
+    MultiFDPages_t *pages = multifd_send_state->pages;
+    uint64_t transferred;
+
+    if (atomic_read(&multifd_send_state->exiting)) {
+        return -1;
+    }
+
+    qemu_sem_wait(&multifd_send_state->channels_ready);
+    for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
+        p = &multifd_send_state->params[i];
+
+        qemu_mutex_lock(&p->mutex);
+        if (p->quit) {
+            error_report("%s: channel %d has already quit!", __func__, i);
+            qemu_mutex_unlock(&p->mutex);
+            return -1;
+        }
+        if (!p->pending_job) {
+            p->pending_job++;
+            next_channel = (i + 1) % migrate_multifd_channels();
+            break;
+        }
+        qemu_mutex_unlock(&p->mutex);
+    }
+    assert(!p->pages->used);
+    assert(!p->pages->block);
+
+    p->packet_num = multifd_send_state->packet_num++;
+    multifd_send_state->pages = p->pages;
+    p->pages = pages;
+    transferred = ((uint64_t) pages->used) * qemu_target_page_size()
+                + p->packet_len;
+    qemu_file_update_transfer(f, transferred);
+    ram_counters.multifd_bytes += transferred;
+    ram_counters.transferred += transferred;;
+    qemu_mutex_unlock(&p->mutex);
+    qemu_sem_post(&p->sem);
+
+    return 1;
+}
+
+int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
+{
+    MultiFDPages_t *pages = multifd_send_state->pages;
+
+    if (!pages->block) {
+        pages->block = block;
+    }
+
+    if (pages->block == block) {
+        pages->offset[pages->used] = offset;
+        pages->iov[pages->used].iov_base = block->host + offset;
+        pages->iov[pages->used].iov_len = qemu_target_page_size();
+        pages->used++;
+
+        if (pages->used < pages->allocated) {
+            return 1;
+        }
+    }
+
+    if (multifd_send_pages(f) < 0) {
+        return -1;
+    }
+
+    if (pages->block != block) {
+        return  multifd_queue_page(f, block, offset);
+    }
+
+    return 1;
+}
+
+static void multifd_send_terminate_threads(Error *err)
+{
+    int i;
+
+    trace_multifd_send_terminate_threads(err != NULL);
+
+    if (err) {
+        MigrationState *s = migrate_get_current();
+        migrate_set_error(s, err);
+        if (s->state == MIGRATION_STATUS_SETUP ||
+            s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
+            s->state == MIGRATION_STATUS_DEVICE ||
+            s->state == MIGRATION_STATUS_ACTIVE) {
+            migrate_set_state(&s->state, s->state,
+                              MIGRATION_STATUS_FAILED);
+        }
+    }
+
+    /*
+     * We don't want to exit each threads twice.  Depending on where
+     * we get the error, or if there are two independent errors in two
+     * threads at the same time, we can end calling this function
+     * twice.
+     */
+    if (atomic_xchg(&multifd_send_state->exiting, 1)) {
+        return;
+    }
+
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDSendParams *p = &multifd_send_state->params[i];
+
+        qemu_mutex_lock(&p->mutex);
+        p->quit = true;
+        qemu_sem_post(&p->sem);
+        qemu_mutex_unlock(&p->mutex);
+    }
+}
+
+void multifd_save_cleanup(void)
+{
+    int i;
+
+    if (!migrate_use_multifd()) {
+        return;
+    }
+    multifd_send_terminate_threads(NULL);
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDSendParams *p = &multifd_send_state->params[i];
+
+        if (p->running) {
+            qemu_thread_join(&p->thread);
+        }
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDSendParams *p = &multifd_send_state->params[i];
+
+        socket_send_channel_destroy(p->c);
+        p->c = NULL;
+        qemu_mutex_destroy(&p->mutex);
+        qemu_sem_destroy(&p->sem);
+        qemu_sem_destroy(&p->sem_sync);
+        g_free(p->name);
+        p->name = NULL;
+        multifd_pages_clear(p->pages);
+        p->pages = NULL;
+        p->packet_len = 0;
+        g_free(p->packet);
+        p->packet = NULL;
+    }
+    qemu_sem_destroy(&multifd_send_state->channels_ready);
+    g_free(multifd_send_state->params);
+    multifd_send_state->params = NULL;
+    multifd_pages_clear(multifd_send_state->pages);
+    multifd_send_state->pages = NULL;
+    g_free(multifd_send_state);
+    multifd_send_state = NULL;
+}
+
+void multifd_send_sync_main(QEMUFile *f)
+{
+    int i;
+
+    if (!migrate_use_multifd()) {
+        return;
+    }
+    if (multifd_send_state->pages->used) {
+        if (multifd_send_pages(f) < 0) {
+            error_report("%s: multifd_send_pages fail", __func__);
+            return;
+        }
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDSendParams *p = &multifd_send_state->params[i];
+
+        trace_multifd_send_sync_main_signal(p->id);
+
+        qemu_mutex_lock(&p->mutex);
+
+        if (p->quit) {
+            error_report("%s: channel %d has already quit", __func__, i);
+            qemu_mutex_unlock(&p->mutex);
+            return;
+        }
+
+        p->packet_num = multifd_send_state->packet_num++;
+        p->flags |= MULTIFD_FLAG_SYNC;
+        p->pending_job++;
+        qemu_file_update_transfer(f, p->packet_len);
+        ram_counters.multifd_bytes += p->packet_len;
+        ram_counters.transferred += p->packet_len;
+        qemu_mutex_unlock(&p->mutex);
+        qemu_sem_post(&p->sem);
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDSendParams *p = &multifd_send_state->params[i];
+
+        trace_multifd_send_sync_main_wait(p->id);
+        qemu_sem_wait(&p->sem_sync);
+    }
+    trace_multifd_send_sync_main(multifd_send_state->packet_num);
+}
+
+static void *multifd_send_thread(void *opaque)
+{
+    MultiFDSendParams *p = opaque;
+    Error *local_err = NULL;
+    int ret = 0;
+    uint32_t flags = 0;
+
+    trace_multifd_send_thread_start(p->id);
+    rcu_register_thread();
+
+    if (multifd_send_initial_packet(p, &local_err) < 0) {
+        ret = -1;
+        goto out;
+    }
+    /* initial packet */
+    p->num_packets = 1;
+
+    while (true) {
+        qemu_sem_wait(&p->sem);
+
+        if (atomic_read(&multifd_send_state->exiting)) {
+            break;
+        }
+        qemu_mutex_lock(&p->mutex);
+
+        if (p->pending_job) {
+            uint32_t used = p->pages->used;
+            uint64_t packet_num = p->packet_num;
+            flags = p->flags;
+
+            p->next_packet_size = used * qemu_target_page_size();
+            multifd_send_fill_packet(p);
+            p->flags = 0;
+            p->num_packets++;
+            p->num_pages += used;
+            p->pages->used = 0;
+            p->pages->block = NULL;
+            qemu_mutex_unlock(&p->mutex);
+
+            trace_multifd_send(p->id, packet_num, used, flags,
+                               p->next_packet_size);
+
+            ret = qio_channel_write_all(p->c, (void *)p->packet,
+                                        p->packet_len, &local_err);
+            if (ret != 0) {
+                break;
+            }
+
+            if (used) {
+                ret = qio_channel_writev_all(p->c, p->pages->iov,
+                                             used, &local_err);
+                if (ret != 0) {
+                    break;
+                }
+            }
+
+            qemu_mutex_lock(&p->mutex);
+            p->pending_job--;
+            qemu_mutex_unlock(&p->mutex);
+
+            if (flags & MULTIFD_FLAG_SYNC) {
+                qemu_sem_post(&p->sem_sync);
+            }
+            qemu_sem_post(&multifd_send_state->channels_ready);
+        } else if (p->quit) {
+            qemu_mutex_unlock(&p->mutex);
+            break;
+        } else {
+            qemu_mutex_unlock(&p->mutex);
+            /* sometimes there are spurious wakeups */
+        }
+    }
+
+out:
+    if (local_err) {
+        trace_multifd_send_error(p->id);
+        multifd_send_terminate_threads(local_err);
+    }
+
+    /*
+     * Error happen, I will exit, but I can't just leave, tell
+     * who pay attention to me.
+     */
+    if (ret != 0) {
+        qemu_sem_post(&p->sem_sync);
+        qemu_sem_post(&multifd_send_state->channels_ready);
+    }
+
+    qemu_mutex_lock(&p->mutex);
+    p->running = false;
+    qemu_mutex_unlock(&p->mutex);
+
+    rcu_unregister_thread();
+    trace_multifd_send_thread_end(p->id, p->num_packets, p->num_pages);
+
+    return NULL;
+}
+
+static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
+{
+    MultiFDSendParams *p = opaque;
+    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
+    Error *local_err = NULL;
+
+    trace_multifd_new_send_channel_async(p->id);
+    if (qio_task_propagate_error(task, &local_err)) {
+        migrate_set_error(migrate_get_current(), local_err);
+        multifd_save_cleanup();
+    } else {
+        p->c = QIO_CHANNEL(sioc);
+        qio_channel_set_delay(p->c, false);
+        p->running = true;
+        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
+                           QEMU_THREAD_JOINABLE);
+    }
+}
+
+int multifd_save_setup(Error **errp)
+{
+    int thread_count;
+    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
+    uint8_t i;
+
+    if (!migrate_use_multifd()) {
+        return 0;
+    }
+    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->pages = multifd_pages_init(page_count);
+    qemu_sem_init(&multifd_send_state->channels_ready, 0);
+    atomic_set(&multifd_send_state->exiting, 0);
+
+    for (i = 0; i < thread_count; i++) {
+        MultiFDSendParams *p = &multifd_send_state->params[i];
+
+        qemu_mutex_init(&p->mutex);
+        qemu_sem_init(&p->sem, 0);
+        qemu_sem_init(&p->sem_sync, 0);
+        p->quit = false;
+        p->pending_job = 0;
+        p->id = i;
+        p->pages = multifd_pages_init(page_count);
+        p->packet_len = sizeof(MultiFDPacket_t)
+                      + sizeof(uint64_t) * page_count;
+        p->packet = g_malloc0(p->packet_len);
+        p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
+        p->packet->version = cpu_to_be32(MULTIFD_VERSION);
+        p->name = g_strdup_printf("multifdsend_%d", i);
+        socket_send_channel_create(multifd_new_send_channel_async, p);
+    }
+    return 0;
+}
+
+struct {
+    MultiFDRecvParams *params;
+    /* number of created threads */
+    int count;
+    /* syncs main thread and channels */
+    QemuSemaphore sem_sync;
+    /* global number of generated multifd packets */
+    uint64_t packet_num;
+} *multifd_recv_state;
+
+static void multifd_recv_terminate_threads(Error *err)
+{
+    int i;
+
+    trace_multifd_recv_terminate_threads(err != NULL);
+
+    if (err) {
+        MigrationState *s = migrate_get_current();
+        migrate_set_error(s, err);
+        if (s->state == MIGRATION_STATUS_SETUP ||
+            s->state == MIGRATION_STATUS_ACTIVE) {
+            migrate_set_state(&s->state, s->state,
+                              MIGRATION_STATUS_FAILED);
+        }
+    }
+
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDRecvParams *p = &multifd_recv_state->params[i];
+
+        qemu_mutex_lock(&p->mutex);
+        p->quit = true;
+        /*
+         * We could arrive here for two reasons:
+         *  - normal quit, i.e. everything went fine, just finished
+         *  - error quit: We close the channels so the channel threads
+         *    finish the qio_channel_read_all_eof()
+         */
+        if (p->c) {
+            qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
+        }
+        qemu_mutex_unlock(&p->mutex);
+    }
+}
+
+int multifd_load_cleanup(Error **errp)
+{
+    int i;
+    int ret = 0;
+
+    if (!migrate_use_multifd()) {
+        return 0;
+    }
+    multifd_recv_terminate_threads(NULL);
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDRecvParams *p = &multifd_recv_state->params[i];
+
+        if (p->running) {
+            p->quit = true;
+            /*
+             * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
+             * however try to wakeup it without harm in cleanup phase.
+             */
+            qemu_sem_post(&p->sem_sync);
+            qemu_thread_join(&p->thread);
+        }
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDRecvParams *p = &multifd_recv_state->params[i];
+
+        object_unref(OBJECT(p->c));
+        p->c = NULL;
+        qemu_mutex_destroy(&p->mutex);
+        qemu_sem_destroy(&p->sem_sync);
+        g_free(p->name);
+        p->name = NULL;
+        multifd_pages_clear(p->pages);
+        p->pages = NULL;
+        p->packet_len = 0;
+        g_free(p->packet);
+        p->packet = NULL;
+    }
+    qemu_sem_destroy(&multifd_recv_state->sem_sync);
+    g_free(multifd_recv_state->params);
+    multifd_recv_state->params = NULL;
+    g_free(multifd_recv_state);
+    multifd_recv_state = NULL;
+
+    return ret;
+}
+
+void multifd_recv_sync_main(void)
+{
+    int i;
+
+    if (!migrate_use_multifd()) {
+        return;
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDRecvParams *p = &multifd_recv_state->params[i];
+
+        trace_multifd_recv_sync_main_wait(p->id);
+        qemu_sem_wait(&multifd_recv_state->sem_sync);
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDRecvParams *p = &multifd_recv_state->params[i];
+
+        qemu_mutex_lock(&p->mutex);
+        if (multifd_recv_state->packet_num < p->packet_num) {
+            multifd_recv_state->packet_num = p->packet_num;
+        }
+        qemu_mutex_unlock(&p->mutex);
+        trace_multifd_recv_sync_main_signal(p->id);
+        qemu_sem_post(&p->sem_sync);
+    }
+    trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
+}
+
+static void *multifd_recv_thread(void *opaque)
+{
+    MultiFDRecvParams *p = opaque;
+    Error *local_err = NULL;
+    int ret;
+
+    trace_multifd_recv_thread_start(p->id);
+    rcu_register_thread();
+
+    while (true) {
+        uint32_t used;
+        uint32_t flags;
+
+        if (p->quit) {
+            break;
+        }
+
+        ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
+                                       p->packet_len, &local_err);
+        if (ret == 0) {   /* EOF */
+            break;
+        }
+        if (ret == -1) {   /* Error */
+            break;
+        }
+
+        qemu_mutex_lock(&p->mutex);
+        ret = multifd_recv_unfill_packet(p, &local_err);
+        if (ret) {
+            qemu_mutex_unlock(&p->mutex);
+            break;
+        }
+
+        used = p->pages->used;
+        flags = p->flags;
+        trace_multifd_recv(p->id, p->packet_num, used, flags,
+                           p->next_packet_size);
+        p->num_packets++;
+        p->num_pages += used;
+        qemu_mutex_unlock(&p->mutex);
+
+        if (used) {
+            ret = qio_channel_readv_all(p->c, p->pages->iov,
+                                        used, &local_err);
+            if (ret != 0) {
+                break;
+            }
+        }
+
+        if (flags & MULTIFD_FLAG_SYNC) {
+            qemu_sem_post(&multifd_recv_state->sem_sync);
+            qemu_sem_wait(&p->sem_sync);
+        }
+    }
+
+    if (local_err) {
+        multifd_recv_terminate_threads(local_err);
+    }
+    qemu_mutex_lock(&p->mutex);
+    p->running = false;
+    qemu_mutex_unlock(&p->mutex);
+
+    rcu_unregister_thread();
+    trace_multifd_recv_thread_end(p->id, p->num_packets, p->num_pages);
+
+    return NULL;
+}
+
+int multifd_load_setup(Error **errp)
+{
+    int thread_count;
+    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
+    uint8_t i;
+
+    if (!migrate_use_multifd()) {
+        return 0;
+    }
+    thread_count = migrate_multifd_channels();
+    multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
+    multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
+    atomic_set(&multifd_recv_state->count, 0);
+    qemu_sem_init(&multifd_recv_state->sem_sync, 0);
+
+    for (i = 0; i < thread_count; i++) {
+        MultiFDRecvParams *p = &multifd_recv_state->params[i];
+
+        qemu_mutex_init(&p->mutex);
+        qemu_sem_init(&p->sem_sync, 0);
+        p->quit = false;
+        p->id = i;
+        p->pages = multifd_pages_init(page_count);
+        p->packet_len = sizeof(MultiFDPacket_t)
+                      + sizeof(uint64_t) * page_count;
+        p->packet = g_malloc0(p->packet_len);
+        p->name = g_strdup_printf("multifdrecv_%d", i);
+    }
+    return 0;
+}
+
+bool multifd_recv_all_channels_created(void)
+{
+    int thread_count = migrate_multifd_channels();
+
+    if (!migrate_use_multifd()) {
+        return true;
+    }
+
+    return thread_count == atomic_read(&multifd_recv_state->count);
+}
+
+/*
+ * Try to receive all multifd channels to get ready for the migration.
+ * - Return true and do not set @errp when correctly receving all channels;
+ * - Return false and do not set @errp when correctly receiving the current one;
+ * - Return false and set @errp when failing to receive the current channel.
+ */
+bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
+{
+    MultiFDRecvParams *p;
+    Error *local_err = NULL;
+    int id;
+
+    id = multifd_recv_initial_packet(ioc, &local_err);
+    if (id < 0) {
+        multifd_recv_terminate_threads(local_err);
+        error_propagate_prepend(errp, local_err,
+                                "failed to receive packet"
+                                " via multifd channel %d: ",
+                                atomic_read(&multifd_recv_state->count));
+        return false;
+    }
+    trace_multifd_recv_new_channel(id);
+
+    p = &multifd_recv_state->params[id];
+    if (p->c != NULL) {
+        error_setg(&local_err, "multifd: received id '%d' already setup'",
+                   id);
+        multifd_recv_terminate_threads(local_err);
+        error_propagate(errp, local_err);
+        return false;
+    }
+    p->c = ioc;
+    object_ref(OBJECT(ioc));
+    /* initial packet */
+    p->num_packets = 1;
+
+    p->running = true;
+    qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
+                       QEMU_THREAD_JOINABLE);
+    atomic_inc(&multifd_recv_state->count);
+    return atomic_read(&multifd_recv_state->count) ==
+           migrate_multifd_channels();
+}
+
diff --git a/migration/multifd.h b/migration/multifd.h
new file mode 100644
index 0000000000..d8b0205977
--- /dev/null
+++ b/migration/multifd.h
@@ -0,0 +1,139 @@ 
+/*
+ * Multifd common functions
+ *
+ * Copyright (c) 2019-2020 Red Hat Inc
+ *
+ * Authors:
+ *  Juan Quintela <quintela@redhat.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or later.
+ * See the COPYING file in the top-level directory.
+ */
+
+#ifndef QEMU_MIGRATION_MULTIFD_H
+#define QEMU_MIGRATION_MULTIFD_H
+
+int multifd_save_setup(Error **errp);
+void multifd_save_cleanup(void);
+int multifd_load_setup(Error **errp);
+int multifd_load_cleanup(Error **errp);
+bool multifd_recv_all_channels_created(void);
+bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp);
+void multifd_recv_sync_main(void);
+void multifd_send_sync_main(QEMUFile *f);
+int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset);
+
+#define MULTIFD_FLAG_SYNC (1 << 0)
+
+/* This value needs to be a multiple of qemu_target_page_size() */
+#define MULTIFD_PACKET_SIZE (512 * 1024)
+
+typedef struct {
+    uint32_t magic;
+    uint32_t version;
+    uint32_t flags;
+    /* maximum number of allocated pages */
+    uint32_t pages_alloc;
+    uint32_t pages_used;
+    /* size of the next packet that contains pages */
+    uint32_t next_packet_size;
+    uint64_t packet_num;
+    uint64_t unused[4];    /* Reserved for future use */
+    char ramblock[256];
+    uint64_t offset[];
+} __attribute__((packed)) MultiFDPacket_t;
+
+typedef struct {
+    /* number of used pages */
+    uint32_t used;
+    /* number of allocated pages */
+    uint32_t allocated;
+    /* global number of generated multifd packets */
+    uint64_t packet_num;
+    /* offset of each page */
+    ram_addr_t *offset;
+    /* pointer to each page */
+    struct iovec *iov;
+    RAMBlock *block;
+} MultiFDPages_t;
+
+typedef struct {
+    /* this fields are not changed once the thread is created */
+    /* channel number */
+    uint8_t id;
+    /* channel thread name */
+    char *name;
+    /* channel thread id */
+    QemuThread thread;
+    /* communication channel */
+    QIOChannel *c;
+    /* sem where to wait for more work */
+    QemuSemaphore sem;
+    /* this mutex protects the following parameters */
+    QemuMutex mutex;
+    /* is this channel thread running */
+    bool running;
+    /* should this thread finish */
+    bool quit;
+    /* thread has work to do */
+    int pending_job;
+    /* array of pages to sent */
+    MultiFDPages_t *pages;
+    /* packet allocated len */
+    uint32_t packet_len;
+    /* pointer to the packet */
+    MultiFDPacket_t *packet;
+    /* multifd flags for each packet */
+    uint32_t flags;
+    /* size of the next packet that contains pages */
+    uint32_t next_packet_size;
+    /* global number of generated multifd packets */
+    uint64_t packet_num;
+    /* thread local variables */
+    /* packets sent through this channel */
+    uint64_t num_packets;
+    /* pages sent through this channel */
+    uint64_t num_pages;
+    /* syncs main thread and channels */
+    QemuSemaphore sem_sync;
+}  MultiFDSendParams;
+
+typedef struct {
+    /* this fields are not changed once the thread is created */
+    /* channel number */
+    uint8_t id;
+    /* channel thread name */
+    char *name;
+    /* channel thread id */
+    QemuThread thread;
+    /* communication channel */
+    QIOChannel *c;
+    /* this mutex protects the following parameters */
+    QemuMutex mutex;
+    /* is this channel thread running */
+    bool running;
+    /* should this thread finish */
+    bool quit;
+    /* array of pages to receive */
+    MultiFDPages_t *pages;
+    /* packet allocated len */
+    uint32_t packet_len;
+    /* pointer to the packet */
+    MultiFDPacket_t *packet;
+    /* multifd flags for each packet */
+    uint32_t flags;
+    /* global number of generated multifd packets */
+    uint64_t packet_num;
+    /* thread local variables */
+    /* size of the next packet that contains pages */
+    uint32_t next_packet_size;
+    /* packets sent through this channel */
+    uint64_t num_packets;
+    /* pages sent through this channel */
+    uint64_t num_pages;
+    /* syncs main thread and channels */
+    QemuSemaphore sem_sync;
+} MultiFDRecvParams;
+
+#endif
+
diff --git a/migration/ram.c b/migration/ram.c
index c24b4cc771..ed23ed1c7c 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -36,7 +36,6 @@ 
 #include "xbzrle.h"
 #include "ram.h"
 #include "migration.h"
-#include "socket.h"
 #include "migration/register.h"
 #include "migration/misc.h"
 #include "qemu-file.h"
@@ -53,9 +52,9 @@ 
 #include "migration/colo.h"
 #include "block.h"
 #include "sysemu/sysemu.h"
-#include "qemu/uuid.h"
 #include "savevm.h"
 #include "qemu/iov.h"
+#include "multifd.h"
 
 /***********************************************************/
 /* ram save/restore */
@@ -575,983 +574,6 @@  exit:
     return -1;
 }
 
-/* Multiple fd's */
-
-#define MULTIFD_MAGIC 0x11223344U
-#define MULTIFD_VERSION 1
-
-#define MULTIFD_FLAG_SYNC (1 << 0)
-
-/* This value needs to be a multiple of qemu_target_page_size() */
-#define MULTIFD_PACKET_SIZE (512 * 1024)
-
-typedef struct {
-    uint32_t magic;
-    uint32_t version;
-    unsigned char uuid[16]; /* QemuUUID */
-    uint8_t id;
-    uint8_t unused1[7];     /* Reserved for future use */
-    uint64_t unused2[4];    /* Reserved for future use */
-} __attribute__((packed)) MultiFDInit_t;
-
-typedef struct {
-    uint32_t magic;
-    uint32_t version;
-    uint32_t flags;
-    /* maximum number of allocated pages */
-    uint32_t pages_alloc;
-    uint32_t pages_used;
-    /* size of the next packet that contains pages */
-    uint32_t next_packet_size;
-    uint64_t packet_num;
-    uint64_t unused[4];    /* Reserved for future use */
-    char ramblock[256];
-    uint64_t offset[];
-} __attribute__((packed)) MultiFDPacket_t;
-
-typedef struct {
-    /* number of used pages */
-    uint32_t used;
-    /* number of allocated pages */
-    uint32_t allocated;
-    /* global number of generated multifd packets */
-    uint64_t packet_num;
-    /* offset of each page */
-    ram_addr_t *offset;
-    /* pointer to each page */
-    struct iovec *iov;
-    RAMBlock *block;
-} MultiFDPages_t;
-
-typedef struct {
-    /* this fields are not changed once the thread is created */
-    /* channel number */
-    uint8_t id;
-    /* channel thread name */
-    char *name;
-    /* channel thread id */
-    QemuThread thread;
-    /* communication channel */
-    QIOChannel *c;
-    /* sem where to wait for more work */
-    QemuSemaphore sem;
-    /* this mutex protects the following parameters */
-    QemuMutex mutex;
-    /* is this channel thread running */
-    bool running;
-    /* should this thread finish */
-    bool quit;
-    /* thread has work to do */
-    int pending_job;
-    /* array of pages to sent */
-    MultiFDPages_t *pages;
-    /* packet allocated len */
-    uint32_t packet_len;
-    /* pointer to the packet */
-    MultiFDPacket_t *packet;
-    /* multifd flags for each packet */
-    uint32_t flags;
-    /* size of the next packet that contains pages */
-    uint32_t next_packet_size;
-    /* global number of generated multifd packets */
-    uint64_t packet_num;
-    /* thread local variables */
-    /* packets sent through this channel */
-    uint64_t num_packets;
-    /* pages sent through this channel */
-    uint64_t num_pages;
-    /* syncs main thread and channels */
-    QemuSemaphore sem_sync;
-}  MultiFDSendParams;
-
-typedef struct {
-    /* this fields are not changed once the thread is created */
-    /* channel number */
-    uint8_t id;
-    /* channel thread name */
-    char *name;
-    /* channel thread id */
-    QemuThread thread;
-    /* communication channel */
-    QIOChannel *c;
-    /* this mutex protects the following parameters */
-    QemuMutex mutex;
-    /* is this channel thread running */
-    bool running;
-    /* should this thread finish */
-    bool quit;
-    /* array of pages to receive */
-    MultiFDPages_t *pages;
-    /* packet allocated len */
-    uint32_t packet_len;
-    /* pointer to the packet */
-    MultiFDPacket_t *packet;
-    /* multifd flags for each packet */
-    uint32_t flags;
-    /* global number of generated multifd packets */
-    uint64_t packet_num;
-    /* thread local variables */
-    /* size of the next packet that contains pages */
-    uint32_t next_packet_size;
-    /* packets sent through this channel */
-    uint64_t num_packets;
-    /* pages sent through this channel */
-    uint64_t num_pages;
-    /* syncs main thread and channels */
-    QemuSemaphore sem_sync;
-} MultiFDRecvParams;
-
-static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
-{
-    MultiFDInit_t msg = {};
-    int ret;
-
-    msg.magic = cpu_to_be32(MULTIFD_MAGIC);
-    msg.version = cpu_to_be32(MULTIFD_VERSION);
-    msg.id = p->id;
-    memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
-
-    ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), errp);
-    if (ret != 0) {
-        return -1;
-    }
-    return 0;
-}
-
-static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
-{
-    MultiFDInit_t msg;
-    int ret;
-
-    ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
-    if (ret != 0) {
-        return -1;
-    }
-
-    msg.magic = be32_to_cpu(msg.magic);
-    msg.version = be32_to_cpu(msg.version);
-
-    if (msg.magic != MULTIFD_MAGIC) {
-        error_setg(errp, "multifd: received packet magic %x "
-                   "expected %x", msg.magic, MULTIFD_MAGIC);
-        return -1;
-    }
-
-    if (msg.version != MULTIFD_VERSION) {
-        error_setg(errp, "multifd: received packet version %d "
-                   "expected %d", msg.version, MULTIFD_VERSION);
-        return -1;
-    }
-
-    if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
-        char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
-        char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
-
-        error_setg(errp, "multifd: received uuid '%s' and expected "
-                   "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
-        g_free(uuid);
-        g_free(msg_uuid);
-        return -1;
-    }
-
-    if (msg.id > migrate_multifd_channels()) {
-        error_setg(errp, "multifd: received channel version %d "
-                   "expected %d", msg.version, MULTIFD_VERSION);
-        return -1;
-    }
-
-    return msg.id;
-}
-
-static MultiFDPages_t *multifd_pages_init(size_t size)
-{
-    MultiFDPages_t *pages = g_new0(MultiFDPages_t, 1);
-
-    pages->allocated = size;
-    pages->iov = g_new0(struct iovec, size);
-    pages->offset = g_new0(ram_addr_t, size);
-
-    return pages;
-}
-
-static void multifd_pages_clear(MultiFDPages_t *pages)
-{
-    pages->used = 0;
-    pages->allocated = 0;
-    pages->packet_num = 0;
-    pages->block = NULL;
-    g_free(pages->iov);
-    pages->iov = NULL;
-    g_free(pages->offset);
-    pages->offset = NULL;
-    g_free(pages);
-}
-
-static void multifd_send_fill_packet(MultiFDSendParams *p)
-{
-    MultiFDPacket_t *packet = p->packet;
-    int i;
-
-    packet->flags = cpu_to_be32(p->flags);
-    packet->pages_alloc = cpu_to_be32(p->pages->allocated);
-    packet->pages_used = cpu_to_be32(p->pages->used);
-    packet->next_packet_size = cpu_to_be32(p->next_packet_size);
-    packet->packet_num = cpu_to_be64(p->packet_num);
-
-    if (p->pages->block) {
-        strncpy(packet->ramblock, p->pages->block->idstr, 256);
-    }
-
-    for (i = 0; i < p->pages->used; i++) {
-        /* there are architectures where ram_addr_t is 32 bit */
-        uint64_t temp = p->pages->offset[i];
-
-        packet->offset[i] = cpu_to_be64(temp);
-    }
-}
-
-static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
-{
-    MultiFDPacket_t *packet = p->packet;
-    uint32_t pages_max = MULTIFD_PACKET_SIZE / qemu_target_page_size();
-    RAMBlock *block;
-    int i;
-
-    packet->magic = be32_to_cpu(packet->magic);
-    if (packet->magic != MULTIFD_MAGIC) {
-        error_setg(errp, "multifd: received packet "
-                   "magic %x and expected magic %x",
-                   packet->magic, MULTIFD_MAGIC);
-        return -1;
-    }
-
-    packet->version = be32_to_cpu(packet->version);
-    if (packet->version != MULTIFD_VERSION) {
-        error_setg(errp, "multifd: received packet "
-                   "version %d and expected version %d",
-                   packet->version, MULTIFD_VERSION);
-        return -1;
-    }
-
-    p->flags = be32_to_cpu(packet->flags);
-
-    packet->pages_alloc = be32_to_cpu(packet->pages_alloc);
-    /*
-     * If we received a packet that is 100 times bigger than expected
-     * just stop migration.  It is a magic number.
-     */
-    if (packet->pages_alloc > pages_max * 100) {
-        error_setg(errp, "multifd: received packet "
-                   "with size %d and expected a maximum size of %d",
-                   packet->pages_alloc, pages_max * 100) ;
-        return -1;
-    }
-    /*
-     * We received a packet that is bigger than expected but inside
-     * reasonable limits (see previous comment).  Just reallocate.
-     */
-    if (packet->pages_alloc > p->pages->allocated) {
-        multifd_pages_clear(p->pages);
-        p->pages = multifd_pages_init(packet->pages_alloc);
-    }
-
-    p->pages->used = be32_to_cpu(packet->pages_used);
-    if (p->pages->used > packet->pages_alloc) {
-        error_setg(errp, "multifd: received packet "
-                   "with %d pages and expected maximum pages are %d",
-                   p->pages->used, packet->pages_alloc) ;
-        return -1;
-    }
-
-    p->next_packet_size = be32_to_cpu(packet->next_packet_size);
-    p->packet_num = be64_to_cpu(packet->packet_num);
-
-    if (p->pages->used == 0) {
-        return 0;
-    }
-
-    /* make sure that ramblock is 0 terminated */
-    packet->ramblock[255] = 0;
-    block = qemu_ram_block_by_name(packet->ramblock);
-    if (!block) {
-        error_setg(errp, "multifd: unknown ram block %s",
-                   packet->ramblock);
-        return -1;
-    }
-
-    for (i = 0; i < p->pages->used; i++) {
-        uint64_t offset = be64_to_cpu(packet->offset[i]);
-
-        if (offset > (block->used_length - qemu_target_page_size())) {
-            error_setg(errp, "multifd: offset too long %" PRIu64
-                       " (max " RAM_ADDR_FMT ")",
-                       offset, block->max_length);
-            return -1;
-        }
-        p->pages->iov[i].iov_base = block->host + offset;
-        p->pages->iov[i].iov_len = qemu_target_page_size();
-    }
-
-    return 0;
-}
-
-struct {
-    MultiFDSendParams *params;
-    /* array of pages to sent */
-    MultiFDPages_t *pages;
-    /* global number of generated multifd packets */
-    uint64_t packet_num;
-    /* send channels ready */
-    QemuSemaphore channels_ready;
-    /*
-     * Have we already run terminate threads.  There is a race when it
-     * happens that we got one error while we are exiting.
-     * We will use atomic operations.  Only valid values are 0 and 1.
-     */
-    int exiting;
-} *multifd_send_state;
-
-/*
- * How we use multifd_send_state->pages and channel->pages?
- *
- * We create a pages for each channel, and a main one.  Each time that
- * we need to send a batch of pages we interchange the ones between
- * multifd_send_state and the channel that is sending it.  There are
- * two reasons for that:
- *    - to not have to do so many mallocs during migration
- *    - to make easier to know what to free at the end of migration
- *
- * This way we always know who is the owner of each "pages" struct,
- * and we don't need any locking.  It belongs to the migration thread
- * or to the channel thread.  Switching is safe because the migration
- * thread is using the channel mutex when changing it, and the channel
- * have to had finish with its own, otherwise pending_job can't be
- * false.
- */
-
-static int multifd_send_pages(QEMUFile *f)
-{
-    int i;
-    static int next_channel;
-    MultiFDSendParams *p = NULL; /* make happy gcc */
-    MultiFDPages_t *pages = multifd_send_state->pages;
-    uint64_t transferred;
-
-    if (atomic_read(&multifd_send_state->exiting)) {
-        return -1;
-    }
-
-    qemu_sem_wait(&multifd_send_state->channels_ready);
-    for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
-        p = &multifd_send_state->params[i];
-
-        qemu_mutex_lock(&p->mutex);
-        if (p->quit) {
-            error_report("%s: channel %d has already quit!", __func__, i);
-            qemu_mutex_unlock(&p->mutex);
-            return -1;
-        }
-        if (!p->pending_job) {
-            p->pending_job++;
-            next_channel = (i + 1) % migrate_multifd_channels();
-            break;
-        }
-        qemu_mutex_unlock(&p->mutex);
-    }
-    assert(!p->pages->used);
-    assert(!p->pages->block);
-
-    p->packet_num = multifd_send_state->packet_num++;
-    multifd_send_state->pages = p->pages;
-    p->pages = pages;
-    transferred = ((uint64_t) pages->used) * qemu_target_page_size()
-                + p->packet_len;
-    qemu_file_update_transfer(f, transferred);
-    ram_counters.multifd_bytes += transferred;
-    ram_counters.transferred += transferred;;
-    qemu_mutex_unlock(&p->mutex);
-    qemu_sem_post(&p->sem);
-
-    return 1;
-}
-
-static int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
-{
-    MultiFDPages_t *pages = multifd_send_state->pages;
-
-    if (!pages->block) {
-        pages->block = block;
-    }
-
-    if (pages->block == block) {
-        pages->offset[pages->used] = offset;
-        pages->iov[pages->used].iov_base = block->host + offset;
-        pages->iov[pages->used].iov_len = qemu_target_page_size();
-        pages->used++;
-
-        if (pages->used < pages->allocated) {
-            return 1;
-        }
-    }
-
-    if (multifd_send_pages(f) < 0) {
-        return -1;
-    }
-
-    if (pages->block != block) {
-        return  multifd_queue_page(f, block, offset);
-    }
-
-    return 1;
-}
-
-static void multifd_send_terminate_threads(Error *err)
-{
-    int i;
-
-    trace_multifd_send_terminate_threads(err != NULL);
-
-    if (err) {
-        MigrationState *s = migrate_get_current();
-        migrate_set_error(s, err);
-        if (s->state == MIGRATION_STATUS_SETUP ||
-            s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
-            s->state == MIGRATION_STATUS_DEVICE ||
-            s->state == MIGRATION_STATUS_ACTIVE) {
-            migrate_set_state(&s->state, s->state,
-                              MIGRATION_STATUS_FAILED);
-        }
-    }
-
-    /*
-     * We don't want to exit each threads twice.  Depending on where
-     * we get the error, or if there are two independent errors in two
-     * threads at the same time, we can end calling this function
-     * twice.
-     */
-    if (atomic_xchg(&multifd_send_state->exiting, 1)) {
-        return;
-    }
-
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDSendParams *p = &multifd_send_state->params[i];
-
-        qemu_mutex_lock(&p->mutex);
-        p->quit = true;
-        qemu_sem_post(&p->sem);
-        qemu_mutex_unlock(&p->mutex);
-    }
-}
-
-void multifd_save_cleanup(void)
-{
-    int i;
-
-    if (!migrate_use_multifd()) {
-        return;
-    }
-    multifd_send_terminate_threads(NULL);
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDSendParams *p = &multifd_send_state->params[i];
-
-        if (p->running) {
-            qemu_thread_join(&p->thread);
-        }
-    }
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDSendParams *p = &multifd_send_state->params[i];
-
-        socket_send_channel_destroy(p->c);
-        p->c = NULL;
-        qemu_mutex_destroy(&p->mutex);
-        qemu_sem_destroy(&p->sem);
-        qemu_sem_destroy(&p->sem_sync);
-        g_free(p->name);
-        p->name = NULL;
-        multifd_pages_clear(p->pages);
-        p->pages = NULL;
-        p->packet_len = 0;
-        g_free(p->packet);
-        p->packet = NULL;
-    }
-    qemu_sem_destroy(&multifd_send_state->channels_ready);
-    g_free(multifd_send_state->params);
-    multifd_send_state->params = NULL;
-    multifd_pages_clear(multifd_send_state->pages);
-    multifd_send_state->pages = NULL;
-    g_free(multifd_send_state);
-    multifd_send_state = NULL;
-}
-
-static void multifd_send_sync_main(QEMUFile *f)
-{
-    int i;
-
-    if (!migrate_use_multifd()) {
-        return;
-    }
-    if (multifd_send_state->pages->used) {
-        if (multifd_send_pages(f) < 0) {
-            error_report("%s: multifd_send_pages fail", __func__);
-            return;
-        }
-    }
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDSendParams *p = &multifd_send_state->params[i];
-
-        trace_multifd_send_sync_main_signal(p->id);
-
-        qemu_mutex_lock(&p->mutex);
-
-        if (p->quit) {
-            error_report("%s: channel %d has already quit", __func__, i);
-            qemu_mutex_unlock(&p->mutex);
-            return;
-        }
-
-        p->packet_num = multifd_send_state->packet_num++;
-        p->flags |= MULTIFD_FLAG_SYNC;
-        p->pending_job++;
-        qemu_file_update_transfer(f, p->packet_len);
-        ram_counters.multifd_bytes += p->packet_len;
-        ram_counters.transferred += p->packet_len;
-        qemu_mutex_unlock(&p->mutex);
-        qemu_sem_post(&p->sem);
-    }
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDSendParams *p = &multifd_send_state->params[i];
-
-        trace_multifd_send_sync_main_wait(p->id);
-        qemu_sem_wait(&p->sem_sync);
-    }
-    trace_multifd_send_sync_main(multifd_send_state->packet_num);
-}
-
-static void *multifd_send_thread(void *opaque)
-{
-    MultiFDSendParams *p = opaque;
-    Error *local_err = NULL;
-    int ret = 0;
-    uint32_t flags = 0;
-
-    trace_multifd_send_thread_start(p->id);
-    rcu_register_thread();
-
-    if (multifd_send_initial_packet(p, &local_err) < 0) {
-        ret = -1;
-        goto out;
-    }
-    /* initial packet */
-    p->num_packets = 1;
-
-    while (true) {
-        qemu_sem_wait(&p->sem);
-
-        if (atomic_read(&multifd_send_state->exiting)) {
-            break;
-        }
-        qemu_mutex_lock(&p->mutex);
-
-        if (p->pending_job) {
-            uint32_t used = p->pages->used;
-            uint64_t packet_num = p->packet_num;
-            flags = p->flags;
-
-            p->next_packet_size = used * qemu_target_page_size();
-            multifd_send_fill_packet(p);
-            p->flags = 0;
-            p->num_packets++;
-            p->num_pages += used;
-            p->pages->used = 0;
-            p->pages->block = NULL;
-            qemu_mutex_unlock(&p->mutex);
-
-            trace_multifd_send(p->id, packet_num, used, flags,
-                               p->next_packet_size);
-
-            ret = qio_channel_write_all(p->c, (void *)p->packet,
-                                        p->packet_len, &local_err);
-            if (ret != 0) {
-                break;
-            }
-
-            if (used) {
-                ret = qio_channel_writev_all(p->c, p->pages->iov,
-                                             used, &local_err);
-                if (ret != 0) {
-                    break;
-                }
-            }
-
-            qemu_mutex_lock(&p->mutex);
-            p->pending_job--;
-            qemu_mutex_unlock(&p->mutex);
-
-            if (flags & MULTIFD_FLAG_SYNC) {
-                qemu_sem_post(&p->sem_sync);
-            }
-            qemu_sem_post(&multifd_send_state->channels_ready);
-        } else if (p->quit) {
-            qemu_mutex_unlock(&p->mutex);
-            break;
-        } else {
-            qemu_mutex_unlock(&p->mutex);
-            /* sometimes there are spurious wakeups */
-        }
-    }
-
-out:
-    if (local_err) {
-        trace_multifd_send_error(p->id);
-        multifd_send_terminate_threads(local_err);
-    }
-
-    /*
-     * Error happen, I will exit, but I can't just leave, tell
-     * who pay attention to me.
-     */
-    if (ret != 0) {
-        qemu_sem_post(&p->sem_sync);
-        qemu_sem_post(&multifd_send_state->channels_ready);
-    }
-
-    qemu_mutex_lock(&p->mutex);
-    p->running = false;
-    qemu_mutex_unlock(&p->mutex);
-
-    rcu_unregister_thread();
-    trace_multifd_send_thread_end(p->id, p->num_packets, p->num_pages);
-
-    return NULL;
-}
-
-static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
-{
-    MultiFDSendParams *p = opaque;
-    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
-    Error *local_err = NULL;
-
-    trace_multifd_new_send_channel_async(p->id);
-    if (qio_task_propagate_error(task, &local_err)) {
-        migrate_set_error(migrate_get_current(), local_err);
-        multifd_save_cleanup();
-    } else {
-        p->c = QIO_CHANNEL(sioc);
-        qio_channel_set_delay(p->c, false);
-        p->running = true;
-        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
-                           QEMU_THREAD_JOINABLE);
-    }
-}
-
-int multifd_save_setup(Error **errp)
-{
-    int thread_count;
-    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
-    uint8_t i;
-
-    if (!migrate_use_multifd()) {
-        return 0;
-    }
-    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->pages = multifd_pages_init(page_count);
-    qemu_sem_init(&multifd_send_state->channels_ready, 0);
-    atomic_set(&multifd_send_state->exiting, 0);
-
-    for (i = 0; i < thread_count; i++) {
-        MultiFDSendParams *p = &multifd_send_state->params[i];
-
-        qemu_mutex_init(&p->mutex);
-        qemu_sem_init(&p->sem, 0);
-        qemu_sem_init(&p->sem_sync, 0);
-        p->quit = false;
-        p->pending_job = 0;
-        p->id = i;
-        p->pages = multifd_pages_init(page_count);
-        p->packet_len = sizeof(MultiFDPacket_t)
-                      + sizeof(uint64_t) * page_count;
-        p->packet = g_malloc0(p->packet_len);
-        p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
-        p->packet->version = cpu_to_be32(MULTIFD_VERSION);
-        p->name = g_strdup_printf("multifdsend_%d", i);
-        socket_send_channel_create(multifd_new_send_channel_async, p);
-    }
-    return 0;
-}
-
-struct {
-    MultiFDRecvParams *params;
-    /* number of created threads */
-    int count;
-    /* syncs main thread and channels */
-    QemuSemaphore sem_sync;
-    /* global number of generated multifd packets */
-    uint64_t packet_num;
-} *multifd_recv_state;
-
-static void multifd_recv_terminate_threads(Error *err)
-{
-    int i;
-
-    trace_multifd_recv_terminate_threads(err != NULL);
-
-    if (err) {
-        MigrationState *s = migrate_get_current();
-        migrate_set_error(s, err);
-        if (s->state == MIGRATION_STATUS_SETUP ||
-            s->state == MIGRATION_STATUS_ACTIVE) {
-            migrate_set_state(&s->state, s->state,
-                              MIGRATION_STATUS_FAILED);
-        }
-    }
-
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDRecvParams *p = &multifd_recv_state->params[i];
-
-        qemu_mutex_lock(&p->mutex);
-        p->quit = true;
-        /*
-         * We could arrive here for two reasons:
-         *  - normal quit, i.e. everything went fine, just finished
-         *  - error quit: We close the channels so the channel threads
-         *    finish the qio_channel_read_all_eof()
-         */
-        if (p->c) {
-            qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
-        }
-        qemu_mutex_unlock(&p->mutex);
-    }
-}
-
-int multifd_load_cleanup(Error **errp)
-{
-    int i;
-    int ret = 0;
-
-    if (!migrate_use_multifd()) {
-        return 0;
-    }
-    multifd_recv_terminate_threads(NULL);
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDRecvParams *p = &multifd_recv_state->params[i];
-
-        if (p->running) {
-            p->quit = true;
-            /*
-             * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
-             * however try to wakeup it without harm in cleanup phase.
-             */
-            qemu_sem_post(&p->sem_sync);
-            qemu_thread_join(&p->thread);
-        }
-    }
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDRecvParams *p = &multifd_recv_state->params[i];
-
-        object_unref(OBJECT(p->c));
-        p->c = NULL;
-        qemu_mutex_destroy(&p->mutex);
-        qemu_sem_destroy(&p->sem_sync);
-        g_free(p->name);
-        p->name = NULL;
-        multifd_pages_clear(p->pages);
-        p->pages = NULL;
-        p->packet_len = 0;
-        g_free(p->packet);
-        p->packet = NULL;
-    }
-    qemu_sem_destroy(&multifd_recv_state->sem_sync);
-    g_free(multifd_recv_state->params);
-    multifd_recv_state->params = NULL;
-    g_free(multifd_recv_state);
-    multifd_recv_state = NULL;
-
-    return ret;
-}
-
-static void multifd_recv_sync_main(void)
-{
-    int i;
-
-    if (!migrate_use_multifd()) {
-        return;
-    }
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDRecvParams *p = &multifd_recv_state->params[i];
-
-        trace_multifd_recv_sync_main_wait(p->id);
-        qemu_sem_wait(&multifd_recv_state->sem_sync);
-    }
-    for (i = 0; i < migrate_multifd_channels(); i++) {
-        MultiFDRecvParams *p = &multifd_recv_state->params[i];
-
-        qemu_mutex_lock(&p->mutex);
-        if (multifd_recv_state->packet_num < p->packet_num) {
-            multifd_recv_state->packet_num = p->packet_num;
-        }
-        qemu_mutex_unlock(&p->mutex);
-        trace_multifd_recv_sync_main_signal(p->id);
-        qemu_sem_post(&p->sem_sync);
-    }
-    trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
-}
-
-static void *multifd_recv_thread(void *opaque)
-{
-    MultiFDRecvParams *p = opaque;
-    Error *local_err = NULL;
-    int ret;
-
-    trace_multifd_recv_thread_start(p->id);
-    rcu_register_thread();
-
-    while (true) {
-        uint32_t used;
-        uint32_t flags;
-
-        if (p->quit) {
-            break;
-        }
-
-        ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
-                                       p->packet_len, &local_err);
-        if (ret == 0) {   /* EOF */
-            break;
-        }
-        if (ret == -1) {   /* Error */
-            break;
-        }
-
-        qemu_mutex_lock(&p->mutex);
-        ret = multifd_recv_unfill_packet(p, &local_err);
-        if (ret) {
-            qemu_mutex_unlock(&p->mutex);
-            break;
-        }
-
-        used = p->pages->used;
-        flags = p->flags;
-        trace_multifd_recv(p->id, p->packet_num, used, flags,
-                           p->next_packet_size);
-        p->num_packets++;
-        p->num_pages += used;
-        qemu_mutex_unlock(&p->mutex);
-
-        if (used) {
-            ret = qio_channel_readv_all(p->c, p->pages->iov,
-                                        used, &local_err);
-            if (ret != 0) {
-                break;
-            }
-        }
-
-        if (flags & MULTIFD_FLAG_SYNC) {
-            qemu_sem_post(&multifd_recv_state->sem_sync);
-            qemu_sem_wait(&p->sem_sync);
-        }
-    }
-
-    if (local_err) {
-        multifd_recv_terminate_threads(local_err);
-    }
-    qemu_mutex_lock(&p->mutex);
-    p->running = false;
-    qemu_mutex_unlock(&p->mutex);
-
-    rcu_unregister_thread();
-    trace_multifd_recv_thread_end(p->id, p->num_packets, p->num_pages);
-
-    return NULL;
-}
-
-int multifd_load_setup(Error **errp)
-{
-    int thread_count;
-    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
-    uint8_t i;
-
-    if (!migrate_use_multifd()) {
-        return 0;
-    }
-    thread_count = migrate_multifd_channels();
-    multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
-    multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
-    atomic_set(&multifd_recv_state->count, 0);
-    qemu_sem_init(&multifd_recv_state->sem_sync, 0);
-
-    for (i = 0; i < thread_count; i++) {
-        MultiFDRecvParams *p = &multifd_recv_state->params[i];
-
-        qemu_mutex_init(&p->mutex);
-        qemu_sem_init(&p->sem_sync, 0);
-        p->quit = false;
-        p->id = i;
-        p->pages = multifd_pages_init(page_count);
-        p->packet_len = sizeof(MultiFDPacket_t)
-                      + sizeof(uint64_t) * page_count;
-        p->packet = g_malloc0(p->packet_len);
-        p->name = g_strdup_printf("multifdrecv_%d", i);
-    }
-    return 0;
-}
-
-bool multifd_recv_all_channels_created(void)
-{
-    int thread_count = migrate_multifd_channels();
-
-    if (!migrate_use_multifd()) {
-        return true;
-    }
-
-    return thread_count == atomic_read(&multifd_recv_state->count);
-}
-
-/*
- * Try to receive all multifd channels to get ready for the migration.
- * - Return true and do not set @errp when correctly receving all channels;
- * - Return false and do not set @errp when correctly receiving the current one;
- * - Return false and set @errp when failing to receive the current channel.
- */
-bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
-{
-    MultiFDRecvParams *p;
-    Error *local_err = NULL;
-    int id;
-
-    id = multifd_recv_initial_packet(ioc, &local_err);
-    if (id < 0) {
-        multifd_recv_terminate_threads(local_err);
-        error_propagate_prepend(errp, local_err,
-                                "failed to receive packet"
-                                " via multifd channel %d: ",
-                                atomic_read(&multifd_recv_state->count));
-        return false;
-    }
-    trace_multifd_recv_new_channel(id);
-
-    p = &multifd_recv_state->params[id];
-    if (p->c != NULL) {
-        error_setg(&local_err, "multifd: received id '%d' already setup'",
-                   id);
-        multifd_recv_terminate_threads(local_err);
-        error_propagate(errp, local_err);
-        return false;
-    }
-    p->c = ioc;
-    object_ref(OBJECT(ioc));
-    /* initial packet */
-    p->num_packets = 1;
-
-    p->running = true;
-    qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
-                       QEMU_THREAD_JOINABLE);
-    atomic_inc(&multifd_recv_state->count);
-    return atomic_read(&multifd_recv_state->count) ==
-           migrate_multifd_channels();
-}
-
 /**
  * save_page_header: write page header to wire
  *
diff --git a/migration/ram.h b/migration/ram.h
index 42be471d52..a553d40751 100644
--- a/migration/ram.h
+++ b/migration/ram.h
@@ -41,13 +41,6 @@  int xbzrle_cache_resize(int64_t new_size, Error **errp);
 uint64_t ram_bytes_remaining(void);
 uint64_t ram_bytes_total(void);
 
-int multifd_save_setup(Error **errp);
-void multifd_save_cleanup(void);
-int multifd_load_setup(Error **errp);
-int multifd_load_cleanup(Error **errp);
-bool multifd_recv_all_channels_created(void);
-bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp);
-
 uint64_t ram_pagesize_summary(void);
 int ram_save_queue_pages(const char *rbname, ram_addr_t start, ram_addr_t len);
 void acct_update_position(QEMUFile *f, size_t size, bool zero);