migration/multifd.c

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