xref: /openbmc/qemu/migration/migration.c (revision cddafd8f353d2d251b1a5c6c948a577a85838582)
1 /*
2  * QEMU live migration
3  *
4  * Copyright IBM, Corp. 2008
5  *
6  * Authors:
7  *  Anthony Liguori   <aliguori@us.ibm.com>
8  *
9  * This work is licensed under the terms of the GNU GPL, version 2.  See
10  * the COPYING file in the top-level directory.
11  *
12  * Contributions after 2012-01-13 are licensed under the terms of the
13  * GNU GPL, version 2 or (at your option) any later version.
14  */
15 
16 #include "qemu/osdep.h"
17 #include "qemu/cutils.h"
18 #include "qemu/error-report.h"
19 #include "qemu/main-loop.h"
20 #include "migration/blocker.h"
21 #include "exec.h"
22 #include "fd.h"
23 #include "socket.h"
24 #include "rdma.h"
25 #include "ram.h"
26 #include "migration/global_state.h"
27 #include "migration/misc.h"
28 #include "migration.h"
29 #include "savevm.h"
30 #include "qemu-file-channel.h"
31 #include "qemu-file.h"
32 #include "migration/vmstate.h"
33 #include "sysemu/sysemu.h"
34 #include "block/block.h"
35 #include "qapi/qmp/qerror.h"
36 #include "qapi/util.h"
37 #include "qemu/sockets.h"
38 #include "qemu/rcu.h"
39 #include "block.h"
40 #include "postcopy-ram.h"
41 #include "qemu/thread.h"
42 #include "qmp-commands.h"
43 #include "trace.h"
44 #include "qapi-event.h"
45 #include "qom/cpu.h"
46 #include "exec/memory.h"
47 #include "exec/address-spaces.h"
48 #include "exec/target_page.h"
49 #include "io/channel-buffer.h"
50 #include "migration/colo.h"
51 
52 #define MAX_THROTTLE  (32 << 20)      /* Migration transfer speed throttling */
53 
54 /* Amount of time to allocate to each "chunk" of bandwidth-throttled
55  * data. */
56 #define BUFFER_DELAY     100
57 #define XFER_LIMIT_RATIO (1000 / BUFFER_DELAY)
58 
59 /* Time in milliseconds we are allowed to stop the source,
60  * for sending the last part */
61 #define DEFAULT_MIGRATE_SET_DOWNTIME 300
62 
63 /* Maximum migrate downtime set to 2000 seconds */
64 #define MAX_MIGRATE_DOWNTIME_SECONDS 2000
65 #define MAX_MIGRATE_DOWNTIME (MAX_MIGRATE_DOWNTIME_SECONDS * 1000)
66 
67 /* Default compression thread count */
68 #define DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT 8
69 /* Default decompression thread count, usually decompression is at
70  * least 4 times as fast as compression.*/
71 #define DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT 2
72 /*0: means nocompress, 1: best speed, ... 9: best compress ratio */
73 #define DEFAULT_MIGRATE_COMPRESS_LEVEL 1
74 /* Define default autoconverge cpu throttle migration parameters */
75 #define DEFAULT_MIGRATE_CPU_THROTTLE_INITIAL 20
76 #define DEFAULT_MIGRATE_CPU_THROTTLE_INCREMENT 10
77 
78 /* Migration XBZRLE default cache size */
79 #define DEFAULT_MIGRATE_CACHE_SIZE (64 * 1024 * 1024)
80 
81 /* The delay time (in ms) between two COLO checkpoints
82  * Note: Please change this default value to 10000 when we support hybrid mode.
83  */
84 #define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY 200
85 
86 static NotifierList migration_state_notifiers =
87     NOTIFIER_LIST_INITIALIZER(migration_state_notifiers);
88 
89 static bool deferred_incoming;
90 
91 /* Messages sent on the return path from destination to source */
92 enum mig_rp_message_type {
93     MIG_RP_MSG_INVALID = 0,  /* Must be 0 */
94     MIG_RP_MSG_SHUT,         /* sibling will not send any more RP messages */
95     MIG_RP_MSG_PONG,         /* Response to a PING; data (seq: be32 ) */
96 
97     MIG_RP_MSG_REQ_PAGES_ID, /* data (start: be64, len: be32, id: string) */
98     MIG_RP_MSG_REQ_PAGES,    /* data (start: be64, len: be32) */
99 
100     MIG_RP_MSG_MAX
101 };
102 
103 /* When we add fault tolerance, we could have several
104    migrations at once.  For now we don't need to add
105    dynamic creation of migration */
106 
107 /* For outgoing */
108 MigrationState *migrate_get_current(void)
109 {
110     static bool once;
111     static MigrationState current_migration = {
112         .state = MIGRATION_STATUS_NONE,
113         .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
114         .mbps = -1,
115         .parameters = {
116             .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
117             .compress_threads = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
118             .decompress_threads = DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
119             .cpu_throttle_initial = DEFAULT_MIGRATE_CPU_THROTTLE_INITIAL,
120             .cpu_throttle_increment = DEFAULT_MIGRATE_CPU_THROTTLE_INCREMENT,
121             .max_bandwidth = MAX_THROTTLE,
122             .downtime_limit = DEFAULT_MIGRATE_SET_DOWNTIME,
123             .x_checkpoint_delay = DEFAULT_MIGRATE_X_CHECKPOINT_DELAY,
124         },
125     };
126 
127     if (!once) {
128         current_migration.parameters.tls_creds = g_strdup("");
129         current_migration.parameters.tls_hostname = g_strdup("");
130         once = true;
131     }
132     return &current_migration;
133 }
134 
135 MigrationIncomingState *migration_incoming_get_current(void)
136 {
137     static bool once;
138     static MigrationIncomingState mis_current;
139 
140     if (!once) {
141         mis_current.state = MIGRATION_STATUS_NONE;
142         memset(&mis_current, 0, sizeof(MigrationIncomingState));
143         qemu_mutex_init(&mis_current.rp_mutex);
144         qemu_event_init(&mis_current.main_thread_load_event, false);
145         once = true;
146     }
147     return &mis_current;
148 }
149 
150 void migration_incoming_state_destroy(void)
151 {
152     struct MigrationIncomingState *mis = migration_incoming_get_current();
153 
154     if (mis->to_src_file) {
155         /* Tell source that we are done */
156         migrate_send_rp_shut(mis, qemu_file_get_error(mis->from_src_file) != 0);
157         qemu_fclose(mis->to_src_file);
158         mis->to_src_file = NULL;
159     }
160 
161     if (mis->from_src_file) {
162         qemu_fclose(mis->from_src_file);
163         mis->from_src_file = NULL;
164     }
165 
166     qemu_event_destroy(&mis->main_thread_load_event);
167 }
168 
169 static void migrate_generate_event(int new_state)
170 {
171     if (migrate_use_events()) {
172         qapi_event_send_migration(new_state, &error_abort);
173     }
174 }
175 
176 /*
177  * Called on -incoming with a defer: uri.
178  * The migration can be started later after any parameters have been
179  * changed.
180  */
181 static void deferred_incoming_migration(Error **errp)
182 {
183     if (deferred_incoming) {
184         error_setg(errp, "Incoming migration already deferred");
185     }
186     deferred_incoming = true;
187 }
188 
189 /*
190  * Send a message on the return channel back to the source
191  * of the migration.
192  */
193 static void migrate_send_rp_message(MigrationIncomingState *mis,
194                                     enum mig_rp_message_type message_type,
195                                     uint16_t len, void *data)
196 {
197     trace_migrate_send_rp_message((int)message_type, len);
198     qemu_mutex_lock(&mis->rp_mutex);
199     qemu_put_be16(mis->to_src_file, (unsigned int)message_type);
200     qemu_put_be16(mis->to_src_file, len);
201     qemu_put_buffer(mis->to_src_file, data, len);
202     qemu_fflush(mis->to_src_file);
203     qemu_mutex_unlock(&mis->rp_mutex);
204 }
205 
206 /* Request a range of pages from the source VM at the given
207  * start address.
208  *   rbname: Name of the RAMBlock to request the page in, if NULL it's the same
209  *           as the last request (a name must have been given previously)
210  *   Start: Address offset within the RB
211  *   Len: Length in bytes required - must be a multiple of pagesize
212  */
213 void migrate_send_rp_req_pages(MigrationIncomingState *mis, const char *rbname,
214                                ram_addr_t start, size_t len)
215 {
216     uint8_t bufc[12 + 1 + 255]; /* start (8), len (4), rbname up to 256 */
217     size_t msglen = 12; /* start + len */
218 
219     *(uint64_t *)bufc = cpu_to_be64((uint64_t)start);
220     *(uint32_t *)(bufc + 8) = cpu_to_be32((uint32_t)len);
221 
222     if (rbname) {
223         int rbname_len = strlen(rbname);
224         assert(rbname_len < 256);
225 
226         bufc[msglen++] = rbname_len;
227         memcpy(bufc + msglen, rbname, rbname_len);
228         msglen += rbname_len;
229         migrate_send_rp_message(mis, MIG_RP_MSG_REQ_PAGES_ID, msglen, bufc);
230     } else {
231         migrate_send_rp_message(mis, MIG_RP_MSG_REQ_PAGES, msglen, bufc);
232     }
233 }
234 
235 void qemu_start_incoming_migration(const char *uri, Error **errp)
236 {
237     const char *p;
238 
239     qapi_event_send_migration(MIGRATION_STATUS_SETUP, &error_abort);
240     if (!strcmp(uri, "defer")) {
241         deferred_incoming_migration(errp);
242     } else if (strstart(uri, "tcp:", &p)) {
243         tcp_start_incoming_migration(p, errp);
244 #ifdef CONFIG_RDMA
245     } else if (strstart(uri, "rdma:", &p)) {
246         rdma_start_incoming_migration(p, errp);
247 #endif
248     } else if (strstart(uri, "exec:", &p)) {
249         exec_start_incoming_migration(p, errp);
250     } else if (strstart(uri, "unix:", &p)) {
251         unix_start_incoming_migration(p, errp);
252     } else if (strstart(uri, "fd:", &p)) {
253         fd_start_incoming_migration(p, errp);
254     } else {
255         error_setg(errp, "unknown migration protocol: %s", uri);
256     }
257 }
258 
259 static void process_incoming_migration_bh(void *opaque)
260 {
261     Error *local_err = NULL;
262     MigrationIncomingState *mis = opaque;
263 
264     /* Make sure all file formats flush their mutable metadata.
265      * If we get an error here, just don't restart the VM yet. */
266     bdrv_invalidate_cache_all(&local_err);
267     if (local_err) {
268         error_report_err(local_err);
269         local_err = NULL;
270         autostart = false;
271     }
272 
273     /*
274      * This must happen after all error conditions are dealt with and
275      * we're sure the VM is going to be running on this host.
276      */
277     qemu_announce_self();
278 
279     /* If global state section was not received or we are in running
280        state, we need to obey autostart. Any other state is set with
281        runstate_set. */
282 
283     if (!global_state_received() ||
284         global_state_get_runstate() == RUN_STATE_RUNNING) {
285         if (autostart) {
286             vm_start();
287         } else {
288             runstate_set(RUN_STATE_PAUSED);
289         }
290     } else {
291         runstate_set(global_state_get_runstate());
292     }
293     migrate_decompress_threads_join();
294     /*
295      * This must happen after any state changes since as soon as an external
296      * observer sees this event they might start to prod at the VM assuming
297      * it's ready to use.
298      */
299     migrate_set_state(&mis->state, MIGRATION_STATUS_ACTIVE,
300                       MIGRATION_STATUS_COMPLETED);
301     qemu_bh_delete(mis->bh);
302     migration_incoming_state_destroy();
303 }
304 
305 static void process_incoming_migration_co(void *opaque)
306 {
307     QEMUFile *f = opaque;
308     MigrationIncomingState *mis = migration_incoming_get_current();
309     PostcopyState ps;
310     int ret;
311 
312     mis->from_src_file = f;
313     mis->largest_page_size = qemu_ram_pagesize_largest();
314     postcopy_state_set(POSTCOPY_INCOMING_NONE);
315     migrate_set_state(&mis->state, MIGRATION_STATUS_NONE,
316                       MIGRATION_STATUS_ACTIVE);
317     ret = qemu_loadvm_state(f);
318 
319     ps = postcopy_state_get();
320     trace_process_incoming_migration_co_end(ret, ps);
321     if (ps != POSTCOPY_INCOMING_NONE) {
322         if (ps == POSTCOPY_INCOMING_ADVISE) {
323             /*
324              * Where a migration had postcopy enabled (and thus went to advise)
325              * but managed to complete within the precopy period, we can use
326              * the normal exit.
327              */
328             postcopy_ram_incoming_cleanup(mis);
329         } else if (ret >= 0) {
330             /*
331              * Postcopy was started, cleanup should happen at the end of the
332              * postcopy thread.
333              */
334             trace_process_incoming_migration_co_postcopy_end_main();
335             return;
336         }
337         /* Else if something went wrong then just fall out of the normal exit */
338     }
339 
340     /* we get COLO info, and know if we are in COLO mode */
341     if (!ret && migration_incoming_enable_colo()) {
342         mis->migration_incoming_co = qemu_coroutine_self();
343         qemu_thread_create(&mis->colo_incoming_thread, "COLO incoming",
344              colo_process_incoming_thread, mis, QEMU_THREAD_JOINABLE);
345         mis->have_colo_incoming_thread = true;
346         qemu_coroutine_yield();
347 
348         /* Wait checkpoint incoming thread exit before free resource */
349         qemu_thread_join(&mis->colo_incoming_thread);
350     }
351 
352     if (ret < 0) {
353         migrate_set_state(&mis->state, MIGRATION_STATUS_ACTIVE,
354                           MIGRATION_STATUS_FAILED);
355         error_report("load of migration failed: %s", strerror(-ret));
356         migrate_decompress_threads_join();
357         exit(EXIT_FAILURE);
358     }
359 
360     free_xbzrle_decoded_buf();
361 
362     mis->bh = qemu_bh_new(process_incoming_migration_bh, mis);
363     qemu_bh_schedule(mis->bh);
364 }
365 
366 void migration_fd_process_incoming(QEMUFile *f)
367 {
368     Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, f);
369 
370     migrate_decompress_threads_create();
371     qemu_file_set_blocking(f, false);
372     qemu_coroutine_enter(co);
373 }
374 
375 /*
376  * Send a 'SHUT' message on the return channel with the given value
377  * to indicate that we've finished with the RP.  Non-0 value indicates
378  * error.
379  */
380 void migrate_send_rp_shut(MigrationIncomingState *mis,
381                           uint32_t value)
382 {
383     uint32_t buf;
384 
385     buf = cpu_to_be32(value);
386     migrate_send_rp_message(mis, MIG_RP_MSG_SHUT, sizeof(buf), &buf);
387 }
388 
389 /*
390  * Send a 'PONG' message on the return channel with the given value
391  * (normally in response to a 'PING')
392  */
393 void migrate_send_rp_pong(MigrationIncomingState *mis,
394                           uint32_t value)
395 {
396     uint32_t buf;
397 
398     buf = cpu_to_be32(value);
399     migrate_send_rp_message(mis, MIG_RP_MSG_PONG, sizeof(buf), &buf);
400 }
401 
402 MigrationCapabilityStatusList *qmp_query_migrate_capabilities(Error **errp)
403 {
404     MigrationCapabilityStatusList *head = NULL;
405     MigrationCapabilityStatusList *caps;
406     MigrationState *s = migrate_get_current();
407     int i;
408 
409     caps = NULL; /* silence compiler warning */
410     for (i = 0; i < MIGRATION_CAPABILITY__MAX; i++) {
411 #ifndef CONFIG_LIVE_BLOCK_MIGRATION
412         if (i == MIGRATION_CAPABILITY_BLOCK) {
413             continue;
414         }
415 #endif
416         if (i == MIGRATION_CAPABILITY_X_COLO && !colo_supported()) {
417             continue;
418         }
419         if (head == NULL) {
420             head = g_malloc0(sizeof(*caps));
421             caps = head;
422         } else {
423             caps->next = g_malloc0(sizeof(*caps));
424             caps = caps->next;
425         }
426         caps->value =
427             g_malloc(sizeof(*caps->value));
428         caps->value->capability = i;
429         caps->value->state = s->enabled_capabilities[i];
430     }
431 
432     return head;
433 }
434 
435 MigrationParameters *qmp_query_migrate_parameters(Error **errp)
436 {
437     MigrationParameters *params;
438     MigrationState *s = migrate_get_current();
439 
440     params = g_malloc0(sizeof(*params));
441     params->has_compress_level = true;
442     params->compress_level = s->parameters.compress_level;
443     params->has_compress_threads = true;
444     params->compress_threads = s->parameters.compress_threads;
445     params->has_decompress_threads = true;
446     params->decompress_threads = s->parameters.decompress_threads;
447     params->has_cpu_throttle_initial = true;
448     params->cpu_throttle_initial = s->parameters.cpu_throttle_initial;
449     params->has_cpu_throttle_increment = true;
450     params->cpu_throttle_increment = s->parameters.cpu_throttle_increment;
451     params->has_tls_creds = !!s->parameters.tls_creds;
452     params->tls_creds = g_strdup(s->parameters.tls_creds);
453     params->has_tls_hostname = !!s->parameters.tls_hostname;
454     params->tls_hostname = g_strdup(s->parameters.tls_hostname);
455     params->has_max_bandwidth = true;
456     params->max_bandwidth = s->parameters.max_bandwidth;
457     params->has_downtime_limit = true;
458     params->downtime_limit = s->parameters.downtime_limit;
459     params->has_x_checkpoint_delay = true;
460     params->x_checkpoint_delay = s->parameters.x_checkpoint_delay;
461     params->has_block_incremental = true;
462     params->block_incremental = s->parameters.block_incremental;
463 
464     return params;
465 }
466 
467 /*
468  * Return true if we're already in the middle of a migration
469  * (i.e. any of the active or setup states)
470  */
471 static bool migration_is_setup_or_active(int state)
472 {
473     switch (state) {
474     case MIGRATION_STATUS_ACTIVE:
475     case MIGRATION_STATUS_POSTCOPY_ACTIVE:
476     case MIGRATION_STATUS_SETUP:
477         return true;
478 
479     default:
480         return false;
481 
482     }
483 }
484 
485 static void populate_ram_info(MigrationInfo *info, MigrationState *s)
486 {
487     info->has_ram = true;
488     info->ram = g_malloc0(sizeof(*info->ram));
489     info->ram->transferred = ram_counters.transferred;
490     info->ram->total = ram_bytes_total();
491     info->ram->duplicate = ram_counters.duplicate;
492     /* legacy value.  It is not used anymore */
493     info->ram->skipped = 0;
494     info->ram->normal = ram_counters.normal;
495     info->ram->normal_bytes = ram_counters.normal *
496         qemu_target_page_size();
497     info->ram->mbps = s->mbps;
498     info->ram->dirty_sync_count = ram_counters.dirty_sync_count;
499     info->ram->postcopy_requests = ram_counters.postcopy_requests;
500     info->ram->page_size = qemu_target_page_size();
501 
502     if (migrate_use_xbzrle()) {
503         info->has_xbzrle_cache = true;
504         info->xbzrle_cache = g_malloc0(sizeof(*info->xbzrle_cache));
505         info->xbzrle_cache->cache_size = migrate_xbzrle_cache_size();
506         info->xbzrle_cache->bytes = xbzrle_counters.bytes;
507         info->xbzrle_cache->pages = xbzrle_counters.pages;
508         info->xbzrle_cache->cache_miss = xbzrle_counters.cache_miss;
509         info->xbzrle_cache->cache_miss_rate = xbzrle_counters.cache_miss_rate;
510         info->xbzrle_cache->overflow = xbzrle_counters.overflow;
511     }
512 
513     if (cpu_throttle_active()) {
514         info->has_cpu_throttle_percentage = true;
515         info->cpu_throttle_percentage = cpu_throttle_get_percentage();
516     }
517 
518     if (s->state != MIGRATION_STATUS_COMPLETED) {
519         info->ram->remaining = ram_bytes_remaining();
520         info->ram->dirty_pages_rate = ram_counters.dirty_pages_rate;
521     }
522 }
523 
524 static void populate_disk_info(MigrationInfo *info)
525 {
526     if (blk_mig_active()) {
527         info->has_disk = true;
528         info->disk = g_malloc0(sizeof(*info->disk));
529         info->disk->transferred = blk_mig_bytes_transferred();
530         info->disk->remaining = blk_mig_bytes_remaining();
531         info->disk->total = blk_mig_bytes_total();
532     }
533 }
534 
535 MigrationInfo *qmp_query_migrate(Error **errp)
536 {
537     MigrationInfo *info = g_malloc0(sizeof(*info));
538     MigrationState *s = migrate_get_current();
539 
540     switch (s->state) {
541     case MIGRATION_STATUS_NONE:
542         /* no migration has happened ever */
543         break;
544     case MIGRATION_STATUS_SETUP:
545         info->has_status = true;
546         info->has_total_time = false;
547         break;
548     case MIGRATION_STATUS_ACTIVE:
549     case MIGRATION_STATUS_CANCELLING:
550     case MIGRATION_STATUS_POSTCOPY_ACTIVE:
551          /* TODO add some postcopy stats */
552         info->has_status = true;
553         info->has_total_time = true;
554         info->total_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME)
555             - s->total_time;
556         info->has_expected_downtime = true;
557         info->expected_downtime = s->expected_downtime;
558         info->has_setup_time = true;
559         info->setup_time = s->setup_time;
560 
561         populate_ram_info(info, s);
562         populate_disk_info(info);
563         break;
564     case MIGRATION_STATUS_COLO:
565         info->has_status = true;
566         /* TODO: display COLO specific information (checkpoint info etc.) */
567         break;
568     case MIGRATION_STATUS_COMPLETED:
569         info->has_status = true;
570         info->has_total_time = true;
571         info->total_time = s->total_time;
572         info->has_downtime = true;
573         info->downtime = s->downtime;
574         info->has_setup_time = true;
575         info->setup_time = s->setup_time;
576 
577         populate_ram_info(info, s);
578         break;
579     case MIGRATION_STATUS_FAILED:
580         info->has_status = true;
581         if (s->error) {
582             info->has_error_desc = true;
583             info->error_desc = g_strdup(error_get_pretty(s->error));
584         }
585         break;
586     case MIGRATION_STATUS_CANCELLED:
587         info->has_status = true;
588         break;
589     }
590     info->status = s->state;
591 
592     return info;
593 }
594 
595 void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
596                                   Error **errp)
597 {
598     MigrationState *s = migrate_get_current();
599     MigrationCapabilityStatusList *cap;
600     bool old_postcopy_cap = migrate_postcopy_ram();
601 
602     if (migration_is_setup_or_active(s->state)) {
603         error_setg(errp, QERR_MIGRATION_ACTIVE);
604         return;
605     }
606 
607     for (cap = params; cap; cap = cap->next) {
608 #ifndef CONFIG_LIVE_BLOCK_MIGRATION
609         if (cap->value->capability == MIGRATION_CAPABILITY_BLOCK
610             && cap->value->state) {
611             error_setg(errp, "QEMU compiled without old-style (blk/-b, inc/-i) "
612                        "block migration");
613             error_append_hint(errp, "Use drive_mirror+NBD instead.\n");
614             continue;
615         }
616 #endif
617         if (cap->value->capability == MIGRATION_CAPABILITY_X_COLO) {
618             if (!colo_supported()) {
619                 error_setg(errp, "COLO is not currently supported, please"
620                              " configure with --enable-colo option in order to"
621                              " support COLO feature");
622                 continue;
623             }
624         }
625         s->enabled_capabilities[cap->value->capability] = cap->value->state;
626     }
627 
628     if (migrate_postcopy_ram()) {
629         if (migrate_use_compression()) {
630             /* The decompression threads asynchronously write into RAM
631              * rather than use the atomic copies needed to avoid
632              * userfaulting.  It should be possible to fix the decompression
633              * threads for compatibility in future.
634              */
635             error_report("Postcopy is not currently compatible with "
636                          "compression");
637             s->enabled_capabilities[MIGRATION_CAPABILITY_POSTCOPY_RAM] =
638                 false;
639         }
640         /* This check is reasonably expensive, so only when it's being
641          * set the first time, also it's only the destination that needs
642          * special support.
643          */
644         if (!old_postcopy_cap && runstate_check(RUN_STATE_INMIGRATE) &&
645             !postcopy_ram_supported_by_host()) {
646             /* postcopy_ram_supported_by_host will have emitted a more
647              * detailed message
648              */
649             error_report("Postcopy is not supported");
650             s->enabled_capabilities[MIGRATION_CAPABILITY_POSTCOPY_RAM] =
651                 false;
652         }
653     }
654 }
655 
656 void qmp_migrate_set_parameters(MigrationParameters *params, Error **errp)
657 {
658     MigrationState *s = migrate_get_current();
659 
660     if (params->has_compress_level &&
661         (params->compress_level < 0 || params->compress_level > 9)) {
662         error_setg(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
663                    "is invalid, it should be in the range of 0 to 9");
664         return;
665     }
666     if (params->has_compress_threads &&
667         (params->compress_threads < 1 || params->compress_threads > 255)) {
668         error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
669                    "compress_threads",
670                    "is invalid, it should be in the range of 1 to 255");
671         return;
672     }
673     if (params->has_decompress_threads &&
674         (params->decompress_threads < 1 || params->decompress_threads > 255)) {
675         error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
676                    "decompress_threads",
677                    "is invalid, it should be in the range of 1 to 255");
678         return;
679     }
680     if (params->has_cpu_throttle_initial &&
681         (params->cpu_throttle_initial < 1 ||
682          params->cpu_throttle_initial > 99)) {
683         error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
684                    "cpu_throttle_initial",
685                    "an integer in the range of 1 to 99");
686         return;
687     }
688     if (params->has_cpu_throttle_increment &&
689         (params->cpu_throttle_increment < 1 ||
690          params->cpu_throttle_increment > 99)) {
691         error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
692                    "cpu_throttle_increment",
693                    "an integer in the range of 1 to 99");
694         return;
695     }
696     if (params->has_max_bandwidth &&
697         (params->max_bandwidth < 0 || params->max_bandwidth > SIZE_MAX)) {
698         error_setg(errp, "Parameter 'max_bandwidth' expects an integer in the"
699                          " range of 0 to %zu bytes/second", SIZE_MAX);
700         return;
701     }
702     if (params->has_downtime_limit &&
703         (params->downtime_limit < 0 ||
704          params->downtime_limit > MAX_MIGRATE_DOWNTIME)) {
705         error_setg(errp, "Parameter 'downtime_limit' expects an integer in "
706                          "the range of 0 to %d milliseconds",
707                          MAX_MIGRATE_DOWNTIME);
708         return;
709     }
710     if (params->has_x_checkpoint_delay && (params->x_checkpoint_delay < 0)) {
711         error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
712                     "x_checkpoint_delay",
713                     "is invalid, it should be positive");
714     }
715 
716     if (params->has_compress_level) {
717         s->parameters.compress_level = params->compress_level;
718     }
719     if (params->has_compress_threads) {
720         s->parameters.compress_threads = params->compress_threads;
721     }
722     if (params->has_decompress_threads) {
723         s->parameters.decompress_threads = params->decompress_threads;
724     }
725     if (params->has_cpu_throttle_initial) {
726         s->parameters.cpu_throttle_initial = params->cpu_throttle_initial;
727     }
728     if (params->has_cpu_throttle_increment) {
729         s->parameters.cpu_throttle_increment = params->cpu_throttle_increment;
730     }
731     if (params->has_tls_creds) {
732         g_free(s->parameters.tls_creds);
733         s->parameters.tls_creds = g_strdup(params->tls_creds);
734     }
735     if (params->has_tls_hostname) {
736         g_free(s->parameters.tls_hostname);
737         s->parameters.tls_hostname = g_strdup(params->tls_hostname);
738     }
739     if (params->has_max_bandwidth) {
740         s->parameters.max_bandwidth = params->max_bandwidth;
741         if (s->to_dst_file) {
742             qemu_file_set_rate_limit(s->to_dst_file,
743                                 s->parameters.max_bandwidth / XFER_LIMIT_RATIO);
744         }
745     }
746     if (params->has_downtime_limit) {
747         s->parameters.downtime_limit = params->downtime_limit;
748     }
749 
750     if (params->has_x_checkpoint_delay) {
751         s->parameters.x_checkpoint_delay = params->x_checkpoint_delay;
752         if (migration_in_colo_state()) {
753             colo_checkpoint_notify(s);
754         }
755     }
756     if (params->has_block_incremental) {
757         s->parameters.block_incremental = params->block_incremental;
758     }
759 }
760 
761 
762 void qmp_migrate_start_postcopy(Error **errp)
763 {
764     MigrationState *s = migrate_get_current();
765 
766     if (!migrate_postcopy_ram()) {
767         error_setg(errp, "Enable postcopy with migrate_set_capability before"
768                          " the start of migration");
769         return;
770     }
771 
772     if (s->state == MIGRATION_STATUS_NONE) {
773         error_setg(errp, "Postcopy must be started after migration has been"
774                          " started");
775         return;
776     }
777     /*
778      * we don't error if migration has finished since that would be racy
779      * with issuing this command.
780      */
781     atomic_set(&s->start_postcopy, true);
782 }
783 
784 /* shared migration helpers */
785 
786 void migrate_set_state(int *state, int old_state, int new_state)
787 {
788     if (atomic_cmpxchg(state, old_state, new_state) == old_state) {
789         trace_migrate_set_state(new_state);
790         migrate_generate_event(new_state);
791     }
792 }
793 
794 void migrate_set_block_enabled(bool value, Error **errp)
795 {
796     MigrationCapabilityStatusList *cap;
797 
798     cap = g_new0(MigrationCapabilityStatusList, 1);
799     cap->value = g_new0(MigrationCapabilityStatus, 1);
800     cap->value->capability = MIGRATION_CAPABILITY_BLOCK;
801     cap->value->state = value;
802     qmp_migrate_set_capabilities(cap, errp);
803     qapi_free_MigrationCapabilityStatusList(cap);
804 }
805 
806 static void migrate_set_block_incremental(MigrationState *s, bool value)
807 {
808     s->parameters.block_incremental = value;
809 }
810 
811 static void block_cleanup_parameters(MigrationState *s)
812 {
813     if (s->must_remove_block_options) {
814         /* setting to false can never fail */
815         migrate_set_block_enabled(false, &error_abort);
816         migrate_set_block_incremental(s, false);
817         s->must_remove_block_options = false;
818     }
819 }
820 
821 static void migrate_fd_cleanup(void *opaque)
822 {
823     MigrationState *s = opaque;
824 
825     qemu_bh_delete(s->cleanup_bh);
826     s->cleanup_bh = NULL;
827 
828     if (s->to_dst_file) {
829         trace_migrate_fd_cleanup();
830         qemu_mutex_unlock_iothread();
831         if (s->migration_thread_running) {
832             qemu_thread_join(&s->thread);
833             s->migration_thread_running = false;
834         }
835         qemu_mutex_lock_iothread();
836 
837         migrate_compress_threads_join();
838         qemu_fclose(s->to_dst_file);
839         s->to_dst_file = NULL;
840     }
841 
842     assert((s->state != MIGRATION_STATUS_ACTIVE) &&
843            (s->state != MIGRATION_STATUS_POSTCOPY_ACTIVE));
844 
845     if (s->state == MIGRATION_STATUS_CANCELLING) {
846         migrate_set_state(&s->state, MIGRATION_STATUS_CANCELLING,
847                           MIGRATION_STATUS_CANCELLED);
848     }
849 
850     notifier_list_notify(&migration_state_notifiers, s);
851     block_cleanup_parameters(s);
852 }
853 
854 void migrate_fd_error(MigrationState *s, const Error *error)
855 {
856     trace_migrate_fd_error(error_get_pretty(error));
857     assert(s->to_dst_file == NULL);
858     migrate_set_state(&s->state, MIGRATION_STATUS_SETUP,
859                       MIGRATION_STATUS_FAILED);
860     if (!s->error) {
861         s->error = error_copy(error);
862     }
863     notifier_list_notify(&migration_state_notifiers, s);
864     block_cleanup_parameters(s);
865 }
866 
867 static void migrate_fd_cancel(MigrationState *s)
868 {
869     int old_state ;
870     QEMUFile *f = migrate_get_current()->to_dst_file;
871     trace_migrate_fd_cancel();
872 
873     if (s->rp_state.from_dst_file) {
874         /* shutdown the rp socket, so causing the rp thread to shutdown */
875         qemu_file_shutdown(s->rp_state.from_dst_file);
876     }
877 
878     do {
879         old_state = s->state;
880         if (!migration_is_setup_or_active(old_state)) {
881             break;
882         }
883         migrate_set_state(&s->state, old_state, MIGRATION_STATUS_CANCELLING);
884     } while (s->state != MIGRATION_STATUS_CANCELLING);
885 
886     /*
887      * If we're unlucky the migration code might be stuck somewhere in a
888      * send/write while the network has failed and is waiting to timeout;
889      * if we've got shutdown(2) available then we can force it to quit.
890      * The outgoing qemu file gets closed in migrate_fd_cleanup that is
891      * called in a bh, so there is no race against this cancel.
892      */
893     if (s->state == MIGRATION_STATUS_CANCELLING && f) {
894         qemu_file_shutdown(f);
895     }
896     if (s->state == MIGRATION_STATUS_CANCELLING && s->block_inactive) {
897         Error *local_err = NULL;
898 
899         bdrv_invalidate_cache_all(&local_err);
900         if (local_err) {
901             error_report_err(local_err);
902         } else {
903             s->block_inactive = false;
904         }
905     }
906     block_cleanup_parameters(s);
907 }
908 
909 void add_migration_state_change_notifier(Notifier *notify)
910 {
911     notifier_list_add(&migration_state_notifiers, notify);
912 }
913 
914 void remove_migration_state_change_notifier(Notifier *notify)
915 {
916     notifier_remove(notify);
917 }
918 
919 bool migration_in_setup(MigrationState *s)
920 {
921     return s->state == MIGRATION_STATUS_SETUP;
922 }
923 
924 bool migration_has_finished(MigrationState *s)
925 {
926     return s->state == MIGRATION_STATUS_COMPLETED;
927 }
928 
929 bool migration_has_failed(MigrationState *s)
930 {
931     return (s->state == MIGRATION_STATUS_CANCELLED ||
932             s->state == MIGRATION_STATUS_FAILED);
933 }
934 
935 bool migration_in_postcopy(void)
936 {
937     MigrationState *s = migrate_get_current();
938 
939     return (s->state == MIGRATION_STATUS_POSTCOPY_ACTIVE);
940 }
941 
942 bool migration_in_postcopy_after_devices(MigrationState *s)
943 {
944     return migration_in_postcopy() && s->postcopy_after_devices;
945 }
946 
947 bool migration_is_idle(void)
948 {
949     MigrationState *s = migrate_get_current();
950 
951     switch (s->state) {
952     case MIGRATION_STATUS_NONE:
953     case MIGRATION_STATUS_CANCELLED:
954     case MIGRATION_STATUS_COMPLETED:
955     case MIGRATION_STATUS_FAILED:
956         return true;
957     case MIGRATION_STATUS_SETUP:
958     case MIGRATION_STATUS_CANCELLING:
959     case MIGRATION_STATUS_ACTIVE:
960     case MIGRATION_STATUS_POSTCOPY_ACTIVE:
961     case MIGRATION_STATUS_COLO:
962         return false;
963     case MIGRATION_STATUS__MAX:
964         g_assert_not_reached();
965     }
966 
967     return false;
968 }
969 
970 MigrationState *migrate_init(void)
971 {
972     MigrationState *s = migrate_get_current();
973 
974     /*
975      * Reinitialise all migration state, except
976      * parameters/capabilities that the user set, and
977      * locks.
978      */
979     s->bytes_xfer = 0;
980     s->xfer_limit = 0;
981     s->cleanup_bh = 0;
982     s->to_dst_file = NULL;
983     s->state = MIGRATION_STATUS_NONE;
984     s->rp_state.from_dst_file = NULL;
985     s->rp_state.error = false;
986     s->mbps = 0.0;
987     s->downtime = 0;
988     s->expected_downtime = 0;
989     s->setup_time = 0;
990     s->start_postcopy = false;
991     s->postcopy_after_devices = false;
992     s->migration_thread_running = false;
993     error_free(s->error);
994     s->error = NULL;
995 
996     migrate_set_state(&s->state, MIGRATION_STATUS_NONE, MIGRATION_STATUS_SETUP);
997 
998     s->total_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
999     return s;
1000 }
1001 
1002 static GSList *migration_blockers;
1003 
1004 int migrate_add_blocker(Error *reason, Error **errp)
1005 {
1006     if (only_migratable) {
1007         error_propagate(errp, error_copy(reason));
1008         error_prepend(errp, "disallowing migration blocker "
1009                           "(--only_migratable) for: ");
1010         return -EACCES;
1011     }
1012 
1013     if (migration_is_idle()) {
1014         migration_blockers = g_slist_prepend(migration_blockers, reason);
1015         return 0;
1016     }
1017 
1018     error_propagate(errp, error_copy(reason));
1019     error_prepend(errp, "disallowing migration blocker (migration in "
1020                       "progress) for: ");
1021     return -EBUSY;
1022 }
1023 
1024 void migrate_del_blocker(Error *reason)
1025 {
1026     migration_blockers = g_slist_remove(migration_blockers, reason);
1027 }
1028 
1029 void qmp_migrate_incoming(const char *uri, Error **errp)
1030 {
1031     Error *local_err = NULL;
1032     static bool once = true;
1033 
1034     if (!deferred_incoming) {
1035         error_setg(errp, "For use with '-incoming defer'");
1036         return;
1037     }
1038     if (!once) {
1039         error_setg(errp, "The incoming migration has already been started");
1040     }
1041 
1042     qemu_start_incoming_migration(uri, &local_err);
1043 
1044     if (local_err) {
1045         error_propagate(errp, local_err);
1046         return;
1047     }
1048 
1049     once = false;
1050 }
1051 
1052 bool migration_is_blocked(Error **errp)
1053 {
1054     if (qemu_savevm_state_blocked(errp)) {
1055         return true;
1056     }
1057 
1058     if (migration_blockers) {
1059         error_propagate(errp, error_copy(migration_blockers->data));
1060         return true;
1061     }
1062 
1063     return false;
1064 }
1065 
1066 void qmp_migrate(const char *uri, bool has_blk, bool blk,
1067                  bool has_inc, bool inc, bool has_detach, bool detach,
1068                  Error **errp)
1069 {
1070     Error *local_err = NULL;
1071     MigrationState *s = migrate_get_current();
1072     const char *p;
1073 
1074     if (migration_is_setup_or_active(s->state) ||
1075         s->state == MIGRATION_STATUS_CANCELLING ||
1076         s->state == MIGRATION_STATUS_COLO) {
1077         error_setg(errp, QERR_MIGRATION_ACTIVE);
1078         return;
1079     }
1080     if (runstate_check(RUN_STATE_INMIGRATE)) {
1081         error_setg(errp, "Guest is waiting for an incoming migration");
1082         return;
1083     }
1084 
1085     if (migration_is_blocked(errp)) {
1086         return;
1087     }
1088 
1089     if ((has_blk && blk) || (has_inc && inc)) {
1090         if (migrate_use_block() || migrate_use_block_incremental()) {
1091             error_setg(errp, "Command options are incompatible with "
1092                        "current migration capabilities");
1093             return;
1094         }
1095         migrate_set_block_enabled(true, &local_err);
1096         if (local_err) {
1097             error_propagate(errp, local_err);
1098             return;
1099         }
1100         s->must_remove_block_options = true;
1101     }
1102 
1103     if (has_inc && inc) {
1104         migrate_set_block_incremental(s, true);
1105     }
1106 
1107     s = migrate_init();
1108 
1109     if (strstart(uri, "tcp:", &p)) {
1110         tcp_start_outgoing_migration(s, p, &local_err);
1111 #ifdef CONFIG_RDMA
1112     } else if (strstart(uri, "rdma:", &p)) {
1113         rdma_start_outgoing_migration(s, p, &local_err);
1114 #endif
1115     } else if (strstart(uri, "exec:", &p)) {
1116         exec_start_outgoing_migration(s, p, &local_err);
1117     } else if (strstart(uri, "unix:", &p)) {
1118         unix_start_outgoing_migration(s, p, &local_err);
1119     } else if (strstart(uri, "fd:", &p)) {
1120         fd_start_outgoing_migration(s, p, &local_err);
1121     } else {
1122         error_setg(errp, QERR_INVALID_PARAMETER_VALUE, "uri",
1123                    "a valid migration protocol");
1124         migrate_set_state(&s->state, MIGRATION_STATUS_SETUP,
1125                           MIGRATION_STATUS_FAILED);
1126         return;
1127     }
1128 
1129     if (local_err) {
1130         migrate_fd_error(s, local_err);
1131         error_propagate(errp, local_err);
1132         return;
1133     }
1134 }
1135 
1136 void qmp_migrate_cancel(Error **errp)
1137 {
1138     migrate_fd_cancel(migrate_get_current());
1139 }
1140 
1141 void qmp_migrate_set_cache_size(int64_t value, Error **errp)
1142 {
1143     MigrationState *s = migrate_get_current();
1144     int64_t new_size;
1145 
1146     /* Check for truncation */
1147     if (value != (size_t)value) {
1148         error_setg(errp, QERR_INVALID_PARAMETER_VALUE, "cache size",
1149                    "exceeding address space");
1150         return;
1151     }
1152 
1153     /* Cache should not be larger than guest ram size */
1154     if (value > ram_bytes_total()) {
1155         error_setg(errp, QERR_INVALID_PARAMETER_VALUE, "cache size",
1156                    "exceeds guest ram size ");
1157         return;
1158     }
1159 
1160     new_size = xbzrle_cache_resize(value);
1161     if (new_size < 0) {
1162         error_setg(errp, QERR_INVALID_PARAMETER_VALUE, "cache size",
1163                    "is smaller than page size");
1164         return;
1165     }
1166 
1167     s->xbzrle_cache_size = new_size;
1168 }
1169 
1170 int64_t qmp_query_migrate_cache_size(Error **errp)
1171 {
1172     return migrate_xbzrle_cache_size();
1173 }
1174 
1175 void qmp_migrate_set_speed(int64_t value, Error **errp)
1176 {
1177     MigrationParameters p = {
1178         .has_max_bandwidth = true,
1179         .max_bandwidth = value,
1180     };
1181 
1182     qmp_migrate_set_parameters(&p, errp);
1183 }
1184 
1185 void qmp_migrate_set_downtime(double value, Error **errp)
1186 {
1187     if (value < 0 || value > MAX_MIGRATE_DOWNTIME_SECONDS) {
1188         error_setg(errp, "Parameter 'downtime_limit' expects an integer in "
1189                          "the range of 0 to %d seconds",
1190                          MAX_MIGRATE_DOWNTIME_SECONDS);
1191         return;
1192     }
1193 
1194     value *= 1000; /* Convert to milliseconds */
1195     value = MAX(0, MIN(INT64_MAX, value));
1196 
1197     MigrationParameters p = {
1198         .has_downtime_limit = true,
1199         .downtime_limit = value,
1200     };
1201 
1202     qmp_migrate_set_parameters(&p, errp);
1203 }
1204 
1205 bool migrate_release_ram(void)
1206 {
1207     MigrationState *s;
1208 
1209     s = migrate_get_current();
1210 
1211     return s->enabled_capabilities[MIGRATION_CAPABILITY_RELEASE_RAM];
1212 }
1213 
1214 bool migrate_postcopy_ram(void)
1215 {
1216     MigrationState *s;
1217 
1218     s = migrate_get_current();
1219 
1220     return s->enabled_capabilities[MIGRATION_CAPABILITY_POSTCOPY_RAM];
1221 }
1222 
1223 bool migrate_auto_converge(void)
1224 {
1225     MigrationState *s;
1226 
1227     s = migrate_get_current();
1228 
1229     return s->enabled_capabilities[MIGRATION_CAPABILITY_AUTO_CONVERGE];
1230 }
1231 
1232 bool migrate_zero_blocks(void)
1233 {
1234     MigrationState *s;
1235 
1236     s = migrate_get_current();
1237 
1238     return s->enabled_capabilities[MIGRATION_CAPABILITY_ZERO_BLOCKS];
1239 }
1240 
1241 bool migrate_use_compression(void)
1242 {
1243     MigrationState *s;
1244 
1245     s = migrate_get_current();
1246 
1247     return s->enabled_capabilities[MIGRATION_CAPABILITY_COMPRESS];
1248 }
1249 
1250 int migrate_compress_level(void)
1251 {
1252     MigrationState *s;
1253 
1254     s = migrate_get_current();
1255 
1256     return s->parameters.compress_level;
1257 }
1258 
1259 int migrate_compress_threads(void)
1260 {
1261     MigrationState *s;
1262 
1263     s = migrate_get_current();
1264 
1265     return s->parameters.compress_threads;
1266 }
1267 
1268 int migrate_decompress_threads(void)
1269 {
1270     MigrationState *s;
1271 
1272     s = migrate_get_current();
1273 
1274     return s->parameters.decompress_threads;
1275 }
1276 
1277 bool migrate_use_events(void)
1278 {
1279     MigrationState *s;
1280 
1281     s = migrate_get_current();
1282 
1283     return s->enabled_capabilities[MIGRATION_CAPABILITY_EVENTS];
1284 }
1285 
1286 int migrate_use_xbzrle(void)
1287 {
1288     MigrationState *s;
1289 
1290     s = migrate_get_current();
1291 
1292     return s->enabled_capabilities[MIGRATION_CAPABILITY_XBZRLE];
1293 }
1294 
1295 int64_t migrate_xbzrle_cache_size(void)
1296 {
1297     MigrationState *s;
1298 
1299     s = migrate_get_current();
1300 
1301     return s->xbzrle_cache_size;
1302 }
1303 
1304 bool migrate_use_block(void)
1305 {
1306     MigrationState *s;
1307 
1308     s = migrate_get_current();
1309 
1310     return s->enabled_capabilities[MIGRATION_CAPABILITY_BLOCK];
1311 }
1312 
1313 bool migrate_use_block_incremental(void)
1314 {
1315     MigrationState *s;
1316 
1317     s = migrate_get_current();
1318 
1319     return s->parameters.block_incremental;
1320 }
1321 
1322 /* migration thread support */
1323 /*
1324  * Something bad happened to the RP stream, mark an error
1325  * The caller shall print or trace something to indicate why
1326  */
1327 static void mark_source_rp_bad(MigrationState *s)
1328 {
1329     s->rp_state.error = true;
1330 }
1331 
1332 static struct rp_cmd_args {
1333     ssize_t     len; /* -1 = variable */
1334     const char *name;
1335 } rp_cmd_args[] = {
1336     [MIG_RP_MSG_INVALID]        = { .len = -1, .name = "INVALID" },
1337     [MIG_RP_MSG_SHUT]           = { .len =  4, .name = "SHUT" },
1338     [MIG_RP_MSG_PONG]           = { .len =  4, .name = "PONG" },
1339     [MIG_RP_MSG_REQ_PAGES]      = { .len = 12, .name = "REQ_PAGES" },
1340     [MIG_RP_MSG_REQ_PAGES_ID]   = { .len = -1, .name = "REQ_PAGES_ID" },
1341     [MIG_RP_MSG_MAX]            = { .len = -1, .name = "MAX" },
1342 };
1343 
1344 /*
1345  * Process a request for pages received on the return path,
1346  * We're allowed to send more than requested (e.g. to round to our page size)
1347  * and we don't need to send pages that have already been sent.
1348  */
1349 static void migrate_handle_rp_req_pages(MigrationState *ms, const char* rbname,
1350                                        ram_addr_t start, size_t len)
1351 {
1352     long our_host_ps = getpagesize();
1353 
1354     trace_migrate_handle_rp_req_pages(rbname, start, len);
1355 
1356     /*
1357      * Since we currently insist on matching page sizes, just sanity check
1358      * we're being asked for whole host pages.
1359      */
1360     if (start & (our_host_ps-1) ||
1361        (len & (our_host_ps-1))) {
1362         error_report("%s: Misaligned page request, start: " RAM_ADDR_FMT
1363                      " len: %zd", __func__, start, len);
1364         mark_source_rp_bad(ms);
1365         return;
1366     }
1367 
1368     if (ram_save_queue_pages(rbname, start, len)) {
1369         mark_source_rp_bad(ms);
1370     }
1371 }
1372 
1373 /*
1374  * Handles messages sent on the return path towards the source VM
1375  *
1376  */
1377 static void *source_return_path_thread(void *opaque)
1378 {
1379     MigrationState *ms = opaque;
1380     QEMUFile *rp = ms->rp_state.from_dst_file;
1381     uint16_t header_len, header_type;
1382     uint8_t buf[512];
1383     uint32_t tmp32, sibling_error;
1384     ram_addr_t start = 0; /* =0 to silence warning */
1385     size_t  len = 0, expected_len;
1386     int res;
1387 
1388     trace_source_return_path_thread_entry();
1389     while (!ms->rp_state.error && !qemu_file_get_error(rp) &&
1390            migration_is_setup_or_active(ms->state)) {
1391         trace_source_return_path_thread_loop_top();
1392         header_type = qemu_get_be16(rp);
1393         header_len = qemu_get_be16(rp);
1394 
1395         if (header_type >= MIG_RP_MSG_MAX ||
1396             header_type == MIG_RP_MSG_INVALID) {
1397             error_report("RP: Received invalid message 0x%04x length 0x%04x",
1398                     header_type, header_len);
1399             mark_source_rp_bad(ms);
1400             goto out;
1401         }
1402 
1403         if ((rp_cmd_args[header_type].len != -1 &&
1404             header_len != rp_cmd_args[header_type].len) ||
1405             header_len > sizeof(buf)) {
1406             error_report("RP: Received '%s' message (0x%04x) with"
1407                     "incorrect length %d expecting %zu",
1408                     rp_cmd_args[header_type].name, header_type, header_len,
1409                     (size_t)rp_cmd_args[header_type].len);
1410             mark_source_rp_bad(ms);
1411             goto out;
1412         }
1413 
1414         /* We know we've got a valid header by this point */
1415         res = qemu_get_buffer(rp, buf, header_len);
1416         if (res != header_len) {
1417             error_report("RP: Failed reading data for message 0x%04x"
1418                          " read %d expected %d",
1419                          header_type, res, header_len);
1420             mark_source_rp_bad(ms);
1421             goto out;
1422         }
1423 
1424         /* OK, we have the message and the data */
1425         switch (header_type) {
1426         case MIG_RP_MSG_SHUT:
1427             sibling_error = ldl_be_p(buf);
1428             trace_source_return_path_thread_shut(sibling_error);
1429             if (sibling_error) {
1430                 error_report("RP: Sibling indicated error %d", sibling_error);
1431                 mark_source_rp_bad(ms);
1432             }
1433             /*
1434              * We'll let the main thread deal with closing the RP
1435              * we could do a shutdown(2) on it, but we're the only user
1436              * anyway, so there's nothing gained.
1437              */
1438             goto out;
1439 
1440         case MIG_RP_MSG_PONG:
1441             tmp32 = ldl_be_p(buf);
1442             trace_source_return_path_thread_pong(tmp32);
1443             break;
1444 
1445         case MIG_RP_MSG_REQ_PAGES:
1446             start = ldq_be_p(buf);
1447             len = ldl_be_p(buf + 8);
1448             migrate_handle_rp_req_pages(ms, NULL, start, len);
1449             break;
1450 
1451         case MIG_RP_MSG_REQ_PAGES_ID:
1452             expected_len = 12 + 1; /* header + termination */
1453 
1454             if (header_len >= expected_len) {
1455                 start = ldq_be_p(buf);
1456                 len = ldl_be_p(buf + 8);
1457                 /* Now we expect an idstr */
1458                 tmp32 = buf[12]; /* Length of the following idstr */
1459                 buf[13 + tmp32] = '\0';
1460                 expected_len += tmp32;
1461             }
1462             if (header_len != expected_len) {
1463                 error_report("RP: Req_Page_id with length %d expecting %zd",
1464                         header_len, expected_len);
1465                 mark_source_rp_bad(ms);
1466                 goto out;
1467             }
1468             migrate_handle_rp_req_pages(ms, (char *)&buf[13], start, len);
1469             break;
1470 
1471         default:
1472             break;
1473         }
1474     }
1475     if (qemu_file_get_error(rp)) {
1476         trace_source_return_path_thread_bad_end();
1477         mark_source_rp_bad(ms);
1478     }
1479 
1480     trace_source_return_path_thread_end();
1481 out:
1482     ms->rp_state.from_dst_file = NULL;
1483     qemu_fclose(rp);
1484     return NULL;
1485 }
1486 
1487 static int open_return_path_on_source(MigrationState *ms)
1488 {
1489 
1490     ms->rp_state.from_dst_file = qemu_file_get_return_path(ms->to_dst_file);
1491     if (!ms->rp_state.from_dst_file) {
1492         return -1;
1493     }
1494 
1495     trace_open_return_path_on_source();
1496     qemu_thread_create(&ms->rp_state.rp_thread, "return path",
1497                        source_return_path_thread, ms, QEMU_THREAD_JOINABLE);
1498 
1499     trace_open_return_path_on_source_continue();
1500 
1501     return 0;
1502 }
1503 
1504 /* Returns 0 if the RP was ok, otherwise there was an error on the RP */
1505 static int await_return_path_close_on_source(MigrationState *ms)
1506 {
1507     /*
1508      * If this is a normal exit then the destination will send a SHUT and the
1509      * rp_thread will exit, however if there's an error we need to cause
1510      * it to exit.
1511      */
1512     if (qemu_file_get_error(ms->to_dst_file) && ms->rp_state.from_dst_file) {
1513         /*
1514          * shutdown(2), if we have it, will cause it to unblock if it's stuck
1515          * waiting for the destination.
1516          */
1517         qemu_file_shutdown(ms->rp_state.from_dst_file);
1518         mark_source_rp_bad(ms);
1519     }
1520     trace_await_return_path_close_on_source_joining();
1521     qemu_thread_join(&ms->rp_state.rp_thread);
1522     trace_await_return_path_close_on_source_close();
1523     return ms->rp_state.error;
1524 }
1525 
1526 /*
1527  * Switch from normal iteration to postcopy
1528  * Returns non-0 on error
1529  */
1530 static int postcopy_start(MigrationState *ms, bool *old_vm_running)
1531 {
1532     int ret;
1533     QIOChannelBuffer *bioc;
1534     QEMUFile *fb;
1535     int64_t time_at_stop = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
1536     bool restart_block = false;
1537     migrate_set_state(&ms->state, MIGRATION_STATUS_ACTIVE,
1538                       MIGRATION_STATUS_POSTCOPY_ACTIVE);
1539 
1540     trace_postcopy_start();
1541     qemu_mutex_lock_iothread();
1542     trace_postcopy_start_set_run();
1543 
1544     qemu_system_wakeup_request(QEMU_WAKEUP_REASON_OTHER);
1545     *old_vm_running = runstate_is_running();
1546     global_state_store();
1547     ret = vm_stop_force_state(RUN_STATE_FINISH_MIGRATE);
1548     if (ret < 0) {
1549         goto fail;
1550     }
1551 
1552     ret = bdrv_inactivate_all();
1553     if (ret < 0) {
1554         goto fail;
1555     }
1556     restart_block = true;
1557 
1558     /*
1559      * Cause any non-postcopiable, but iterative devices to
1560      * send out their final data.
1561      */
1562     qemu_savevm_state_complete_precopy(ms->to_dst_file, true);
1563 
1564     /*
1565      * in Finish migrate and with the io-lock held everything should
1566      * be quiet, but we've potentially still got dirty pages and we
1567      * need to tell the destination to throw any pages it's already received
1568      * that are dirty
1569      */
1570     if (ram_postcopy_send_discard_bitmap(ms)) {
1571         error_report("postcopy send discard bitmap failed");
1572         goto fail;
1573     }
1574 
1575     /*
1576      * send rest of state - note things that are doing postcopy
1577      * will notice we're in POSTCOPY_ACTIVE and not actually
1578      * wrap their state up here
1579      */
1580     qemu_file_set_rate_limit(ms->to_dst_file, INT64_MAX);
1581     /* Ping just for debugging, helps line traces up */
1582     qemu_savevm_send_ping(ms->to_dst_file, 2);
1583 
1584     /*
1585      * While loading the device state we may trigger page transfer
1586      * requests and the fd must be free to process those, and thus
1587      * the destination must read the whole device state off the fd before
1588      * it starts processing it.  Unfortunately the ad-hoc migration format
1589      * doesn't allow the destination to know the size to read without fully
1590      * parsing it through each devices load-state code (especially the open
1591      * coded devices that use get/put).
1592      * So we wrap the device state up in a package with a length at the start;
1593      * to do this we use a qemu_buf to hold the whole of the device state.
1594      */
1595     bioc = qio_channel_buffer_new(4096);
1596     qio_channel_set_name(QIO_CHANNEL(bioc), "migration-postcopy-buffer");
1597     fb = qemu_fopen_channel_output(QIO_CHANNEL(bioc));
1598     object_unref(OBJECT(bioc));
1599 
1600     /*
1601      * Make sure the receiver can get incoming pages before we send the rest
1602      * of the state
1603      */
1604     qemu_savevm_send_postcopy_listen(fb);
1605 
1606     qemu_savevm_state_complete_precopy(fb, false);
1607     qemu_savevm_send_ping(fb, 3);
1608 
1609     qemu_savevm_send_postcopy_run(fb);
1610 
1611     /* <><> end of stuff going into the package */
1612 
1613     /* Last point of recovery; as soon as we send the package the destination
1614      * can open devices and potentially start running.
1615      * Lets just check again we've not got any errors.
1616      */
1617     ret = qemu_file_get_error(ms->to_dst_file);
1618     if (ret) {
1619         error_report("postcopy_start: Migration stream errored (pre package)");
1620         goto fail_closefb;
1621     }
1622 
1623     restart_block = false;
1624 
1625     /* Now send that blob */
1626     if (qemu_savevm_send_packaged(ms->to_dst_file, bioc->data, bioc->usage)) {
1627         goto fail_closefb;
1628     }
1629     qemu_fclose(fb);
1630 
1631     /* Send a notify to give a chance for anything that needs to happen
1632      * at the transition to postcopy and after the device state; in particular
1633      * spice needs to trigger a transition now
1634      */
1635     ms->postcopy_after_devices = true;
1636     notifier_list_notify(&migration_state_notifiers, ms);
1637 
1638     ms->downtime =  qemu_clock_get_ms(QEMU_CLOCK_REALTIME) - time_at_stop;
1639 
1640     qemu_mutex_unlock_iothread();
1641 
1642     /*
1643      * Although this ping is just for debug, it could potentially be
1644      * used for getting a better measurement of downtime at the source.
1645      */
1646     qemu_savevm_send_ping(ms->to_dst_file, 4);
1647 
1648     if (migrate_release_ram()) {
1649         ram_postcopy_migrated_memory_release(ms);
1650     }
1651 
1652     ret = qemu_file_get_error(ms->to_dst_file);
1653     if (ret) {
1654         error_report("postcopy_start: Migration stream errored");
1655         migrate_set_state(&ms->state, MIGRATION_STATUS_POSTCOPY_ACTIVE,
1656                               MIGRATION_STATUS_FAILED);
1657     }
1658 
1659     return ret;
1660 
1661 fail_closefb:
1662     qemu_fclose(fb);
1663 fail:
1664     migrate_set_state(&ms->state, MIGRATION_STATUS_POSTCOPY_ACTIVE,
1665                           MIGRATION_STATUS_FAILED);
1666     if (restart_block) {
1667         /* A failure happened early enough that we know the destination hasn't
1668          * accessed block devices, so we're safe to recover.
1669          */
1670         Error *local_err = NULL;
1671 
1672         bdrv_invalidate_cache_all(&local_err);
1673         if (local_err) {
1674             error_report_err(local_err);
1675         }
1676     }
1677     qemu_mutex_unlock_iothread();
1678     return -1;
1679 }
1680 
1681 /**
1682  * migration_completion: Used by migration_thread when there's not much left.
1683  *   The caller 'breaks' the loop when this returns.
1684  *
1685  * @s: Current migration state
1686  * @current_active_state: The migration state we expect to be in
1687  * @*old_vm_running: Pointer to old_vm_running flag
1688  * @*start_time: Pointer to time to update
1689  */
1690 static void migration_completion(MigrationState *s, int current_active_state,
1691                                  bool *old_vm_running,
1692                                  int64_t *start_time)
1693 {
1694     int ret;
1695 
1696     if (s->state == MIGRATION_STATUS_ACTIVE) {
1697         qemu_mutex_lock_iothread();
1698         *start_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
1699         qemu_system_wakeup_request(QEMU_WAKEUP_REASON_OTHER);
1700         *old_vm_running = runstate_is_running();
1701         ret = global_state_store();
1702 
1703         if (!ret) {
1704             ret = vm_stop_force_state(RUN_STATE_FINISH_MIGRATE);
1705             if (ret >= 0) {
1706                 qemu_file_set_rate_limit(s->to_dst_file, INT64_MAX);
1707                 qemu_savevm_state_complete_precopy(s->to_dst_file, false);
1708             }
1709             /*
1710              * Don't mark the image with BDRV_O_INACTIVE flag if
1711              * we will go into COLO stage later.
1712              */
1713             if (ret >= 0 && !migrate_colo_enabled()) {
1714                 ret = bdrv_inactivate_all();
1715                 if (ret >= 0) {
1716                     s->block_inactive = true;
1717                 }
1718             }
1719         }
1720         qemu_mutex_unlock_iothread();
1721 
1722         if (ret < 0) {
1723             goto fail;
1724         }
1725     } else if (s->state == MIGRATION_STATUS_POSTCOPY_ACTIVE) {
1726         trace_migration_completion_postcopy_end();
1727 
1728         qemu_savevm_state_complete_postcopy(s->to_dst_file);
1729         trace_migration_completion_postcopy_end_after_complete();
1730     }
1731 
1732     /*
1733      * If rp was opened we must clean up the thread before
1734      * cleaning everything else up (since if there are no failures
1735      * it will wait for the destination to send it's status in
1736      * a SHUT command).
1737      */
1738     if (s->rp_state.from_dst_file) {
1739         int rp_error;
1740         trace_migration_return_path_end_before();
1741         rp_error = await_return_path_close_on_source(s);
1742         trace_migration_return_path_end_after(rp_error);
1743         if (rp_error) {
1744             goto fail_invalidate;
1745         }
1746     }
1747 
1748     if (qemu_file_get_error(s->to_dst_file)) {
1749         trace_migration_completion_file_err();
1750         goto fail_invalidate;
1751     }
1752 
1753     if (!migrate_colo_enabled()) {
1754         migrate_set_state(&s->state, current_active_state,
1755                           MIGRATION_STATUS_COMPLETED);
1756     }
1757 
1758     return;
1759 
1760 fail_invalidate:
1761     /* If not doing postcopy, vm_start() will be called: let's regain
1762      * control on images.
1763      */
1764     if (s->state == MIGRATION_STATUS_ACTIVE) {
1765         Error *local_err = NULL;
1766 
1767         qemu_mutex_lock_iothread();
1768         bdrv_invalidate_cache_all(&local_err);
1769         if (local_err) {
1770             error_report_err(local_err);
1771         } else {
1772             s->block_inactive = false;
1773         }
1774         qemu_mutex_unlock_iothread();
1775     }
1776 
1777 fail:
1778     migrate_set_state(&s->state, current_active_state,
1779                       MIGRATION_STATUS_FAILED);
1780 }
1781 
1782 bool migrate_colo_enabled(void)
1783 {
1784     MigrationState *s = migrate_get_current();
1785     return s->enabled_capabilities[MIGRATION_CAPABILITY_X_COLO];
1786 }
1787 
1788 /*
1789  * Master migration thread on the source VM.
1790  * It drives the migration and pumps the data down the outgoing channel.
1791  */
1792 static void *migration_thread(void *opaque)
1793 {
1794     MigrationState *s = opaque;
1795     /* Used by the bandwidth calcs, updated later */
1796     int64_t initial_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
1797     int64_t setup_start = qemu_clock_get_ms(QEMU_CLOCK_HOST);
1798     int64_t initial_bytes = 0;
1799     /*
1800      * The final stage happens when the remaining data is smaller than
1801      * this threshold; it's calculated from the requested downtime and
1802      * measured bandwidth
1803      */
1804     int64_t threshold_size = 0;
1805     int64_t start_time = initial_time;
1806     int64_t end_time;
1807     bool old_vm_running = false;
1808     bool entered_postcopy = false;
1809     /* The active state we expect to be in; ACTIVE or POSTCOPY_ACTIVE */
1810     enum MigrationStatus current_active_state = MIGRATION_STATUS_ACTIVE;
1811     bool enable_colo = migrate_colo_enabled();
1812 
1813     rcu_register_thread();
1814 
1815     qemu_savevm_state_header(s->to_dst_file);
1816 
1817     if (s->to_dst_file) {
1818         /* Now tell the dest that it should open its end so it can reply */
1819         qemu_savevm_send_open_return_path(s->to_dst_file);
1820 
1821         /* And do a ping that will make stuff easier to debug */
1822         qemu_savevm_send_ping(s->to_dst_file, 1);
1823     }
1824 
1825     if (migrate_postcopy_ram()) {
1826         /*
1827          * Tell the destination that we *might* want to do postcopy later;
1828          * if the other end can't do postcopy it should fail now, nice and
1829          * early.
1830          */
1831         qemu_savevm_send_postcopy_advise(s->to_dst_file);
1832     }
1833 
1834     qemu_savevm_state_begin(s->to_dst_file);
1835 
1836     s->setup_time = qemu_clock_get_ms(QEMU_CLOCK_HOST) - setup_start;
1837     migrate_set_state(&s->state, MIGRATION_STATUS_SETUP,
1838                       MIGRATION_STATUS_ACTIVE);
1839 
1840     trace_migration_thread_setup_complete();
1841 
1842     while (s->state == MIGRATION_STATUS_ACTIVE ||
1843            s->state == MIGRATION_STATUS_POSTCOPY_ACTIVE) {
1844         int64_t current_time;
1845         uint64_t pending_size;
1846 
1847         if (!qemu_file_rate_limit(s->to_dst_file)) {
1848             uint64_t pend_post, pend_nonpost;
1849 
1850             qemu_savevm_state_pending(s->to_dst_file, threshold_size,
1851                                       &pend_nonpost, &pend_post);
1852             pending_size = pend_nonpost + pend_post;
1853             trace_migrate_pending(pending_size, threshold_size,
1854                                   pend_post, pend_nonpost);
1855             if (pending_size && pending_size >= threshold_size) {
1856                 /* Still a significant amount to transfer */
1857 
1858                 if (migrate_postcopy_ram() &&
1859                     s->state != MIGRATION_STATUS_POSTCOPY_ACTIVE &&
1860                     pend_nonpost <= threshold_size &&
1861                     atomic_read(&s->start_postcopy)) {
1862 
1863                     if (!postcopy_start(s, &old_vm_running)) {
1864                         current_active_state = MIGRATION_STATUS_POSTCOPY_ACTIVE;
1865                         entered_postcopy = true;
1866                     }
1867 
1868                     continue;
1869                 }
1870                 /* Just another iteration step */
1871                 qemu_savevm_state_iterate(s->to_dst_file, entered_postcopy);
1872             } else {
1873                 trace_migration_thread_low_pending(pending_size);
1874                 migration_completion(s, current_active_state,
1875                                      &old_vm_running, &start_time);
1876                 break;
1877             }
1878         }
1879 
1880         if (qemu_file_get_error(s->to_dst_file)) {
1881             migrate_set_state(&s->state, current_active_state,
1882                               MIGRATION_STATUS_FAILED);
1883             trace_migration_thread_file_err();
1884             break;
1885         }
1886         current_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
1887         if (current_time >= initial_time + BUFFER_DELAY) {
1888             uint64_t transferred_bytes = qemu_ftell(s->to_dst_file) -
1889                                          initial_bytes;
1890             uint64_t time_spent = current_time - initial_time;
1891             double bandwidth = (double)transferred_bytes / time_spent;
1892             threshold_size = bandwidth * s->parameters.downtime_limit;
1893 
1894             s->mbps = (((double) transferred_bytes * 8.0) /
1895                     ((double) time_spent / 1000.0)) / 1000.0 / 1000.0;
1896 
1897             trace_migrate_transferred(transferred_bytes, time_spent,
1898                                       bandwidth, threshold_size);
1899             /* if we haven't sent anything, we don't want to recalculate
1900                10000 is a small enough number for our purposes */
1901             if (ram_counters.dirty_pages_rate && transferred_bytes > 10000) {
1902                 s->expected_downtime = ram_counters.dirty_pages_rate *
1903                     qemu_target_page_size() / bandwidth;
1904             }
1905 
1906             qemu_file_reset_rate_limit(s->to_dst_file);
1907             initial_time = current_time;
1908             initial_bytes = qemu_ftell(s->to_dst_file);
1909         }
1910         if (qemu_file_rate_limit(s->to_dst_file)) {
1911             /* usleep expects microseconds */
1912             g_usleep((initial_time + BUFFER_DELAY - current_time)*1000);
1913         }
1914     }
1915 
1916     trace_migration_thread_after_loop();
1917     /* If we enabled cpu throttling for auto-converge, turn it off. */
1918     cpu_throttle_stop();
1919     end_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
1920 
1921     qemu_mutex_lock_iothread();
1922     /*
1923      * The resource has been allocated by migration will be reused in COLO
1924      * process, so don't release them.
1925      */
1926     if (!enable_colo) {
1927         qemu_savevm_state_cleanup();
1928     }
1929     if (s->state == MIGRATION_STATUS_COMPLETED) {
1930         uint64_t transferred_bytes = qemu_ftell(s->to_dst_file);
1931         s->total_time = end_time - s->total_time;
1932         if (!entered_postcopy) {
1933             s->downtime = end_time - start_time;
1934         }
1935         if (s->total_time) {
1936             s->mbps = (((double) transferred_bytes * 8.0) /
1937                        ((double) s->total_time)) / 1000;
1938         }
1939         runstate_set(RUN_STATE_POSTMIGRATE);
1940     } else {
1941         if (s->state == MIGRATION_STATUS_ACTIVE && enable_colo) {
1942             migrate_start_colo_process(s);
1943             qemu_savevm_state_cleanup();
1944             /*
1945             * Fixme: we will run VM in COLO no matter its old running state.
1946             * After exited COLO, we will keep running.
1947             */
1948             old_vm_running = true;
1949         }
1950         if (old_vm_running && !entered_postcopy) {
1951             vm_start();
1952         } else {
1953             if (runstate_check(RUN_STATE_FINISH_MIGRATE)) {
1954                 runstate_set(RUN_STATE_POSTMIGRATE);
1955             }
1956         }
1957     }
1958     qemu_bh_schedule(s->cleanup_bh);
1959     qemu_mutex_unlock_iothread();
1960 
1961     rcu_unregister_thread();
1962     return NULL;
1963 }
1964 
1965 void migrate_fd_connect(MigrationState *s)
1966 {
1967     s->expected_downtime = s->parameters.downtime_limit;
1968     s->cleanup_bh = qemu_bh_new(migrate_fd_cleanup, s);
1969 
1970     qemu_file_set_blocking(s->to_dst_file, true);
1971     qemu_file_set_rate_limit(s->to_dst_file,
1972                              s->parameters.max_bandwidth / XFER_LIMIT_RATIO);
1973 
1974     /* Notify before starting migration thread */
1975     notifier_list_notify(&migration_state_notifiers, s);
1976 
1977     /*
1978      * Open the return path; currently for postcopy but other things might
1979      * also want it.
1980      */
1981     if (migrate_postcopy_ram()) {
1982         if (open_return_path_on_source(s)) {
1983             error_report("Unable to open return-path for postcopy");
1984             migrate_set_state(&s->state, MIGRATION_STATUS_SETUP,
1985                               MIGRATION_STATUS_FAILED);
1986             migrate_fd_cleanup(s);
1987             return;
1988         }
1989     }
1990 
1991     migrate_compress_threads_create();
1992     qemu_thread_create(&s->thread, "live_migration", migration_thread, s,
1993                        QEMU_THREAD_JOINABLE);
1994     s->migration_thread_running = true;
1995 }
1996 
1997