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