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
  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     MultiFDPages_t *pages = p->pages;
  90     size_t page_size = qemu_target_page_size();
  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 = page_size;
  95         p->iovs_num++;
  96     }
  97
  98     p->next_packet_size = p->normal_num * 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     size_t page_size = qemu_target_page_size();
 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 = 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     int ret;
 179
 180     msg.magic = cpu_to_be32(MULTIFD_MAGIC);
 181     msg.version = cpu_to_be32(MULTIFD_VERSION);
 182     msg.id = p->id;
 183     memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
 184
 185     ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), errp);
 186     if (ret != 0) {
 187         return -1;
 188     }
 189     return 0;
 190 }
 191
 192 static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
 193 {
 194     MultiFDInit_t msg;
 195     int ret;
 196
 197     ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
 198     if (ret != 0) {
 199         return -1;
 200     }
 201
 202     msg.magic = be32_to_cpu(msg.magic);
 203     msg.version = be32_to_cpu(msg.version);
 204
 205     if (msg.magic != MULTIFD_MAGIC) {
 206         error_setg(errp, "multifd: received packet magic %x "
 207                    "expected %x", msg.magic, MULTIFD_MAGIC);
 208         return -1;
 209     }
 210
 211     if (msg.version != MULTIFD_VERSION) {
 212         error_setg(errp, "multifd: received packet version %u "
 213                    "expected %u", msg.version, MULTIFD_VERSION);
 214         return -1;
 215     }
 216
 217     if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
 218         char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
 219         char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
 220
 221         error_setg(errp, "multifd: received uuid '%s' and expected "
 222                    "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
 223         g_free(uuid);
 224         g_free(msg_uuid);
 225         return -1;
 226     }
 227
 228     if (msg.id > migrate_multifd_channels()) {
 229         error_setg(errp, "multifd: received channel version %u "
 230                    "expected %u", msg.version, MULTIFD_VERSION);
 231         return -1;
 232     }
 233
 234     return msg.id;
 235 }
 236
 237 static MultiFDPages_t *multifd_pages_init(size_t size)
 238 {
 239     MultiFDPages_t *pages = g_new0(MultiFDPages_t, 1);
 240
 241     pages->allocated = size;
 242     pages->offset = g_new0(ram_addr_t, size);
 243
 244     return pages;
 245 }
 246
 247 static void multifd_pages_clear(MultiFDPages_t *pages)
 248 {
 249     pages->num = 0;
 250     pages->allocated = 0;
 251     pages->packet_num = 0;
 252     pages->block = NULL;
 253     g_free(pages->offset);
 254     pages->offset = NULL;
 255     g_free(pages);
 256 }
 257
 258 static void multifd_send_fill_packet(MultiFDSendParams *p)
 259 {
 260     MultiFDPacket_t *packet = p->packet;
 261     int i;
 262
 263     packet->flags = cpu_to_be32(p->flags);
 264     packet->pages_alloc = cpu_to_be32(p->pages->allocated);
 265     packet->normal_pages = cpu_to_be32(p->normal_num);
 266     packet->next_packet_size = cpu_to_be32(p->next_packet_size);
 267     packet->packet_num = cpu_to_be64(p->packet_num);
 268
 269     if (p->pages->block) {
 270         strncpy(packet->ramblock, p->pages->block->idstr, 256);
 271     }
 272
 273     for (i = 0; i < p->normal_num; i++) {
 274         /* there are architectures where ram_addr_t is 32 bit */
 275         uint64_t temp = p->normal[i];
 276
 277         packet->offset[i] = cpu_to_be64(temp);
 278     }
 279 }
 280
 281 static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
 282 {
 283     MultiFDPacket_t *packet = p->packet;
 284     size_t page_size = qemu_target_page_size();
 285     uint32_t page_count = MULTIFD_PACKET_SIZE / page_size;
 286     RAMBlock *block;
 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 > page_count) {
 313         error_setg(errp, "multifd: received packet "
 314                    "with size %u and expected a size of %u",
 315                    packet->pages_alloc, 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     block = qemu_ram_block_by_name(packet->ramblock);
 337     if (!block) {
 338         error_setg(errp, "multifd: unknown ram block %s",
 339                    packet->ramblock);
 340         return -1;
 341     }
 342
 343     p->host = 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 > (block->used_length - page_size)) {
 348             error_setg(errp, "multifd: offset too long %" PRIu64
 349                        " (max " RAM_ADDR_FMT ")",
 350                        offset, 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     uint64_t transferred;
 402
 403     if (qatomic_read(&multifd_send_state->exiting)) {
 404         return -1;
 405     }
 406
 407     qemu_sem_wait(&multifd_send_state->channels_ready);
 408     /*
 409      * next_channel can remain from a previous migration that was
 410      * using more channels, so ensure it doesn't overflow if the
 411      * limit is lower now.
 412      */
 413     next_channel %= migrate_multifd_channels();
 414     for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
 415         p = &multifd_send_state->params[i];
 416
 417         qemu_mutex_lock(&p->mutex);
 418         if (p->quit) {
 419             error_report("%s: channel %d has already quit!", __func__, i);
 420             qemu_mutex_unlock(&p->mutex);
 421             return -1;
 422         }
 423         if (!p->pending_job) {
 424             p->pending_job++;
 425             next_channel = (i + 1) % migrate_multifd_channels();
 426             break;
 427         }
 428         qemu_mutex_unlock(&p->mutex);
 429     }
 430     assert(!p->pages->num);
 431     assert(!p->pages->block);
 432
 433     p->packet_num = multifd_send_state->packet_num++;
 434     multifd_send_state->pages = p->pages;
 435     p->pages = pages;
 436     transferred = ((uint64_t) pages->num) * qemu_target_page_size()
 437                 + p->packet_len;
 438     qemu_file_acct_rate_limit(f, transferred);
 439     ram_counters.multifd_bytes += transferred;
 440     ram_counters.transferred += transferred;
 441     qemu_mutex_unlock(&p->mutex);
 442     qemu_sem_post(&p->sem);
 443
 444     return 1;
 445 }
 446
 447 int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
 448 {
 449     MultiFDPages_t *pages = multifd_send_state->pages;
 450
 451     if (!pages->block) {
 452         pages->block = block;
 453     }
 454
 455     if (pages->block == block) {
 456         pages->offset[pages->num] = offset;
 457         pages->num++;
 458
 459         if (pages->num < pages->allocated) {
 460             return 1;
 461         }
 462     }
 463
 464     if (multifd_send_pages(f) < 0) {
 465         return -1;
 466     }
 467
 468     if (pages->block != block) {
 469         return  multifd_queue_page(f, block, offset);
 470     }
 471
 472     return 1;
 473 }
 474
 475 static void multifd_send_terminate_threads(Error *err)
 476 {
 477     int i;
 478
 479     trace_multifd_send_terminate_threads(err != NULL);
 480
 481     if (err) {
 482         MigrationState *s = migrate_get_current();
 483         migrate_set_error(s, err);
 484         if (s->state == MIGRATION_STATUS_SETUP ||
 485             s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
 486             s->state == MIGRATION_STATUS_DEVICE ||
 487             s->state == MIGRATION_STATUS_ACTIVE) {
 488             migrate_set_state(&s->state, s->state,
 489                               MIGRATION_STATUS_FAILED);
 490         }
 491     }
 492
 493     /*
 494      * We don't want to exit each threads twice.  Depending on where
 495      * we get the error, or if there are two independent errors in two
 496      * threads at the same time, we can end calling this function
 497      * twice.
 498      */
 499     if (qatomic_xchg(&multifd_send_state->exiting, 1)) {
 500         return;
 501     }
 502
 503     for (i = 0; i < migrate_multifd_channels(); i++) {
 504         MultiFDSendParams *p = &multifd_send_state->params[i];
 505
 506         qemu_mutex_lock(&p->mutex);
 507         p->quit = true;
 508         qemu_sem_post(&p->sem);
 509         if (p->c) {
 510             qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
 511         }
 512         qemu_mutex_unlock(&p->mutex);
 513     }
 514 }
 515
 516 void multifd_save_cleanup(void)
 517 {
 518     int i;
 519
 520     if (!migrate_use_multifd() || !migrate_multi_channels_is_allowed()) {
 521         return;
 522     }
 523     multifd_send_terminate_threads(NULL);
 524     for (i = 0; i < migrate_multifd_channels(); i++) {
 525         MultiFDSendParams *p = &multifd_send_state->params[i];
 526
 527         if (p->running) {
 528             qemu_thread_join(&p->thread);
 529         }
 530     }
 531     for (i = 0; i < migrate_multifd_channels(); i++) {
 532         MultiFDSendParams *p = &multifd_send_state->params[i];
 533         Error *local_err = NULL;
 534
 535         if (p->registered_yank) {
 536             migration_ioc_unregister_yank(p->c);
 537         }
 538         socket_send_channel_destroy(p->c);
 539         p->c = NULL;
 540         qemu_mutex_destroy(&p->mutex);
 541         qemu_sem_destroy(&p->sem);
 542         qemu_sem_destroy(&p->sem_sync);
 543         g_free(p->name);
 544         p->name = NULL;
 545         multifd_pages_clear(p->pages);
 546         p->pages = NULL;
 547         p->packet_len = 0;
 548         g_free(p->packet);
 549         p->packet = NULL;
 550         g_free(p->iov);
 551         p->iov = NULL;
 552         g_free(p->normal);
 553         p->normal = NULL;
 554         multifd_send_state->ops->send_cleanup(p, &local_err);
 555         if (local_err) {
 556             migrate_set_error(migrate_get_current(), local_err);
 557             error_free(local_err);
 558         }
 559     }
 560     qemu_sem_destroy(&multifd_send_state->channels_ready);
 561     g_free(multifd_send_state->params);
 562     multifd_send_state->params = NULL;
 563     multifd_pages_clear(multifd_send_state->pages);
 564     multifd_send_state->pages = NULL;
 565     g_free(multifd_send_state);
 566     multifd_send_state = NULL;
 567 }
 568
 569 static int multifd_zero_copy_flush(QIOChannel *c)
 570 {
 571     int ret;
 572     Error *err = NULL;
 573
 574     ret = qio_channel_flush(c, &err);
 575     if (ret < 0) {
 576         error_report_err(err);
 577         return -1;
 578     }
 579     if (ret == 1) {
 580         dirty_sync_missed_zero_copy();
 581     }
 582
 583     return ret;
 584 }
 585
 586 int multifd_send_sync_main(QEMUFile *f)
 587 {
 588     int i;
 589     bool flush_zero_copy;
 590
 591     if (!migrate_use_multifd()) {
 592         return 0;
 593     }
 594     if (multifd_send_state->pages->num) {
 595         if (multifd_send_pages(f) < 0) {
 596             error_report("%s: multifd_send_pages fail", __func__);
 597             return -1;
 598         }
 599     }
 600
 601     /*
 602      * When using zero-copy, it's necessary to flush the pages before any of
 603      * the pages can be sent again, so we'll make sure the new version of the
 604      * pages will always arrive _later_ than the old pages.
 605      *
 606      * Currently we achieve this by flushing the zero-page requested writes
 607      * per ram iteration, but in the future we could potentially optimize it
 608      * to be less frequent, e.g. only after we finished one whole scanning of
 609      * all the dirty bitmaps.
 610      */
 611
 612     flush_zero_copy = migrate_use_zero_copy_send();
 613
 614     for (i = 0; i < migrate_multifd_channels(); i++) {
 615         MultiFDSendParams *p = &multifd_send_state->params[i];
 616
 617         trace_multifd_send_sync_main_signal(p->id);
 618
 619         qemu_mutex_lock(&p->mutex);
 620
 621         if (p->quit) {
 622             error_report("%s: channel %d has already quit", __func__, i);
 623             qemu_mutex_unlock(&p->mutex);
 624             return -1;
 625         }
 626
 627         p->packet_num = multifd_send_state->packet_num++;
 628         p->flags |= MULTIFD_FLAG_SYNC;
 629         p->pending_job++;
 630         qemu_file_acct_rate_limit(f, p->packet_len);
 631         ram_counters.multifd_bytes += p->packet_len;
 632         ram_counters.transferred += p->packet_len;
 633         qemu_mutex_unlock(&p->mutex);
 634         qemu_sem_post(&p->sem);
 635
 636         if (flush_zero_copy && p->c && (multifd_zero_copy_flush(p->c) < 0)) {
 637             return -1;
 638         }
 639     }
 640     for (i = 0; i < migrate_multifd_channels(); i++) {
 641         MultiFDSendParams *p = &multifd_send_state->params[i];
 642
 643         trace_multifd_send_sync_main_wait(p->id);
 644         qemu_sem_wait(&p->sem_sync);
 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     Error *local_err = NULL;
 655     int ret = 0;
 656     bool use_zero_copy_send = migrate_use_zero_copy_send();
 657
 658     trace_multifd_send_thread_start(p->id);
 659     rcu_register_thread();
 660
 661     if (multifd_send_initial_packet(p, &local_err) < 0) {
 662         ret = -1;
 663         goto out;
 664     }
 665     /* initial packet */
 666     p->num_packets = 1;
 667
 668     while (true) {
 669         qemu_sem_wait(&p->sem);
 670
 671         if (qatomic_read(&multifd_send_state->exiting)) {
 672             break;
 673         }
 674         qemu_mutex_lock(&p->mutex);
 675
 676         if (p->pending_job) {
 677             uint64_t packet_num = p->packet_num;
 678             uint32_t flags = p->flags;
 679             p->normal_num = 0;
 680
 681             if (use_zero_copy_send) {
 682                 p->iovs_num = 0;
 683             } else {
 684                 p->iovs_num = 1;
 685             }
 686
 687             for (int i = 0; i < p->pages->num; i++) {
 688                 p->normal[p->normal_num] = p->pages->offset[i];
 689                 p->normal_num++;
 690             }
 691
 692             if (p->normal_num) {
 693                 ret = multifd_send_state->ops->send_prepare(p, &local_err);
 694                 if (ret != 0) {
 695                     qemu_mutex_unlock(&p->mutex);
 696                     break;
 697                 }
 698             }
 699             multifd_send_fill_packet(p);
 700             p->flags = 0;
 701             p->num_packets++;
 702             p->total_normal_pages += p->normal_num;
 703             p->pages->num = 0;
 704             p->pages->block = NULL;
 705             qemu_mutex_unlock(&p->mutex);
 706
 707             trace_multifd_send(p->id, packet_num, p->normal_num, flags,
 708                                p->next_packet_size);
 709
 710             if (use_zero_copy_send) {
 711                 /* Send header first, without zerocopy */
 712                 ret = qio_channel_write_all(p->c, (void *)p->packet,
 713                                             p->packet_len, &local_err);
 714                 if (ret != 0) {
 715                     break;
 716                 }
 717             } else {
 718                 /* Send header using the same writev call */
 719                 p->iov[0].iov_len = p->packet_len;
 720                 p->iov[0].iov_base = p->packet;
 721             }
 722
 723             ret = qio_channel_writev_full_all(p->c, p->iov, p->iovs_num, NULL,
 724                                               0, p->write_flags, &local_err);
 725             if (ret != 0) {
 726                 break;
 727             }
 728
 729             qemu_mutex_lock(&p->mutex);
 730             p->pending_job--;
 731             qemu_mutex_unlock(&p->mutex);
 732
 733             if (flags & MULTIFD_FLAG_SYNC) {
 734                 qemu_sem_post(&p->sem_sync);
 735             }
 736             qemu_sem_post(&multifd_send_state->channels_ready);
 737         } else if (p->quit) {
 738             qemu_mutex_unlock(&p->mutex);
 739             break;
 740         } else {
 741             qemu_mutex_unlock(&p->mutex);
 742             /* sometimes there are spurious wakeups */
 743         }
 744     }
 745
 746 out:
 747     if (local_err) {
 748         trace_multifd_send_error(p->id);
 749         multifd_send_terminate_threads(local_err);
 750         error_free(local_err);
 751     }
 752
 753     /*
 754      * Error happen, I will exit, but I can't just leave, tell
 755      * who pay attention to me.
 756      */
 757     if (ret != 0) {
 758         qemu_sem_post(&p->sem_sync);
 759         qemu_sem_post(&multifd_send_state->channels_ready);
 760     }
 761
 762     qemu_mutex_lock(&p->mutex);
 763     p->running = false;
 764     qemu_mutex_unlock(&p->mutex);
 765
 766     rcu_unregister_thread();
 767     trace_multifd_send_thread_end(p->id, p->num_packets, p->total_normal_pages);
 768
 769     return NULL;
 770 }
 771
 772 static bool multifd_channel_connect(MultiFDSendParams *p,
 773                                     QIOChannel *ioc,
 774                                     Error *error);
 775
 776 static void multifd_tls_outgoing_handshake(QIOTask *task,
 777                                            gpointer opaque)
 778 {
 779     MultiFDSendParams *p = opaque;
 780     QIOChannel *ioc = QIO_CHANNEL(qio_task_get_source(task));
 781     Error *err = NULL;
 782
 783     if (qio_task_propagate_error(task, &err)) {
 784         trace_multifd_tls_outgoing_handshake_error(ioc, error_get_pretty(err));
 785     } else {
 786         trace_multifd_tls_outgoing_handshake_complete(ioc);
 787     }
 788
 789     if (!multifd_channel_connect(p, ioc, err)) {
 790         /*
 791          * Error happen, mark multifd_send_thread status as 'quit' although it
 792          * is not created, and then tell who pay attention to me.
 793          */
 794         p->quit = true;
 795         qemu_sem_post(&multifd_send_state->channels_ready);
 796         qemu_sem_post(&p->sem_sync);
 797     }
 798 }
 799
 800 static void *multifd_tls_handshake_thread(void *opaque)
 801 {
 802     MultiFDSendParams *p = opaque;
 803     QIOChannelTLS *tioc = QIO_CHANNEL_TLS(p->c);
 804
 805     qio_channel_tls_handshake(tioc,
 806                               multifd_tls_outgoing_handshake,
 807                               p,
 808                               NULL,
 809                               NULL);
 810     return NULL;
 811 }
 812
 813 static void multifd_tls_channel_connect(MultiFDSendParams *p,
 814                                         QIOChannel *ioc,
 815                                         Error **errp)
 816 {
 817     MigrationState *s = migrate_get_current();
 818     const char *hostname = s->hostname;
 819     QIOChannelTLS *tioc;
 820
 821     tioc = migration_tls_client_create(s, ioc, hostname, errp);
 822     if (!tioc) {
 823         return;
 824     }
 825
 826     object_unref(OBJECT(ioc));
 827     trace_multifd_tls_outgoing_handshake_start(ioc, tioc, hostname);
 828     qio_channel_set_name(QIO_CHANNEL(tioc), "multifd-tls-outgoing");
 829     p->c = QIO_CHANNEL(tioc);
 830     qemu_thread_create(&p->thread, "multifd-tls-handshake-worker",
 831                        multifd_tls_handshake_thread, p,
 832                        QEMU_THREAD_JOINABLE);
 833 }
 834
 835 static bool multifd_channel_connect(MultiFDSendParams *p,
 836                                     QIOChannel *ioc,
 837                                     Error *error)
 838 {
 839     trace_multifd_set_outgoing_channel(
 840         ioc, object_get_typename(OBJECT(ioc)),
 841         migrate_get_current()->hostname, error);
 842
 843     if (!error) {
 844         if (migrate_channel_requires_tls_upgrade(ioc)) {
 845             multifd_tls_channel_connect(p, ioc, &error);
 846             if (!error) {
 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 true;
 853             } else {
 854                 return false;
 855             }
 856         } else {
 857             migration_ioc_register_yank(ioc);
 858             p->registered_yank = true;
 859             p->c = ioc;
 860             qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
 861                                    QEMU_THREAD_JOINABLE);
 862        }
 863        return true;
 864     }
 865
 866     return false;
 867 }
 868
 869 static void multifd_new_send_channel_cleanup(MultiFDSendParams *p,
 870                                              QIOChannel *ioc, Error *err)
 871 {
 872      migrate_set_error(migrate_get_current(), err);
 873      /* Error happen, we need to tell who pay attention to me */
 874      qemu_sem_post(&multifd_send_state->channels_ready);
 875      qemu_sem_post(&p->sem_sync);
 876      /*
 877       * Although multifd_send_thread is not created, but main migration
 878       * thread neet to judge whether it is running, so we need to mark
 879       * its status.
 880       */
 881      p->quit = true;
 882      object_unref(OBJECT(ioc));
 883      error_free(err);
 884 }
 885
 886 static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
 887 {
 888     MultiFDSendParams *p = opaque;
 889     QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
 890     Error *local_err = NULL;
 891
 892     trace_multifd_new_send_channel_async(p->id);
 893     if (qio_task_propagate_error(task, &local_err)) {
 894         goto cleanup;
 895     } else {
 896         p->c = QIO_CHANNEL(sioc);
 897         qio_channel_set_delay(p->c, false);
 898         p->running = true;
 899         if (!multifd_channel_connect(p, sioc, local_err)) {
 900             goto cleanup;
 901         }
 902         return;
 903     }
 904
 905 cleanup:
 906     multifd_new_send_channel_cleanup(p, sioc, local_err);
 907 }
 908
 909 int multifd_save_setup(Error **errp)
 910 {
 911     int thread_count;
 912     uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
 913     uint8_t i;
 914
 915     if (!migrate_use_multifd()) {
 916         return 0;
 917     }
 918     if (!migrate_multi_channels_is_allowed()) {
 919         error_setg(errp, "multifd is not supported by current protocol");
 920         return -1;
 921     }
 922
 923     thread_count = migrate_multifd_channels();
 924     multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
 925     multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
 926     multifd_send_state->pages = multifd_pages_init(page_count);
 927     qemu_sem_init(&multifd_send_state->channels_ready, 0);
 928     qatomic_set(&multifd_send_state->exiting, 0);
 929     multifd_send_state->ops = multifd_ops[migrate_multifd_compression()];
 930
 931     for (i = 0; i < thread_count; i++) {
 932         MultiFDSendParams *p = &multifd_send_state->params[i];
 933
 934         qemu_mutex_init(&p->mutex);
 935         qemu_sem_init(&p->sem, 0);
 936         qemu_sem_init(&p->sem_sync, 0);
 937         p->quit = false;
 938         p->pending_job = 0;
 939         p->id = i;
 940         p->pages = multifd_pages_init(page_count);
 941         p->packet_len = sizeof(MultiFDPacket_t)
 942                       + sizeof(uint64_t) * page_count;
 943         p->packet = g_malloc0(p->packet_len);
 944         p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
 945         p->packet->version = cpu_to_be32(MULTIFD_VERSION);
 946         p->name = g_strdup_printf("multifdsend_%d", i);
 947         /* We need one extra place for the packet header */
 948         p->iov = g_new0(struct iovec, page_count + 1);
 949         p->normal = g_new0(ram_addr_t, page_count);
 950
 951         if (migrate_use_zero_copy_send()) {
 952             p->write_flags = QIO_CHANNEL_WRITE_FLAG_ZERO_COPY;
 953         } else {
 954             p->write_flags = 0;
 955         }
 956
 957         socket_send_channel_create(multifd_new_send_channel_async, p);
 958     }
 959
 960     for (i = 0; i < thread_count; i++) {
 961         MultiFDSendParams *p = &multifd_send_state->params[i];
 962         Error *local_err = NULL;
 963         int ret;
 964
 965         ret = multifd_send_state->ops->send_setup(p, &local_err);
 966         if (ret) {
 967             error_propagate(errp, local_err);
 968             return ret;
 969         }
 970     }
 971     return 0;
 972 }
 973
 974 struct {
 975     MultiFDRecvParams *params;
 976     /* number of created threads */
 977     int count;
 978     /* syncs main thread and channels */
 979     QemuSemaphore sem_sync;
 980     /* global number of generated multifd packets */
 981     uint64_t packet_num;
 982     /* multifd ops */
 983     MultiFDMethods *ops;
 984 } *multifd_recv_state;
 985
 986 static void multifd_recv_terminate_threads(Error *err)
 987 {
 988     int i;
 989
 990     trace_multifd_recv_terminate_threads(err != NULL);
 991
 992     if (err) {
 993         MigrationState *s = migrate_get_current();
 994         migrate_set_error(s, err);
 995         if (s->state == MIGRATION_STATUS_SETUP ||
 996             s->state == MIGRATION_STATUS_ACTIVE) {
 997             migrate_set_state(&s->state, s->state,
 998                               MIGRATION_STATUS_FAILED);
 999         }
1000     }
1001
1002     for (i = 0; i < migrate_multifd_channels(); i++) {
1003         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1004
1005         qemu_mutex_lock(&p->mutex);
1006         p->quit = true;
1007         /*
1008          * We could arrive here for two reasons:
1009          *  - normal quit, i.e. everything went fine, just finished
1010          *  - error quit: We close the channels so the channel threads
1011          *    finish the qio_channel_read_all_eof()
1012          */
1013         if (p->c) {
1014             qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
1015         }
1016         qemu_mutex_unlock(&p->mutex);
1017     }
1018 }
1019
1020 int multifd_load_cleanup(Error **errp)
1021 {
1022     int i;
1023
1024     if (!migrate_use_multifd() || !migrate_multi_channels_is_allowed()) {
1025         return 0;
1026     }
1027     multifd_recv_terminate_threads(NULL);
1028     for (i = 0; i < migrate_multifd_channels(); i++) {
1029         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1030
1031         if (p->running) {
1032             p->quit = true;
1033             /*
1034              * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
1035              * however try to wakeup it without harm in cleanup phase.
1036              */
1037             qemu_sem_post(&p->sem_sync);
1038             qemu_thread_join(&p->thread);
1039         }
1040     }
1041     for (i = 0; i < migrate_multifd_channels(); i++) {
1042         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1043
1044         migration_ioc_unregister_yank(p->c);
1045         object_unref(OBJECT(p->c));
1046         p->c = NULL;
1047         qemu_mutex_destroy(&p->mutex);
1048         qemu_sem_destroy(&p->sem_sync);
1049         g_free(p->name);
1050         p->name = NULL;
1051         p->packet_len = 0;
1052         g_free(p->packet);
1053         p->packet = NULL;
1054         g_free(p->iov);
1055         p->iov = NULL;
1056         g_free(p->normal);
1057         p->normal = NULL;
1058         multifd_recv_state->ops->recv_cleanup(p);
1059     }
1060     qemu_sem_destroy(&multifd_recv_state->sem_sync);
1061     g_free(multifd_recv_state->params);
1062     multifd_recv_state->params = NULL;
1063     g_free(multifd_recv_state);
1064     multifd_recv_state = NULL;
1065
1066     return 0;
1067 }
1068
1069 void multifd_recv_sync_main(void)
1070 {
1071     int i;
1072
1073     if (!migrate_use_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) {   /* EOF */
1115             break;
1116         }
1117         if (ret == -1) {   /* Error */
1118             break;
1119         }
1120
1121         qemu_mutex_lock(&p->mutex);
1122         ret = multifd_recv_unfill_packet(p, &local_err);
1123         if (ret) {
1124             qemu_mutex_unlock(&p->mutex);
1125             break;
1126         }
1127
1128         flags = p->flags;
1129         /* recv methods don't know how to handle the SYNC flag */
1130         p->flags &= ~MULTIFD_FLAG_SYNC;
1131         trace_multifd_recv(p->id, p->packet_num, p->normal_num, flags,
1132                            p->next_packet_size);
1133         p->num_packets++;
1134         p->total_normal_pages += p->normal_num;
1135         qemu_mutex_unlock(&p->mutex);
1136
1137         if (p->normal_num) {
1138             ret = multifd_recv_state->ops->recv_pages(p, &local_err);
1139             if (ret != 0) {
1140                 break;
1141             }
1142         }
1143
1144         if (flags & MULTIFD_FLAG_SYNC) {
1145             qemu_sem_post(&multifd_recv_state->sem_sync);
1146             qemu_sem_wait(&p->sem_sync);
1147         }
1148     }
1149
1150     if (local_err) {
1151         multifd_recv_terminate_threads(local_err);
1152         error_free(local_err);
1153     }
1154     qemu_mutex_lock(&p->mutex);
1155     p->running = false;
1156     qemu_mutex_unlock(&p->mutex);
1157
1158     rcu_unregister_thread();
1159     trace_multifd_recv_thread_end(p->id, p->num_packets, p->total_normal_pages);
1160
1161     return NULL;
1162 }
1163
1164 int multifd_load_setup(Error **errp)
1165 {
1166     int thread_count;
1167     uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
1168     uint8_t i;
1169
1170     if (!migrate_use_multifd()) {
1171         return 0;
1172     }
1173     if (!migrate_multi_channels_is_allowed()) {
1174         error_setg(errp, "multifd is not supported by current protocol");
1175         return -1;
1176     }
1177     thread_count = migrate_multifd_channels();
1178     multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
1179     multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
1180     qatomic_set(&multifd_recv_state->count, 0);
1181     qemu_sem_init(&multifd_recv_state->sem_sync, 0);
1182     multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()];
1183
1184     for (i = 0; i < thread_count; i++) {
1185         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1186
1187         qemu_mutex_init(&p->mutex);
1188         qemu_sem_init(&p->sem_sync, 0);
1189         p->quit = false;
1190         p->id = i;
1191         p->packet_len = sizeof(MultiFDPacket_t)
1192                       + sizeof(uint64_t) * page_count;
1193         p->packet = g_malloc0(p->packet_len);
1194         p->name = g_strdup_printf("multifdrecv_%d", i);
1195         p->iov = g_new0(struct iovec, page_count);
1196         p->normal = g_new0(ram_addr_t, page_count);
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_use_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  * - Return true and do not set @errp when correctly receiving all channels;
1232  * - Return false and do not set @errp when correctly receiving the current one;
1233  * - Return false and set @errp when failing to receive the current channel.
1234  */
1235 bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
1236 {
1237     MultiFDRecvParams *p;
1238     Error *local_err = NULL;
1239     int id;
1240
1241     id = multifd_recv_initial_packet(ioc, &local_err);
1242     if (id < 0) {
1243         multifd_recv_terminate_threads(local_err);
1244         error_propagate_prepend(errp, local_err,
1245                                 "failed to receive packet"
1246                                 " via multifd channel %d: ",
1247                                 qatomic_read(&multifd_recv_state->count));
1248         return false;
1249     }
1250     trace_multifd_recv_new_channel(id);
1251
1252     p = &multifd_recv_state->params[id];
1253     if (p->c != NULL) {
1254         error_setg(&local_err, "multifd: received id '%d' already setup'",
1255                    id);
1256         multifd_recv_terminate_threads(local_err);
1257         error_propagate(errp, local_err);
1258         return false;
1259     }
1260     p->c = ioc;
1261     object_ref(OBJECT(ioc));
1262     /* initial packet */
1263     p->num_packets = 1;
1264
1265     p->running = true;
1266     qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
1267                        QEMU_THREAD_JOINABLE);
1268     qatomic_inc(&multifd_recv_state->count);
1269     return qatomic_read(&multifd_recv_state->count) ==
1270            migrate_multifd_channels();
1271 }