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
  23 #define COLO_BUFFER_BASE_SIZE (4 * 1024 * 1024)
  24
  25 bool colo_supported(void)
  26 {
  27     return true;
  28 }
  29
  30 bool migration_in_colo_state(void)
  31 {
  32     MigrationState *s = migrate_get_current();
  33
  34     return (s->state == MIGRATION_STATUS_COLO);
  35 }
  36
  37 bool migration_incoming_in_colo_state(void)
  38 {
  39     MigrationIncomingState *mis = migration_incoming_get_current();
  40
  41     return mis && (mis->state == MIGRATION_STATUS_COLO);
  42 }
  43
  44 static bool colo_runstate_is_stopped(void)
  45 {
  46     return runstate_check(RUN_STATE_COLO) || !runstate_is_running();
  47 }
  48
  49 static void secondary_vm_do_failover(void)
  50 {
  51     int old_state;
  52     MigrationIncomingState *mis = migration_incoming_get_current();
  53
  54     migrate_set_state(&mis->state, MIGRATION_STATUS_COLO,
  55                       MIGRATION_STATUS_COMPLETED);
  56
  57     if (!autostart) {
  58         error_report("\"-S\" qemu option will be ignored in secondary side");
  59         /* recover runstate to normal migration finish state */
  60         autostart = true;
  61     }
  62
  63     old_state = failover_set_state(FAILOVER_STATUS_ACTIVE,
  64                                    FAILOVER_STATUS_COMPLETED);
  65     if (old_state != FAILOVER_STATUS_ACTIVE) {
  66         error_report("Incorrect state (%s) while doing failover for "
  67                      "secondary VM", FailoverStatus_lookup[old_state]);
  68         return;
  69     }
  70     /* For Secondary VM, jump to incoming co */
  71     if (mis->migration_incoming_co) {
  72         qemu_coroutine_enter(mis->migration_incoming_co);
  73     }
  74 }
  75
  76 static void primary_vm_do_failover(void)
  77 {
  78     MigrationState *s = migrate_get_current();
  79     int old_state;
  80
  81     migrate_set_state(&s->state, MIGRATION_STATUS_COLO,
  82                       MIGRATION_STATUS_COMPLETED);
  83
  84     old_state = failover_set_state(FAILOVER_STATUS_ACTIVE,
  85                                    FAILOVER_STATUS_COMPLETED);
  86     if (old_state != FAILOVER_STATUS_ACTIVE) {
  87         error_report("Incorrect state (%s) while doing failover for Primary VM",
  88                      FailoverStatus_lookup[old_state]);
  89         return;
  90     }
  91 }
  92
  93 void colo_do_failover(MigrationState *s)
  94 {
  95     /* Make sure VM stopped while failover happened. */
  96     if (!colo_runstate_is_stopped()) {
  97         vm_stop_force_state(RUN_STATE_COLO);
  98     }
  99
 100     if (get_colo_mode() == COLO_MODE_PRIMARY) {
 101         primary_vm_do_failover();
 102     } else {
 103         secondary_vm_do_failover();
 104     }
 105 }
 106
 107 static void colo_send_message(QEMUFile *f, COLOMessage msg,
 108                               Error **errp)
 109 {
 110     int ret;
 111
 112     if (msg >= COLO_MESSAGE__MAX) {
 113         error_setg(errp, "%s: Invalid message", __func__);
 114         return;
 115     }
 116     qemu_put_be32(f, msg);
 117     qemu_fflush(f);
 118
 119     ret = qemu_file_get_error(f);
 120     if (ret < 0) {
 121         error_setg_errno(errp, -ret, "Can't send COLO message");
 122     }
 123     trace_colo_send_message(COLOMessage_lookup[msg]);
 124 }
 125
 126 static void colo_send_message_value(QEMUFile *f, COLOMessage msg,
 127                                     uint64_t value, Error **errp)
 128 {
 129     Error *local_err = NULL;
 130     int ret;
 131
 132     colo_send_message(f, msg, &local_err);
 133     if (local_err) {
 134         error_propagate(errp, local_err);
 135         return;
 136     }
 137     qemu_put_be64(f, value);
 138     qemu_fflush(f);
 139
 140     ret = qemu_file_get_error(f);
 141     if (ret < 0) {
 142         error_setg_errno(errp, -ret, "Failed to send value for message:%s",
 143                          COLOMessage_lookup[msg]);
 144     }
 145 }
 146
 147 static COLOMessage colo_receive_message(QEMUFile *f, Error **errp)
 148 {
 149     COLOMessage msg;
 150     int ret;
 151
 152     msg = qemu_get_be32(f);
 153     ret = qemu_file_get_error(f);
 154     if (ret < 0) {
 155         error_setg_errno(errp, -ret, "Can't receive COLO message");
 156         return msg;
 157     }
 158     if (msg >= COLO_MESSAGE__MAX) {
 159         error_setg(errp, "%s: Invalid message", __func__);
 160         return msg;
 161     }
 162     trace_colo_receive_message(COLOMessage_lookup[msg]);
 163     return msg;
 164 }
 165
 166 static void colo_receive_check_message(QEMUFile *f, COLOMessage expect_msg,
 167                                        Error **errp)
 168 {
 169     COLOMessage msg;
 170     Error *local_err = NULL;
 171
 172     msg = colo_receive_message(f, &local_err);
 173     if (local_err) {
 174         error_propagate(errp, local_err);
 175         return;
 176     }
 177     if (msg != expect_msg) {
 178         error_setg(errp, "Unexpected COLO message %d, expected %d",
 179                           msg, expect_msg);
 180     }
 181 }
 182
 183 static uint64_t colo_receive_message_value(QEMUFile *f, uint32_t expect_msg,
 184                                            Error **errp)
 185 {
 186     Error *local_err = NULL;
 187     uint64_t value;
 188     int ret;
 189
 190     colo_receive_check_message(f, expect_msg, &local_err);
 191     if (local_err) {
 192         error_propagate(errp, local_err);
 193         return 0;
 194     }
 195
 196     value = qemu_get_be64(f);
 197     ret = qemu_file_get_error(f);
 198     if (ret < 0) {
 199         error_setg_errno(errp, -ret, "Failed to get value for COLO message: %s",
 200                          COLOMessage_lookup[expect_msg]);
 201     }
 202     return value;
 203 }
 204
 205 static int colo_do_checkpoint_transaction(MigrationState *s,
 206                                           QIOChannelBuffer *bioc,
 207                                           QEMUFile *fb)
 208 {
 209     Error *local_err = NULL;
 210     int ret = -1;
 211
 212     colo_send_message(s->to_dst_file, COLO_MESSAGE_CHECKPOINT_REQUEST,
 213                       &local_err);
 214     if (local_err) {
 215         goto out;
 216     }
 217
 218     colo_receive_check_message(s->rp_state.from_dst_file,
 219                     COLO_MESSAGE_CHECKPOINT_REPLY, &local_err);
 220     if (local_err) {
 221         goto out;
 222     }
 223     /* Reset channel-buffer directly */
 224     qio_channel_io_seek(QIO_CHANNEL(bioc), 0, 0, NULL);
 225     bioc->usage = 0;
 226
 227     qemu_mutex_lock_iothread();
 228     if (failover_get_state() != FAILOVER_STATUS_NONE) {
 229         qemu_mutex_unlock_iothread();
 230         goto out;
 231     }
 232     vm_stop_force_state(RUN_STATE_COLO);
 233     qemu_mutex_unlock_iothread();
 234     trace_colo_vm_state_change("run", "stop");
 235     /*
 236      * Failover request bh could be called after vm_stop_force_state(),
 237      * So we need check failover_request_is_active() again.
 238      */
 239     if (failover_get_state() != FAILOVER_STATUS_NONE) {
 240         goto out;
 241     }
 242
 243     /* Disable block migration */
 244     s->params.blk = 0;
 245     s->params.shared = 0;
 246     qemu_savevm_state_header(fb);
 247     qemu_savevm_state_begin(fb, &s->params);
 248     qemu_mutex_lock_iothread();
 249     qemu_savevm_state_complete_precopy(fb, false);
 250     qemu_mutex_unlock_iothread();
 251
 252     qemu_fflush(fb);
 253
 254     colo_send_message(s->to_dst_file, COLO_MESSAGE_VMSTATE_SEND, &local_err);
 255     if (local_err) {
 256         goto out;
 257     }
 258     /*
 259      * We need the size of the VMstate data in Secondary side,
 260      * With which we can decide how much data should be read.
 261      */
 262     colo_send_message_value(s->to_dst_file, COLO_MESSAGE_VMSTATE_SIZE,
 263                             bioc->usage, &local_err);
 264     if (local_err) {
 265         goto out;
 266     }
 267
 268     qemu_put_buffer(s->to_dst_file, bioc->data, bioc->usage);
 269     qemu_fflush(s->to_dst_file);
 270     ret = qemu_file_get_error(s->to_dst_file);
 271     if (ret < 0) {
 272         goto out;
 273     }
 274
 275     colo_receive_check_message(s->rp_state.from_dst_file,
 276                        COLO_MESSAGE_VMSTATE_RECEIVED, &local_err);
 277     if (local_err) {
 278         goto out;
 279     }
 280
 281     colo_receive_check_message(s->rp_state.from_dst_file,
 282                        COLO_MESSAGE_VMSTATE_LOADED, &local_err);
 283     if (local_err) {
 284         goto out;
 285     }
 286
 287     ret = 0;
 288
 289     qemu_mutex_lock_iothread();
 290     vm_start();
 291     qemu_mutex_unlock_iothread();
 292     trace_colo_vm_state_change("stop", "run");
 293
 294 out:
 295     if (local_err) {
 296         error_report_err(local_err);
 297     }
 298     return ret;
 299 }
 300
 301 static void colo_process_checkpoint(MigrationState *s)
 302 {
 303     QIOChannelBuffer *bioc;
 304     QEMUFile *fb = NULL;
 305     int64_t current_time, checkpoint_time = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 306     Error *local_err = NULL;
 307     int ret;
 308
 309     failover_init_state();
 310
 311     s->rp_state.from_dst_file = qemu_file_get_return_path(s->to_dst_file);
 312     if (!s->rp_state.from_dst_file) {
 313         error_report("Open QEMUFile from_dst_file failed");
 314         goto out;
 315     }
 316
 317     /*
 318      * Wait for Secondary finish loading VM states and enter COLO
 319      * restore.
 320      */
 321     colo_receive_check_message(s->rp_state.from_dst_file,
 322                        COLO_MESSAGE_CHECKPOINT_READY, &local_err);
 323     if (local_err) {
 324         goto out;
 325     }
 326     bioc = qio_channel_buffer_new(COLO_BUFFER_BASE_SIZE);
 327     fb = qemu_fopen_channel_output(QIO_CHANNEL(bioc));
 328     object_unref(OBJECT(bioc));
 329
 330     qemu_mutex_lock_iothread();
 331     vm_start();
 332     qemu_mutex_unlock_iothread();
 333     trace_colo_vm_state_change("stop", "run");
 334
 335     while (s->state == MIGRATION_STATUS_COLO) {
 336         if (failover_get_state() != FAILOVER_STATUS_NONE) {
 337             error_report("failover request");
 338             goto out;
 339         }
 340
 341         current_time = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 342         if (current_time - checkpoint_time <
 343             s->parameters.x_checkpoint_delay) {
 344             int64_t delay_ms;
 345
 346             delay_ms = s->parameters.x_checkpoint_delay -
 347                        (current_time - checkpoint_time);
 348             g_usleep(delay_ms * 1000);
 349         }
 350         ret = colo_do_checkpoint_transaction(s, bioc, fb);
 351         if (ret < 0) {
 352             goto out;
 353         }
 354         checkpoint_time = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 355     }
 356
 357 out:
 358     /* Throw the unreported error message after exited from loop */
 359     if (local_err) {
 360         error_report_err(local_err);
 361     }
 362
 363     if (fb) {
 364         qemu_fclose(fb);
 365     }
 366
 367     if (s->rp_state.from_dst_file) {
 368         qemu_fclose(s->rp_state.from_dst_file);
 369     }
 370 }
 371
 372 void migrate_start_colo_process(MigrationState *s)
 373 {
 374     qemu_mutex_unlock_iothread();
 375     migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
 376                       MIGRATION_STATUS_COLO);
 377     colo_process_checkpoint(s);
 378     qemu_mutex_lock_iothread();
 379 }
 380
 381 static void colo_wait_handle_message(QEMUFile *f, int *checkpoint_request,
 382                                      Error **errp)
 383 {
 384     COLOMessage msg;
 385     Error *local_err = NULL;
 386
 387     msg = colo_receive_message(f, &local_err);
 388     if (local_err) {
 389         error_propagate(errp, local_err);
 390         return;
 391     }
 392
 393     switch (msg) {
 394     case COLO_MESSAGE_CHECKPOINT_REQUEST:
 395         *checkpoint_request = 1;
 396         break;
 397     default:
 398         *checkpoint_request = 0;
 399         error_setg(errp, "Got unknown COLO message: %d", msg);
 400         break;
 401     }
 402 }
 403
 404 void *colo_process_incoming_thread(void *opaque)
 405 {
 406     MigrationIncomingState *mis = opaque;
 407     QEMUFile *fb = NULL;
 408     QIOChannelBuffer *bioc = NULL; /* Cache incoming device state */
 409     uint64_t total_size;
 410     uint64_t value;
 411     Error *local_err = NULL;
 412
 413     migrate_set_state(&mis->state, MIGRATION_STATUS_ACTIVE,
 414                       MIGRATION_STATUS_COLO);
 415
 416     failover_init_state();
 417
 418     mis->to_src_file = qemu_file_get_return_path(mis->from_src_file);
 419     if (!mis->to_src_file) {
 420         error_report("COLO incoming thread: Open QEMUFile to_src_file failed");
 421         goto out;
 422     }
 423     /*
 424      * Note: the communication between Primary side and Secondary side
 425      * should be sequential, we set the fd to unblocked in migration incoming
 426      * coroutine, and here we are in the COLO incoming thread, so it is ok to
 427      * set the fd back to blocked.
 428      */
 429     qemu_file_set_blocking(mis->from_src_file, true);
 430
 431     bioc = qio_channel_buffer_new(COLO_BUFFER_BASE_SIZE);
 432     fb = qemu_fopen_channel_input(QIO_CHANNEL(bioc));
 433     object_unref(OBJECT(bioc));
 434
 435     colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_READY,
 436                       &local_err);
 437     if (local_err) {
 438         goto out;
 439     }
 440
 441     while (mis->state == MIGRATION_STATUS_COLO) {
 442         int request = 0;
 443
 444         colo_wait_handle_message(mis->from_src_file, &request, &local_err);
 445         if (local_err) {
 446             goto out;
 447         }
 448         assert(request);
 449         if (failover_get_state() != FAILOVER_STATUS_NONE) {
 450             error_report("failover request");
 451             goto out;
 452         }
 453
 454         /* FIXME: This is unnecessary for periodic checkpoint mode */
 455         colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_REPLY,
 456                      &local_err);
 457         if (local_err) {
 458             goto out;
 459         }
 460
 461         colo_receive_check_message(mis->from_src_file,
 462                            COLO_MESSAGE_VMSTATE_SEND, &local_err);
 463         if (local_err) {
 464             goto out;
 465         }
 466
 467         value = colo_receive_message_value(mis->from_src_file,
 468                                  COLO_MESSAGE_VMSTATE_SIZE, &local_err);
 469         if (local_err) {
 470             goto out;
 471         }
 472
 473         /*
 474          * Read VM device state data into channel buffer,
 475          * It's better to re-use the memory allocated.
 476          * Here we need to handle the channel buffer directly.
 477          */
 478         if (value > bioc->capacity) {
 479             bioc->capacity = value;
 480             bioc->data = g_realloc(bioc->data, bioc->capacity);
 481         }
 482         total_size = qemu_get_buffer(mis->from_src_file, bioc->data, value);
 483         if (total_size != value) {
 484             error_report("Got %" PRIu64 " VMState data, less than expected"
 485                         " %" PRIu64, total_size, value);
 486             goto out;
 487         }
 488         bioc->usage = total_size;
 489         qio_channel_io_seek(QIO_CHANNEL(bioc), 0, 0, NULL);
 490
 491         colo_send_message(mis->to_src_file, COLO_MESSAGE_VMSTATE_RECEIVED,
 492                      &local_err);
 493         if (local_err) {
 494             goto out;
 495         }
 496
 497         qemu_mutex_lock_iothread();
 498         qemu_system_reset(VMRESET_SILENT);
 499         if (qemu_loadvm_state(fb) < 0) {
 500             error_report("COLO: loadvm failed");
 501             qemu_mutex_unlock_iothread();
 502             goto out;
 503         }
 504         qemu_mutex_unlock_iothread();
 505
 506         colo_send_message(mis->to_src_file, COLO_MESSAGE_VMSTATE_LOADED,
 507                      &local_err);
 508         if (local_err) {
 509             goto out;
 510         }
 511     }
 512
 513 out:
 514     /* Throw the unreported error message after exited from loop */
 515     if (local_err) {
 516         error_report_err(local_err);
 517     }
 518
 519     if (fb) {
 520         qemu_fclose(fb);
 521     }
 522
 523     if (mis->to_src_file) {
 524         qemu_fclose(mis->to_src_file);
 525     }
 526     migration_incoming_exit_colo();
 527
 528     return NULL;
 529 }