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