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