qapi2texi: Fix up output around #optional
[qemu.git] / 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 }