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