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