4 * Copyright (c) 2019-2020 Red Hat Inc
7 * Juan Quintela <quintela@redhat.com>
9 * This work is licensed under the terms of the GNU GPL, version 2 or later.
10 * See the COPYING file in the top-level directory.
13 #include "qemu/osdep.h"
15 #include "exec/target_page.h"
16 #include "sysemu/sysemu.h"
17 #include "exec/ramblock.h"
18 #include "qemu/error-report.h"
19 #include "qapi/error.h"
21 #include "migration.h"
24 #include "qemu-file.h"
27 #include "threadinfo.h"
29 #include "qemu/yank.h"
30 #include "io/channel-socket.h"
31 #include "yank_functions.h"
35 #define MULTIFD_MAGIC 0x11223344U
36 #define MULTIFD_VERSION 1
41 unsigned char uuid[16]; /* QemuUUID */
43 uint8_t unused1[7]; /* Reserved for future use */
44 uint64_t unused2[4]; /* Reserved for future use */
45 } __attribute__((packed)) MultiFDInit_t;
47 /* Multifd without compression */
50 * nocomp_send_setup: setup send side
52 * For no compression this function does nothing.
54 * Returns 0 for success or -1 for error
56 * @p: Params for the channel that we are using
57 * @errp: pointer to an error
59 static int nocomp_send_setup(MultiFDSendParams *p, Error **errp)
65 * nocomp_send_cleanup: cleanup send side
67 * For no compression this function does nothing.
69 * @p: Params for the channel that we are using
70 * @errp: pointer to an error
72 static void nocomp_send_cleanup(MultiFDSendParams *p, Error **errp)
78 * nocomp_send_prepare: prepare date to be able to send
80 * For no compression we just have to calculate the size of the
83 * Returns 0 for success or -1 for error
85 * @p: Params for the channel that we are using
86 * @errp: pointer to an error
88 static int nocomp_send_prepare(MultiFDSendParams *p, Error **errp)
90 MultiFDPages_t *pages = p->pages;
92 for (int i = 0; i < p->normal_num; i++) {
93 p->iov[p->iovs_num].iov_base = pages->block->host + p->normal[i];
94 p->iov[p->iovs_num].iov_len = p->page_size;
98 p->next_packet_size = p->normal_num * p->page_size;
99 p->flags |= MULTIFD_FLAG_NOCOMP;
104 * nocomp_recv_setup: setup receive side
106 * For no compression this function does nothing.
108 * Returns 0 for success or -1 for error
110 * @p: Params for the channel that we are using
111 * @errp: pointer to an error
113 static int nocomp_recv_setup(MultiFDRecvParams *p, Error **errp)
119 * nocomp_recv_cleanup: setup receive side
121 * For no compression this function does nothing.
123 * @p: Params for the channel that we are using
125 static void nocomp_recv_cleanup(MultiFDRecvParams *p)
130 * nocomp_recv_pages: read the data from the channel into actual pages
132 * For no compression we just need to read things into the correct place.
134 * Returns 0 for success or -1 for error
136 * @p: Params for the channel that we are using
137 * @errp: pointer to an error
139 static int nocomp_recv_pages(MultiFDRecvParams *p, Error **errp)
141 uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
143 if (flags != MULTIFD_FLAG_NOCOMP) {
144 error_setg(errp, "multifd %u: flags received %x flags expected %x",
145 p->id, flags, MULTIFD_FLAG_NOCOMP);
148 for (int i = 0; i < p->normal_num; i++) {
149 p->iov[i].iov_base = p->host + p->normal[i];
150 p->iov[i].iov_len = p->page_size;
152 return qio_channel_readv_all(p->c, p->iov, p->normal_num, errp);
155 static MultiFDMethods multifd_nocomp_ops = {
156 .send_setup = nocomp_send_setup,
157 .send_cleanup = nocomp_send_cleanup,
158 .send_prepare = nocomp_send_prepare,
159 .recv_setup = nocomp_recv_setup,
160 .recv_cleanup = nocomp_recv_cleanup,
161 .recv_pages = nocomp_recv_pages
164 static MultiFDMethods *multifd_ops[MULTIFD_COMPRESSION__MAX] = {
165 [MULTIFD_COMPRESSION_NONE] = &multifd_nocomp_ops,
168 void multifd_register_ops(int method, MultiFDMethods *ops)
170 assert(0 < method && method < MULTIFD_COMPRESSION__MAX);
171 multifd_ops[method] = ops;
174 static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
176 MultiFDInit_t msg = {};
179 msg.magic = cpu_to_be32(MULTIFD_MAGIC);
180 msg.version = cpu_to_be32(MULTIFD_VERSION);
182 memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
184 ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), errp);
191 static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
196 ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
201 msg.magic = be32_to_cpu(msg.magic);
202 msg.version = be32_to_cpu(msg.version);
204 if (msg.magic != MULTIFD_MAGIC) {
205 error_setg(errp, "multifd: received packet magic %x "
206 "expected %x", msg.magic, MULTIFD_MAGIC);
210 if (msg.version != MULTIFD_VERSION) {
211 error_setg(errp, "multifd: received packet version %u "
212 "expected %u", msg.version, MULTIFD_VERSION);
216 if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
217 char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
218 char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
220 error_setg(errp, "multifd: received uuid '%s' and expected "
221 "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
227 if (msg.id > migrate_multifd_channels()) {
228 error_setg(errp, "multifd: received channel version %u "
229 "expected %u", msg.version, MULTIFD_VERSION);
236 static MultiFDPages_t *multifd_pages_init(size_t size)
238 MultiFDPages_t *pages = g_new0(MultiFDPages_t, 1);
240 pages->allocated = size;
241 pages->offset = g_new0(ram_addr_t, size);
246 static void multifd_pages_clear(MultiFDPages_t *pages)
249 pages->allocated = 0;
250 pages->packet_num = 0;
252 g_free(pages->offset);
253 pages->offset = NULL;
257 static void multifd_send_fill_packet(MultiFDSendParams *p)
259 MultiFDPacket_t *packet = p->packet;
262 packet->flags = cpu_to_be32(p->flags);
263 packet->pages_alloc = cpu_to_be32(p->pages->allocated);
264 packet->normal_pages = cpu_to_be32(p->normal_num);
265 packet->next_packet_size = cpu_to_be32(p->next_packet_size);
266 packet->packet_num = cpu_to_be64(p->packet_num);
268 if (p->pages->block) {
269 strncpy(packet->ramblock, p->pages->block->idstr, 256);
272 for (i = 0; i < p->normal_num; i++) {
273 /* there are architectures where ram_addr_t is 32 bit */
274 uint64_t temp = p->normal[i];
276 packet->offset[i] = cpu_to_be64(temp);
280 static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
282 MultiFDPacket_t *packet = p->packet;
286 packet->magic = be32_to_cpu(packet->magic);
287 if (packet->magic != MULTIFD_MAGIC) {
288 error_setg(errp, "multifd: received packet "
289 "magic %x and expected magic %x",
290 packet->magic, MULTIFD_MAGIC);
294 packet->version = be32_to_cpu(packet->version);
295 if (packet->version != MULTIFD_VERSION) {
296 error_setg(errp, "multifd: received packet "
297 "version %u and expected version %u",
298 packet->version, MULTIFD_VERSION);
302 p->flags = be32_to_cpu(packet->flags);
304 packet->pages_alloc = be32_to_cpu(packet->pages_alloc);
306 * If we received a packet that is 100 times bigger than expected
307 * just stop migration. It is a magic number.
309 if (packet->pages_alloc > p->page_count) {
310 error_setg(errp, "multifd: received packet "
311 "with size %u and expected a size of %u",
312 packet->pages_alloc, p->page_count) ;
316 p->normal_num = be32_to_cpu(packet->normal_pages);
317 if (p->normal_num > packet->pages_alloc) {
318 error_setg(errp, "multifd: received packet "
319 "with %u pages and expected maximum pages are %u",
320 p->normal_num, packet->pages_alloc) ;
324 p->next_packet_size = be32_to_cpu(packet->next_packet_size);
325 p->packet_num = be64_to_cpu(packet->packet_num);
327 if (p->normal_num == 0) {
331 /* make sure that ramblock is 0 terminated */
332 packet->ramblock[255] = 0;
333 block = qemu_ram_block_by_name(packet->ramblock);
335 error_setg(errp, "multifd: unknown ram block %s",
340 p->host = block->host;
341 for (i = 0; i < p->normal_num; i++) {
342 uint64_t offset = be64_to_cpu(packet->offset[i]);
344 if (offset > (block->used_length - p->page_size)) {
345 error_setg(errp, "multifd: offset too long %" PRIu64
346 " (max " RAM_ADDR_FMT ")",
347 offset, block->used_length);
350 p->normal[i] = offset;
357 MultiFDSendParams *params;
358 /* array of pages to sent */
359 MultiFDPages_t *pages;
360 /* global number of generated multifd packets */
362 /* send channels ready */
363 QemuSemaphore channels_ready;
365 * Have we already run terminate threads. There is a race when it
366 * happens that we got one error while we are exiting.
367 * We will use atomic operations. Only valid values are 0 and 1.
372 } *multifd_send_state;
375 * How we use multifd_send_state->pages and channel->pages?
377 * We create a pages for each channel, and a main one. Each time that
378 * we need to send a batch of pages we interchange the ones between
379 * multifd_send_state and the channel that is sending it. There are
380 * two reasons for that:
381 * - to not have to do so many mallocs during migration
382 * - to make easier to know what to free at the end of migration
384 * This way we always know who is the owner of each "pages" struct,
385 * and we don't need any locking. It belongs to the migration thread
386 * or to the channel thread. Switching is safe because the migration
387 * thread is using the channel mutex when changing it, and the channel
388 * have to had finish with its own, otherwise pending_job can't be
392 static int multifd_send_pages(QEMUFile *f)
395 static int next_channel;
396 MultiFDSendParams *p = NULL; /* make happy gcc */
397 MultiFDPages_t *pages = multifd_send_state->pages;
398 uint64_t transferred;
400 if (qatomic_read(&multifd_send_state->exiting)) {
404 qemu_sem_wait(&multifd_send_state->channels_ready);
406 * next_channel can remain from a previous migration that was
407 * using more channels, so ensure it doesn't overflow if the
408 * limit is lower now.
410 next_channel %= migrate_multifd_channels();
411 for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
412 p = &multifd_send_state->params[i];
414 qemu_mutex_lock(&p->mutex);
416 error_report("%s: channel %d has already quit!", __func__, i);
417 qemu_mutex_unlock(&p->mutex);
420 if (!p->pending_job) {
422 next_channel = (i + 1) % migrate_multifd_channels();
425 qemu_mutex_unlock(&p->mutex);
427 assert(!p->pages->num);
428 assert(!p->pages->block);
430 p->packet_num = multifd_send_state->packet_num++;
431 multifd_send_state->pages = p->pages;
433 transferred = ((uint64_t) pages->num) * p->page_size + p->packet_len;
434 qemu_file_acct_rate_limit(f, transferred);
435 qemu_mutex_unlock(&p->mutex);
436 stat64_add(&ram_counters.transferred, transferred);
437 stat64_add(&ram_counters.multifd_bytes, transferred);
438 qemu_sem_post(&p->sem);
443 int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
445 MultiFDPages_t *pages = multifd_send_state->pages;
446 bool changed = false;
449 pages->block = block;
452 if (pages->block == block) {
453 pages->offset[pages->num] = offset;
456 if (pages->num < pages->allocated) {
463 if (multifd_send_pages(f) < 0) {
468 return multifd_queue_page(f, block, offset);
474 static void multifd_send_terminate_threads(Error *err)
478 trace_multifd_send_terminate_threads(err != NULL);
481 MigrationState *s = migrate_get_current();
482 migrate_set_error(s, err);
483 if (s->state == MIGRATION_STATUS_SETUP ||
484 s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
485 s->state == MIGRATION_STATUS_DEVICE ||
486 s->state == MIGRATION_STATUS_ACTIVE) {
487 migrate_set_state(&s->state, s->state,
488 MIGRATION_STATUS_FAILED);
493 * We don't want to exit each threads twice. Depending on where
494 * we get the error, or if there are two independent errors in two
495 * threads at the same time, we can end calling this function
498 if (qatomic_xchg(&multifd_send_state->exiting, 1)) {
502 for (i = 0; i < migrate_multifd_channels(); i++) {
503 MultiFDSendParams *p = &multifd_send_state->params[i];
505 qemu_mutex_lock(&p->mutex);
507 qemu_sem_post(&p->sem);
509 qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
511 qemu_mutex_unlock(&p->mutex);
515 void multifd_save_cleanup(void)
519 if (!migrate_multifd()) {
522 multifd_send_terminate_threads(NULL);
523 for (i = 0; i < migrate_multifd_channels(); i++) {
524 MultiFDSendParams *p = &multifd_send_state->params[i];
527 qemu_thread_join(&p->thread);
530 for (i = 0; i < migrate_multifd_channels(); i++) {
531 MultiFDSendParams *p = &multifd_send_state->params[i];
532 Error *local_err = NULL;
534 if (p->registered_yank) {
535 migration_ioc_unregister_yank(p->c);
537 socket_send_channel_destroy(p->c);
539 qemu_mutex_destroy(&p->mutex);
540 qemu_sem_destroy(&p->sem);
541 qemu_sem_destroy(&p->sem_sync);
544 multifd_pages_clear(p->pages);
553 multifd_send_state->ops->send_cleanup(p, &local_err);
555 migrate_set_error(migrate_get_current(), local_err);
556 error_free(local_err);
559 qemu_sem_destroy(&multifd_send_state->channels_ready);
560 g_free(multifd_send_state->params);
561 multifd_send_state->params = NULL;
562 multifd_pages_clear(multifd_send_state->pages);
563 multifd_send_state->pages = NULL;
564 g_free(multifd_send_state);
565 multifd_send_state = NULL;
568 static int multifd_zero_copy_flush(QIOChannel *c)
573 ret = qio_channel_flush(c, &err);
575 error_report_err(err);
579 stat64_add(&ram_counters.dirty_sync_missed_zero_copy, 1);
585 int multifd_send_sync_main(QEMUFile *f)
588 bool flush_zero_copy;
590 if (!migrate_multifd()) {
593 if (multifd_send_state->pages->num) {
594 if (multifd_send_pages(f) < 0) {
595 error_report("%s: multifd_send_pages fail", __func__);
601 * When using zero-copy, it's necessary to flush the pages before any of
602 * the pages can be sent again, so we'll make sure the new version of the
603 * pages will always arrive _later_ than the old pages.
605 * Currently we achieve this by flushing the zero-page requested writes
606 * per ram iteration, but in the future we could potentially optimize it
607 * to be less frequent, e.g. only after we finished one whole scanning of
608 * all the dirty bitmaps.
611 flush_zero_copy = migrate_zero_copy_send();
613 for (i = 0; i < migrate_multifd_channels(); i++) {
614 MultiFDSendParams *p = &multifd_send_state->params[i];
616 trace_multifd_send_sync_main_signal(p->id);
618 qemu_mutex_lock(&p->mutex);
621 error_report("%s: channel %d has already quit", __func__, i);
622 qemu_mutex_unlock(&p->mutex);
626 p->packet_num = multifd_send_state->packet_num++;
627 p->flags |= MULTIFD_FLAG_SYNC;
629 qemu_file_acct_rate_limit(f, p->packet_len);
630 qemu_mutex_unlock(&p->mutex);
631 stat64_add(&ram_counters.transferred, p->packet_len);
632 stat64_add(&ram_counters.multifd_bytes, p->packet_len);
633 qemu_sem_post(&p->sem);
635 for (i = 0; i < migrate_multifd_channels(); i++) {
636 MultiFDSendParams *p = &multifd_send_state->params[i];
638 qemu_sem_wait(&multifd_send_state->channels_ready);
639 trace_multifd_send_sync_main_wait(p->id);
640 qemu_sem_wait(&p->sem_sync);
642 if (flush_zero_copy && p->c && (multifd_zero_copy_flush(p->c) < 0)) {
646 trace_multifd_send_sync_main(multifd_send_state->packet_num);
651 static void *multifd_send_thread(void *opaque)
653 MultiFDSendParams *p = opaque;
654 MigrationThread *thread = NULL;
655 Error *local_err = NULL;
657 bool use_zero_copy_send = migrate_zero_copy_send();
659 thread = MigrationThreadAdd(p->name, qemu_get_thread_id());
661 trace_multifd_send_thread_start(p->id);
662 rcu_register_thread();
664 if (multifd_send_initial_packet(p, &local_err) < 0) {
672 qemu_sem_post(&multifd_send_state->channels_ready);
673 qemu_sem_wait(&p->sem);
675 if (qatomic_read(&multifd_send_state->exiting)) {
678 qemu_mutex_lock(&p->mutex);
680 if (p->pending_job) {
681 uint64_t packet_num = p->packet_num;
685 if (use_zero_copy_send) {
691 for (int i = 0; i < p->pages->num; i++) {
692 p->normal[p->normal_num] = p->pages->offset[i];
697 ret = multifd_send_state->ops->send_prepare(p, &local_err);
699 qemu_mutex_unlock(&p->mutex);
703 multifd_send_fill_packet(p);
707 p->total_normal_pages += p->normal_num;
709 p->pages->block = NULL;
710 qemu_mutex_unlock(&p->mutex);
712 trace_multifd_send(p->id, packet_num, p->normal_num, flags,
713 p->next_packet_size);
715 if (use_zero_copy_send) {
716 /* Send header first, without zerocopy */
717 ret = qio_channel_write_all(p->c, (void *)p->packet,
718 p->packet_len, &local_err);
723 /* Send header using the same writev call */
724 p->iov[0].iov_len = p->packet_len;
725 p->iov[0].iov_base = p->packet;
728 ret = qio_channel_writev_full_all(p->c, p->iov, p->iovs_num, NULL,
729 0, p->write_flags, &local_err);
734 qemu_mutex_lock(&p->mutex);
736 qemu_mutex_unlock(&p->mutex);
738 if (flags & MULTIFD_FLAG_SYNC) {
739 qemu_sem_post(&p->sem_sync);
741 } else if (p->quit) {
742 qemu_mutex_unlock(&p->mutex);
745 qemu_mutex_unlock(&p->mutex);
746 /* sometimes there are spurious wakeups */
752 trace_multifd_send_error(p->id);
753 multifd_send_terminate_threads(local_err);
754 error_free(local_err);
758 * Error happen, I will exit, but I can't just leave, tell
759 * who pay attention to me.
762 qemu_sem_post(&p->sem_sync);
763 qemu_sem_post(&multifd_send_state->channels_ready);
766 qemu_mutex_lock(&p->mutex);
768 qemu_mutex_unlock(&p->mutex);
770 rcu_unregister_thread();
771 MigrationThreadDel(thread);
772 trace_multifd_send_thread_end(p->id, p->num_packets, p->total_normal_pages);
777 static bool multifd_channel_connect(MultiFDSendParams *p,
781 static void multifd_tls_outgoing_handshake(QIOTask *task,
784 MultiFDSendParams *p = opaque;
785 QIOChannel *ioc = QIO_CHANNEL(qio_task_get_source(task));
788 if (qio_task_propagate_error(task, &err)) {
789 trace_multifd_tls_outgoing_handshake_error(ioc, error_get_pretty(err));
791 trace_multifd_tls_outgoing_handshake_complete(ioc);
794 if (!multifd_channel_connect(p, ioc, err)) {
796 * Error happen, mark multifd_send_thread status as 'quit' although it
797 * is not created, and then tell who pay attention to me.
800 qemu_sem_post(&multifd_send_state->channels_ready);
801 qemu_sem_post(&p->sem_sync);
805 static void *multifd_tls_handshake_thread(void *opaque)
807 MultiFDSendParams *p = opaque;
808 QIOChannelTLS *tioc = QIO_CHANNEL_TLS(p->c);
810 qio_channel_tls_handshake(tioc,
811 multifd_tls_outgoing_handshake,
818 static void multifd_tls_channel_connect(MultiFDSendParams *p,
822 MigrationState *s = migrate_get_current();
823 const char *hostname = s->hostname;
826 tioc = migration_tls_client_create(s, ioc, hostname, errp);
831 object_unref(OBJECT(ioc));
832 trace_multifd_tls_outgoing_handshake_start(ioc, tioc, hostname);
833 qio_channel_set_name(QIO_CHANNEL(tioc), "multifd-tls-outgoing");
834 p->c = QIO_CHANNEL(tioc);
835 qemu_thread_create(&p->thread, "multifd-tls-handshake-worker",
836 multifd_tls_handshake_thread, p,
837 QEMU_THREAD_JOINABLE);
840 static bool multifd_channel_connect(MultiFDSendParams *p,
844 trace_multifd_set_outgoing_channel(
845 ioc, object_get_typename(OBJECT(ioc)),
846 migrate_get_current()->hostname, error);
851 if (migrate_channel_requires_tls_upgrade(ioc)) {
852 multifd_tls_channel_connect(p, ioc, &error);
855 * tls_channel_connect will call back to this
856 * function after the TLS handshake,
857 * so we mustn't call multifd_send_thread until then
864 migration_ioc_register_yank(ioc);
865 p->registered_yank = true;
867 qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
868 QEMU_THREAD_JOINABLE);
873 static void multifd_new_send_channel_cleanup(MultiFDSendParams *p,
874 QIOChannel *ioc, Error *err)
876 migrate_set_error(migrate_get_current(), err);
877 /* Error happen, we need to tell who pay attention to me */
878 qemu_sem_post(&multifd_send_state->channels_ready);
879 qemu_sem_post(&p->sem_sync);
881 * Although multifd_send_thread is not created, but main migration
882 * thread neet to judge whether it is running, so we need to mark
886 object_unref(OBJECT(ioc));
890 static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
892 MultiFDSendParams *p = opaque;
893 QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
894 Error *local_err = NULL;
896 trace_multifd_new_send_channel_async(p->id);
897 if (!qio_task_propagate_error(task, &local_err)) {
898 p->c = QIO_CHANNEL(sioc);
899 qio_channel_set_delay(p->c, false);
901 if (multifd_channel_connect(p, sioc, local_err)) {
906 multifd_new_send_channel_cleanup(p, sioc, local_err);
909 int multifd_save_setup(Error **errp)
912 uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
915 if (!migrate_multifd()) {
919 thread_count = migrate_multifd_channels();
920 multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
921 multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
922 multifd_send_state->pages = multifd_pages_init(page_count);
923 qemu_sem_init(&multifd_send_state->channels_ready, 0);
924 qatomic_set(&multifd_send_state->exiting, 0);
925 multifd_send_state->ops = multifd_ops[migrate_multifd_compression()];
927 for (i = 0; i < thread_count; i++) {
928 MultiFDSendParams *p = &multifd_send_state->params[i];
930 qemu_mutex_init(&p->mutex);
931 qemu_sem_init(&p->sem, 0);
932 qemu_sem_init(&p->sem_sync, 0);
936 p->pages = multifd_pages_init(page_count);
937 p->packet_len = sizeof(MultiFDPacket_t)
938 + sizeof(uint64_t) * page_count;
939 p->packet = g_malloc0(p->packet_len);
940 p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
941 p->packet->version = cpu_to_be32(MULTIFD_VERSION);
942 p->name = g_strdup_printf("multifdsend_%d", i);
943 /* We need one extra place for the packet header */
944 p->iov = g_new0(struct iovec, page_count + 1);
945 p->normal = g_new0(ram_addr_t, page_count);
946 p->page_size = qemu_target_page_size();
947 p->page_count = page_count;
949 if (migrate_zero_copy_send()) {
950 p->write_flags = QIO_CHANNEL_WRITE_FLAG_ZERO_COPY;
955 socket_send_channel_create(multifd_new_send_channel_async, p);
958 for (i = 0; i < thread_count; i++) {
959 MultiFDSendParams *p = &multifd_send_state->params[i];
960 Error *local_err = NULL;
963 ret = multifd_send_state->ops->send_setup(p, &local_err);
965 error_propagate(errp, local_err);
973 MultiFDRecvParams *params;
974 /* number of created threads */
976 /* syncs main thread and channels */
977 QemuSemaphore sem_sync;
978 /* global number of generated multifd packets */
982 } *multifd_recv_state;
984 static void multifd_recv_terminate_threads(Error *err)
988 trace_multifd_recv_terminate_threads(err != NULL);
991 MigrationState *s = migrate_get_current();
992 migrate_set_error(s, err);
993 if (s->state == MIGRATION_STATUS_SETUP ||
994 s->state == MIGRATION_STATUS_ACTIVE) {
995 migrate_set_state(&s->state, s->state,
996 MIGRATION_STATUS_FAILED);
1000 for (i = 0; i < migrate_multifd_channels(); i++) {
1001 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1003 qemu_mutex_lock(&p->mutex);
1006 * We could arrive here for two reasons:
1007 * - normal quit, i.e. everything went fine, just finished
1008 * - error quit: We close the channels so the channel threads
1009 * finish the qio_channel_read_all_eof()
1012 qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
1014 qemu_mutex_unlock(&p->mutex);
1018 void multifd_load_shutdown(void)
1020 if (migrate_multifd()) {
1021 multifd_recv_terminate_threads(NULL);
1025 void multifd_load_cleanup(void)
1029 if (!migrate_multifd()) {
1032 multifd_recv_terminate_threads(NULL);
1033 for (i = 0; i < migrate_multifd_channels(); i++) {
1034 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1038 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
1039 * however try to wakeup it without harm in cleanup phase.
1041 qemu_sem_post(&p->sem_sync);
1044 qemu_thread_join(&p->thread);
1046 for (i = 0; i < migrate_multifd_channels(); i++) {
1047 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1049 migration_ioc_unregister_yank(p->c);
1050 object_unref(OBJECT(p->c));
1052 qemu_mutex_destroy(&p->mutex);
1053 qemu_sem_destroy(&p->sem_sync);
1063 multifd_recv_state->ops->recv_cleanup(p);
1065 qemu_sem_destroy(&multifd_recv_state->sem_sync);
1066 g_free(multifd_recv_state->params);
1067 multifd_recv_state->params = NULL;
1068 g_free(multifd_recv_state);
1069 multifd_recv_state = NULL;
1072 void multifd_recv_sync_main(void)
1076 if (!migrate_multifd()) {
1079 for (i = 0; i < migrate_multifd_channels(); i++) {
1080 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1082 trace_multifd_recv_sync_main_wait(p->id);
1083 qemu_sem_wait(&multifd_recv_state->sem_sync);
1085 for (i = 0; i < migrate_multifd_channels(); i++) {
1086 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1088 WITH_QEMU_LOCK_GUARD(&p->mutex) {
1089 if (multifd_recv_state->packet_num < p->packet_num) {
1090 multifd_recv_state->packet_num = p->packet_num;
1093 trace_multifd_recv_sync_main_signal(p->id);
1094 qemu_sem_post(&p->sem_sync);
1096 trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
1099 static void *multifd_recv_thread(void *opaque)
1101 MultiFDRecvParams *p = opaque;
1102 Error *local_err = NULL;
1105 trace_multifd_recv_thread_start(p->id);
1106 rcu_register_thread();
1115 ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
1116 p->packet_len, &local_err);
1117 if (ret == 0 || ret == -1) { /* 0: EOF -1: Error */
1121 qemu_mutex_lock(&p->mutex);
1122 ret = multifd_recv_unfill_packet(p, &local_err);
1124 qemu_mutex_unlock(&p->mutex);
1129 /* recv methods don't know how to handle the SYNC flag */
1130 p->flags &= ~MULTIFD_FLAG_SYNC;
1131 trace_multifd_recv(p->id, p->packet_num, p->normal_num, flags,
1132 p->next_packet_size);
1134 p->total_normal_pages += p->normal_num;
1135 qemu_mutex_unlock(&p->mutex);
1137 if (p->normal_num) {
1138 ret = multifd_recv_state->ops->recv_pages(p, &local_err);
1144 if (flags & MULTIFD_FLAG_SYNC) {
1145 qemu_sem_post(&multifd_recv_state->sem_sync);
1146 qemu_sem_wait(&p->sem_sync);
1151 multifd_recv_terminate_threads(local_err);
1152 error_free(local_err);
1154 qemu_mutex_lock(&p->mutex);
1156 qemu_mutex_unlock(&p->mutex);
1158 rcu_unregister_thread();
1159 trace_multifd_recv_thread_end(p->id, p->num_packets, p->total_normal_pages);
1164 int multifd_load_setup(Error **errp)
1167 uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
1171 * Return successfully if multiFD recv state is already initialised
1172 * or multiFD is not enabled.
1174 if (multifd_recv_state || !migrate_multifd()) {
1178 thread_count = migrate_multifd_channels();
1179 multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
1180 multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
1181 qatomic_set(&multifd_recv_state->count, 0);
1182 qemu_sem_init(&multifd_recv_state->sem_sync, 0);
1183 multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()];
1185 for (i = 0; i < thread_count; i++) {
1186 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1188 qemu_mutex_init(&p->mutex);
1189 qemu_sem_init(&p->sem_sync, 0);
1192 p->packet_len = sizeof(MultiFDPacket_t)
1193 + sizeof(uint64_t) * page_count;
1194 p->packet = g_malloc0(p->packet_len);
1195 p->name = g_strdup_printf("multifdrecv_%d", i);
1196 p->iov = g_new0(struct iovec, page_count);
1197 p->normal = g_new0(ram_addr_t, page_count);
1198 p->page_count = page_count;
1199 p->page_size = qemu_target_page_size();
1202 for (i = 0; i < thread_count; i++) {
1203 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1204 Error *local_err = NULL;
1207 ret = multifd_recv_state->ops->recv_setup(p, &local_err);
1209 error_propagate(errp, local_err);
1216 bool multifd_recv_all_channels_created(void)
1218 int thread_count = migrate_multifd_channels();
1220 if (!migrate_multifd()) {
1224 if (!multifd_recv_state) {
1225 /* Called before any connections created */
1229 return thread_count == qatomic_read(&multifd_recv_state->count);
1233 * Try to receive all multifd channels to get ready for the migration.
1234 * Sets @errp when failing to receive the current channel.
1236 void multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
1238 MultiFDRecvParams *p;
1239 Error *local_err = NULL;
1242 id = multifd_recv_initial_packet(ioc, &local_err);
1244 multifd_recv_terminate_threads(local_err);
1245 error_propagate_prepend(errp, local_err,
1246 "failed to receive packet"
1247 " via multifd channel %d: ",
1248 qatomic_read(&multifd_recv_state->count));
1251 trace_multifd_recv_new_channel(id);
1253 p = &multifd_recv_state->params[id];
1255 error_setg(&local_err, "multifd: received id '%d' already setup'",
1257 multifd_recv_terminate_threads(local_err);
1258 error_propagate(errp, local_err);
1262 object_ref(OBJECT(ioc));
1263 /* initial packet */
1267 qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
1268 QEMU_THREAD_JOINABLE);
1269 qatomic_inc(&multifd_recv_state->count);