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