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 "qemu/main-loop.h"
27 #include "qemu/rcu.h"
28 #include "migration/failover.h"
29 #include "migration/ram.h"
30 #ifdef CONFIG_REPLICATION
31 #include "replication.h"
32 #endif
33 #include "net/colo-compare.h"
34 #include "net/colo.h"
35 #include "block/block.h"
36 #include "qapi/qapi-events-migration.h"
37 #include "qapi/qmp/qerror.h"
38 #include "sysemu/cpus.h"
39 #include "sysemu/runstate.h"
40 #include "net/filter.h"
41
42 static bool vmstate_loading;
43 static Notifier packets_compare_notifier;
44
45 /* User need to know colo mode after COLO failover */
46 static COLOMode last_colo_mode;
47
48 #define COLO_BUFFER_BASE_SIZE (4 * 1024 * 1024)
49
migration_in_colo_state(void)50 bool migration_in_colo_state(void)
51 {
52 MigrationState *s = migrate_get_current();
53
54 return (s->state == MIGRATION_STATUS_COLO);
55 }
56
migration_incoming_in_colo_state(void)57 bool migration_incoming_in_colo_state(void)
58 {
59 MigrationIncomingState *mis = migration_incoming_get_current();
60
61 return mis && (mis->state == MIGRATION_STATUS_COLO);
62 }
63
colo_runstate_is_stopped(void)64 static bool colo_runstate_is_stopped(void)
65 {
66 return runstate_check(RUN_STATE_COLO) || !runstate_is_running();
67 }
68
secondary_vm_do_failover(void)69 static void secondary_vm_do_failover(void)
70 {
71 /* COLO needs enable block-replication */
72 #ifdef CONFIG_REPLICATION
73 int old_state;
74 MigrationIncomingState *mis = migration_incoming_get_current();
75 Error *local_err = NULL;
76
77 /* Can not do failover during the process of VM's loading VMstate, Or
78 * it will break the secondary VM.
79 */
80 if (vmstate_loading) {
81 old_state = failover_set_state(FAILOVER_STATUS_ACTIVE,
82 FAILOVER_STATUS_RELAUNCH);
83 if (old_state != FAILOVER_STATUS_ACTIVE) {
84 error_report("Unknown error while do failover for secondary VM,"
85 "old_state: %s", FailoverStatus_str(old_state));
86 }
87 return;
88 }
89
90 migrate_set_state(&mis->state, MIGRATION_STATUS_COLO,
91 MIGRATION_STATUS_COMPLETED);
92
93 replication_stop_all(true, &local_err);
94 if (local_err) {
95 error_report_err(local_err);
96 }
97
98 /* Notify all filters of all NIC to do checkpoint */
99 colo_notify_filters_event(COLO_EVENT_FAILOVER, &local_err);
100 if (local_err) {
101 error_report_err(local_err);
102 }
103
104 if (!autostart) {
105 error_report("\"-S\" qemu option will be ignored in secondary side");
106 /* recover runstate to normal migration finish state */
107 autostart = true;
108 }
109 /*
110 * Make sure COLO incoming thread not block in recv or send,
111 * If mis->from_src_file and mis->to_src_file use the same fd,
112 * The second shutdown() will return -1, we ignore this value,
113 * It is harmless.
114 */
115 if (mis->from_src_file) {
116 qemu_file_shutdown(mis->from_src_file);
117 }
118 if (mis->to_src_file) {
119 qemu_file_shutdown(mis->to_src_file);
120 }
121
122 old_state = failover_set_state(FAILOVER_STATUS_ACTIVE,
123 FAILOVER_STATUS_COMPLETED);
124 if (old_state != FAILOVER_STATUS_ACTIVE) {
125 error_report("Incorrect state (%s) while doing failover for "
126 "secondary VM", FailoverStatus_str(old_state));
127 return;
128 }
129 /* Notify COLO incoming thread that failover work is finished */
130 qemu_sem_post(&mis->colo_incoming_sem);
131
132 /* For Secondary VM, jump to incoming co */
133 if (mis->migration_incoming_co) {
134 qemu_coroutine_enter(mis->migration_incoming_co);
135 }
136 #else
137 abort();
138 #endif
139 }
140
primary_vm_do_failover(void)141 static void primary_vm_do_failover(void)
142 {
143 #ifdef CONFIG_REPLICATION
144 MigrationState *s = migrate_get_current();
145 int old_state;
146 Error *local_err = NULL;
147
148 migrate_set_state(&s->state, MIGRATION_STATUS_COLO,
149 MIGRATION_STATUS_COMPLETED);
150 /*
151 * kick COLO thread which might wait at
152 * qemu_sem_wait(&s->colo_checkpoint_sem).
153 */
154 colo_checkpoint_notify(migrate_get_current());
155
156 /*
157 * Wake up COLO thread which may blocked in recv() or send(),
158 * The s->rp_state.from_dst_file and s->to_dst_file may use the
159 * same fd, but we still shutdown the fd for twice, it is harmless.
160 */
161 if (s->to_dst_file) {
162 qemu_file_shutdown(s->to_dst_file);
163 }
164 if (s->rp_state.from_dst_file) {
165 qemu_file_shutdown(s->rp_state.from_dst_file);
166 }
167
168 old_state = failover_set_state(FAILOVER_STATUS_ACTIVE,
169 FAILOVER_STATUS_COMPLETED);
170 if (old_state != FAILOVER_STATUS_ACTIVE) {
171 error_report("Incorrect state (%s) while doing failover for Primary VM",
172 FailoverStatus_str(old_state));
173 return;
174 }
175
176 replication_stop_all(true, &local_err);
177 if (local_err) {
178 error_report_err(local_err);
179 local_err = NULL;
180 }
181
182 /* Notify COLO thread that failover work is finished */
183 qemu_sem_post(&s->colo_exit_sem);
184 #else
185 abort();
186 #endif
187 }
188
get_colo_mode(void)189 COLOMode get_colo_mode(void)
190 {
191 if (migration_in_colo_state()) {
192 return COLO_MODE_PRIMARY;
193 } else if (migration_incoming_in_colo_state()) {
194 return COLO_MODE_SECONDARY;
195 } else {
196 return COLO_MODE_NONE;
197 }
198 }
199
colo_do_failover(void)200 void colo_do_failover(void)
201 {
202 /* Make sure VM stopped while failover happened. */
203 if (!colo_runstate_is_stopped()) {
204 vm_stop_force_state(RUN_STATE_COLO);
205 }
206
207 switch (get_colo_mode()) {
208 case COLO_MODE_PRIMARY:
209 primary_vm_do_failover();
210 break;
211 case COLO_MODE_SECONDARY:
212 secondary_vm_do_failover();
213 break;
214 default:
215 error_report("colo_do_failover failed because the colo mode"
216 " could not be obtained");
217 }
218 }
219
220 #ifdef CONFIG_REPLICATION
qmp_xen_set_replication(bool enable,bool primary,bool has_failover,bool failover,Error ** errp)221 void qmp_xen_set_replication(bool enable, bool primary,
222 bool has_failover, bool failover,
223 Error **errp)
224 {
225 ReplicationMode mode = primary ?
226 REPLICATION_MODE_PRIMARY :
227 REPLICATION_MODE_SECONDARY;
228
229 if (has_failover && enable) {
230 error_setg(errp, "Parameter 'failover' is only for"
231 " stopping replication");
232 return;
233 }
234
235 if (enable) {
236 replication_start_all(mode, errp);
237 } else {
238 if (!has_failover) {
239 failover = NULL;
240 }
241 replication_stop_all(failover, failover ? NULL : errp);
242 }
243 }
244
qmp_query_xen_replication_status(Error ** errp)245 ReplicationStatus *qmp_query_xen_replication_status(Error **errp)
246 {
247 Error *err = NULL;
248 ReplicationStatus *s = g_new0(ReplicationStatus, 1);
249
250 replication_get_error_all(&err);
251 if (err) {
252 s->error = true;
253 s->has_desc = true;
254 s->desc = g_strdup(error_get_pretty(err));
255 } else {
256 s->error = false;
257 }
258
259 error_free(err);
260 return s;
261 }
262
qmp_xen_colo_do_checkpoint(Error ** errp)263 void qmp_xen_colo_do_checkpoint(Error **errp)
264 {
265 replication_do_checkpoint_all(errp);
266 /* Notify all filters of all NIC to do checkpoint */
267 colo_notify_filters_event(COLO_EVENT_CHECKPOINT, errp);
268 }
269 #endif
270
qmp_query_colo_status(Error ** errp)271 COLOStatus *qmp_query_colo_status(Error **errp)
272 {
273 COLOStatus *s = g_new0(COLOStatus, 1);
274
275 s->mode = get_colo_mode();
276 s->last_mode = last_colo_mode;
277
278 switch (failover_get_state()) {
279 case FAILOVER_STATUS_NONE:
280 s->reason = COLO_EXIT_REASON_NONE;
281 break;
282 case FAILOVER_STATUS_COMPLETED:
283 s->reason = COLO_EXIT_REASON_REQUEST;
284 break;
285 default:
286 if (migration_in_colo_state()) {
287 s->reason = COLO_EXIT_REASON_PROCESSING;
288 } else {
289 s->reason = COLO_EXIT_REASON_ERROR;
290 }
291 }
292
293 return s;
294 }
295
colo_send_message(QEMUFile * f,COLOMessage msg,Error ** errp)296 static void colo_send_message(QEMUFile *f, COLOMessage msg,
297 Error **errp)
298 {
299 int ret;
300
301 if (msg >= COLO_MESSAGE__MAX) {
302 error_setg(errp, "%s: Invalid message", __func__);
303 return;
304 }
305 qemu_put_be32(f, msg);
306 qemu_fflush(f);
307
308 ret = qemu_file_get_error(f);
309 if (ret < 0) {
310 error_setg_errno(errp, -ret, "Can't send COLO message");
311 }
312 trace_colo_send_message(COLOMessage_str(msg));
313 }
314
colo_send_message_value(QEMUFile * f,COLOMessage msg,uint64_t value,Error ** errp)315 static void colo_send_message_value(QEMUFile *f, COLOMessage msg,
316 uint64_t value, Error **errp)
317 {
318 Error *local_err = NULL;
319 int ret;
320
321 colo_send_message(f, msg, &local_err);
322 if (local_err) {
323 error_propagate(errp, local_err);
324 return;
325 }
326 qemu_put_be64(f, value);
327 qemu_fflush(f);
328
329 ret = qemu_file_get_error(f);
330 if (ret < 0) {
331 error_setg_errno(errp, -ret, "Failed to send value for message:%s",
332 COLOMessage_str(msg));
333 }
334 }
335
colo_receive_message(QEMUFile * f,Error ** errp)336 static COLOMessage colo_receive_message(QEMUFile *f, Error **errp)
337 {
338 COLOMessage msg;
339 int ret;
340
341 msg = qemu_get_be32(f);
342 ret = qemu_file_get_error(f);
343 if (ret < 0) {
344 error_setg_errno(errp, -ret, "Can't receive COLO message");
345 return msg;
346 }
347 if (msg >= COLO_MESSAGE__MAX) {
348 error_setg(errp, "%s: Invalid message", __func__);
349 return msg;
350 }
351 trace_colo_receive_message(COLOMessage_str(msg));
352 return msg;
353 }
354
colo_receive_check_message(QEMUFile * f,COLOMessage expect_msg,Error ** errp)355 static void colo_receive_check_message(QEMUFile *f, COLOMessage expect_msg,
356 Error **errp)
357 {
358 COLOMessage msg;
359 Error *local_err = NULL;
360
361 msg = colo_receive_message(f, &local_err);
362 if (local_err) {
363 error_propagate(errp, local_err);
364 return;
365 }
366 if (msg != expect_msg) {
367 error_setg(errp, "Unexpected COLO message %d, expected %d",
368 msg, expect_msg);
369 }
370 }
371
colo_receive_message_value(QEMUFile * f,uint32_t expect_msg,Error ** errp)372 static uint64_t colo_receive_message_value(QEMUFile *f, uint32_t expect_msg,
373 Error **errp)
374 {
375 Error *local_err = NULL;
376 uint64_t value;
377 int ret;
378
379 colo_receive_check_message(f, expect_msg, &local_err);
380 if (local_err) {
381 error_propagate(errp, local_err);
382 return 0;
383 }
384
385 value = qemu_get_be64(f);
386 ret = qemu_file_get_error(f);
387 if (ret < 0) {
388 error_setg_errno(errp, -ret, "Failed to get value for COLO message: %s",
389 COLOMessage_str(expect_msg));
390 }
391 return value;
392 }
393
colo_do_checkpoint_transaction(MigrationState * s,QIOChannelBuffer * bioc,QEMUFile * fb)394 static int colo_do_checkpoint_transaction(MigrationState *s,
395 QIOChannelBuffer *bioc,
396 QEMUFile *fb)
397 {
398 Error *local_err = NULL;
399 int ret = -1;
400
401 colo_send_message(s->to_dst_file, COLO_MESSAGE_CHECKPOINT_REQUEST,
402 &local_err);
403 if (local_err) {
404 goto out;
405 }
406
407 colo_receive_check_message(s->rp_state.from_dst_file,
408 COLO_MESSAGE_CHECKPOINT_REPLY, &local_err);
409 if (local_err) {
410 goto out;
411 }
412 /* Reset channel-buffer directly */
413 qio_channel_io_seek(QIO_CHANNEL(bioc), 0, 0, NULL);
414 bioc->usage = 0;
415
416 qemu_mutex_lock_iothread();
417 if (failover_get_state() != FAILOVER_STATUS_NONE) {
418 qemu_mutex_unlock_iothread();
419 goto out;
420 }
421 vm_stop_force_state(RUN_STATE_COLO);
422 qemu_mutex_unlock_iothread();
423 trace_colo_vm_state_change("run", "stop");
424 /*
425 * Failover request bh could be called after vm_stop_force_state(),
426 * So we need check failover_request_is_active() again.
427 */
428 if (failover_get_state() != FAILOVER_STATUS_NONE) {
429 goto out;
430 }
431
432 colo_notify_compares_event(NULL, COLO_EVENT_CHECKPOINT, &local_err);
433 if (local_err) {
434 goto out;
435 }
436
437 /* Disable block migration */
438 migrate_set_block_enabled(false, &local_err);
439 qemu_mutex_lock_iothread();
440
441 #ifdef CONFIG_REPLICATION
442 replication_do_checkpoint_all(&local_err);
443 if (local_err) {
444 qemu_mutex_unlock_iothread();
445 goto out;
446 }
447 #else
448 abort();
449 #endif
450
451 colo_send_message(s->to_dst_file, COLO_MESSAGE_VMSTATE_SEND, &local_err);
452 if (local_err) {
453 qemu_mutex_unlock_iothread();
454 goto out;
455 }
456 /* Note: device state is saved into buffer */
457 ret = qemu_save_device_state(fb);
458
459 qemu_mutex_unlock_iothread();
460 if (ret < 0) {
461 goto out;
462 }
463 /*
464 * Only save VM's live state, which not including device state.
465 * TODO: We may need a timeout mechanism to prevent COLO process
466 * to be blocked here.
467 */
468 qemu_savevm_live_state(s->to_dst_file);
469
470 qemu_fflush(fb);
471
472 /*
473 * We need the size of the VMstate data in Secondary side,
474 * With which we can decide how much data should be read.
475 */
476 colo_send_message_value(s->to_dst_file, COLO_MESSAGE_VMSTATE_SIZE,
477 bioc->usage, &local_err);
478 if (local_err) {
479 goto out;
480 }
481
482 qemu_put_buffer(s->to_dst_file, bioc->data, bioc->usage);
483 qemu_fflush(s->to_dst_file);
484 ret = qemu_file_get_error(s->to_dst_file);
485 if (ret < 0) {
486 goto out;
487 }
488
489 colo_receive_check_message(s->rp_state.from_dst_file,
490 COLO_MESSAGE_VMSTATE_RECEIVED, &local_err);
491 if (local_err) {
492 goto out;
493 }
494
495 colo_receive_check_message(s->rp_state.from_dst_file,
496 COLO_MESSAGE_VMSTATE_LOADED, &local_err);
497 if (local_err) {
498 goto out;
499 }
500
501 ret = 0;
502
503 qemu_mutex_lock_iothread();
504 vm_start();
505 qemu_mutex_unlock_iothread();
506 trace_colo_vm_state_change("stop", "run");
507
508 out:
509 if (local_err) {
510 error_report_err(local_err);
511 }
512 return ret;
513 }
514
colo_compare_notify_checkpoint(Notifier * notifier,void * data)515 static void colo_compare_notify_checkpoint(Notifier *notifier, void *data)
516 {
517 colo_checkpoint_notify(data);
518 }
519
colo_process_checkpoint(MigrationState * s)520 static void colo_process_checkpoint(MigrationState *s)
521 {
522 QIOChannelBuffer *bioc;
523 QEMUFile *fb = NULL;
524 int64_t current_time = qemu_clock_get_ms(QEMU_CLOCK_HOST);
525 Error *local_err = NULL;
526 int ret;
527
528 last_colo_mode = get_colo_mode();
529 if (last_colo_mode != COLO_MODE_PRIMARY) {
530 error_report("COLO mode must be COLO_MODE_PRIMARY");
531 return;
532 }
533
534 failover_init_state();
535
536 s->rp_state.from_dst_file = qemu_file_get_return_path(s->to_dst_file);
537 if (!s->rp_state.from_dst_file) {
538 error_report("Open QEMUFile from_dst_file failed");
539 goto out;
540 }
541
542 packets_compare_notifier.notify = colo_compare_notify_checkpoint;
543 colo_compare_register_notifier(&packets_compare_notifier);
544
545 /*
546 * Wait for Secondary finish loading VM states and enter COLO
547 * restore.
548 */
549 colo_receive_check_message(s->rp_state.from_dst_file,
550 COLO_MESSAGE_CHECKPOINT_READY, &local_err);
551 if (local_err) {
552 goto out;
553 }
554 bioc = qio_channel_buffer_new(COLO_BUFFER_BASE_SIZE);
555 fb = qemu_fopen_channel_output(QIO_CHANNEL(bioc));
556 object_unref(OBJECT(bioc));
557
558 qemu_mutex_lock_iothread();
559 #ifdef CONFIG_REPLICATION
560 replication_start_all(REPLICATION_MODE_PRIMARY, &local_err);
561 if (local_err) {
562 qemu_mutex_unlock_iothread();
563 goto out;
564 }
565 #else
566 abort();
567 #endif
568
569 vm_start();
570 qemu_mutex_unlock_iothread();
571 trace_colo_vm_state_change("stop", "run");
572
573 timer_mod(s->colo_delay_timer,
574 current_time + s->parameters.x_checkpoint_delay);
575
576 while (s->state == MIGRATION_STATUS_COLO) {
577 if (failover_get_state() != FAILOVER_STATUS_NONE) {
578 error_report("failover request");
579 goto out;
580 }
581
582 qemu_sem_wait(&s->colo_checkpoint_sem);
583
584 if (s->state != MIGRATION_STATUS_COLO) {
585 goto out;
586 }
587 ret = colo_do_checkpoint_transaction(s, bioc, fb);
588 if (ret < 0) {
589 goto out;
590 }
591 }
592
593 out:
594 /* Throw the unreported error message after exited from loop */
595 if (local_err) {
596 error_report_err(local_err);
597 }
598
599 if (fb) {
600 qemu_fclose(fb);
601 }
602
603 /*
604 * There are only two reasons we can get here, some error happened
605 * or the user triggered failover.
606 */
607 switch (failover_get_state()) {
608 case FAILOVER_STATUS_COMPLETED:
609 qapi_event_send_colo_exit(COLO_MODE_PRIMARY,
610 COLO_EXIT_REASON_REQUEST);
611 break;
612 default:
613 qapi_event_send_colo_exit(COLO_MODE_PRIMARY,
614 COLO_EXIT_REASON_ERROR);
615 }
616
617 /* Hope this not to be too long to wait here */
618 qemu_sem_wait(&s->colo_exit_sem);
619 qemu_sem_destroy(&s->colo_exit_sem);
620
621 /*
622 * It is safe to unregister notifier after failover finished.
623 * Besides, colo_delay_timer and colo_checkpoint_sem can't be
624 * released befor unregister notifier, or there will be use-after-free
625 * error.
626 */
627 colo_compare_unregister_notifier(&packets_compare_notifier);
628 timer_del(s->colo_delay_timer);
629 timer_free(s->colo_delay_timer);
630 qemu_sem_destroy(&s->colo_checkpoint_sem);
631
632 /*
633 * Must be called after failover BH is completed,
634 * Or the failover BH may shutdown the wrong fd that
635 * re-used by other threads after we release here.
636 */
637 if (s->rp_state.from_dst_file) {
638 qemu_fclose(s->rp_state.from_dst_file);
639 }
640 }
641
colo_checkpoint_notify(void * opaque)642 void colo_checkpoint_notify(void *opaque)
643 {
644 MigrationState *s = opaque;
645 int64_t next_notify_time;
646
647 qemu_sem_post(&s->colo_checkpoint_sem);
648 s->colo_checkpoint_time = qemu_clock_get_ms(QEMU_CLOCK_HOST);
649 next_notify_time = s->colo_checkpoint_time +
650 s->parameters.x_checkpoint_delay;
651 timer_mod(s->colo_delay_timer, next_notify_time);
652 }
653
migrate_start_colo_process(MigrationState * s)654 void migrate_start_colo_process(MigrationState *s)
655 {
656 qemu_mutex_unlock_iothread();
657 qemu_sem_init(&s->colo_checkpoint_sem, 0);
658 s->colo_delay_timer = timer_new_ms(QEMU_CLOCK_HOST,
659 colo_checkpoint_notify, s);
660
661 qemu_sem_init(&s->colo_exit_sem, 0);
662 migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
663 MIGRATION_STATUS_COLO);
664 colo_process_checkpoint(s);
665 qemu_mutex_lock_iothread();
666 }
667
colo_incoming_process_checkpoint(MigrationIncomingState * mis,QEMUFile * fb,QIOChannelBuffer * bioc,Error ** errp)668 static void colo_incoming_process_checkpoint(MigrationIncomingState *mis,
669 QEMUFile *fb, QIOChannelBuffer *bioc, Error **errp)
670 {
671 uint64_t total_size;
672 uint64_t value;
673 Error *local_err = NULL;
674 int ret;
675
676 qemu_mutex_lock_iothread();
677 vm_stop_force_state(RUN_STATE_COLO);
678 trace_colo_vm_state_change("run", "stop");
679 qemu_mutex_unlock_iothread();
680
681 /* FIXME: This is unnecessary for periodic checkpoint mode */
682 colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_REPLY,
683 &local_err);
684 if (local_err) {
685 error_propagate(errp, local_err);
686 return;
687 }
688
689 colo_receive_check_message(mis->from_src_file,
690 COLO_MESSAGE_VMSTATE_SEND, &local_err);
691 if (local_err) {
692 error_propagate(errp, local_err);
693 return;
694 }
695
696 qemu_mutex_lock_iothread();
697 cpu_synchronize_all_pre_loadvm();
698 ret = qemu_loadvm_state_main(mis->from_src_file, mis);
699 qemu_mutex_unlock_iothread();
700
701 if (ret < 0) {
702 error_setg(errp, "Load VM's live state (ram) error");
703 return;
704 }
705
706 value = colo_receive_message_value(mis->from_src_file,
707 COLO_MESSAGE_VMSTATE_SIZE, &local_err);
708 if (local_err) {
709 error_propagate(errp, local_err);
710 return;
711 }
712
713 /*
714 * Read VM device state data into channel buffer,
715 * It's better to re-use the memory allocated.
716 * Here we need to handle the channel buffer directly.
717 */
718 if (value > bioc->capacity) {
719 bioc->capacity = value;
720 bioc->data = g_realloc(bioc->data, bioc->capacity);
721 }
722 total_size = qemu_get_buffer(mis->from_src_file, bioc->data, value);
723 if (total_size != value) {
724 error_setg(errp, "Got %" PRIu64 " VMState data, less than expected"
725 " %" PRIu64, total_size, value);
726 return;
727 }
728 bioc->usage = total_size;
729 qio_channel_io_seek(QIO_CHANNEL(bioc), 0, 0, NULL);
730
731 colo_send_message(mis->to_src_file, COLO_MESSAGE_VMSTATE_RECEIVED,
732 &local_err);
733 if (local_err) {
734 error_propagate(errp, local_err);
735 return;
736 }
737
738 qemu_mutex_lock_iothread();
739 vmstate_loading = true;
740 ret = qemu_load_device_state(fb);
741 if (ret < 0) {
742 error_setg(errp, "COLO: load device state failed");
743 qemu_mutex_unlock_iothread();
744 return;
745 }
746
747 #ifdef CONFIG_REPLICATION
748 replication_get_error_all(&local_err);
749 if (local_err) {
750 error_propagate(errp, local_err);
751 qemu_mutex_unlock_iothread();
752 return;
753 }
754
755 /* discard colo disk buffer */
756 replication_do_checkpoint_all(&local_err);
757 if (local_err) {
758 error_propagate(errp, local_err);
759 qemu_mutex_unlock_iothread();
760 return;
761 }
762 #else
763 abort();
764 #endif
765 /* Notify all filters of all NIC to do checkpoint */
766 colo_notify_filters_event(COLO_EVENT_CHECKPOINT, &local_err);
767
768 if (local_err) {
769 error_propagate(errp, local_err);
770 qemu_mutex_unlock_iothread();
771 return;
772 }
773
774 vmstate_loading = false;
775 vm_start();
776 trace_colo_vm_state_change("stop", "run");
777 qemu_mutex_unlock_iothread();
778
779 if (failover_get_state() == FAILOVER_STATUS_RELAUNCH) {
780 failover_set_state(FAILOVER_STATUS_RELAUNCH,
781 FAILOVER_STATUS_NONE);
782 failover_request_active(NULL);
783 return;
784 }
785
786 colo_send_message(mis->to_src_file, COLO_MESSAGE_VMSTATE_LOADED,
787 &local_err);
788 if (local_err) {
789 error_propagate(errp, local_err);
790 }
791 }
792
colo_wait_handle_message(MigrationIncomingState * mis,QEMUFile * fb,QIOChannelBuffer * bioc,Error ** errp)793 static void colo_wait_handle_message(MigrationIncomingState *mis,
794 QEMUFile *fb, QIOChannelBuffer *bioc, Error **errp)
795 {
796 COLOMessage msg;
797 Error *local_err = NULL;
798
799 msg = colo_receive_message(mis->from_src_file, &local_err);
800 if (local_err) {
801 error_propagate(errp, local_err);
802 return;
803 }
804
805 switch (msg) {
806 case COLO_MESSAGE_CHECKPOINT_REQUEST:
807 colo_incoming_process_checkpoint(mis, fb, bioc, errp);
808 break;
809 default:
810 error_setg(errp, "Got unknown COLO message: %d", msg);
811 break;
812 }
813 }
814
colo_process_incoming_thread(void * opaque)815 void *colo_process_incoming_thread(void *opaque)
816 {
817 MigrationIncomingState *mis = opaque;
818 QEMUFile *fb = NULL;
819 QIOChannelBuffer *bioc = NULL; /* Cache incoming device state */
820 Error *local_err = NULL;
821
822 rcu_register_thread();
823 qemu_sem_init(&mis->colo_incoming_sem, 0);
824
825 migrate_set_state(&mis->state, MIGRATION_STATUS_ACTIVE,
826 MIGRATION_STATUS_COLO);
827
828 last_colo_mode = get_colo_mode();
829 if (last_colo_mode != COLO_MODE_SECONDARY) {
830 error_report("COLO mode must be COLO_MODE_SECONDARY");
831 return NULL;
832 }
833
834 failover_init_state();
835
836 mis->to_src_file = qemu_file_get_return_path(mis->from_src_file);
837 if (!mis->to_src_file) {
838 error_report("COLO incoming thread: Open QEMUFile to_src_file failed");
839 goto out;
840 }
841 /*
842 * Note: the communication between Primary side and Secondary side
843 * should be sequential, we set the fd to unblocked in migration incoming
844 * coroutine, and here we are in the COLO incoming thread, so it is ok to
845 * set the fd back to blocked.
846 */
847 qemu_file_set_blocking(mis->from_src_file, true);
848
849 colo_incoming_start_dirty_log();
850
851 bioc = qio_channel_buffer_new(COLO_BUFFER_BASE_SIZE);
852 fb = qemu_fopen_channel_input(QIO_CHANNEL(bioc));
853 object_unref(OBJECT(bioc));
854
855 qemu_mutex_lock_iothread();
856 #ifdef CONFIG_REPLICATION
857 replication_start_all(REPLICATION_MODE_SECONDARY, &local_err);
858 if (local_err) {
859 qemu_mutex_unlock_iothread();
860 goto out;
861 }
862 #else
863 abort();
864 #endif
865 vm_start();
866 trace_colo_vm_state_change("stop", "run");
867 qemu_mutex_unlock_iothread();
868
869 colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_READY,
870 &local_err);
871 if (local_err) {
872 goto out;
873 }
874
875 while (mis->state == MIGRATION_STATUS_COLO) {
876 colo_wait_handle_message(mis, fb, bioc, &local_err);
877 if (local_err) {
878 error_report_err(local_err);
879 break;
880 }
881 if (failover_get_state() != FAILOVER_STATUS_NONE) {
882 error_report("failover request");
883 break;
884 }
885 }
886
887 out:
888 vmstate_loading = false;
889
890 /*
891 * There are only two reasons we can get here, some error happened
892 * or the user triggered failover.
893 */
894 switch (failover_get_state()) {
895 case FAILOVER_STATUS_COMPLETED:
896 qapi_event_send_colo_exit(COLO_MODE_SECONDARY,
897 COLO_EXIT_REASON_REQUEST);
898 break;
899 default:
900 qapi_event_send_colo_exit(COLO_MODE_SECONDARY,
901 COLO_EXIT_REASON_ERROR);
902 }
903
904 if (fb) {
905 qemu_fclose(fb);
906 }
907
908 /* Hope this not to be too long to loop here */
909 qemu_sem_wait(&mis->colo_incoming_sem);
910 qemu_sem_destroy(&mis->colo_incoming_sem);
911 /* Must be called after failover BH is completed */
912 if (mis->to_src_file) {
913 qemu_fclose(mis->to_src_file);
914 mis->to_src_file = NULL;
915 }
916
917 rcu_unregister_thread();
918 return NULL;
919 }
920