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