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 /* 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_channel_connect(MultiFDSendParams *p,
 743                                         QIOChannel *ioc,
 744                                         Error **errp)
 745 {
 746     MigrationState *s = migrate_get_current();
 747     const char *hostname = p->tls_hostname;
 748     QIOChannelTLS *tioc;
 749
 750     tioc = migration_tls_client_create(s, ioc, hostname, errp);
 751     if (!tioc) {
 752         return;
 753     }
 754
 755     trace_multifd_tls_outgoing_handshake_start(ioc, tioc, hostname);
 756     qio_channel_set_name(QIO_CHANNEL(tioc), "multifd-tls-outgoing");
 757     qio_channel_tls_handshake(tioc,
 758                               multifd_tls_outgoing_handshake,
 759                               p,
 760                               NULL,
 761                               NULL);
 762
 763 }
 764
 765 static bool multifd_channel_connect(MultiFDSendParams *p,
 766                                     QIOChannel *ioc,
 767                                     Error *error)
 768 {
 769     MigrationState *s = migrate_get_current();
 770
 771     trace_multifd_set_outgoing_channel(
 772         ioc, object_get_typename(OBJECT(ioc)), p->tls_hostname, error);
 773
 774     if (!error) {
 775         if (s->parameters.tls_creds &&
 776             *s->parameters.tls_creds &&
 777             !object_dynamic_cast(OBJECT(ioc),
 778                                  TYPE_QIO_CHANNEL_TLS)) {
 779             multifd_tls_channel_connect(p, ioc, &error);
 780             if (!error) {
 781                 /*
 782                  * tls_channel_connect will call back to this
 783                  * function after the TLS handshake,
 784                  * so we mustn't call multifd_send_thread until then
 785                  */
 786                 return false;
 787             } else {
 788                 return true;
 789             }
 790         } else {
 791             /* update for tls qio channel */
 792             p->c = ioc;
 793             qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
 794                                    QEMU_THREAD_JOINABLE);
 795        }
 796        return false;
 797     }
 798
 799     return true;
 800 }
 801
 802 static void multifd_new_send_channel_cleanup(MultiFDSendParams *p,
 803                                              QIOChannel *ioc, Error *err)
 804 {
 805      migrate_set_error(migrate_get_current(), err);
 806      /* Error happen, we need to tell who pay attention to me */
 807      qemu_sem_post(&multifd_send_state->channels_ready);
 808      qemu_sem_post(&p->sem_sync);
 809      /*
 810       * Although multifd_send_thread is not created, but main migration
 811       * thread neet to judge whether it is running, so we need to mark
 812       * its status.
 813       */
 814      p->quit = true;
 815      object_unref(OBJECT(ioc));
 816      error_free(err);
 817 }
 818
 819 static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
 820 {
 821     MultiFDSendParams *p = opaque;
 822     QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
 823     Error *local_err = NULL;
 824
 825     trace_multifd_new_send_channel_async(p->id);
 826     if (qio_task_propagate_error(task, &local_err)) {
 827         goto cleanup;
 828     } else {
 829         p->c = QIO_CHANNEL(sioc);
 830         qio_channel_set_delay(p->c, false);
 831         p->running = true;
 832         if (multifd_channel_connect(p, sioc, local_err)) {
 833             goto cleanup;
 834         }
 835         return;
 836     }
 837
 838 cleanup:
 839     multifd_new_send_channel_cleanup(p, sioc, local_err);
 840 }
 841
 842 int multifd_save_setup(Error **errp)
 843 {
 844     int thread_count;
 845     uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
 846     uint8_t i;
 847     MigrationState *s;
 848
 849     if (!migrate_use_multifd()) {
 850         return 0;
 851     }
 852     s = migrate_get_current();
 853     thread_count = migrate_multifd_channels();
 854     multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
 855     multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
 856     multifd_send_state->pages = multifd_pages_init(page_count);
 857     qemu_sem_init(&multifd_send_state->channels_ready, 0);
 858     qatomic_set(&multifd_send_state->exiting, 0);
 859     multifd_send_state->ops = multifd_ops[migrate_multifd_compression()];
 860
 861     for (i = 0; i < thread_count; i++) {
 862         MultiFDSendParams *p = &multifd_send_state->params[i];
 863
 864         qemu_mutex_init(&p->mutex);
 865         qemu_sem_init(&p->sem, 0);
 866         qemu_sem_init(&p->sem_sync, 0);
 867         p->quit = false;
 868         p->pending_job = 0;
 869         p->id = i;
 870         p->pages = multifd_pages_init(page_count);
 871         p->packet_len = sizeof(MultiFDPacket_t)
 872                       + sizeof(uint64_t) * page_count;
 873         p->packet = g_malloc0(p->packet_len);
 874         p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
 875         p->packet->version = cpu_to_be32(MULTIFD_VERSION);
 876         p->name = g_strdup_printf("multifdsend_%d", i);
 877         p->tls_hostname = g_strdup(s->hostname);
 878         socket_send_channel_create(multifd_new_send_channel_async, p);
 879     }
 880
 881     for (i = 0; i < thread_count; i++) {
 882         MultiFDSendParams *p = &multifd_send_state->params[i];
 883         Error *local_err = NULL;
 884         int ret;
 885
 886         ret = multifd_send_state->ops->send_setup(p, &local_err);
 887         if (ret) {
 888             error_propagate(errp, local_err);
 889             return ret;
 890         }
 891     }
 892     return 0;
 893 }
 894
 895 struct {
 896     MultiFDRecvParams *params;
 897     /* number of created threads */
 898     int count;
 899     /* syncs main thread and channels */
 900     QemuSemaphore sem_sync;
 901     /* global number of generated multifd packets */
 902     uint64_t packet_num;
 903     /* multifd ops */
 904     MultiFDMethods *ops;
 905 } *multifd_recv_state;
 906
 907 static void multifd_recv_terminate_threads(Error *err)
 908 {
 909     int i;
 910
 911     trace_multifd_recv_terminate_threads(err != NULL);
 912
 913     if (err) {
 914         MigrationState *s = migrate_get_current();
 915         migrate_set_error(s, err);
 916         if (s->state == MIGRATION_STATUS_SETUP ||
 917             s->state == MIGRATION_STATUS_ACTIVE) {
 918             migrate_set_state(&s->state, s->state,
 919                               MIGRATION_STATUS_FAILED);
 920         }
 921     }
 922
 923     for (i = 0; i < migrate_multifd_channels(); i++) {
 924         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 925
 926         qemu_mutex_lock(&p->mutex);
 927         p->quit = true;
 928         /*
 929          * We could arrive here for two reasons:
 930          *  - normal quit, i.e. everything went fine, just finished
 931          *  - error quit: We close the channels so the channel threads
 932          *    finish the qio_channel_read_all_eof()
 933          */
 934         if (p->c) {
 935             qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
 936         }
 937         qemu_mutex_unlock(&p->mutex);
 938     }
 939 }
 940
 941 int multifd_load_cleanup(Error **errp)
 942 {
 943     int i;
 944
 945     if (!migrate_use_multifd()) {
 946         return 0;
 947     }
 948     multifd_recv_terminate_threads(NULL);
 949     for (i = 0; i < migrate_multifd_channels(); i++) {
 950         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 951
 952         if (p->running) {
 953             p->quit = true;
 954             /*
 955              * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
 956              * however try to wakeup it without harm in cleanup phase.
 957              */
 958             qemu_sem_post(&p->sem_sync);
 959             qemu_thread_join(&p->thread);
 960         }
 961     }
 962     for (i = 0; i < migrate_multifd_channels(); i++) {
 963         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 964
 965         object_unref(OBJECT(p->c));
 966         p->c = NULL;
 967         qemu_mutex_destroy(&p->mutex);
 968         qemu_sem_destroy(&p->sem_sync);
 969         g_free(p->name);
 970         p->name = NULL;
 971         multifd_pages_clear(p->pages);
 972         p->pages = NULL;
 973         p->packet_len = 0;
 974         g_free(p->packet);
 975         p->packet = NULL;
 976         multifd_recv_state->ops->recv_cleanup(p);
 977     }
 978     qemu_sem_destroy(&multifd_recv_state->sem_sync);
 979     g_free(multifd_recv_state->params);
 980     multifd_recv_state->params = NULL;
 981     g_free(multifd_recv_state);
 982     multifd_recv_state = NULL;
 983
 984     return 0;
 985 }
 986
 987 void multifd_recv_sync_main(void)
 988 {
 989     int i;
 990
 991     if (!migrate_use_multifd()) {
 992         return;
 993     }
 994     for (i = 0; i < migrate_multifd_channels(); i++) {
 995         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 996
 997         trace_multifd_recv_sync_main_wait(p->id);
 998         qemu_sem_wait(&multifd_recv_state->sem_sync);
 999     }
1000     for (i = 0; i < migrate_multifd_channels(); i++) {
1001         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1002
1003         WITH_QEMU_LOCK_GUARD(&p->mutex) {
1004             if (multifd_recv_state->packet_num < p->packet_num) {
1005                 multifd_recv_state->packet_num = p->packet_num;
1006             }
1007         }
1008         trace_multifd_recv_sync_main_signal(p->id);
1009         qemu_sem_post(&p->sem_sync);
1010     }
1011     trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
1012 }
1013
1014 static void *multifd_recv_thread(void *opaque)
1015 {
1016     MultiFDRecvParams *p = opaque;
1017     Error *local_err = NULL;
1018     int ret;
1019
1020     trace_multifd_recv_thread_start(p->id);
1021     rcu_register_thread();
1022
1023     while (true) {
1024         uint32_t used;
1025         uint32_t flags;
1026
1027         if (p->quit) {
1028             break;
1029         }
1030
1031         ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
1032                                        p->packet_len, &local_err);
1033         if (ret == 0) {   /* EOF */
1034             break;
1035         }
1036         if (ret == -1) {   /* Error */
1037             break;
1038         }
1039
1040         qemu_mutex_lock(&p->mutex);
1041         ret = multifd_recv_unfill_packet(p, &local_err);
1042         if (ret) {
1043             qemu_mutex_unlock(&p->mutex);
1044             break;
1045         }
1046
1047         used = p->pages->used;
1048         flags = p->flags;
1049         /* recv methods don't know how to handle the SYNC flag */
1050         p->flags &= ~MULTIFD_FLAG_SYNC;
1051         trace_multifd_recv(p->id, p->packet_num, used, flags,
1052                            p->next_packet_size);
1053         p->num_packets++;
1054         p->num_pages += used;
1055         qemu_mutex_unlock(&p->mutex);
1056
1057         if (used) {
1058             ret = multifd_recv_state->ops->recv_pages(p, used, &local_err);
1059             if (ret != 0) {
1060                 break;
1061             }
1062         }
1063
1064         if (flags & MULTIFD_FLAG_SYNC) {
1065             qemu_sem_post(&multifd_recv_state->sem_sync);
1066             qemu_sem_wait(&p->sem_sync);
1067         }
1068     }
1069
1070     if (local_err) {
1071         multifd_recv_terminate_threads(local_err);
1072         error_free(local_err);
1073     }
1074     qemu_mutex_lock(&p->mutex);
1075     p->running = false;
1076     qemu_mutex_unlock(&p->mutex);
1077
1078     rcu_unregister_thread();
1079     trace_multifd_recv_thread_end(p->id, p->num_packets, p->num_pages);
1080
1081     return NULL;
1082 }
1083
1084 int multifd_load_setup(Error **errp)
1085 {
1086     int thread_count;
1087     uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
1088     uint8_t i;
1089
1090     if (!migrate_use_multifd()) {
1091         return 0;
1092     }
1093     thread_count = migrate_multifd_channels();
1094     multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
1095     multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
1096     qatomic_set(&multifd_recv_state->count, 0);
1097     qemu_sem_init(&multifd_recv_state->sem_sync, 0);
1098     multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()];
1099
1100     for (i = 0; i < thread_count; i++) {
1101         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1102
1103         qemu_mutex_init(&p->mutex);
1104         qemu_sem_init(&p->sem_sync, 0);
1105         p->quit = false;
1106         p->id = i;
1107         p->pages = multifd_pages_init(page_count);
1108         p->packet_len = sizeof(MultiFDPacket_t)
1109                       + sizeof(uint64_t) * page_count;
1110         p->packet = g_malloc0(p->packet_len);
1111         p->name = g_strdup_printf("multifdrecv_%d", i);
1112     }
1113
1114     for (i = 0; i < thread_count; i++) {
1115         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1116         Error *local_err = NULL;
1117         int ret;
1118
1119         ret = multifd_recv_state->ops->recv_setup(p, &local_err);
1120         if (ret) {
1121             error_propagate(errp, local_err);
1122             return ret;
1123         }
1124     }
1125     return 0;
1126 }
1127
1128 bool multifd_recv_all_channels_created(void)
1129 {
1130     int thread_count = migrate_multifd_channels();
1131
1132     if (!migrate_use_multifd()) {
1133         return true;
1134     }
1135
1136     return thread_count == qatomic_read(&multifd_recv_state->count);
1137 }
1138
1139 /*
1140  * Try to receive all multifd channels to get ready for the migration.
1141  * - Return true and do not set @errp when correctly receiving all channels;
1142  * - Return false and do not set @errp when correctly receiving the current one;
1143  * - Return false and set @errp when failing to receive the current channel.
1144  */
1145 bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
1146 {
1147     MultiFDRecvParams *p;
1148     Error *local_err = NULL;
1149     int id;
1150
1151     id = multifd_recv_initial_packet(ioc, &local_err);
1152     if (id < 0) {
1153         multifd_recv_terminate_threads(local_err);
1154         error_propagate_prepend(errp, local_err,
1155                                 "failed to receive packet"
1156                                 " via multifd channel %d: ",
1157                                 qatomic_read(&multifd_recv_state->count));
1158         return false;
1159     }
1160     trace_multifd_recv_new_channel(id);
1161
1162     p = &multifd_recv_state->params[id];
1163     if (p->c != NULL) {
1164         error_setg(&local_err, "multifd: received id '%d' already setup'",
1165                    id);
1166         multifd_recv_terminate_threads(local_err);
1167         error_propagate(errp, local_err);
1168         return false;
1169     }
1170     p->c = ioc;
1171     object_ref(OBJECT(ioc));
1172     /* initial packet */
1173     p->num_packets = 1;
1174
1175     p->running = true;
1176     qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
1177                        QEMU_THREAD_JOINABLE);
1178     qatomic_inc(&multifd_recv_state->count);
1179     return qatomic_read(&multifd_recv_state->count) ==
1180            migrate_multifd_channels();
1181 }