qemu/migration/multifd.c
<<
>>
Prefs
   1/*
   2 * Multifd common code
   3 *
   4 * Copyright (c) 2019-2020 Red Hat Inc
   5 *
   6 * Authors:
   7 *  Juan Quintela <quintela@redhat.com>
   8 *
   9 * This work is licensed under the terms of the GNU GPL, version 2 or later.
  10 * See the COPYING file in the top-level directory.
  11 */
  12
  13#include "qemu/osdep.h"
  14#include "qemu/rcu.h"
  15#include "exec/target_page.h"
  16#include "sysemu/sysemu.h"
  17#include "exec/ramblock.h"
  18#include "qemu/error-report.h"
  19#include "qapi/error.h"
  20#include "ram.h"
  21#include "migration.h"
  22#include "socket.h"
  23#include "tls.h"
  24#include "qemu-file.h"
  25#include "trace.h"
  26#include "multifd.h"
  27
  28#include "qemu/yank.h"
  29#include "io/channel-socket.h"
  30#include "yank_functions.h"
  31
  32/* Multiple fd's */
  33
  34#define MULTIFD_MAGIC 0x11223344U
  35#define MULTIFD_VERSION 1
  36
  37typedef struct {
  38    uint32_t magic;
  39    uint32_t version;
  40    unsigned char uuid[16]; /* QemuUUID */
  41    uint8_t id;
  42    uint8_t unused1[7];     /* Reserved for future use */
  43    uint64_t unused2[4];    /* Reserved for future use */
  44} __attribute__((packed)) MultiFDInit_t;
  45
  46/* Multifd without compression */
  47
  48/**
  49 * nocomp_send_setup: setup send side
  50 *
  51 * For no compression this function does nothing.
  52 *
  53 * Returns 0 for success or -1 for error
  54 *
  55 * @p: Params for the channel that we are using
  56 * @errp: pointer to an error
  57 */
  58static int nocomp_send_setup(MultiFDSendParams *p, Error **errp)
  59{
  60    return 0;
  61}
  62
  63/**
  64 * nocomp_send_cleanup: cleanup send side
  65 *
  66 * For no compression this function does nothing.
  67 *
  68 * @p: Params for the channel that we are using
  69 */
  70static void nocomp_send_cleanup(MultiFDSendParams *p, Error **errp)
  71{
  72    return;
  73}
  74
  75/**
  76 * nocomp_send_prepare: prepare date to be able to send
  77 *
  78 * For no compression we just have to calculate the size of the
  79 * packet.
  80 *
  81 * Returns 0 for success or -1 for error
  82 *
  83 * @p: Params for the channel that we are using
  84 * @used: number of pages used
  85 * @errp: pointer to an error
  86 */
  87static int nocomp_send_prepare(MultiFDSendParams *p, uint32_t used,
  88                               Error **errp)
  89{
  90    p->next_packet_size = used * qemu_target_page_size();
  91    p->flags |= MULTIFD_FLAG_NOCOMP;
  92    return 0;
  93}
  94
  95/**
  96 * nocomp_send_write: do the actual write of the data
  97 *
  98 * For no compression we just have to write the data.
  99 *
 100 * Returns 0 for success or -1 for error
 101 *
 102 * @p: Params for the channel that we are using
 103 * @used: number of pages used
 104 * @errp: pointer to an error
 105 */
 106static int nocomp_send_write(MultiFDSendParams *p, uint32_t used, Error **errp)
 107{
 108    return qio_channel_writev_all(p->c, p->pages->iov, used, errp);
 109}
 110
 111/**
 112 * nocomp_recv_setup: setup receive side
 113 *
 114 * For no compression this function does nothing.
 115 *
 116 * Returns 0 for success or -1 for error
 117 *
 118 * @p: Params for the channel that we are using
 119 * @errp: pointer to an error
 120 */
 121static int nocomp_recv_setup(MultiFDRecvParams *p, Error **errp)
 122{
 123    return 0;
 124}
 125
 126/**
 127 * nocomp_recv_cleanup: setup receive side
 128 *
 129 * For no compression this function does nothing.
 130 *
 131 * @p: Params for the channel that we are using
 132 */
 133static void nocomp_recv_cleanup(MultiFDRecvParams *p)
 134{
 135}
 136
 137/**
 138 * nocomp_recv_pages: read the data from the channel into actual pages
 139 *
 140 * For no compression we just need to read things into the correct place.
 141 *
 142 * Returns 0 for success or -1 for error
 143 *
 144 * @p: Params for the channel that we are using
 145 * @used: number of pages used
 146 * @errp: pointer to an error
 147 */
 148static int nocomp_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
 149{
 150    uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
 151
 152    if (flags != MULTIFD_FLAG_NOCOMP) {
 153        error_setg(errp, "multifd %d: flags received %x flags expected %x",
 154                   p->id, flags, MULTIFD_FLAG_NOCOMP);
 155        return -1;
 156    }
 157    return qio_channel_readv_all(p->c, p->pages->iov, used, errp);
 158}
 159
 160static MultiFDMethods multifd_nocomp_ops = {
 161    .send_setup = nocomp_send_setup,
 162    .send_cleanup = nocomp_send_cleanup,
 163    .send_prepare = nocomp_send_prepare,
 164    .send_write = nocomp_send_write,
 165    .recv_setup = nocomp_recv_setup,
 166    .recv_cleanup = nocomp_recv_cleanup,
 167    .recv_pages = nocomp_recv_pages
 168};
 169
 170static MultiFDMethods *multifd_ops[MULTIFD_COMPRESSION__MAX] = {
 171    [MULTIFD_COMPRESSION_NONE] = &multifd_nocomp_ops,
 172};
 173
 174void multifd_register_ops(int method, MultiFDMethods *ops)
 175{
 176    assert(0 < method && method < MULTIFD_COMPRESSION__MAX);
 177    multifd_ops[method] = ops;
 178}
 179
 180static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
 181{
 182    MultiFDInit_t msg = {};
 183    int ret;
 184
 185    msg.magic = cpu_to_be32(MULTIFD_MAGIC);
 186    msg.version = cpu_to_be32(MULTIFD_VERSION);
 187    msg.id = p->id;
 188    memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
 189
 190    ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), errp);
 191    if (ret != 0) {
 192        return -1;
 193    }
 194    return 0;
 195}
 196
 197static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
 198{
 199    MultiFDInit_t msg;
 200    int ret;
 201
 202    ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
 203    if (ret != 0) {
 204        return -1;
 205    }
 206
 207    msg.magic = be32_to_cpu(msg.magic);
 208    msg.version = be32_to_cpu(msg.version);
 209
 210    if (msg.magic != MULTIFD_MAGIC) {
 211        error_setg(errp, "multifd: received packet magic %x "
 212                   "expected %x", msg.magic, MULTIFD_MAGIC);
 213        return -1;
 214    }
 215
 216    if (msg.version != MULTIFD_VERSION) {
 217        error_setg(errp, "multifd: received packet version %d "
 218                   "expected %d", msg.version, MULTIFD_VERSION);
 219        return -1;
 220    }
 221
 222    if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
 223        char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
 224        char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
 225
 226        error_setg(errp, "multifd: received uuid '%s' and expected "
 227                   "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
 228        g_free(uuid);
 229        g_free(msg_uuid);
 230        return -1;
 231    }
 232
 233    if (msg.id > migrate_multifd_channels()) {
 234        error_setg(errp, "multifd: received channel version %d "
 235                   "expected %d", msg.version, MULTIFD_VERSION);
 236        return -1;
 237    }
 238
 239    return msg.id;
 240}
 241
 242static MultiFDPages_t *multifd_pages_init(size_t size)
 243{
 244    MultiFDPages_t *pages = g_new0(MultiFDPages_t, 1);
 245
 246    pages->allocated = size;
 247    pages->iov = g_new0(struct iovec, size);
 248    pages->offset = g_new0(ram_addr_t, size);
 249
 250    return pages;
 251}
 252
 253static void multifd_pages_clear(MultiFDPages_t *pages)
 254{
 255    pages->used = 0;
 256    pages->allocated = 0;
 257    pages->packet_num = 0;
 258    pages->block = NULL;
 259    g_free(pages->iov);
 260    pages->iov = NULL;
 261    g_free(pages->offset);
 262    pages->offset = NULL;
 263    g_free(pages);
 264}
 265
 266static void multifd_send_fill_packet(MultiFDSendParams *p)
 267{
 268    MultiFDPacket_t *packet = p->packet;
 269    int i;
 270
 271    packet->flags = cpu_to_be32(p->flags);
 272    packet->pages_alloc = cpu_to_be32(p->pages->allocated);
 273    packet->pages_used = cpu_to_be32(p->pages->used);
 274    packet->next_packet_size = cpu_to_be32(p->next_packet_size);
 275    packet->packet_num = cpu_to_be64(p->packet_num);
 276
 277    if (p->pages->block) {
 278        strncpy(packet->ramblock, p->pages->block->idstr, 256);
 279    }
 280
 281    for (i = 0; i < p->pages->used; i++) {
 282        /* there are architectures where ram_addr_t is 32 bit */
 283        uint64_t temp = p->pages->offset[i];
 284
 285        packet->offset[i] = cpu_to_be64(temp);
 286    }
 287}
 288
 289static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
 290{
 291    MultiFDPacket_t *packet = p->packet;
 292    uint32_t pages_max = MULTIFD_PACKET_SIZE / qemu_target_page_size();
 293    RAMBlock *block;
 294    int i;
 295
 296    packet->magic = be32_to_cpu(packet->magic);
 297    if (packet->magic != MULTIFD_MAGIC) {
 298        error_setg(errp, "multifd: received packet "
 299                   "magic %x and expected magic %x",
 300                   packet->magic, MULTIFD_MAGIC);
 301        return -1;
 302    }
 303
 304    packet->version = be32_to_cpu(packet->version);
 305    if (packet->version != MULTIFD_VERSION) {
 306        error_setg(errp, "multifd: received packet "
 307                   "version %d and expected version %d",
 308                   packet->version, MULTIFD_VERSION);
 309        return -1;
 310    }
 311
 312    p->flags = be32_to_cpu(packet->flags);
 313
 314    packet->pages_alloc = be32_to_cpu(packet->pages_alloc);
 315    /*
 316     * If we received a packet that is 100 times bigger than expected
 317     * just stop migration.  It is a magic number.
 318     */
 319    if (packet->pages_alloc > pages_max * 100) {
 320        error_setg(errp, "multifd: received packet "
 321                   "with size %d and expected a maximum size of %d",
 322                   packet->pages_alloc, pages_max * 100) ;
 323        return -1;
 324    }
 325    /*
 326     * We received a packet that is bigger than expected but inside
 327     * reasonable limits (see previous comment).  Just reallocate.
 328     */
 329    if (packet->pages_alloc > p->pages->allocated) {
 330        multifd_pages_clear(p->pages);
 331        p->pages = multifd_pages_init(packet->pages_alloc);
 332    }
 333
 334    p->pages->used = be32_to_cpu(packet->pages_used);
 335    if (p->pages->used > packet->pages_alloc) {
 336        error_setg(errp, "multifd: received packet "
 337                   "with %d pages and expected maximum pages are %d",
 338                   p->pages->used, packet->pages_alloc) ;
 339        return -1;
 340    }
 341
 342    p->next_packet_size = be32_to_cpu(packet->next_packet_size);
 343    p->packet_num = be64_to_cpu(packet->packet_num);
 344
 345    if (p->pages->used == 0) {
 346        return 0;
 347    }
 348
 349    /* make sure that ramblock is 0 terminated */
 350    packet->ramblock[255] = 0;
 351    block = qemu_ram_block_by_name(packet->ramblock);
 352    if (!block) {
 353        error_setg(errp, "multifd: unknown ram block %s",
 354                   packet->ramblock);
 355        return -1;
 356    }
 357
 358    for (i = 0; i < p->pages->used; i++) {
 359        uint64_t offset = be64_to_cpu(packet->offset[i]);
 360
 361        if (offset > (block->used_length - qemu_target_page_size())) {
 362            error_setg(errp, "multifd: offset too long %" PRIu64
 363                       " (max " RAM_ADDR_FMT ")",
 364                       offset, block->used_length);
 365            return -1;
 366        }
 367        p->pages->iov[i].iov_base = block->host + offset;
 368        p->pages->iov[i].iov_len = qemu_target_page_size();
 369    }
 370
 371    return 0;
 372}
 373
 374struct {
 375    MultiFDSendParams *params;
 376    /* array of pages to sent */
 377    MultiFDPages_t *pages;
 378    /* global number of generated multifd packets */
 379    uint64_t packet_num;
 380    /* send channels ready */
 381    QemuSemaphore channels_ready;
 382    /*
 383     * Have we already run terminate threads.  There is a race when it
 384     * happens that we got one error while we are exiting.
 385     * We will use atomic operations.  Only valid values are 0 and 1.
 386     */
 387    int exiting;
 388    /* multifd ops */
 389    MultiFDMethods *ops;
 390} *multifd_send_state;
 391
 392/*
 393 * How we use multifd_send_state->pages and channel->pages?
 394 *
 395 * We create a pages for each channel, and a main one.  Each time that
 396 * we need to send a batch of pages we interchange the ones between
 397 * multifd_send_state and the channel that is sending it.  There are
 398 * two reasons for that:
 399 *    - to not have to do so many mallocs during migration
 400 *    - to make easier to know what to free at the end of migration
 401 *
 402 * This way we always know who is the owner of each "pages" struct,
 403 * and we don't need any locking.  It belongs to the migration thread
 404 * or to the channel thread.  Switching is safe because the migration
 405 * thread is using the channel mutex when changing it, and the channel
 406 * have to had finish with its own, otherwise pending_job can't be
 407 * false.
 408 */
 409
 410static int multifd_send_pages(QEMUFile *f)
 411{
 412    int i;
 413    static int next_channel;
 414    MultiFDSendParams *p = NULL; /* make happy gcc */
 415    MultiFDPages_t *pages = multifd_send_state->pages;
 416    uint64_t transferred;
 417
 418    if (qatomic_read(&multifd_send_state->exiting)) {
 419        return -1;
 420    }
 421
 422    qemu_sem_wait(&multifd_send_state->channels_ready);
 423    /*
 424     * next_channel can remain from a previous migration that was
 425     * using more channels, so ensure it doesn't overflow if the
 426     * limit is lower now.
 427     */
 428    next_channel %= migrate_multifd_channels();
 429    for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
 430        p = &multifd_send_state->params[i];
 431
 432        qemu_mutex_lock(&p->mutex);
 433        if (p->quit) {
 434            error_report("%s: channel %d has already quit!", __func__, i);
 435            qemu_mutex_unlock(&p->mutex);
 436            return -1;
 437        }
 438        if (!p->pending_job) {
 439            p->pending_job++;
 440            next_channel = (i + 1) % migrate_multifd_channels();
 441            break;
 442        }
 443        qemu_mutex_unlock(&p->mutex);
 444    }
 445    assert(!p->pages->used);
 446    assert(!p->pages->block);
 447
 448    p->packet_num = multifd_send_state->packet_num++;
 449    multifd_send_state->pages = p->pages;
 450    p->pages = pages;
 451    transferred = ((uint64_t) pages->used) * qemu_target_page_size()
 452                + p->packet_len;
 453    qemu_file_update_transfer(f, transferred);
 454    ram_counters.multifd_bytes += transferred;
 455    ram_counters.transferred += transferred;
 456    qemu_mutex_unlock(&p->mutex);
 457    qemu_sem_post(&p->sem);
 458
 459    return 1;
 460}
 461
 462int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
 463{
 464    MultiFDPages_t *pages = multifd_send_state->pages;
 465
 466    if (!pages->block) {
 467        pages->block = block;
 468    }
 469
 470    if (pages->block == block) {
 471        pages->offset[pages->used] = offset;
 472        pages->iov[pages->used].iov_base = block->host + offset;
 473        pages->iov[pages->used].iov_len = qemu_target_page_size();
 474        pages->used++;
 475
 476        if (pages->used < pages->allocated) {
 477            return 1;
 478        }
 479    }
 480
 481    if (multifd_send_pages(f) < 0) {
 482        return -1;
 483    }
 484
 485    if (pages->block != block) {
 486        return  multifd_queue_page(f, block, offset);
 487    }
 488
 489    return 1;
 490}
 491
 492static void multifd_send_terminate_threads(Error *err)
 493{
 494    int i;
 495
 496    trace_multifd_send_terminate_threads(err != NULL);
 497
 498    if (err) {
 499        MigrationState *s = migrate_get_current();
 500        migrate_set_error(s, err);
 501        if (s->state == MIGRATION_STATUS_SETUP ||
 502            s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
 503            s->state == MIGRATION_STATUS_DEVICE ||
 504            s->state == MIGRATION_STATUS_ACTIVE) {
 505            migrate_set_state(&s->state, s->state,
 506                              MIGRATION_STATUS_FAILED);
 507        }
 508    }
 509
 510    /*
 511     * We don't want to exit each threads twice.  Depending on where
 512     * we get the error, or if there are two independent errors in two
 513     * threads at the same time, we can end calling this function
 514     * twice.
 515     */
 516    if (qatomic_xchg(&multifd_send_state->exiting, 1)) {
 517        return;
 518    }
 519
 520    for (i = 0; i < migrate_multifd_channels(); i++) {
 521        MultiFDSendParams *p = &multifd_send_state->params[i];
 522
 523        qemu_mutex_lock(&p->mutex);
 524        p->quit = true;
 525        qemu_sem_post(&p->sem);
 526        qemu_mutex_unlock(&p->mutex);
 527    }
 528}
 529
 530void multifd_save_cleanup(void)
 531{
 532    int i;
 533
 534    if (!migrate_use_multifd() || !migrate_multifd_is_allowed()) {
 535        return;
 536    }
 537    multifd_send_terminate_threads(NULL);
 538    for (i = 0; i < migrate_multifd_channels(); i++) {
 539        MultiFDSendParams *p = &multifd_send_state->params[i];
 540
 541        if (p->running) {
 542            qemu_thread_join(&p->thread);
 543        }
 544    }
 545    for (i = 0; i < migrate_multifd_channels(); i++) {
 546        MultiFDSendParams *p = &multifd_send_state->params[i];
 547        Error *local_err = NULL;
 548
 549        if (p->registered_yank) {
 550            migration_ioc_unregister_yank(p->c);
 551        }
 552        socket_send_channel_destroy(p->c);
 553        p->c = NULL;
 554        qemu_mutex_destroy(&p->mutex);
 555        qemu_sem_destroy(&p->sem);
 556        qemu_sem_destroy(&p->sem_sync);
 557        g_free(p->name);
 558        p->name = NULL;
 559        g_free(p->tls_hostname);
 560        p->tls_hostname = NULL;
 561        multifd_pages_clear(p->pages);
 562        p->pages = NULL;
 563        p->packet_len = 0;
 564        g_free(p->packet);
 565        p->packet = NULL;
 566        multifd_send_state->ops->send_cleanup(p, &local_err);
 567        if (local_err) {
 568            migrate_set_error(migrate_get_current(), local_err);
 569            error_free(local_err);
 570        }
 571    }
 572    qemu_sem_destroy(&multifd_send_state->channels_ready);
 573    g_free(multifd_send_state->params);
 574    multifd_send_state->params = NULL;
 575    multifd_pages_clear(multifd_send_state->pages);
 576    multifd_send_state->pages = NULL;
 577    g_free(multifd_send_state);
 578    multifd_send_state = NULL;
 579}
 580
 581void multifd_send_sync_main(QEMUFile *f)
 582{
 583    int i;
 584
 585    if (!migrate_use_multifd()) {
 586        return;
 587    }
 588    if (multifd_send_state->pages->used) {
 589        if (multifd_send_pages(f) < 0) {
 590            error_report("%s: multifd_send_pages fail", __func__);
 591            return;
 592        }
 593    }
 594    for (i = 0; i < migrate_multifd_channels(); i++) {
 595        MultiFDSendParams *p = &multifd_send_state->params[i];
 596
 597        trace_multifd_send_sync_main_signal(p->id);
 598
 599        qemu_mutex_lock(&p->mutex);
 600
 601        if (p->quit) {
 602            error_report("%s: channel %d has already quit", __func__, i);
 603            qemu_mutex_unlock(&p->mutex);
 604            return;
 605        }
 606
 607        p->packet_num = multifd_send_state->packet_num++;
 608        p->flags |= MULTIFD_FLAG_SYNC;
 609        p->pending_job++;
 610        qemu_file_update_transfer(f, p->packet_len);
 611        ram_counters.multifd_bytes += p->packet_len;
 612        ram_counters.transferred += p->packet_len;
 613        qemu_mutex_unlock(&p->mutex);
 614        qemu_sem_post(&p->sem);
 615    }
 616    for (i = 0; i < migrate_multifd_channels(); i++) {
 617        MultiFDSendParams *p = &multifd_send_state->params[i];
 618
 619        trace_multifd_send_sync_main_wait(p->id);
 620        qemu_sem_wait(&p->sem_sync);
 621    }
 622    trace_multifd_send_sync_main(multifd_send_state->packet_num);
 623}
 624
 625static void *multifd_send_thread(void *opaque)
 626{
 627    MultiFDSendParams *p = opaque;
 628    Error *local_err = NULL;
 629    int ret = 0;
 630    uint32_t flags = 0;
 631
 632    trace_multifd_send_thread_start(p->id);
 633    rcu_register_thread();
 634
 635    if (multifd_send_initial_packet(p, &local_err) < 0) {
 636        ret = -1;
 637        goto out;
 638    }
 639    /* initial packet */
 640    p->num_packets = 1;
 641
 642    while (true) {
 643        qemu_sem_wait(&p->sem);
 644
 645        if (qatomic_read(&multifd_send_state->exiting)) {
 646            break;
 647        }
 648        qemu_mutex_lock(&p->mutex);
 649
 650        if (p->pending_job) {
 651            uint32_t used = p->pages->used;
 652            uint64_t packet_num = p->packet_num;
 653            flags = p->flags;
 654
 655            if (used) {
 656                ret = multifd_send_state->ops->send_prepare(p, used,
 657                                                            &local_err);
 658                if (ret != 0) {
 659                    qemu_mutex_unlock(&p->mutex);
 660                    break;
 661                }
 662            }
 663            multifd_send_fill_packet(p);
 664            p->flags = 0;
 665            p->num_packets++;
 666            p->num_pages += used;
 667            p->pages->used = 0;
 668            p->pages->block = NULL;
 669            qemu_mutex_unlock(&p->mutex);
 670
 671            trace_multifd_send(p->id, packet_num, used, flags,
 672                               p->next_packet_size);
 673
 674            ret = qio_channel_write_all(p->c, (void *)p->packet,
 675                                        p->packet_len, &local_err);
 676            if (ret != 0) {
 677                break;
 678            }
 679
 680            if (used) {
 681                ret = multifd_send_state->ops->send_write(p, used, &local_err);
 682                if (ret != 0) {
 683                    break;
 684                }
 685            }
 686
 687            qemu_mutex_lock(&p->mutex);
 688            p->pending_job--;
 689            qemu_mutex_unlock(&p->mutex);
 690
 691            if (flags & MULTIFD_FLAG_SYNC) {
 692                qemu_sem_post(&p->sem_sync);
 693            }
 694            qemu_sem_post(&multifd_send_state->channels_ready);
 695        } else if (p->quit) {
 696            qemu_mutex_unlock(&p->mutex);
 697            break;
 698        } else {
 699            qemu_mutex_unlock(&p->mutex);
 700            /* sometimes there are spurious wakeups */
 701        }
 702    }
 703
 704out:
 705    if (local_err) {
 706        trace_multifd_send_error(p->id);
 707        multifd_send_terminate_threads(local_err);
 708        error_free(local_err);
 709    }
 710
 711    /*
 712     * Error happen, I will exit, but I can't just leave, tell
 713     * who pay attention to me.
 714     */
 715    if (ret != 0) {
 716        qemu_sem_post(&p->sem_sync);
 717        qemu_sem_post(&multifd_send_state->channels_ready);
 718    }
 719
 720    qemu_mutex_lock(&p->mutex);
 721    p->running = false;
 722    qemu_mutex_unlock(&p->mutex);
 723
 724    rcu_unregister_thread();
 725    trace_multifd_send_thread_end(p->id, p->num_packets, p->num_pages);
 726
 727    return NULL;
 728}
 729
 730static bool multifd_channel_connect(MultiFDSendParams *p,
 731                                    QIOChannel *ioc,
 732                                    Error *error);
 733
 734static void multifd_tls_outgoing_handshake(QIOTask *task,
 735                                           gpointer opaque)
 736{
 737    MultiFDSendParams *p = opaque;
 738    QIOChannel *ioc = QIO_CHANNEL(qio_task_get_source(task));
 739    Error *err = NULL;
 740
 741    if (qio_task_propagate_error(task, &err)) {
 742        trace_multifd_tls_outgoing_handshake_error(ioc, error_get_pretty(err));
 743    } else {
 744        trace_multifd_tls_outgoing_handshake_complete(ioc);
 745    }
 746
 747    if (!multifd_channel_connect(p, ioc, err)) {
 748        /*
 749         * Error happen, mark multifd_send_thread status as 'quit' although it
 750         * is not created, and then tell who pay attention to me.
 751         */
 752        p->quit = true;
 753        qemu_sem_post(&multifd_send_state->channels_ready);
 754        qemu_sem_post(&p->sem_sync);
 755    }
 756}
 757
 758static void *multifd_tls_handshake_thread(void *opaque)
 759{
 760    MultiFDSendParams *p = opaque;
 761    QIOChannelTLS *tioc = QIO_CHANNEL_TLS(p->c);
 762
 763    qio_channel_tls_handshake(tioc,
 764                              multifd_tls_outgoing_handshake,
 765                              p,
 766                              NULL,
 767                              NULL);
 768    return NULL;
 769}
 770
 771static void multifd_tls_channel_connect(MultiFDSendParams *p,
 772                                        QIOChannel *ioc,
 773                                        Error **errp)
 774{
 775    MigrationState *s = migrate_get_current();
 776    const char *hostname = p->tls_hostname;
 777    QIOChannelTLS *tioc;
 778
 779    tioc = migration_tls_client_create(s, ioc, hostname, errp);
 780    if (!tioc) {
 781        return;
 782    }
 783
 784    object_unref(OBJECT(ioc));
 785    trace_multifd_tls_outgoing_handshake_start(ioc, tioc, hostname);
 786    qio_channel_set_name(QIO_CHANNEL(tioc), "multifd-tls-outgoing");
 787    p->c = QIO_CHANNEL(tioc);
 788    qemu_thread_create(&p->thread, "multifd-tls-handshake-worker",
 789                       multifd_tls_handshake_thread, p,
 790                       QEMU_THREAD_JOINABLE);
 791}
 792
 793static bool multifd_channel_connect(MultiFDSendParams *p,
 794                                    QIOChannel *ioc,
 795                                    Error *error)
 796{
 797    MigrationState *s = migrate_get_current();
 798
 799    trace_multifd_set_outgoing_channel(
 800        ioc, object_get_typename(OBJECT(ioc)), p->tls_hostname, error);
 801
 802    if (!error) {
 803        if (s->parameters.tls_creds &&
 804            *s->parameters.tls_creds &&
 805            !object_dynamic_cast(OBJECT(ioc),
 806                                 TYPE_QIO_CHANNEL_TLS)) {
 807            multifd_tls_channel_connect(p, ioc, &error);
 808            if (!error) {
 809                /*
 810                 * tls_channel_connect will call back to this
 811                 * function after the TLS handshake,
 812                 * so we mustn't call multifd_send_thread until then
 813                 */
 814                return true;
 815            } else {
 816                return false;
 817            }
 818        } else {
 819            migration_ioc_register_yank(ioc);
 820            p->registered_yank = true;
 821            p->c = ioc;
 822            qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
 823                                   QEMU_THREAD_JOINABLE);
 824       }
 825       return true;
 826    }
 827
 828    return false;
 829}
 830
 831static void multifd_new_send_channel_cleanup(MultiFDSendParams *p,
 832                                             QIOChannel *ioc, Error *err)
 833{
 834     migrate_set_error(migrate_get_current(), err);
 835     /* Error happen, we need to tell who pay attention to me */
 836     qemu_sem_post(&multifd_send_state->channels_ready);
 837     qemu_sem_post(&p->sem_sync);
 838     /*
 839      * Although multifd_send_thread is not created, but main migration
 840      * thread neet to judge whether it is running, so we need to mark
 841      * its status.
 842      */
 843     p->quit = true;
 844     object_unref(OBJECT(ioc));
 845     error_free(err);
 846}
 847
 848static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
 849{
 850    MultiFDSendParams *p = opaque;
 851    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
 852    Error *local_err = NULL;
 853
 854    trace_multifd_new_send_channel_async(p->id);
 855    if (qio_task_propagate_error(task, &local_err)) {
 856        goto cleanup;
 857    } else {
 858        p->c = QIO_CHANNEL(sioc);
 859        qio_channel_set_delay(p->c, false);
 860        p->running = true;
 861        if (!multifd_channel_connect(p, sioc, local_err)) {
 862            goto cleanup;
 863        }
 864        return;
 865    }
 866
 867cleanup:
 868    multifd_new_send_channel_cleanup(p, sioc, local_err);
 869}
 870
 871static bool migrate_allow_multifd = true;
 872void migrate_protocol_allow_multifd(bool allow)
 873{
 874    migrate_allow_multifd = allow;
 875}
 876
 877bool migrate_multifd_is_allowed(void)
 878{
 879    return migrate_allow_multifd;
 880}
 881
 882int multifd_save_setup(Error **errp)
 883{
 884    int thread_count;
 885    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
 886    uint8_t i;
 887    MigrationState *s;
 888
 889    if (!migrate_use_multifd()) {
 890        return 0;
 891    }
 892    if (!migrate_multifd_is_allowed()) {
 893        error_setg(errp, "multifd is not supported by current protocol");
 894        return -1;
 895    }
 896
 897    s = migrate_get_current();
 898    thread_count = migrate_multifd_channels();
 899    multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
 900    multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
 901    multifd_send_state->pages = multifd_pages_init(page_count);
 902    qemu_sem_init(&multifd_send_state->channels_ready, 0);
 903    qatomic_set(&multifd_send_state->exiting, 0);
 904    multifd_send_state->ops = multifd_ops[migrate_multifd_compression()];
 905
 906    for (i = 0; i < thread_count; i++) {
 907        MultiFDSendParams *p = &multifd_send_state->params[i];
 908
 909        qemu_mutex_init(&p->mutex);
 910        qemu_sem_init(&p->sem, 0);
 911        qemu_sem_init(&p->sem_sync, 0);
 912        p->quit = false;
 913        p->pending_job = 0;
 914        p->id = i;
 915        p->pages = multifd_pages_init(page_count);
 916        p->packet_len = sizeof(MultiFDPacket_t)
 917                      + sizeof(uint64_t) * page_count;
 918        p->packet = g_malloc0(p->packet_len);
 919        p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
 920        p->packet->version = cpu_to_be32(MULTIFD_VERSION);
 921        p->name = g_strdup_printf("multifdsend_%d", i);
 922        p->tls_hostname = g_strdup(s->hostname);
 923        socket_send_channel_create(multifd_new_send_channel_async, p);
 924    }
 925
 926    for (i = 0; i < thread_count; i++) {
 927        MultiFDSendParams *p = &multifd_send_state->params[i];
 928        Error *local_err = NULL;
 929        int ret;
 930
 931        ret = multifd_send_state->ops->send_setup(p, &local_err);
 932        if (ret) {
 933            error_propagate(errp, local_err);
 934            return ret;
 935        }
 936    }
 937    return 0;
 938}
 939
 940struct {
 941    MultiFDRecvParams *params;
 942    /* number of created threads */
 943    int count;
 944    /* syncs main thread and channels */
 945    QemuSemaphore sem_sync;
 946    /* global number of generated multifd packets */
 947    uint64_t packet_num;
 948    /* multifd ops */
 949    MultiFDMethods *ops;
 950} *multifd_recv_state;
 951
 952static void multifd_recv_terminate_threads(Error *err)
 953{
 954    int i;
 955
 956    trace_multifd_recv_terminate_threads(err != NULL);
 957
 958    if (err) {
 959        MigrationState *s = migrate_get_current();
 960        migrate_set_error(s, err);
 961        if (s->state == MIGRATION_STATUS_SETUP ||
 962            s->state == MIGRATION_STATUS_ACTIVE) {
 963            migrate_set_state(&s->state, s->state,
 964                              MIGRATION_STATUS_FAILED);
 965        }
 966    }
 967
 968    for (i = 0; i < migrate_multifd_channels(); i++) {
 969        MultiFDRecvParams *p = &multifd_recv_state->params[i];
 970
 971        qemu_mutex_lock(&p->mutex);
 972        p->quit = true;
 973        /*
 974         * We could arrive here for two reasons:
 975         *  - normal quit, i.e. everything went fine, just finished
 976         *  - error quit: We close the channels so the channel threads
 977         *    finish the qio_channel_read_all_eof()
 978         */
 979        if (p->c) {
 980            qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
 981        }
 982        qemu_mutex_unlock(&p->mutex);
 983    }
 984}
 985
 986int multifd_load_cleanup(Error **errp)
 987{
 988    int i;
 989
 990    if (!migrate_use_multifd() || !migrate_multifd_is_allowed()) {
 991        return 0;
 992    }
 993    multifd_recv_terminate_threads(NULL);
 994    for (i = 0; i < migrate_multifd_channels(); i++) {
 995        MultiFDRecvParams *p = &multifd_recv_state->params[i];
 996
 997        if (p->running) {
 998            p->quit = true;
 999            /*
1000             * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
1001             * however try to wakeup it without harm in cleanup phase.
1002             */
1003            qemu_sem_post(&p->sem_sync);
1004            qemu_thread_join(&p->thread);
1005        }
1006    }
1007    for (i = 0; i < migrate_multifd_channels(); i++) {
1008        MultiFDRecvParams *p = &multifd_recv_state->params[i];
1009
1010        migration_ioc_unregister_yank(p->c);
1011        object_unref(OBJECT(p->c));
1012        p->c = NULL;
1013        qemu_mutex_destroy(&p->mutex);
1014        qemu_sem_destroy(&p->sem_sync);
1015        g_free(p->name);
1016        p->name = NULL;
1017        multifd_pages_clear(p->pages);
1018        p->pages = NULL;
1019        p->packet_len = 0;
1020        g_free(p->packet);
1021        p->packet = NULL;
1022        multifd_recv_state->ops->recv_cleanup(p);
1023    }
1024    qemu_sem_destroy(&multifd_recv_state->sem_sync);
1025    g_free(multifd_recv_state->params);
1026    multifd_recv_state->params = NULL;
1027    g_free(multifd_recv_state);
1028    multifd_recv_state = NULL;
1029
1030    return 0;
1031}
1032
1033void multifd_recv_sync_main(void)
1034{
1035    int i;
1036
1037    if (!migrate_use_multifd()) {
1038        return;
1039    }
1040    for (i = 0; i < migrate_multifd_channels(); i++) {
1041        MultiFDRecvParams *p = &multifd_recv_state->params[i];
1042
1043        trace_multifd_recv_sync_main_wait(p->id);
1044        qemu_sem_wait(&multifd_recv_state->sem_sync);
1045    }
1046    for (i = 0; i < migrate_multifd_channels(); i++) {
1047        MultiFDRecvParams *p = &multifd_recv_state->params[i];
1048
1049        WITH_QEMU_LOCK_GUARD(&p->mutex) {
1050            if (multifd_recv_state->packet_num < p->packet_num) {
1051                multifd_recv_state->packet_num = p->packet_num;
1052            }
1053        }
1054        trace_multifd_recv_sync_main_signal(p->id);
1055        qemu_sem_post(&p->sem_sync);
1056    }
1057    trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
1058}
1059
1060static void *multifd_recv_thread(void *opaque)
1061{
1062    MultiFDRecvParams *p = opaque;
1063    Error *local_err = NULL;
1064    int ret;
1065
1066    trace_multifd_recv_thread_start(p->id);
1067    rcu_register_thread();
1068
1069    while (true) {
1070        uint32_t used;
1071        uint32_t flags;
1072
1073        if (p->quit) {
1074            break;
1075        }
1076
1077        ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
1078                                       p->packet_len, &local_err);
1079        if (ret == 0) {   /* EOF */
1080            break;
1081        }
1082        if (ret == -1) {   /* Error */
1083            break;
1084        }
1085
1086        qemu_mutex_lock(&p->mutex);
1087        ret = multifd_recv_unfill_packet(p, &local_err);
1088        if (ret) {
1089            qemu_mutex_unlock(&p->mutex);
1090            break;
1091        }
1092
1093        used = p->pages->used;
1094        flags = p->flags;
1095        /* recv methods don't know how to handle the SYNC flag */
1096        p->flags &= ~MULTIFD_FLAG_SYNC;
1097        trace_multifd_recv(p->id, p->packet_num, used, flags,
1098                           p->next_packet_size);
1099        p->num_packets++;
1100        p->num_pages += used;
1101        qemu_mutex_unlock(&p->mutex);
1102
1103        if (used) {
1104            ret = multifd_recv_state->ops->recv_pages(p, used, &local_err);
1105            if (ret != 0) {
1106                break;
1107            }
1108        }
1109
1110        if (flags & MULTIFD_FLAG_SYNC) {
1111            qemu_sem_post(&multifd_recv_state->sem_sync);
1112            qemu_sem_wait(&p->sem_sync);
1113        }
1114    }
1115
1116    if (local_err) {
1117        multifd_recv_terminate_threads(local_err);
1118        error_free(local_err);
1119    }
1120    qemu_mutex_lock(&p->mutex);
1121    p->running = false;
1122    qemu_mutex_unlock(&p->mutex);
1123
1124    rcu_unregister_thread();
1125    trace_multifd_recv_thread_end(p->id, p->num_packets, p->num_pages);
1126
1127    return NULL;
1128}
1129
1130int multifd_load_setup(Error **errp)
1131{
1132    int thread_count;
1133    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
1134    uint8_t i;
1135
1136    if (!migrate_use_multifd()) {
1137        return 0;
1138    }
1139    if (!migrate_multifd_is_allowed()) {
1140        error_setg(errp, "multifd is not supported by current protocol");
1141        return -1;
1142    }
1143    thread_count = migrate_multifd_channels();
1144    multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
1145    multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
1146    qatomic_set(&multifd_recv_state->count, 0);
1147    qemu_sem_init(&multifd_recv_state->sem_sync, 0);
1148    multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()];
1149
1150    for (i = 0; i < thread_count; i++) {
1151        MultiFDRecvParams *p = &multifd_recv_state->params[i];
1152
1153        qemu_mutex_init(&p->mutex);
1154        qemu_sem_init(&p->sem_sync, 0);
1155        p->quit = false;
1156        p->id = i;
1157        p->pages = multifd_pages_init(page_count);
1158        p->packet_len = sizeof(MultiFDPacket_t)
1159                      + sizeof(uint64_t) * page_count;
1160        p->packet = g_malloc0(p->packet_len);
1161        p->name = g_strdup_printf("multifdrecv_%d", i);
1162    }
1163
1164    for (i = 0; i < thread_count; i++) {
1165        MultiFDRecvParams *p = &multifd_recv_state->params[i];
1166        Error *local_err = NULL;
1167        int ret;
1168
1169        ret = multifd_recv_state->ops->recv_setup(p, &local_err);
1170        if (ret) {
1171            error_propagate(errp, local_err);
1172            return ret;
1173        }
1174    }
1175    return 0;
1176}
1177
1178bool multifd_recv_all_channels_created(void)
1179{
1180    int thread_count = migrate_multifd_channels();
1181
1182    if (!migrate_use_multifd()) {
1183        return true;
1184    }
1185
1186    if (!multifd_recv_state) {
1187        /* Called before any connections created */
1188        return false;
1189    }
1190
1191    return thread_count == qatomic_read(&multifd_recv_state->count);
1192}
1193
1194/*
1195 * Try to receive all multifd channels to get ready for the migration.
1196 * - Return true and do not set @errp when correctly receiving all channels;
1197 * - Return false and do not set @errp when correctly receiving the current one;
1198 * - Return false and set @errp when failing to receive the current channel.
1199 */
1200bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
1201{
1202    MultiFDRecvParams *p;
1203    Error *local_err = NULL;
1204    int id;
1205
1206    id = multifd_recv_initial_packet(ioc, &local_err);
1207    if (id < 0) {
1208        multifd_recv_terminate_threads(local_err);
1209        error_propagate_prepend(errp, local_err,
1210                                "failed to receive packet"
1211                                " via multifd channel %d: ",
1212                                qatomic_read(&multifd_recv_state->count));
1213        return false;
1214    }
1215    trace_multifd_recv_new_channel(id);
1216
1217    p = &multifd_recv_state->params[id];
1218    if (p->c != NULL) {
1219        error_setg(&local_err, "multifd: received id '%d' already setup'",
1220                   id);
1221        multifd_recv_terminate_threads(local_err);
1222        error_propagate(errp, local_err);
1223        return false;
1224    }
1225    p->c = ioc;
1226    object_ref(OBJECT(ioc));
1227    /* initial packet */
1228    p->num_packets = 1;
1229
1230    p->running = true;
1231    qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
1232                       QEMU_THREAD_JOINABLE);
1233    qatomic_inc(&multifd_recv_state->count);
1234    return qatomic_read(&multifd_recv_state->count) ==
1235           migrate_multifd_channels();
1236}
1237