migration/multifd.c

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