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"
22 #include "migration-stats.h"
25 #include "qemu-file.h"
28 #include "threadinfo.h"
30 #include "qemu/yank.h"
31 #include "io/channel-socket.h"
32 #include "yank_functions.h"
36 #define MULTIFD_MAGIC 0x11223344U
37 #define MULTIFD_VERSION 1
42 unsigned char uuid
[16]; /* QemuUUID */
44 uint8_t unused1
[7]; /* Reserved for future use */
45 uint64_t unused2
[4]; /* Reserved for future use */
46 } __attribute__((packed
)) MultiFDInit_t
;
48 /* Multifd without compression */
51 * nocomp_send_setup: setup send side
53 * For no compression this function does nothing.
55 * Returns 0 for success or -1 for error
57 * @p: Params for the channel that we are using
58 * @errp: pointer to an error
60 static int nocomp_send_setup(MultiFDSendParams
*p
, Error
**errp
)
66 * nocomp_send_cleanup: cleanup send side
68 * For no compression this function does nothing.
70 * @p: Params for the channel that we are using
71 * @errp: pointer to an error
73 static void nocomp_send_cleanup(MultiFDSendParams
*p
, Error
**errp
)
79 * nocomp_send_prepare: prepare date to be able to send
81 * For no compression we just have to calculate the size of the
84 * Returns 0 for success or -1 for error
86 * @p: Params for the channel that we are using
87 * @errp: pointer to an error
89 static int nocomp_send_prepare(MultiFDSendParams
*p
, Error
**errp
)
91 MultiFDPages_t
*pages
= p
->pages
;
93 for (int i
= 0; i
< p
->normal_num
; i
++) {
94 p
->iov
[p
->iovs_num
].iov_base
= pages
->block
->host
+ p
->normal
[i
];
95 p
->iov
[p
->iovs_num
].iov_len
= p
->page_size
;
99 p
->next_packet_size
= p
->normal_num
* p
->page_size
;
100 p
->flags
|= MULTIFD_FLAG_NOCOMP
;
105 * nocomp_recv_setup: setup receive side
107 * For no compression this function does nothing.
109 * Returns 0 for success or -1 for error
111 * @p: Params for the channel that we are using
112 * @errp: pointer to an error
114 static int nocomp_recv_setup(MultiFDRecvParams
*p
, Error
**errp
)
120 * nocomp_recv_cleanup: setup receive side
122 * For no compression this function does nothing.
124 * @p: Params for the channel that we are using
126 static void nocomp_recv_cleanup(MultiFDRecvParams
*p
)
131 * nocomp_recv_pages: read the data from the channel into actual pages
133 * For no compression we just need to read things into the correct place.
135 * Returns 0 for success or -1 for error
137 * @p: Params for the channel that we are using
138 * @errp: pointer to an error
140 static int nocomp_recv_pages(MultiFDRecvParams
*p
, Error
**errp
)
142 uint32_t flags
= p
->flags
& MULTIFD_FLAG_COMPRESSION_MASK
;
144 if (flags
!= MULTIFD_FLAG_NOCOMP
) {
145 error_setg(errp
, "multifd %u: flags received %x flags expected %x",
146 p
->id
, flags
, MULTIFD_FLAG_NOCOMP
);
149 for (int i
= 0; i
< p
->normal_num
; i
++) {
150 p
->iov
[i
].iov_base
= p
->host
+ p
->normal
[i
];
151 p
->iov
[i
].iov_len
= p
->page_size
;
153 return qio_channel_readv_all(p
->c
, p
->iov
, p
->normal_num
, errp
);
156 static MultiFDMethods multifd_nocomp_ops
= {
157 .send_setup
= nocomp_send_setup
,
158 .send_cleanup
= nocomp_send_cleanup
,
159 .send_prepare
= nocomp_send_prepare
,
160 .recv_setup
= nocomp_recv_setup
,
161 .recv_cleanup
= nocomp_recv_cleanup
,
162 .recv_pages
= nocomp_recv_pages
165 static MultiFDMethods
*multifd_ops
[MULTIFD_COMPRESSION__MAX
] = {
166 [MULTIFD_COMPRESSION_NONE
] = &multifd_nocomp_ops
,
169 void multifd_register_ops(int method
, MultiFDMethods
*ops
)
171 assert(0 < method
&& method
< MULTIFD_COMPRESSION__MAX
);
172 multifd_ops
[method
] = ops
;
175 static int multifd_send_initial_packet(MultiFDSendParams
*p
, Error
**errp
)
177 MultiFDInit_t msg
= {};
180 msg
.magic
= cpu_to_be32(MULTIFD_MAGIC
);
181 msg
.version
= cpu_to_be32(MULTIFD_VERSION
);
183 memcpy(msg
.uuid
, &qemu_uuid
.data
, sizeof(msg
.uuid
));
185 ret
= qio_channel_write_all(p
->c
, (char *)&msg
, sizeof(msg
), errp
);
192 static int multifd_recv_initial_packet(QIOChannel
*c
, Error
**errp
)
197 ret
= qio_channel_read_all(c
, (char *)&msg
, sizeof(msg
), errp
);
202 msg
.magic
= be32_to_cpu(msg
.magic
);
203 msg
.version
= be32_to_cpu(msg
.version
);
205 if (msg
.magic
!= MULTIFD_MAGIC
) {
206 error_setg(errp
, "multifd: received packet magic %x "
207 "expected %x", msg
.magic
, MULTIFD_MAGIC
);
211 if (msg
.version
!= MULTIFD_VERSION
) {
212 error_setg(errp
, "multifd: received packet version %u "
213 "expected %u", msg
.version
, MULTIFD_VERSION
);
217 if (memcmp(msg
.uuid
, &qemu_uuid
, sizeof(qemu_uuid
))) {
218 char *uuid
= qemu_uuid_unparse_strdup(&qemu_uuid
);
219 char *msg_uuid
= qemu_uuid_unparse_strdup((const QemuUUID
*)msg
.uuid
);
221 error_setg(errp
, "multifd: received uuid '%s' and expected "
222 "uuid '%s' for channel %hhd", msg_uuid
, uuid
, msg
.id
);
228 if (msg
.id
> migrate_multifd_channels()) {
229 error_setg(errp
, "multifd: received channel version %u "
230 "expected %u", msg
.version
, MULTIFD_VERSION
);
237 static MultiFDPages_t
*multifd_pages_init(size_t size
)
239 MultiFDPages_t
*pages
= g_new0(MultiFDPages_t
, 1);
241 pages
->allocated
= size
;
242 pages
->offset
= g_new0(ram_addr_t
, size
);
247 static void multifd_pages_clear(MultiFDPages_t
*pages
)
250 pages
->allocated
= 0;
251 pages
->packet_num
= 0;
253 g_free(pages
->offset
);
254 pages
->offset
= NULL
;
258 static void multifd_send_fill_packet(MultiFDSendParams
*p
)
260 MultiFDPacket_t
*packet
= p
->packet
;
263 packet
->flags
= cpu_to_be32(p
->flags
);
264 packet
->pages_alloc
= cpu_to_be32(p
->pages
->allocated
);
265 packet
->normal_pages
= cpu_to_be32(p
->normal_num
);
266 packet
->next_packet_size
= cpu_to_be32(p
->next_packet_size
);
267 packet
->packet_num
= cpu_to_be64(p
->packet_num
);
269 if (p
->pages
->block
) {
270 strncpy(packet
->ramblock
, p
->pages
->block
->idstr
, 256);
273 for (i
= 0; i
< p
->normal_num
; i
++) {
274 /* there are architectures where ram_addr_t is 32 bit */
275 uint64_t temp
= p
->normal
[i
];
277 packet
->offset
[i
] = cpu_to_be64(temp
);
281 static int multifd_recv_unfill_packet(MultiFDRecvParams
*p
, Error
**errp
)
283 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 p
->block
= qemu_ram_block_by_name(packet
->ramblock
);
335 error_setg(errp
, "multifd: unknown ram block %s",
340 p
->host
= p
->block
->host
;
341 for (i
= 0; i
< p
->normal_num
; i
++) {
342 uint64_t offset
= be64_to_cpu(packet
->offset
[i
]);
344 if (offset
> (p
->block
->used_length
- p
->page_size
)) {
345 error_setg(errp
, "multifd: offset too long %" PRIu64
346 " (max " RAM_ADDR_FMT
")",
347 offset
, p
->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 migration_rate_account(transferred
);
435 qemu_mutex_unlock(&p
->mutex
);
436 stat64_add(&mig_stats
.transferred
, transferred
);
437 stat64_add(&mig_stats
.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(&mig_stats
.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_mutex_unlock(&p
->mutex
);
630 qemu_sem_post(&p
->sem
);
632 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
633 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
635 qemu_sem_wait(&multifd_send_state
->channels_ready
);
636 trace_multifd_send_sync_main_wait(p
->id
);
637 qemu_sem_wait(&p
->sem_sync
);
639 if (flush_zero_copy
&& p
->c
&& (multifd_zero_copy_flush(p
->c
) < 0)) {
643 trace_multifd_send_sync_main(multifd_send_state
->packet_num
);
648 static void *multifd_send_thread(void *opaque
)
650 MultiFDSendParams
*p
= opaque
;
651 MigrationThread
*thread
= NULL
;
652 Error
*local_err
= NULL
;
654 bool use_zero_copy_send
= migrate_zero_copy_send();
656 thread
= MigrationThreadAdd(p
->name
, qemu_get_thread_id());
658 trace_multifd_send_thread_start(p
->id
);
659 rcu_register_thread();
661 if (multifd_send_initial_packet(p
, &local_err
) < 0) {
669 qemu_sem_post(&multifd_send_state
->channels_ready
);
670 qemu_sem_wait(&p
->sem
);
672 if (qatomic_read(&multifd_send_state
->exiting
)) {
675 qemu_mutex_lock(&p
->mutex
);
677 if (p
->pending_job
) {
678 uint64_t packet_num
= p
->packet_num
;
682 if (use_zero_copy_send
) {
688 for (int i
= 0; i
< p
->pages
->num
; i
++) {
689 p
->normal
[p
->normal_num
] = p
->pages
->offset
[i
];
694 ret
= multifd_send_state
->ops
->send_prepare(p
, &local_err
);
696 qemu_mutex_unlock(&p
->mutex
);
700 multifd_send_fill_packet(p
);
704 p
->total_normal_pages
+= p
->normal_num
;
706 p
->pages
->block
= NULL
;
707 qemu_mutex_unlock(&p
->mutex
);
709 trace_multifd_send(p
->id
, packet_num
, p
->normal_num
, flags
,
710 p
->next_packet_size
);
712 if (use_zero_copy_send
) {
713 /* Send header first, without zerocopy */
714 ret
= qio_channel_write_all(p
->c
, (void *)p
->packet
,
715 p
->packet_len
, &local_err
);
720 /* Send header using the same writev call */
721 p
->iov
[0].iov_len
= p
->packet_len
;
722 p
->iov
[0].iov_base
= p
->packet
;
725 ret
= qio_channel_writev_full_all(p
->c
, p
->iov
, p
->iovs_num
, NULL
,
726 0, p
->write_flags
, &local_err
);
731 qemu_mutex_lock(&p
->mutex
);
733 qemu_mutex_unlock(&p
->mutex
);
735 if (flags
& MULTIFD_FLAG_SYNC
) {
736 qemu_sem_post(&p
->sem_sync
);
738 } else if (p
->quit
) {
739 qemu_mutex_unlock(&p
->mutex
);
742 qemu_mutex_unlock(&p
->mutex
);
743 /* sometimes there are spurious wakeups */
749 trace_multifd_send_error(p
->id
);
750 multifd_send_terminate_threads(local_err
);
751 error_free(local_err
);
755 * Error happen, I will exit, but I can't just leave, tell
756 * who pay attention to me.
759 qemu_sem_post(&p
->sem_sync
);
760 qemu_sem_post(&multifd_send_state
->channels_ready
);
763 qemu_mutex_lock(&p
->mutex
);
765 qemu_mutex_unlock(&p
->mutex
);
767 rcu_unregister_thread();
768 MigrationThreadDel(thread
);
769 trace_multifd_send_thread_end(p
->id
, p
->num_packets
, p
->total_normal_pages
);
774 static bool multifd_channel_connect(MultiFDSendParams
*p
,
778 static void multifd_tls_outgoing_handshake(QIOTask
*task
,
781 MultiFDSendParams
*p
= opaque
;
782 QIOChannel
*ioc
= QIO_CHANNEL(qio_task_get_source(task
));
785 if (qio_task_propagate_error(task
, &err
)) {
786 trace_multifd_tls_outgoing_handshake_error(ioc
, error_get_pretty(err
));
788 trace_multifd_tls_outgoing_handshake_complete(ioc
);
791 if (!multifd_channel_connect(p
, ioc
, err
)) {
793 * Error happen, mark multifd_send_thread status as 'quit' although it
794 * is not created, and then tell who pay attention to me.
797 qemu_sem_post(&multifd_send_state
->channels_ready
);
798 qemu_sem_post(&p
->sem_sync
);
802 static void *multifd_tls_handshake_thread(void *opaque
)
804 MultiFDSendParams
*p
= opaque
;
805 QIOChannelTLS
*tioc
= QIO_CHANNEL_TLS(p
->c
);
807 qio_channel_tls_handshake(tioc
,
808 multifd_tls_outgoing_handshake
,
815 static void multifd_tls_channel_connect(MultiFDSendParams
*p
,
819 MigrationState
*s
= migrate_get_current();
820 const char *hostname
= s
->hostname
;
823 tioc
= migration_tls_client_create(ioc
, hostname
, errp
);
828 object_unref(OBJECT(ioc
));
829 trace_multifd_tls_outgoing_handshake_start(ioc
, tioc
, hostname
);
830 qio_channel_set_name(QIO_CHANNEL(tioc
), "multifd-tls-outgoing");
831 p
->c
= QIO_CHANNEL(tioc
);
832 qemu_thread_create(&p
->thread
, "multifd-tls-handshake-worker",
833 multifd_tls_handshake_thread
, p
,
834 QEMU_THREAD_JOINABLE
);
837 static bool multifd_channel_connect(MultiFDSendParams
*p
,
841 trace_multifd_set_outgoing_channel(
842 ioc
, object_get_typename(OBJECT(ioc
)),
843 migrate_get_current()->hostname
, error
);
848 if (migrate_channel_requires_tls_upgrade(ioc
)) {
849 multifd_tls_channel_connect(p
, ioc
, &error
);
852 * tls_channel_connect will call back to this
853 * function after the TLS handshake,
854 * so we mustn't call multifd_send_thread until then
861 migration_ioc_register_yank(ioc
);
862 p
->registered_yank
= true;
864 qemu_thread_create(&p
->thread
, p
->name
, multifd_send_thread
, p
,
865 QEMU_THREAD_JOINABLE
);
870 static void multifd_new_send_channel_cleanup(MultiFDSendParams
*p
,
871 QIOChannel
*ioc
, Error
*err
)
873 migrate_set_error(migrate_get_current(), err
);
874 /* Error happen, we need to tell who pay attention to me */
875 qemu_sem_post(&multifd_send_state
->channels_ready
);
876 qemu_sem_post(&p
->sem_sync
);
878 * Although multifd_send_thread is not created, but main migration
879 * thread neet to judge whether it is running, so we need to mark
883 object_unref(OBJECT(ioc
));
887 static void multifd_new_send_channel_async(QIOTask
*task
, gpointer opaque
)
889 MultiFDSendParams
*p
= opaque
;
890 QIOChannel
*sioc
= QIO_CHANNEL(qio_task_get_source(task
));
891 Error
*local_err
= NULL
;
893 trace_multifd_new_send_channel_async(p
->id
);
894 if (!qio_task_propagate_error(task
, &local_err
)) {
895 p
->c
= QIO_CHANNEL(sioc
);
896 qio_channel_set_delay(p
->c
, false);
898 if (multifd_channel_connect(p
, sioc
, local_err
)) {
903 multifd_new_send_channel_cleanup(p
, sioc
, local_err
);
906 int multifd_save_setup(Error
**errp
)
909 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
912 if (!migrate_multifd()) {
916 thread_count
= migrate_multifd_channels();
917 multifd_send_state
= g_malloc0(sizeof(*multifd_send_state
));
918 multifd_send_state
->params
= g_new0(MultiFDSendParams
, thread_count
);
919 multifd_send_state
->pages
= multifd_pages_init(page_count
);
920 qemu_sem_init(&multifd_send_state
->channels_ready
, 0);
921 qatomic_set(&multifd_send_state
->exiting
, 0);
922 multifd_send_state
->ops
= multifd_ops
[migrate_multifd_compression()];
924 for (i
= 0; i
< thread_count
; i
++) {
925 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
927 qemu_mutex_init(&p
->mutex
);
928 qemu_sem_init(&p
->sem
, 0);
929 qemu_sem_init(&p
->sem_sync
, 0);
933 p
->pages
= multifd_pages_init(page_count
);
934 p
->packet_len
= sizeof(MultiFDPacket_t
)
935 + sizeof(uint64_t) * page_count
;
936 p
->packet
= g_malloc0(p
->packet_len
);
937 p
->packet
->magic
= cpu_to_be32(MULTIFD_MAGIC
);
938 p
->packet
->version
= cpu_to_be32(MULTIFD_VERSION
);
939 p
->name
= g_strdup_printf("multifdsend_%d", i
);
940 /* We need one extra place for the packet header */
941 p
->iov
= g_new0(struct iovec
, page_count
+ 1);
942 p
->normal
= g_new0(ram_addr_t
, page_count
);
943 p
->page_size
= qemu_target_page_size();
944 p
->page_count
= page_count
;
946 if (migrate_zero_copy_send()) {
947 p
->write_flags
= QIO_CHANNEL_WRITE_FLAG_ZERO_COPY
;
952 socket_send_channel_create(multifd_new_send_channel_async
, p
);
955 for (i
= 0; i
< thread_count
; i
++) {
956 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
957 Error
*local_err
= NULL
;
960 ret
= multifd_send_state
->ops
->send_setup(p
, &local_err
);
962 error_propagate(errp
, local_err
);
970 MultiFDRecvParams
*params
;
971 /* number of created threads */
973 /* syncs main thread and channels */
974 QemuSemaphore sem_sync
;
975 /* global number of generated multifd packets */
979 } *multifd_recv_state
;
981 static void multifd_recv_terminate_threads(Error
*err
)
985 trace_multifd_recv_terminate_threads(err
!= NULL
);
988 MigrationState
*s
= migrate_get_current();
989 migrate_set_error(s
, err
);
990 if (s
->state
== MIGRATION_STATUS_SETUP
||
991 s
->state
== MIGRATION_STATUS_ACTIVE
) {
992 migrate_set_state(&s
->state
, s
->state
,
993 MIGRATION_STATUS_FAILED
);
997 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
998 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1000 qemu_mutex_lock(&p
->mutex
);
1003 * We could arrive here for two reasons:
1004 * - normal quit, i.e. everything went fine, just finished
1005 * - error quit: We close the channels so the channel threads
1006 * finish the qio_channel_read_all_eof()
1009 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
1011 qemu_mutex_unlock(&p
->mutex
);
1015 void multifd_load_shutdown(void)
1017 if (migrate_multifd()) {
1018 multifd_recv_terminate_threads(NULL
);
1022 void multifd_load_cleanup(void)
1026 if (!migrate_multifd()) {
1029 multifd_recv_terminate_threads(NULL
);
1030 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1031 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1035 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
1036 * however try to wakeup it without harm in cleanup phase.
1038 qemu_sem_post(&p
->sem_sync
);
1041 qemu_thread_join(&p
->thread
);
1043 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1044 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1046 migration_ioc_unregister_yank(p
->c
);
1047 object_unref(OBJECT(p
->c
));
1049 qemu_mutex_destroy(&p
->mutex
);
1050 qemu_sem_destroy(&p
->sem_sync
);
1060 multifd_recv_state
->ops
->recv_cleanup(p
);
1062 qemu_sem_destroy(&multifd_recv_state
->sem_sync
);
1063 g_free(multifd_recv_state
->params
);
1064 multifd_recv_state
->params
= NULL
;
1065 g_free(multifd_recv_state
);
1066 multifd_recv_state
= NULL
;
1069 void multifd_recv_sync_main(void)
1073 if (!migrate_multifd()) {
1076 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1077 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1079 trace_multifd_recv_sync_main_wait(p
->id
);
1080 qemu_sem_wait(&multifd_recv_state
->sem_sync
);
1082 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1083 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1085 WITH_QEMU_LOCK_GUARD(&p
->mutex
) {
1086 if (multifd_recv_state
->packet_num
< p
->packet_num
) {
1087 multifd_recv_state
->packet_num
= p
->packet_num
;
1090 trace_multifd_recv_sync_main_signal(p
->id
);
1091 qemu_sem_post(&p
->sem_sync
);
1093 trace_multifd_recv_sync_main(multifd_recv_state
->packet_num
);
1096 static void *multifd_recv_thread(void *opaque
)
1098 MultiFDRecvParams
*p
= opaque
;
1099 Error
*local_err
= NULL
;
1102 trace_multifd_recv_thread_start(p
->id
);
1103 rcu_register_thread();
1112 ret
= qio_channel_read_all_eof(p
->c
, (void *)p
->packet
,
1113 p
->packet_len
, &local_err
);
1114 if (ret
== 0 || ret
== -1) { /* 0: EOF -1: Error */
1118 qemu_mutex_lock(&p
->mutex
);
1119 ret
= multifd_recv_unfill_packet(p
, &local_err
);
1121 qemu_mutex_unlock(&p
->mutex
);
1126 /* recv methods don't know how to handle the SYNC flag */
1127 p
->flags
&= ~MULTIFD_FLAG_SYNC
;
1128 trace_multifd_recv(p
->id
, p
->packet_num
, p
->normal_num
, flags
,
1129 p
->next_packet_size
);
1131 p
->total_normal_pages
+= p
->normal_num
;
1132 qemu_mutex_unlock(&p
->mutex
);
1134 if (p
->normal_num
) {
1135 ret
= multifd_recv_state
->ops
->recv_pages(p
, &local_err
);
1141 if (flags
& MULTIFD_FLAG_SYNC
) {
1142 qemu_sem_post(&multifd_recv_state
->sem_sync
);
1143 qemu_sem_wait(&p
->sem_sync
);
1148 multifd_recv_terminate_threads(local_err
);
1149 error_free(local_err
);
1151 qemu_mutex_lock(&p
->mutex
);
1153 qemu_mutex_unlock(&p
->mutex
);
1155 rcu_unregister_thread();
1156 trace_multifd_recv_thread_end(p
->id
, p
->num_packets
, p
->total_normal_pages
);
1161 int multifd_load_setup(Error
**errp
)
1164 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
1168 * Return successfully if multiFD recv state is already initialised
1169 * or multiFD is not enabled.
1171 if (multifd_recv_state
|| !migrate_multifd()) {
1175 thread_count
= migrate_multifd_channels();
1176 multifd_recv_state
= g_malloc0(sizeof(*multifd_recv_state
));
1177 multifd_recv_state
->params
= g_new0(MultiFDRecvParams
, thread_count
);
1178 qatomic_set(&multifd_recv_state
->count
, 0);
1179 qemu_sem_init(&multifd_recv_state
->sem_sync
, 0);
1180 multifd_recv_state
->ops
= multifd_ops
[migrate_multifd_compression()];
1182 for (i
= 0; i
< thread_count
; i
++) {
1183 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1185 qemu_mutex_init(&p
->mutex
);
1186 qemu_sem_init(&p
->sem_sync
, 0);
1189 p
->packet_len
= sizeof(MultiFDPacket_t
)
1190 + sizeof(uint64_t) * page_count
;
1191 p
->packet
= g_malloc0(p
->packet_len
);
1192 p
->name
= g_strdup_printf("multifdrecv_%d", i
);
1193 p
->iov
= g_new0(struct iovec
, page_count
);
1194 p
->normal
= g_new0(ram_addr_t
, page_count
);
1195 p
->page_count
= page_count
;
1196 p
->page_size
= qemu_target_page_size();
1199 for (i
= 0; i
< thread_count
; i
++) {
1200 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1201 Error
*local_err
= NULL
;
1204 ret
= multifd_recv_state
->ops
->recv_setup(p
, &local_err
);
1206 error_propagate(errp
, local_err
);
1213 bool multifd_recv_all_channels_created(void)
1215 int thread_count
= migrate_multifd_channels();
1217 if (!migrate_multifd()) {
1221 if (!multifd_recv_state
) {
1222 /* Called before any connections created */
1226 return thread_count
== qatomic_read(&multifd_recv_state
->count
);
1230 * Try to receive all multifd channels to get ready for the migration.
1231 * Sets @errp when failing to receive the current channel.
1233 void multifd_recv_new_channel(QIOChannel
*ioc
, Error
**errp
)
1235 MultiFDRecvParams
*p
;
1236 Error
*local_err
= NULL
;
1239 id
= multifd_recv_initial_packet(ioc
, &local_err
);
1241 multifd_recv_terminate_threads(local_err
);
1242 error_propagate_prepend(errp
, local_err
,
1243 "failed to receive packet"
1244 " via multifd channel %d: ",
1245 qatomic_read(&multifd_recv_state
->count
));
1248 trace_multifd_recv_new_channel(id
);
1250 p
= &multifd_recv_state
->params
[id
];
1252 error_setg(&local_err
, "multifd: received id '%d' already setup'",
1254 multifd_recv_terminate_threads(local_err
);
1255 error_propagate(errp
, local_err
);
1259 object_ref(OBJECT(ioc
));
1260 /* initial packet */
1264 qemu_thread_create(&p
->thread
, p
->name
, multifd_recv_thread
, p
,
1265 QEMU_THREAD_JOINABLE
);
1266 qatomic_inc(&multifd_recv_state
->count
);