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