1 /* 2 * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO) 3 * (a.k.a. Fault Tolerance or Continuous Replication) 4 * 5 * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD. 6 * Copyright (c) 2016 FUJITSU LIMITED 7 * Copyright (c) 2016 Intel Corporation 8 * 9 * This work is licensed under the terms of the GNU GPL, version 2 or 10 * later. See the COPYING file in the top-level directory. 11 */ 12 13 #include "qemu/osdep.h" 14 #include "sysemu/sysemu.h" 15 #include "qapi/error.h" 16 #include "qapi/qapi-commands-migration.h" 17 #include "migration.h" 18 #include "qemu-file.h" 19 #include "savevm.h" 20 #include "migration/colo.h" 21 #include "block.h" 22 #include "io/channel-buffer.h" 23 #include "trace.h" 24 #include "qemu/error-report.h" 25 #include "qemu/main-loop.h" 26 #include "qemu/rcu.h" 27 #include "migration/failover.h" 28 #include "migration/ram.h" 29 #include "block/replication.h" 30 #include "net/colo-compare.h" 31 #include "net/colo.h" 32 #include "block/block.h" 33 #include "qapi/qapi-events-migration.h" 34 #include "sysemu/cpus.h" 35 #include "sysemu/runstate.h" 36 #include "net/filter.h" 37 #include "options.h" 38 39 static bool vmstate_loading; 40 static Notifier packets_compare_notifier; 41 42 /* User need to know colo mode after COLO failover */ 43 static COLOMode last_colo_mode; 44 45 #define COLO_BUFFER_BASE_SIZE (4 * 1024 * 1024) 46 47 bool migration_in_colo_state(void) 48 { 49 MigrationState *s = migrate_get_current(); 50 51 return (s->state == MIGRATION_STATUS_COLO); 52 } 53 54 bool migration_incoming_in_colo_state(void) 55 { 56 MigrationIncomingState *mis = migration_incoming_get_current(); 57 58 return mis && (mis->state == MIGRATION_STATUS_COLO); 59 } 60 61 static bool colo_runstate_is_stopped(void) 62 { 63 return runstate_check(RUN_STATE_COLO) || !runstate_is_running(); 64 } 65 66 static void colo_checkpoint_notify(void *opaque) 67 { 68 MigrationState *s = opaque; 69 int64_t next_notify_time; 70 71 qemu_event_set(&s->colo_checkpoint_event); 72 s->colo_checkpoint_time = qemu_clock_get_ms(QEMU_CLOCK_HOST); 73 next_notify_time = s->colo_checkpoint_time + migrate_checkpoint_delay(); 74 timer_mod(s->colo_delay_timer, next_notify_time); 75 } 76 77 void colo_checkpoint_delay_set(void) 78 { 79 if (migration_in_colo_state()) { 80 colo_checkpoint_notify(migrate_get_current()); 81 } 82 } 83 84 static void secondary_vm_do_failover(void) 85 { 86 /* COLO needs enable block-replication */ 87 int old_state; 88 MigrationIncomingState *mis = migration_incoming_get_current(); 89 Error *local_err = NULL; 90 91 /* Can not do failover during the process of VM's loading VMstate, Or 92 * it will break the secondary VM. 93 */ 94 if (vmstate_loading) { 95 old_state = failover_set_state(FAILOVER_STATUS_ACTIVE, 96 FAILOVER_STATUS_RELAUNCH); 97 if (old_state != FAILOVER_STATUS_ACTIVE) { 98 error_report("Unknown error while do failover for secondary VM," 99 "old_state: %s", FailoverStatus_str(old_state)); 100 } 101 return; 102 } 103 104 migrate_set_state(&mis->state, MIGRATION_STATUS_COLO, 105 MIGRATION_STATUS_COMPLETED); 106 107 replication_stop_all(true, &local_err); 108 if (local_err) { 109 error_report_err(local_err); 110 local_err = NULL; 111 } 112 113 /* Notify all filters of all NIC to do checkpoint */ 114 colo_notify_filters_event(COLO_EVENT_FAILOVER, &local_err); 115 if (local_err) { 116 error_report_err(local_err); 117 } 118 119 if (!autostart) { 120 error_report("\"-S\" qemu option will be ignored in secondary side"); 121 /* recover runstate to normal migration finish state */ 122 autostart = true; 123 } 124 /* 125 * Make sure COLO incoming thread not block in recv or send, 126 * If mis->from_src_file and mis->to_src_file use the same fd, 127 * The second shutdown() will return -1, we ignore this value, 128 * It is harmless. 129 */ 130 if (mis->from_src_file) { 131 qemu_file_shutdown(mis->from_src_file); 132 } 133 if (mis->to_src_file) { 134 qemu_file_shutdown(mis->to_src_file); 135 } 136 137 old_state = failover_set_state(FAILOVER_STATUS_ACTIVE, 138 FAILOVER_STATUS_COMPLETED); 139 if (old_state != FAILOVER_STATUS_ACTIVE) { 140 error_report("Incorrect state (%s) while doing failover for " 141 "secondary VM", FailoverStatus_str(old_state)); 142 return; 143 } 144 /* Notify COLO incoming thread that failover work is finished */ 145 qemu_sem_post(&mis->colo_incoming_sem); 146 147 /* For Secondary VM, jump to incoming co */ 148 if (mis->colo_incoming_co) { 149 qemu_coroutine_enter(mis->colo_incoming_co); 150 } 151 } 152 153 static void primary_vm_do_failover(void) 154 { 155 MigrationState *s = migrate_get_current(); 156 int old_state; 157 Error *local_err = NULL; 158 159 migrate_set_state(&s->state, MIGRATION_STATUS_COLO, 160 MIGRATION_STATUS_COMPLETED); 161 /* 162 * kick COLO thread which might wait at 163 * qemu_sem_wait(&s->colo_checkpoint_sem). 164 */ 165 colo_checkpoint_notify(s); 166 167 /* 168 * Wake up COLO thread which may blocked in recv() or send(), 169 * The s->rp_state.from_dst_file and s->to_dst_file may use the 170 * same fd, but we still shutdown the fd for twice, it is harmless. 171 */ 172 if (s->to_dst_file) { 173 qemu_file_shutdown(s->to_dst_file); 174 } 175 if (s->rp_state.from_dst_file) { 176 qemu_file_shutdown(s->rp_state.from_dst_file); 177 } 178 179 old_state = failover_set_state(FAILOVER_STATUS_ACTIVE, 180 FAILOVER_STATUS_COMPLETED); 181 if (old_state != FAILOVER_STATUS_ACTIVE) { 182 error_report("Incorrect state (%s) while doing failover for Primary VM", 183 FailoverStatus_str(old_state)); 184 return; 185 } 186 187 replication_stop_all(true, &local_err); 188 if (local_err) { 189 error_report_err(local_err); 190 local_err = NULL; 191 } 192 193 /* Notify COLO thread that failover work is finished */ 194 qemu_sem_post(&s->colo_exit_sem); 195 } 196 197 COLOMode get_colo_mode(void) 198 { 199 if (migration_in_colo_state()) { 200 return COLO_MODE_PRIMARY; 201 } else if (migration_incoming_in_colo_state()) { 202 return COLO_MODE_SECONDARY; 203 } else { 204 return COLO_MODE_NONE; 205 } 206 } 207 208 void colo_do_failover(void) 209 { 210 /* Make sure VM stopped while failover happened. */ 211 if (!colo_runstate_is_stopped()) { 212 vm_stop_force_state(RUN_STATE_COLO); 213 } 214 215 switch (last_colo_mode = get_colo_mode()) { 216 case COLO_MODE_PRIMARY: 217 primary_vm_do_failover(); 218 break; 219 case COLO_MODE_SECONDARY: 220 secondary_vm_do_failover(); 221 break; 222 default: 223 error_report("colo_do_failover failed because the colo mode" 224 " could not be obtained"); 225 } 226 } 227 228 void qmp_xen_set_replication(bool enable, bool primary, 229 bool has_failover, bool failover, 230 Error **errp) 231 { 232 ReplicationMode mode = primary ? 233 REPLICATION_MODE_PRIMARY : 234 REPLICATION_MODE_SECONDARY; 235 236 if (has_failover && enable) { 237 error_setg(errp, "Parameter 'failover' is only for" 238 " stopping replication"); 239 return; 240 } 241 242 if (enable) { 243 replication_start_all(mode, errp); 244 } else { 245 if (!has_failover) { 246 failover = NULL; 247 } 248 replication_stop_all(failover, failover ? NULL : errp); 249 } 250 } 251 252 ReplicationStatus *qmp_query_xen_replication_status(Error **errp) 253 { 254 Error *err = NULL; 255 ReplicationStatus *s = g_new0(ReplicationStatus, 1); 256 257 replication_get_error_all(&err); 258 if (err) { 259 s->error = true; 260 s->desc = g_strdup(error_get_pretty(err)); 261 } else { 262 s->error = false; 263 } 264 265 error_free(err); 266 return s; 267 } 268 269 void qmp_xen_colo_do_checkpoint(Error **errp) 270 { 271 Error *err = NULL; 272 273 replication_do_checkpoint_all(&err); 274 if (err) { 275 error_propagate(errp, err); 276 return; 277 } 278 /* Notify all filters of all NIC to do checkpoint */ 279 colo_notify_filters_event(COLO_EVENT_CHECKPOINT, errp); 280 } 281 282 COLOStatus *qmp_query_colo_status(Error **errp) 283 { 284 COLOStatus *s = g_new0(COLOStatus, 1); 285 286 s->mode = get_colo_mode(); 287 s->last_mode = last_colo_mode; 288 289 switch (failover_get_state()) { 290 case FAILOVER_STATUS_NONE: 291 s->reason = COLO_EXIT_REASON_NONE; 292 break; 293 case FAILOVER_STATUS_COMPLETED: 294 s->reason = COLO_EXIT_REASON_REQUEST; 295 break; 296 default: 297 if (migration_in_colo_state()) { 298 s->reason = COLO_EXIT_REASON_PROCESSING; 299 } else { 300 s->reason = COLO_EXIT_REASON_ERROR; 301 } 302 } 303 304 return s; 305 } 306 307 static void colo_send_message(QEMUFile *f, COLOMessage msg, 308 Error **errp) 309 { 310 int ret; 311 312 if (msg >= COLO_MESSAGE__MAX) { 313 error_setg(errp, "%s: Invalid message", __func__); 314 return; 315 } 316 qemu_put_be32(f, msg); 317 ret = qemu_fflush(f); 318 if (ret < 0) { 319 error_setg_errno(errp, -ret, "Can't send COLO message"); 320 } 321 trace_colo_send_message(COLOMessage_str(msg)); 322 } 323 324 static void colo_send_message_value(QEMUFile *f, COLOMessage msg, 325 uint64_t value, Error **errp) 326 { 327 Error *local_err = NULL; 328 int ret; 329 330 colo_send_message(f, msg, &local_err); 331 if (local_err) { 332 error_propagate(errp, local_err); 333 return; 334 } 335 qemu_put_be64(f, value); 336 ret = qemu_fflush(f); 337 if (ret < 0) { 338 error_setg_errno(errp, -ret, "Failed to send value for message:%s", 339 COLOMessage_str(msg)); 340 } 341 } 342 343 static COLOMessage colo_receive_message(QEMUFile *f, Error **errp) 344 { 345 COLOMessage msg; 346 int ret; 347 348 msg = qemu_get_be32(f); 349 ret = qemu_file_get_error(f); 350 if (ret < 0) { 351 error_setg_errno(errp, -ret, "Can't receive COLO message"); 352 return msg; 353 } 354 if (msg >= COLO_MESSAGE__MAX) { 355 error_setg(errp, "%s: Invalid message", __func__); 356 return msg; 357 } 358 trace_colo_receive_message(COLOMessage_str(msg)); 359 return msg; 360 } 361 362 static void colo_receive_check_message(QEMUFile *f, COLOMessage expect_msg, 363 Error **errp) 364 { 365 COLOMessage msg; 366 Error *local_err = NULL; 367 368 msg = colo_receive_message(f, &local_err); 369 if (local_err) { 370 error_propagate(errp, local_err); 371 return; 372 } 373 if (msg != expect_msg) { 374 error_setg(errp, "Unexpected COLO message %d, expected %d", 375 msg, expect_msg); 376 } 377 } 378 379 static uint64_t colo_receive_message_value(QEMUFile *f, uint32_t expect_msg, 380 Error **errp) 381 { 382 Error *local_err = NULL; 383 uint64_t value; 384 int ret; 385 386 colo_receive_check_message(f, expect_msg, &local_err); 387 if (local_err) { 388 error_propagate(errp, local_err); 389 return 0; 390 } 391 392 value = qemu_get_be64(f); 393 ret = qemu_file_get_error(f); 394 if (ret < 0) { 395 error_setg_errno(errp, -ret, "Failed to get value for COLO message: %s", 396 COLOMessage_str(expect_msg)); 397 } 398 return value; 399 } 400 401 static int colo_do_checkpoint_transaction(MigrationState *s, 402 QIOChannelBuffer *bioc, 403 QEMUFile *fb) 404 { 405 Error *local_err = NULL; 406 int ret = -1; 407 408 colo_send_message(s->to_dst_file, COLO_MESSAGE_CHECKPOINT_REQUEST, 409 &local_err); 410 if (local_err) { 411 goto out; 412 } 413 414 colo_receive_check_message(s->rp_state.from_dst_file, 415 COLO_MESSAGE_CHECKPOINT_REPLY, &local_err); 416 if (local_err) { 417 goto out; 418 } 419 /* Reset channel-buffer directly */ 420 qio_channel_io_seek(QIO_CHANNEL(bioc), 0, 0, NULL); 421 bioc->usage = 0; 422 423 bql_lock(); 424 if (failover_get_state() != FAILOVER_STATUS_NONE) { 425 bql_unlock(); 426 goto out; 427 } 428 vm_stop_force_state(RUN_STATE_COLO); 429 bql_unlock(); 430 trace_colo_vm_state_change("run", "stop"); 431 /* 432 * Failover request bh could be called after vm_stop_force_state(), 433 * So we need check failover_request_is_active() again. 434 */ 435 if (failover_get_state() != FAILOVER_STATUS_NONE) { 436 goto out; 437 } 438 bql_lock(); 439 440 replication_do_checkpoint_all(&local_err); 441 if (local_err) { 442 bql_unlock(); 443 goto out; 444 } 445 446 colo_send_message(s->to_dst_file, COLO_MESSAGE_VMSTATE_SEND, &local_err); 447 if (local_err) { 448 bql_unlock(); 449 goto out; 450 } 451 /* Note: device state is saved into buffer */ 452 ret = qemu_save_device_state(fb); 453 454 bql_unlock(); 455 if (ret < 0) { 456 goto out; 457 } 458 459 if (migrate_auto_converge()) { 460 mig_throttle_counter_reset(); 461 } 462 /* 463 * Only save VM's live state, which not including device state. 464 * TODO: We may need a timeout mechanism to prevent COLO process 465 * to be blocked here. 466 */ 467 qemu_savevm_live_state(s->to_dst_file); 468 469 qemu_fflush(fb); 470 471 /* 472 * We need the size of the VMstate data in Secondary side, 473 * With which we can decide how much data should be read. 474 */ 475 colo_send_message_value(s->to_dst_file, COLO_MESSAGE_VMSTATE_SIZE, 476 bioc->usage, &local_err); 477 if (local_err) { 478 goto out; 479 } 480 481 qemu_put_buffer(s->to_dst_file, bioc->data, bioc->usage); 482 ret = qemu_fflush(s->to_dst_file); 483 if (ret < 0) { 484 goto out; 485 } 486 487 colo_receive_check_message(s->rp_state.from_dst_file, 488 COLO_MESSAGE_VMSTATE_RECEIVED, &local_err); 489 if (local_err) { 490 goto out; 491 } 492 493 qemu_event_reset(&s->colo_checkpoint_event); 494 colo_notify_compares_event(NULL, COLO_EVENT_CHECKPOINT, &local_err); 495 if (local_err) { 496 goto out; 497 } 498 499 colo_receive_check_message(s->rp_state.from_dst_file, 500 COLO_MESSAGE_VMSTATE_LOADED, &local_err); 501 if (local_err) { 502 goto out; 503 } 504 505 ret = 0; 506 507 bql_lock(); 508 vm_start(); 509 bql_unlock(); 510 trace_colo_vm_state_change("stop", "run"); 511 512 out: 513 if (local_err) { 514 error_report_err(local_err); 515 } 516 return ret; 517 } 518 519 static void colo_compare_notify_checkpoint(Notifier *notifier, void *data) 520 { 521 colo_checkpoint_notify(data); 522 } 523 524 static void colo_process_checkpoint(MigrationState *s) 525 { 526 QIOChannelBuffer *bioc; 527 QEMUFile *fb = NULL; 528 Error *local_err = NULL; 529 int ret; 530 531 if (get_colo_mode() != COLO_MODE_PRIMARY) { 532 error_report("COLO mode must be COLO_MODE_PRIMARY"); 533 return; 534 } 535 536 failover_init_state(); 537 538 s->rp_state.from_dst_file = qemu_file_get_return_path(s->to_dst_file); 539 if (!s->rp_state.from_dst_file) { 540 error_report("Open QEMUFile from_dst_file failed"); 541 goto out; 542 } 543 544 packets_compare_notifier.notify = colo_compare_notify_checkpoint; 545 colo_compare_register_notifier(&packets_compare_notifier); 546 547 /* 548 * Wait for Secondary finish loading VM states and enter COLO 549 * restore. 550 */ 551 colo_receive_check_message(s->rp_state.from_dst_file, 552 COLO_MESSAGE_CHECKPOINT_READY, &local_err); 553 if (local_err) { 554 goto out; 555 } 556 bioc = qio_channel_buffer_new(COLO_BUFFER_BASE_SIZE); 557 fb = qemu_file_new_output(QIO_CHANNEL(bioc)); 558 object_unref(OBJECT(bioc)); 559 560 bql_lock(); 561 replication_start_all(REPLICATION_MODE_PRIMARY, &local_err); 562 if (local_err) { 563 bql_unlock(); 564 goto out; 565 } 566 567 vm_start(); 568 bql_unlock(); 569 trace_colo_vm_state_change("stop", "run"); 570 571 timer_mod(s->colo_delay_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) + 572 migrate_checkpoint_delay()); 573 574 while (s->state == MIGRATION_STATUS_COLO) { 575 if (failover_get_state() != FAILOVER_STATUS_NONE) { 576 error_report("failover request"); 577 goto out; 578 } 579 580 qemu_event_wait(&s->colo_checkpoint_event); 581 582 if (s->state != MIGRATION_STATUS_COLO) { 583 goto out; 584 } 585 ret = colo_do_checkpoint_transaction(s, bioc, fb); 586 if (ret < 0) { 587 goto out; 588 } 589 } 590 591 out: 592 /* Throw the unreported error message after exited from loop */ 593 if (local_err) { 594 error_report_err(local_err); 595 } 596 597 if (fb) { 598 qemu_fclose(fb); 599 } 600 601 /* 602 * There are only two reasons we can get here, some error happened 603 * or the user triggered failover. 604 */ 605 switch (failover_get_state()) { 606 case FAILOVER_STATUS_COMPLETED: 607 qapi_event_send_colo_exit(COLO_MODE_PRIMARY, 608 COLO_EXIT_REASON_REQUEST); 609 break; 610 default: 611 qapi_event_send_colo_exit(COLO_MODE_PRIMARY, 612 COLO_EXIT_REASON_ERROR); 613 } 614 615 /* Hope this not to be too long to wait here */ 616 qemu_sem_wait(&s->colo_exit_sem); 617 qemu_sem_destroy(&s->colo_exit_sem); 618 619 /* 620 * It is safe to unregister notifier after failover finished. 621 * Besides, colo_delay_timer and colo_checkpoint_sem can't be 622 * released before unregister notifier, or there will be use-after-free 623 * error. 624 */ 625 colo_compare_unregister_notifier(&packets_compare_notifier); 626 timer_free(s->colo_delay_timer); 627 qemu_event_destroy(&s->colo_checkpoint_event); 628 629 /* 630 * Must be called after failover BH is completed, 631 * Or the failover BH may shutdown the wrong fd that 632 * re-used by other threads after we release here. 633 */ 634 if (s->rp_state.from_dst_file) { 635 qemu_fclose(s->rp_state.from_dst_file); 636 s->rp_state.from_dst_file = NULL; 637 } 638 } 639 640 void migrate_start_colo_process(MigrationState *s) 641 { 642 bql_unlock(); 643 qemu_event_init(&s->colo_checkpoint_event, false); 644 s->colo_delay_timer = timer_new_ms(QEMU_CLOCK_HOST, 645 colo_checkpoint_notify, s); 646 647 qemu_sem_init(&s->colo_exit_sem, 0); 648 colo_process_checkpoint(s); 649 bql_lock(); 650 } 651 652 static void colo_incoming_process_checkpoint(MigrationIncomingState *mis, 653 QEMUFile *fb, QIOChannelBuffer *bioc, Error **errp) 654 { 655 uint64_t total_size; 656 uint64_t value; 657 Error *local_err = NULL; 658 int ret; 659 660 bql_lock(); 661 vm_stop_force_state(RUN_STATE_COLO); 662 bql_unlock(); 663 trace_colo_vm_state_change("run", "stop"); 664 665 /* FIXME: This is unnecessary for periodic checkpoint mode */ 666 colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_REPLY, 667 &local_err); 668 if (local_err) { 669 error_propagate(errp, local_err); 670 return; 671 } 672 673 colo_receive_check_message(mis->from_src_file, 674 COLO_MESSAGE_VMSTATE_SEND, &local_err); 675 if (local_err) { 676 error_propagate(errp, local_err); 677 return; 678 } 679 680 bql_lock(); 681 cpu_synchronize_all_states(); 682 ret = qemu_loadvm_state_main(mis->from_src_file, mis); 683 bql_unlock(); 684 685 if (ret < 0) { 686 error_setg(errp, "Load VM's live state (ram) error"); 687 return; 688 } 689 690 value = colo_receive_message_value(mis->from_src_file, 691 COLO_MESSAGE_VMSTATE_SIZE, &local_err); 692 if (local_err) { 693 error_propagate(errp, local_err); 694 return; 695 } 696 697 /* 698 * Read VM device state data into channel buffer, 699 * It's better to re-use the memory allocated. 700 * Here we need to handle the channel buffer directly. 701 */ 702 if (value > bioc->capacity) { 703 bioc->capacity = value; 704 bioc->data = g_realloc(bioc->data, bioc->capacity); 705 } 706 total_size = qemu_get_buffer(mis->from_src_file, bioc->data, value); 707 if (total_size != value) { 708 error_setg(errp, "Got %" PRIu64 " VMState data, less than expected" 709 " %" PRIu64, total_size, value); 710 return; 711 } 712 bioc->usage = total_size; 713 qio_channel_io_seek(QIO_CHANNEL(bioc), 0, 0, NULL); 714 715 colo_send_message(mis->to_src_file, COLO_MESSAGE_VMSTATE_RECEIVED, 716 &local_err); 717 if (local_err) { 718 error_propagate(errp, local_err); 719 return; 720 } 721 722 bql_lock(); 723 vmstate_loading = true; 724 colo_flush_ram_cache(); 725 ret = qemu_load_device_state(fb); 726 if (ret < 0) { 727 error_setg(errp, "COLO: load device state failed"); 728 vmstate_loading = false; 729 bql_unlock(); 730 return; 731 } 732 733 replication_get_error_all(&local_err); 734 if (local_err) { 735 error_propagate(errp, local_err); 736 vmstate_loading = false; 737 bql_unlock(); 738 return; 739 } 740 741 /* discard colo disk buffer */ 742 replication_do_checkpoint_all(&local_err); 743 if (local_err) { 744 error_propagate(errp, local_err); 745 vmstate_loading = false; 746 bql_unlock(); 747 return; 748 } 749 /* Notify all filters of all NIC to do checkpoint */ 750 colo_notify_filters_event(COLO_EVENT_CHECKPOINT, &local_err); 751 752 if (local_err) { 753 error_propagate(errp, local_err); 754 vmstate_loading = false; 755 bql_unlock(); 756 return; 757 } 758 759 vmstate_loading = false; 760 vm_start(); 761 bql_unlock(); 762 trace_colo_vm_state_change("stop", "run"); 763 764 if (failover_get_state() == FAILOVER_STATUS_RELAUNCH) { 765 return; 766 } 767 768 colo_send_message(mis->to_src_file, COLO_MESSAGE_VMSTATE_LOADED, 769 &local_err); 770 error_propagate(errp, local_err); 771 } 772 773 static void colo_wait_handle_message(MigrationIncomingState *mis, 774 QEMUFile *fb, QIOChannelBuffer *bioc, Error **errp) 775 { 776 COLOMessage msg; 777 Error *local_err = NULL; 778 779 msg = colo_receive_message(mis->from_src_file, &local_err); 780 if (local_err) { 781 error_propagate(errp, local_err); 782 return; 783 } 784 785 switch (msg) { 786 case COLO_MESSAGE_CHECKPOINT_REQUEST: 787 colo_incoming_process_checkpoint(mis, fb, bioc, errp); 788 break; 789 default: 790 error_setg(errp, "Got unknown COLO message: %d", msg); 791 break; 792 } 793 } 794 795 void colo_shutdown(void) 796 { 797 MigrationIncomingState *mis = NULL; 798 MigrationState *s = NULL; 799 800 switch (get_colo_mode()) { 801 case COLO_MODE_PRIMARY: 802 s = migrate_get_current(); 803 qemu_event_set(&s->colo_checkpoint_event); 804 qemu_sem_post(&s->colo_exit_sem); 805 break; 806 case COLO_MODE_SECONDARY: 807 mis = migration_incoming_get_current(); 808 qemu_sem_post(&mis->colo_incoming_sem); 809 break; 810 default: 811 break; 812 } 813 } 814 815 static void *colo_process_incoming_thread(void *opaque) 816 { 817 MigrationIncomingState *mis = opaque; 818 QEMUFile *fb = NULL; 819 QIOChannelBuffer *bioc = NULL; /* Cache incoming device state */ 820 Error *local_err = NULL; 821 822 rcu_register_thread(); 823 qemu_sem_init(&mis->colo_incoming_sem, 0); 824 825 migrate_set_state(&mis->state, MIGRATION_STATUS_ACTIVE, 826 MIGRATION_STATUS_COLO); 827 828 if (get_colo_mode() != COLO_MODE_SECONDARY) { 829 error_report("COLO mode must be COLO_MODE_SECONDARY"); 830 return NULL; 831 } 832 833 failover_init_state(); 834 835 mis->to_src_file = qemu_file_get_return_path(mis->from_src_file); 836 if (!mis->to_src_file) { 837 error_report("COLO incoming thread: Open QEMUFile to_src_file failed"); 838 goto out; 839 } 840 /* 841 * Note: the communication between Primary side and Secondary side 842 * should be sequential, we set the fd to unblocked in migration incoming 843 * coroutine, and here we are in the COLO incoming thread, so it is ok to 844 * set the fd back to blocked. 845 */ 846 qemu_file_set_blocking(mis->from_src_file, true); 847 848 colo_incoming_start_dirty_log(); 849 850 bioc = qio_channel_buffer_new(COLO_BUFFER_BASE_SIZE); 851 fb = qemu_file_new_input(QIO_CHANNEL(bioc)); 852 object_unref(OBJECT(bioc)); 853 854 bql_lock(); 855 replication_start_all(REPLICATION_MODE_SECONDARY, &local_err); 856 if (local_err) { 857 bql_unlock(); 858 goto out; 859 } 860 vm_start(); 861 bql_unlock(); 862 trace_colo_vm_state_change("stop", "run"); 863 864 colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_READY, 865 &local_err); 866 if (local_err) { 867 goto out; 868 } 869 870 while (mis->state == MIGRATION_STATUS_COLO) { 871 colo_wait_handle_message(mis, fb, bioc, &local_err); 872 if (local_err) { 873 error_report_err(local_err); 874 break; 875 } 876 877 if (failover_get_state() == FAILOVER_STATUS_RELAUNCH) { 878 failover_set_state(FAILOVER_STATUS_RELAUNCH, 879 FAILOVER_STATUS_NONE); 880 failover_request_active(NULL); 881 break; 882 } 883 884 if (failover_get_state() != FAILOVER_STATUS_NONE) { 885 error_report("failover request"); 886 break; 887 } 888 } 889 890 out: 891 /* 892 * There are only two reasons we can get here, some error happened 893 * or the user triggered failover. 894 */ 895 switch (failover_get_state()) { 896 case FAILOVER_STATUS_COMPLETED: 897 qapi_event_send_colo_exit(COLO_MODE_SECONDARY, 898 COLO_EXIT_REASON_REQUEST); 899 break; 900 default: 901 qapi_event_send_colo_exit(COLO_MODE_SECONDARY, 902 COLO_EXIT_REASON_ERROR); 903 } 904 905 if (fb) { 906 qemu_fclose(fb); 907 } 908 909 /* Hope this not to be too long to loop here */ 910 qemu_sem_wait(&mis->colo_incoming_sem); 911 qemu_sem_destroy(&mis->colo_incoming_sem); 912 913 rcu_unregister_thread(); 914 return NULL; 915 } 916 917 int coroutine_fn colo_incoming_co(void) 918 { 919 MigrationIncomingState *mis = migration_incoming_get_current(); 920 Error *local_err = NULL; 921 QemuThread th; 922 923 assert(bql_locked()); 924 925 if (!migration_incoming_colo_enabled()) { 926 return 0; 927 } 928 929 /* Make sure all file formats throw away their mutable metadata */ 930 bdrv_activate_all(&local_err); 931 if (local_err) { 932 error_report_err(local_err); 933 return -EINVAL; 934 } 935 936 qemu_thread_create(&th, "COLO incoming", colo_process_incoming_thread, 937 mis, QEMU_THREAD_JOINABLE); 938 939 mis->colo_incoming_co = qemu_coroutine_self(); 940 qemu_coroutine_yield(); 941 mis->colo_incoming_co = NULL; 942 943 bql_unlock(); 944 /* Wait checkpoint incoming thread exit before free resource */ 945 qemu_thread_join(&th); 946 bql_lock(); 947 948 /* We hold the global BQL, so it is safe here */ 949 colo_release_ram_cache(); 950 951 return 0; 952 } 953