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