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