migration/multifd.c

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