migration/colo.c

   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 "qemu-file-channel.h"
  18 #include "migration.h"
  19 #include "qemu-file.h"
  20 #include "savevm.h"
  21 #include "migration/colo.h"
  22 #include "block.h"
  23 #include "io/channel-buffer.h"
  24 #include "trace.h"
  25 #include "qemu/error-report.h"
  26 #include "migration/failover.h"
  27 #include "replication.h"
  28 #include "net/colo-compare.h"
  29 #include "net/colo.h"
  30 #include "block/block.h"
  31 #include "qapi/qapi-events-migration.h"
  32
  33 static bool vmstate_loading;
  34 static Notifier packets_compare_notifier;
  35
  36 #define COLO_BUFFER_BASE_SIZE (4 * 1024 * 1024)
  37
  38 bool migration_in_colo_state(void)
  39 {
  40     MigrationState *s = migrate_get_current();
  41
  42     return (s->state == MIGRATION_STATUS_COLO);
  43 }
  44
  45 bool migration_incoming_in_colo_state(void)
  46 {
  47     MigrationIncomingState *mis = migration_incoming_get_current();
  48
  49     return mis && (mis->state == MIGRATION_STATUS_COLO);
  50 }
  51
  52 static bool colo_runstate_is_stopped(void)
  53 {
  54     return runstate_check(RUN_STATE_COLO) || !runstate_is_running();
  55 }
  56
  57 static void secondary_vm_do_failover(void)
  58 {
  59     int old_state;
  60     MigrationIncomingState *mis = migration_incoming_get_current();
  61     Error *local_err = NULL;
  62
  63     /* Can not do failover during the process of VM's loading VMstate, Or
  64      * it will break the secondary VM.
  65      */
  66     if (vmstate_loading) {
  67         old_state = failover_set_state(FAILOVER_STATUS_ACTIVE,
  68                         FAILOVER_STATUS_RELAUNCH);
  69         if (old_state != FAILOVER_STATUS_ACTIVE) {
  70             error_report("Unknown error while do failover for secondary VM,"
  71                          "old_state: %s", FailoverStatus_str(old_state));
  72         }
  73         return;
  74     }
  75
  76     migrate_set_state(&mis->state, MIGRATION_STATUS_COLO,
  77                       MIGRATION_STATUS_COMPLETED);
  78
  79     replication_stop_all(true, &local_err);
  80     if (local_err) {
  81         error_report_err(local_err);
  82     }
  83
  84     if (!autostart) {
  85         error_report("\"-S\" qemu option will be ignored in secondary side");
  86         /* recover runstate to normal migration finish state */
  87         autostart = true;
  88     }
  89     /*
  90      * Make sure COLO incoming thread not block in recv or send,
  91      * If mis->from_src_file and mis->to_src_file use the same fd,
  92      * The second shutdown() will return -1, we ignore this value,
  93      * It is harmless.
  94      */
  95     if (mis->from_src_file) {
  96         qemu_file_shutdown(mis->from_src_file);
  97     }
  98     if (mis->to_src_file) {
  99         qemu_file_shutdown(mis->to_src_file);
 100     }
 101
 102     old_state = failover_set_state(FAILOVER_STATUS_ACTIVE,
 103                                    FAILOVER_STATUS_COMPLETED);
 104     if (old_state != FAILOVER_STATUS_ACTIVE) {
 105         error_report("Incorrect state (%s) while doing failover for "
 106                      "secondary VM", FailoverStatus_str(old_state));
 107         return;
 108     }
 109     /* Notify COLO incoming thread that failover work is finished */
 110     qemu_sem_post(&mis->colo_incoming_sem);
 111     /* For Secondary VM, jump to incoming co */
 112     if (mis->migration_incoming_co) {
 113         qemu_coroutine_enter(mis->migration_incoming_co);
 114     }
 115 }
 116
 117 static void primary_vm_do_failover(void)
 118 {
 119     MigrationState *s = migrate_get_current();
 120     int old_state;
 121     Error *local_err = NULL;
 122
 123     migrate_set_state(&s->state, MIGRATION_STATUS_COLO,
 124                       MIGRATION_STATUS_COMPLETED);
 125
 126     /*
 127      * Wake up COLO thread which may blocked in recv() or send(),
 128      * The s->rp_state.from_dst_file and s->to_dst_file may use the
 129      * same fd, but we still shutdown the fd for twice, it is harmless.
 130      */
 131     if (s->to_dst_file) {
 132         qemu_file_shutdown(s->to_dst_file);
 133     }
 134     if (s->rp_state.from_dst_file) {
 135         qemu_file_shutdown(s->rp_state.from_dst_file);
 136     }
 137
 138     old_state = failover_set_state(FAILOVER_STATUS_ACTIVE,
 139                                    FAILOVER_STATUS_COMPLETED);
 140     if (old_state != FAILOVER_STATUS_ACTIVE) {
 141         error_report("Incorrect state (%s) while doing failover for Primary VM",
 142                      FailoverStatus_str(old_state));
 143         return;
 144     }
 145
 146     replication_stop_all(true, &local_err);
 147     if (local_err) {
 148         error_report_err(local_err);
 149         local_err = NULL;
 150     }
 151
 152     /* Notify COLO thread that failover work is finished */
 153     qemu_sem_post(&s->colo_exit_sem);
 154 }
 155
 156 COLOMode get_colo_mode(void)
 157 {
 158     if (migration_in_colo_state()) {
 159         return COLO_MODE_PRIMARY;
 160     } else if (migration_incoming_in_colo_state()) {
 161         return COLO_MODE_SECONDARY;
 162     } else {
 163         return COLO_MODE_NONE;
 164     }
 165 }
 166
 167 void colo_do_failover(MigrationState *s)
 168 {
 169     /* Make sure VM stopped while failover happened. */
 170     if (!colo_runstate_is_stopped()) {
 171         vm_stop_force_state(RUN_STATE_COLO);
 172     }
 173
 174     if (get_colo_mode() == COLO_MODE_PRIMARY) {
 175         primary_vm_do_failover();
 176     } else {
 177         secondary_vm_do_failover();
 178     }
 179 }
 180
 181 void qmp_xen_set_replication(bool enable, bool primary,
 182                              bool has_failover, bool failover,
 183                              Error **errp)
 184 {
 185 #ifdef CONFIG_REPLICATION
 186     ReplicationMode mode = primary ?
 187                            REPLICATION_MODE_PRIMARY :
 188                            REPLICATION_MODE_SECONDARY;
 189
 190     if (has_failover && enable) {
 191         error_setg(errp, "Parameter 'failover' is only for"
 192                    " stopping replication");
 193         return;
 194     }
 195
 196     if (enable) {
 197         replication_start_all(mode, errp);
 198     } else {
 199         if (!has_failover) {
 200             failover = NULL;
 201         }
 202         replication_stop_all(failover, failover ? NULL : errp);
 203     }
 204 #else
 205     abort();
 206 #endif
 207 }
 208
 209 ReplicationStatus *qmp_query_xen_replication_status(Error **errp)
 210 {
 211 #ifdef CONFIG_REPLICATION
 212     Error *err = NULL;
 213     ReplicationStatus *s = g_new0(ReplicationStatus, 1);
 214
 215     replication_get_error_all(&err);
 216     if (err) {
 217         s->error = true;
 218         s->has_desc = true;
 219         s->desc = g_strdup(error_get_pretty(err));
 220     } else {
 221         s->error = false;
 222     }
 223
 224     error_free(err);
 225     return s;
 226 #else
 227     abort();
 228 #endif
 229 }
 230
 231 void qmp_xen_colo_do_checkpoint(Error **errp)
 232 {
 233 #ifdef CONFIG_REPLICATION
 234     replication_do_checkpoint_all(errp);
 235 #else
 236     abort();
 237 #endif
 238 }
 239
 240 static void colo_send_message(QEMUFile *f, COLOMessage msg,
 241                               Error **errp)
 242 {
 243     int ret;
 244
 245     if (msg >= COLO_MESSAGE__MAX) {
 246         error_setg(errp, "%s: Invalid message", __func__);
 247         return;
 248     }
 249     qemu_put_be32(f, msg);
 250     qemu_fflush(f);
 251
 252     ret = qemu_file_get_error(f);
 253     if (ret < 0) {
 254         error_setg_errno(errp, -ret, "Can't send COLO message");
 255     }
 256     trace_colo_send_message(COLOMessage_str(msg));
 257 }
 258
 259 static void colo_send_message_value(QEMUFile *f, COLOMessage msg,
 260                                     uint64_t value, Error **errp)
 261 {
 262     Error *local_err = NULL;
 263     int ret;
 264
 265     colo_send_message(f, msg, &local_err);
 266     if (local_err) {
 267         error_propagate(errp, local_err);
 268         return;
 269     }
 270     qemu_put_be64(f, value);
 271     qemu_fflush(f);
 272
 273     ret = qemu_file_get_error(f);
 274     if (ret < 0) {
 275         error_setg_errno(errp, -ret, "Failed to send value for message:%s",
 276                          COLOMessage_str(msg));
 277     }
 278 }
 279
 280 static COLOMessage colo_receive_message(QEMUFile *f, Error **errp)
 281 {
 282     COLOMessage msg;
 283     int ret;
 284
 285     msg = qemu_get_be32(f);
 286     ret = qemu_file_get_error(f);
 287     if (ret < 0) {
 288         error_setg_errno(errp, -ret, "Can't receive COLO message");
 289         return msg;
 290     }
 291     if (msg >= COLO_MESSAGE__MAX) {
 292         error_setg(errp, "%s: Invalid message", __func__);
 293         return msg;
 294     }
 295     trace_colo_receive_message(COLOMessage_str(msg));
 296     return msg;
 297 }
 298
 299 static void colo_receive_check_message(QEMUFile *f, COLOMessage expect_msg,
 300                                        Error **errp)
 301 {
 302     COLOMessage msg;
 303     Error *local_err = NULL;
 304
 305     msg = colo_receive_message(f, &local_err);
 306     if (local_err) {
 307         error_propagate(errp, local_err);
 308         return;
 309     }
 310     if (msg != expect_msg) {
 311         error_setg(errp, "Unexpected COLO message %d, expected %d",
 312                           msg, expect_msg);
 313     }
 314 }
 315
 316 static uint64_t colo_receive_message_value(QEMUFile *f, uint32_t expect_msg,
 317                                            Error **errp)
 318 {
 319     Error *local_err = NULL;
 320     uint64_t value;
 321     int ret;
 322
 323     colo_receive_check_message(f, expect_msg, &local_err);
 324     if (local_err) {
 325         error_propagate(errp, local_err);
 326         return 0;
 327     }
 328
 329     value = qemu_get_be64(f);
 330     ret = qemu_file_get_error(f);
 331     if (ret < 0) {
 332         error_setg_errno(errp, -ret, "Failed to get value for COLO message: %s",
 333                          COLOMessage_str(expect_msg));
 334     }
 335     return value;
 336 }
 337
 338 static int colo_do_checkpoint_transaction(MigrationState *s,
 339                                           QIOChannelBuffer *bioc,
 340                                           QEMUFile *fb)
 341 {
 342     Error *local_err = NULL;
 343     int ret = -1;
 344
 345     colo_send_message(s->to_dst_file, COLO_MESSAGE_CHECKPOINT_REQUEST,
 346                       &local_err);
 347     if (local_err) {
 348         goto out;
 349     }
 350
 351     colo_receive_check_message(s->rp_state.from_dst_file,
 352                     COLO_MESSAGE_CHECKPOINT_REPLY, &local_err);
 353     if (local_err) {
 354         goto out;
 355     }
 356     /* Reset channel-buffer directly */
 357     qio_channel_io_seek(QIO_CHANNEL(bioc), 0, 0, NULL);
 358     bioc->usage = 0;
 359
 360     qemu_mutex_lock_iothread();
 361     if (failover_get_state() != FAILOVER_STATUS_NONE) {
 362         qemu_mutex_unlock_iothread();
 363         goto out;
 364     }
 365     vm_stop_force_state(RUN_STATE_COLO);
 366     qemu_mutex_unlock_iothread();
 367     trace_colo_vm_state_change("run", "stop");
 368     /*
 369      * Failover request bh could be called after vm_stop_force_state(),
 370      * So we need check failover_request_is_active() again.
 371      */
 372     if (failover_get_state() != FAILOVER_STATUS_NONE) {
 373         goto out;
 374     }
 375
 376     colo_notify_compares_event(NULL, COLO_EVENT_CHECKPOINT, &local_err);
 377     if (local_err) {
 378         goto out;
 379     }
 380
 381     /* Disable block migration */
 382     migrate_set_block_enabled(false, &local_err);
 383     qemu_savevm_state_header(fb);
 384     qemu_savevm_state_setup(fb);
 385     qemu_mutex_lock_iothread();
 386     replication_do_checkpoint_all(&local_err);
 387     if (local_err) {
 388         qemu_mutex_unlock_iothread();
 389         goto out;
 390     }
 391     qemu_savevm_state_complete_precopy(fb, false, false);
 392     qemu_mutex_unlock_iothread();
 393
 394     qemu_fflush(fb);
 395
 396     colo_send_message(s->to_dst_file, COLO_MESSAGE_VMSTATE_SEND, &local_err);
 397     if (local_err) {
 398         goto out;
 399     }
 400     /*
 401      * We need the size of the VMstate data in Secondary side,
 402      * With which we can decide how much data should be read.
 403      */
 404     colo_send_message_value(s->to_dst_file, COLO_MESSAGE_VMSTATE_SIZE,
 405                             bioc->usage, &local_err);
 406     if (local_err) {
 407         goto out;
 408     }
 409
 410     qemu_put_buffer(s->to_dst_file, bioc->data, bioc->usage);
 411     qemu_fflush(s->to_dst_file);
 412     ret = qemu_file_get_error(s->to_dst_file);
 413     if (ret < 0) {
 414         goto out;
 415     }
 416
 417     colo_receive_check_message(s->rp_state.from_dst_file,
 418                        COLO_MESSAGE_VMSTATE_RECEIVED, &local_err);
 419     if (local_err) {
 420         goto out;
 421     }
 422
 423     colo_receive_check_message(s->rp_state.from_dst_file,
 424                        COLO_MESSAGE_VMSTATE_LOADED, &local_err);
 425     if (local_err) {
 426         goto out;
 427     }
 428
 429     ret = 0;
 430
 431     qemu_mutex_lock_iothread();
 432     vm_start();
 433     qemu_mutex_unlock_iothread();
 434     trace_colo_vm_state_change("stop", "run");
 435
 436 out:
 437     if (local_err) {
 438         error_report_err(local_err);
 439     }
 440     return ret;
 441 }
 442
 443 static void colo_compare_notify_checkpoint(Notifier *notifier, void *data)
 444 {
 445     colo_checkpoint_notify(data);
 446 }
 447
 448 static void colo_process_checkpoint(MigrationState *s)
 449 {
 450     QIOChannelBuffer *bioc;
 451     QEMUFile *fb = NULL;
 452     int64_t current_time = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 453     Error *local_err = NULL;
 454     int ret;
 455
 456     failover_init_state();
 457
 458     s->rp_state.from_dst_file = qemu_file_get_return_path(s->to_dst_file);
 459     if (!s->rp_state.from_dst_file) {
 460         error_report("Open QEMUFile from_dst_file failed");
 461         goto out;
 462     }
 463
 464     packets_compare_notifier.notify = colo_compare_notify_checkpoint;
 465     colo_compare_register_notifier(&packets_compare_notifier);
 466
 467     /*
 468      * Wait for Secondary finish loading VM states and enter COLO
 469      * restore.
 470      */
 471     colo_receive_check_message(s->rp_state.from_dst_file,
 472                        COLO_MESSAGE_CHECKPOINT_READY, &local_err);
 473     if (local_err) {
 474         goto out;
 475     }
 476     bioc = qio_channel_buffer_new(COLO_BUFFER_BASE_SIZE);
 477     fb = qemu_fopen_channel_output(QIO_CHANNEL(bioc));
 478     object_unref(OBJECT(bioc));
 479
 480     qemu_mutex_lock_iothread();
 481     replication_start_all(REPLICATION_MODE_PRIMARY, &local_err);
 482     if (local_err) {
 483         qemu_mutex_unlock_iothread();
 484         goto out;
 485     }
 486
 487     vm_start();
 488     qemu_mutex_unlock_iothread();
 489     trace_colo_vm_state_change("stop", "run");
 490
 491     timer_mod(s->colo_delay_timer,
 492             current_time + s->parameters.x_checkpoint_delay);
 493
 494     while (s->state == MIGRATION_STATUS_COLO) {
 495         if (failover_get_state() != FAILOVER_STATUS_NONE) {
 496             error_report("failover request");
 497             goto out;
 498         }
 499
 500         qemu_sem_wait(&s->colo_checkpoint_sem);
 501
 502         ret = colo_do_checkpoint_transaction(s, bioc, fb);
 503         if (ret < 0) {
 504             goto out;
 505         }
 506     }
 507
 508 out:
 509     /* Throw the unreported error message after exited from loop */
 510     if (local_err) {
 511         error_report_err(local_err);
 512     }
 513
 514     if (fb) {
 515         qemu_fclose(fb);
 516     }
 517
 518     /*
 519      * There are only two reasons we can get here, some error happened
 520      * or the user triggered failover.
 521      */
 522     switch (failover_get_state()) {
 523     case FAILOVER_STATUS_NONE:
 524         qapi_event_send_colo_exit(COLO_MODE_PRIMARY,
 525                                   COLO_EXIT_REASON_ERROR);
 526         break;
 527     case FAILOVER_STATUS_REQUIRE:
 528         qapi_event_send_colo_exit(COLO_MODE_PRIMARY,
 529                                   COLO_EXIT_REASON_REQUEST);
 530         break;
 531     default:
 532         abort();
 533     }
 534
 535     /* Hope this not to be too long to wait here */
 536     qemu_sem_wait(&s->colo_exit_sem);
 537     qemu_sem_destroy(&s->colo_exit_sem);
 538
 539     /*
 540      * It is safe to unregister notifier after failover finished.
 541      * Besides, colo_delay_timer and colo_checkpoint_sem can't be
 542      * released befor unregister notifier, or there will be use-after-free
 543      * error.
 544      */
 545     colo_compare_unregister_notifier(&packets_compare_notifier);
 546     timer_del(s->colo_delay_timer);
 547     timer_free(s->colo_delay_timer);
 548     qemu_sem_destroy(&s->colo_checkpoint_sem);
 549
 550     /*
 551      * Must be called after failover BH is completed,
 552      * Or the failover BH may shutdown the wrong fd that
 553      * re-used by other threads after we release here.
 554      */
 555     if (s->rp_state.from_dst_file) {
 556         qemu_fclose(s->rp_state.from_dst_file);
 557     }
 558 }
 559
 560 void colo_checkpoint_notify(void *opaque)
 561 {
 562     MigrationState *s = opaque;
 563     int64_t next_notify_time;
 564
 565     qemu_sem_post(&s->colo_checkpoint_sem);
 566     s->colo_checkpoint_time = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 567     next_notify_time = s->colo_checkpoint_time +
 568                     s->parameters.x_checkpoint_delay;
 569     timer_mod(s->colo_delay_timer, next_notify_time);
 570 }
 571
 572 void migrate_start_colo_process(MigrationState *s)
 573 {
 574     qemu_mutex_unlock_iothread();
 575     qemu_sem_init(&s->colo_checkpoint_sem, 0);
 576     s->colo_delay_timer =  timer_new_ms(QEMU_CLOCK_HOST,
 577                                 colo_checkpoint_notify, s);
 578
 579     qemu_sem_init(&s->colo_exit_sem, 0);
 580     migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
 581                       MIGRATION_STATUS_COLO);
 582     colo_process_checkpoint(s);
 583     qemu_mutex_lock_iothread();
 584 }
 585
 586 static void colo_wait_handle_message(QEMUFile *f, int *checkpoint_request,
 587                                      Error **errp)
 588 {
 589     COLOMessage msg;
 590     Error *local_err = NULL;
 591
 592     msg = colo_receive_message(f, &local_err);
 593     if (local_err) {
 594         error_propagate(errp, local_err);
 595         return;
 596     }
 597
 598     switch (msg) {
 599     case COLO_MESSAGE_CHECKPOINT_REQUEST:
 600         *checkpoint_request = 1;
 601         break;
 602     default:
 603         *checkpoint_request = 0;
 604         error_setg(errp, "Got unknown COLO message: %d", msg);
 605         break;
 606     }
 607 }
 608
 609 void *colo_process_incoming_thread(void *opaque)
 610 {
 611     MigrationIncomingState *mis = opaque;
 612     QEMUFile *fb = NULL;
 613     QIOChannelBuffer *bioc = NULL; /* Cache incoming device state */
 614     uint64_t total_size;
 615     uint64_t value;
 616     Error *local_err = NULL;
 617
 618     rcu_register_thread();
 619     qemu_sem_init(&mis->colo_incoming_sem, 0);
 620
 621     migrate_set_state(&mis->state, MIGRATION_STATUS_ACTIVE,
 622                       MIGRATION_STATUS_COLO);
 623
 624     failover_init_state();
 625
 626     mis->to_src_file = qemu_file_get_return_path(mis->from_src_file);
 627     if (!mis->to_src_file) {
 628         error_report("COLO incoming thread: Open QEMUFile to_src_file failed");
 629         goto out;
 630     }
 631     /*
 632      * Note: the communication between Primary side and Secondary side
 633      * should be sequential, we set the fd to unblocked in migration incoming
 634      * coroutine, and here we are in the COLO incoming thread, so it is ok to
 635      * set the fd back to blocked.
 636      */
 637     qemu_file_set_blocking(mis->from_src_file, true);
 638
 639     bioc = qio_channel_buffer_new(COLO_BUFFER_BASE_SIZE);
 640     fb = qemu_fopen_channel_input(QIO_CHANNEL(bioc));
 641     object_unref(OBJECT(bioc));
 642
 643     qemu_mutex_lock_iothread();
 644     replication_start_all(REPLICATION_MODE_SECONDARY, &local_err);
 645     if (local_err) {
 646         qemu_mutex_unlock_iothread();
 647         goto out;
 648     }
 649     vm_start();
 650     trace_colo_vm_state_change("stop", "run");
 651     qemu_mutex_unlock_iothread();
 652
 653     colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_READY,
 654                       &local_err);
 655     if (local_err) {
 656         goto out;
 657     }
 658
 659     while (mis->state == MIGRATION_STATUS_COLO) {
 660         int request = 0;
 661
 662         colo_wait_handle_message(mis->from_src_file, &request, &local_err);
 663         if (local_err) {
 664             goto out;
 665         }
 666         assert(request);
 667         if (failover_get_state() != FAILOVER_STATUS_NONE) {
 668             error_report("failover request");
 669             goto out;
 670         }
 671
 672         qemu_mutex_lock_iothread();
 673         vm_stop_force_state(RUN_STATE_COLO);
 674         trace_colo_vm_state_change("run", "stop");
 675         qemu_mutex_unlock_iothread();
 676
 677         /* FIXME: This is unnecessary for periodic checkpoint mode */
 678         colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_REPLY,
 679                      &local_err);
 680         if (local_err) {
 681             goto out;
 682         }
 683
 684         colo_receive_check_message(mis->from_src_file,
 685                            COLO_MESSAGE_VMSTATE_SEND, &local_err);
 686         if (local_err) {
 687             goto out;
 688         }
 689
 690         value = colo_receive_message_value(mis->from_src_file,
 691                                  COLO_MESSAGE_VMSTATE_SIZE, &local_err);
 692         if (local_err) {
 693             goto out;
 694         }
 695
 696         /*
 697          * Read VM device state data into channel buffer,
 698          * It's better to re-use the memory allocated.
 699          * Here we need to handle the channel buffer directly.
 700          */
 701         if (value > bioc->capacity) {
 702             bioc->capacity = value;
 703             bioc->data = g_realloc(bioc->data, bioc->capacity);
 704         }
 705         total_size = qemu_get_buffer(mis->from_src_file, bioc->data, value);
 706         if (total_size != value) {
 707             error_report("Got %" PRIu64 " VMState data, less than expected"
 708                         " %" PRIu64, total_size, value);
 709             goto out;
 710         }
 711         bioc->usage = total_size;
 712         qio_channel_io_seek(QIO_CHANNEL(bioc), 0, 0, NULL);
 713
 714         colo_send_message(mis->to_src_file, COLO_MESSAGE_VMSTATE_RECEIVED,
 715                      &local_err);
 716         if (local_err) {
 717             goto out;
 718         }
 719
 720         qemu_mutex_lock_iothread();
 721         qemu_system_reset(SHUTDOWN_CAUSE_NONE);
 722         vmstate_loading = true;
 723         if (qemu_loadvm_state(fb) < 0) {
 724             error_report("COLO: loadvm failed");
 725             qemu_mutex_unlock_iothread();
 726             goto out;
 727         }
 728
 729         replication_get_error_all(&local_err);
 730         if (local_err) {
 731             qemu_mutex_unlock_iothread();
 732             goto out;
 733         }
 734         /* discard colo disk buffer */
 735         replication_do_checkpoint_all(&local_err);
 736         if (local_err) {
 737             qemu_mutex_unlock_iothread();
 738             goto out;
 739         }
 740
 741         vmstate_loading = false;
 742         vm_start();
 743         trace_colo_vm_state_change("stop", "run");
 744         qemu_mutex_unlock_iothread();
 745
 746         if (failover_get_state() == FAILOVER_STATUS_RELAUNCH) {
 747             failover_set_state(FAILOVER_STATUS_RELAUNCH,
 748                             FAILOVER_STATUS_NONE);
 749             failover_request_active(NULL);
 750             goto out;
 751         }
 752
 753         colo_send_message(mis->to_src_file, COLO_MESSAGE_VMSTATE_LOADED,
 754                      &local_err);
 755         if (local_err) {
 756             goto out;
 757         }
 758     }
 759
 760 out:
 761     vmstate_loading = false;
 762     /* Throw the unreported error message after exited from loop */
 763     if (local_err) {
 764         error_report_err(local_err);
 765     }
 766
 767     switch (failover_get_state()) {
 768     case FAILOVER_STATUS_NONE:
 769         qapi_event_send_colo_exit(COLO_MODE_SECONDARY,
 770                                   COLO_EXIT_REASON_ERROR);
 771         break;
 772     case FAILOVER_STATUS_REQUIRE:
 773         qapi_event_send_colo_exit(COLO_MODE_SECONDARY,
 774                                   COLO_EXIT_REASON_REQUEST);
 775         break;
 776     default:
 777         abort();
 778     }
 779
 780     if (fb) {
 781         qemu_fclose(fb);
 782     }
 783
 784     /* Hope this not to be too long to loop here */
 785     qemu_sem_wait(&mis->colo_incoming_sem);
 786     qemu_sem_destroy(&mis->colo_incoming_sem);
 787     /* Must be called after failover BH is completed */
 788     if (mis->to_src_file) {
 789         qemu_fclose(mis->to_src_file);
 790     }
 791     migration_incoming_disable_colo();
 792
 793     rcu_unregister_thread();
 794     return NULL;
 795 }