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