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