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