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
;
287 packet
->magic
= be32_to_cpu(packet
->magic
);
288 if (packet
->magic
!= MULTIFD_MAGIC
) {
289 error_setg(errp
, "multifd: received packet "
290 "magic %x and expected magic %x",
291 packet
->magic
, MULTIFD_MAGIC
);
295 packet
->version
= be32_to_cpu(packet
->version
);
296 if (packet
->version
!= MULTIFD_VERSION
) {
297 error_setg(errp
, "multifd: received packet "
298 "version %u and expected version %u",
299 packet
->version
, MULTIFD_VERSION
);
303 p
->flags
= be32_to_cpu(packet
->flags
);
305 packet
->pages_alloc
= be32_to_cpu(packet
->pages_alloc
);
307 * If we received a packet that is 100 times bigger than expected
308 * just stop migration. It is a magic number.
310 if (packet
->pages_alloc
> p
->page_count
) {
311 error_setg(errp
, "multifd: received packet "
312 "with size %u and expected a size of %u",
313 packet
->pages_alloc
, p
->page_count
) ;
317 p
->normal_num
= be32_to_cpu(packet
->normal_pages
);
318 if (p
->normal_num
> packet
->pages_alloc
) {
319 error_setg(errp
, "multifd: received packet "
320 "with %u pages and expected maximum pages are %u",
321 p
->normal_num
, packet
->pages_alloc
) ;
325 p
->next_packet_size
= be32_to_cpu(packet
->next_packet_size
);
326 p
->packet_num
= be64_to_cpu(packet
->packet_num
);
328 if (p
->normal_num
== 0) {
332 /* make sure that ramblock is 0 terminated */
333 packet
->ramblock
[255] = 0;
334 block
= qemu_ram_block_by_name(packet
->ramblock
);
336 error_setg(errp
, "multifd: unknown ram block %s",
341 p
->host
= block
->host
;
342 for (i
= 0; i
< p
->normal_num
; i
++) {
343 uint64_t offset
= be64_to_cpu(packet
->offset
[i
]);
345 if (offset
> (block
->used_length
- p
->page_size
)) {
346 error_setg(errp
, "multifd: offset too long %" PRIu64
347 " (max " RAM_ADDR_FMT
")",
348 offset
, block
->used_length
);
351 p
->normal
[i
] = offset
;
358 MultiFDSendParams
*params
;
359 /* array of pages to sent */
360 MultiFDPages_t
*pages
;
361 /* global number of generated multifd packets */
363 /* send channels ready */
364 QemuSemaphore channels_ready
;
366 * Have we already run terminate threads. There is a race when it
367 * happens that we got one error while we are exiting.
368 * We will use atomic operations. Only valid values are 0 and 1.
373 } *multifd_send_state
;
376 * How we use multifd_send_state->pages and channel->pages?
378 * We create a pages for each channel, and a main one. Each time that
379 * we need to send a batch of pages we interchange the ones between
380 * multifd_send_state and the channel that is sending it. There are
381 * two reasons for that:
382 * - to not have to do so many mallocs during migration
383 * - to make easier to know what to free at the end of migration
385 * This way we always know who is the owner of each "pages" struct,
386 * and we don't need any locking. It belongs to the migration thread
387 * or to the channel thread. Switching is safe because the migration
388 * thread is using the channel mutex when changing it, and the channel
389 * have to had finish with its own, otherwise pending_job can't be
393 static int multifd_send_pages(QEMUFile
*f
)
396 static int next_channel
;
397 MultiFDSendParams
*p
= NULL
; /* make happy gcc */
398 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
399 uint64_t transferred
;
401 if (qatomic_read(&multifd_send_state
->exiting
)) {
405 qemu_sem_wait(&multifd_send_state
->channels_ready
);
407 * next_channel can remain from a previous migration that was
408 * using more channels, so ensure it doesn't overflow if the
409 * limit is lower now.
411 next_channel
%= migrate_multifd_channels();
412 for (i
= next_channel
;; i
= (i
+ 1) % migrate_multifd_channels()) {
413 p
= &multifd_send_state
->params
[i
];
415 qemu_mutex_lock(&p
->mutex
);
417 error_report("%s: channel %d has already quit!", __func__
, i
);
418 qemu_mutex_unlock(&p
->mutex
);
421 if (!p
->pending_job
) {
423 next_channel
= (i
+ 1) % migrate_multifd_channels();
426 qemu_mutex_unlock(&p
->mutex
);
428 assert(!p
->pages
->num
);
429 assert(!p
->pages
->block
);
431 p
->packet_num
= multifd_send_state
->packet_num
++;
432 multifd_send_state
->pages
= p
->pages
;
434 transferred
= ((uint64_t) pages
->num
) * p
->page_size
+ p
->packet_len
;
435 qemu_file_acct_rate_limit(f
, transferred
);
436 qemu_mutex_unlock(&p
->mutex
);
437 stat64_add(&mig_stats
.transferred
, transferred
);
438 stat64_add(&mig_stats
.multifd_bytes
, transferred
);
439 qemu_sem_post(&p
->sem
);
444 int multifd_queue_page(QEMUFile
*f
, RAMBlock
*block
, ram_addr_t offset
)
446 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
447 bool changed
= false;
450 pages
->block
= block
;
453 if (pages
->block
== block
) {
454 pages
->offset
[pages
->num
] = offset
;
457 if (pages
->num
< pages
->allocated
) {
464 if (multifd_send_pages(f
) < 0) {
469 return multifd_queue_page(f
, block
, offset
);
475 static void multifd_send_terminate_threads(Error
*err
)
479 trace_multifd_send_terminate_threads(err
!= NULL
);
482 MigrationState
*s
= migrate_get_current();
483 migrate_set_error(s
, err
);
484 if (s
->state
== MIGRATION_STATUS_SETUP
||
485 s
->state
== MIGRATION_STATUS_PRE_SWITCHOVER
||
486 s
->state
== MIGRATION_STATUS_DEVICE
||
487 s
->state
== MIGRATION_STATUS_ACTIVE
) {
488 migrate_set_state(&s
->state
, s
->state
,
489 MIGRATION_STATUS_FAILED
);
494 * We don't want to exit each threads twice. Depending on where
495 * we get the error, or if there are two independent errors in two
496 * threads at the same time, we can end calling this function
499 if (qatomic_xchg(&multifd_send_state
->exiting
, 1)) {
503 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
504 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
506 qemu_mutex_lock(&p
->mutex
);
508 qemu_sem_post(&p
->sem
);
510 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
512 qemu_mutex_unlock(&p
->mutex
);
516 void multifd_save_cleanup(void)
520 if (!migrate_multifd()) {
523 multifd_send_terminate_threads(NULL
);
524 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
525 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
528 qemu_thread_join(&p
->thread
);
531 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
532 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
533 Error
*local_err
= NULL
;
535 if (p
->registered_yank
) {
536 migration_ioc_unregister_yank(p
->c
);
538 socket_send_channel_destroy(p
->c
);
540 qemu_mutex_destroy(&p
->mutex
);
541 qemu_sem_destroy(&p
->sem
);
542 qemu_sem_destroy(&p
->sem_sync
);
545 multifd_pages_clear(p
->pages
);
554 multifd_send_state
->ops
->send_cleanup(p
, &local_err
);
556 migrate_set_error(migrate_get_current(), local_err
);
557 error_free(local_err
);
560 qemu_sem_destroy(&multifd_send_state
->channels_ready
);
561 g_free(multifd_send_state
->params
);
562 multifd_send_state
->params
= NULL
;
563 multifd_pages_clear(multifd_send_state
->pages
);
564 multifd_send_state
->pages
= NULL
;
565 g_free(multifd_send_state
);
566 multifd_send_state
= NULL
;
569 static int multifd_zero_copy_flush(QIOChannel
*c
)
574 ret
= qio_channel_flush(c
, &err
);
576 error_report_err(err
);
580 stat64_add(&mig_stats
.dirty_sync_missed_zero_copy
, 1);
586 int multifd_send_sync_main(QEMUFile
*f
)
589 bool flush_zero_copy
;
591 if (!migrate_multifd()) {
594 if (multifd_send_state
->pages
->num
) {
595 if (multifd_send_pages(f
) < 0) {
596 error_report("%s: multifd_send_pages fail", __func__
);
602 * When using zero-copy, it's necessary to flush the pages before any of
603 * the pages can be sent again, so we'll make sure the new version of the
604 * pages will always arrive _later_ than the old pages.
606 * Currently we achieve this by flushing the zero-page requested writes
607 * per ram iteration, but in the future we could potentially optimize it
608 * to be less frequent, e.g. only after we finished one whole scanning of
609 * all the dirty bitmaps.
612 flush_zero_copy
= migrate_zero_copy_send();
614 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
615 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
617 trace_multifd_send_sync_main_signal(p
->id
);
619 qemu_mutex_lock(&p
->mutex
);
622 error_report("%s: channel %d has already quit", __func__
, i
);
623 qemu_mutex_unlock(&p
->mutex
);
627 p
->packet_num
= multifd_send_state
->packet_num
++;
628 p
->flags
|= MULTIFD_FLAG_SYNC
;
630 qemu_mutex_unlock(&p
->mutex
);
631 qemu_sem_post(&p
->sem
);
633 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
634 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
636 qemu_sem_wait(&multifd_send_state
->channels_ready
);
637 trace_multifd_send_sync_main_wait(p
->id
);
638 qemu_sem_wait(&p
->sem_sync
);
640 if (flush_zero_copy
&& p
->c
&& (multifd_zero_copy_flush(p
->c
) < 0)) {
644 trace_multifd_send_sync_main(multifd_send_state
->packet_num
);
649 static void *multifd_send_thread(void *opaque
)
651 MultiFDSendParams
*p
= opaque
;
652 MigrationThread
*thread
= NULL
;
653 Error
*local_err
= NULL
;
655 bool use_zero_copy_send
= migrate_zero_copy_send();
657 thread
= MigrationThreadAdd(p
->name
, qemu_get_thread_id());
659 trace_multifd_send_thread_start(p
->id
);
660 rcu_register_thread();
662 if (multifd_send_initial_packet(p
, &local_err
) < 0) {
670 qemu_sem_post(&multifd_send_state
->channels_ready
);
671 qemu_sem_wait(&p
->sem
);
673 if (qatomic_read(&multifd_send_state
->exiting
)) {
676 qemu_mutex_lock(&p
->mutex
);
678 if (p
->pending_job
) {
679 uint64_t packet_num
= p
->packet_num
;
683 if (use_zero_copy_send
) {
689 for (int i
= 0; i
< p
->pages
->num
; i
++) {
690 p
->normal
[p
->normal_num
] = p
->pages
->offset
[i
];
695 ret
= multifd_send_state
->ops
->send_prepare(p
, &local_err
);
697 qemu_mutex_unlock(&p
->mutex
);
701 multifd_send_fill_packet(p
);
705 p
->total_normal_pages
+= p
->normal_num
;
707 p
->pages
->block
= NULL
;
708 qemu_mutex_unlock(&p
->mutex
);
710 trace_multifd_send(p
->id
, packet_num
, p
->normal_num
, flags
,
711 p
->next_packet_size
);
713 if (use_zero_copy_send
) {
714 /* Send header first, without zerocopy */
715 ret
= qio_channel_write_all(p
->c
, (void *)p
->packet
,
716 p
->packet_len
, &local_err
);
721 /* Send header using the same writev call */
722 p
->iov
[0].iov_len
= p
->packet_len
;
723 p
->iov
[0].iov_base
= p
->packet
;
726 ret
= qio_channel_writev_full_all(p
->c
, p
->iov
, p
->iovs_num
, NULL
,
727 0, p
->write_flags
, &local_err
);
732 qemu_mutex_lock(&p
->mutex
);
734 qemu_mutex_unlock(&p
->mutex
);
736 if (flags
& MULTIFD_FLAG_SYNC
) {
737 qemu_sem_post(&p
->sem_sync
);
739 } else if (p
->quit
) {
740 qemu_mutex_unlock(&p
->mutex
);
743 qemu_mutex_unlock(&p
->mutex
);
744 /* sometimes there are spurious wakeups */
750 trace_multifd_send_error(p
->id
);
751 multifd_send_terminate_threads(local_err
);
752 error_free(local_err
);
756 * Error happen, I will exit, but I can't just leave, tell
757 * who pay attention to me.
760 qemu_sem_post(&p
->sem_sync
);
761 qemu_sem_post(&multifd_send_state
->channels_ready
);
764 qemu_mutex_lock(&p
->mutex
);
766 qemu_mutex_unlock(&p
->mutex
);
768 rcu_unregister_thread();
769 MigrationThreadDel(thread
);
770 trace_multifd_send_thread_end(p
->id
, p
->num_packets
, p
->total_normal_pages
);
775 static bool multifd_channel_connect(MultiFDSendParams
*p
,
779 static void multifd_tls_outgoing_handshake(QIOTask
*task
,
782 MultiFDSendParams
*p
= opaque
;
783 QIOChannel
*ioc
= QIO_CHANNEL(qio_task_get_source(task
));
786 if (qio_task_propagate_error(task
, &err
)) {
787 trace_multifd_tls_outgoing_handshake_error(ioc
, error_get_pretty(err
));
789 trace_multifd_tls_outgoing_handshake_complete(ioc
);
792 if (!multifd_channel_connect(p
, ioc
, err
)) {
794 * Error happen, mark multifd_send_thread status as 'quit' although it
795 * is not created, and then tell who pay attention to me.
798 qemu_sem_post(&multifd_send_state
->channels_ready
);
799 qemu_sem_post(&p
->sem_sync
);
803 static void *multifd_tls_handshake_thread(void *opaque
)
805 MultiFDSendParams
*p
= opaque
;
806 QIOChannelTLS
*tioc
= QIO_CHANNEL_TLS(p
->c
);
808 qio_channel_tls_handshake(tioc
,
809 multifd_tls_outgoing_handshake
,
816 static void multifd_tls_channel_connect(MultiFDSendParams
*p
,
820 MigrationState
*s
= migrate_get_current();
821 const char *hostname
= s
->hostname
;
824 tioc
= migration_tls_client_create(ioc
, hostname
, errp
);
829 object_unref(OBJECT(ioc
));
830 trace_multifd_tls_outgoing_handshake_start(ioc
, tioc
, hostname
);
831 qio_channel_set_name(QIO_CHANNEL(tioc
), "multifd-tls-outgoing");
832 p
->c
= QIO_CHANNEL(tioc
);
833 qemu_thread_create(&p
->thread
, "multifd-tls-handshake-worker",
834 multifd_tls_handshake_thread
, p
,
835 QEMU_THREAD_JOINABLE
);
838 static bool multifd_channel_connect(MultiFDSendParams
*p
,
842 trace_multifd_set_outgoing_channel(
843 ioc
, object_get_typename(OBJECT(ioc
)),
844 migrate_get_current()->hostname
, error
);
849 if (migrate_channel_requires_tls_upgrade(ioc
)) {
850 multifd_tls_channel_connect(p
, ioc
, &error
);
853 * tls_channel_connect will call back to this
854 * function after the TLS handshake,
855 * so we mustn't call multifd_send_thread until then
862 migration_ioc_register_yank(ioc
);
863 p
->registered_yank
= true;
865 qemu_thread_create(&p
->thread
, p
->name
, multifd_send_thread
, p
,
866 QEMU_THREAD_JOINABLE
);
871 static void multifd_new_send_channel_cleanup(MultiFDSendParams
*p
,
872 QIOChannel
*ioc
, Error
*err
)
874 migrate_set_error(migrate_get_current(), err
);
875 /* Error happen, we need to tell who pay attention to me */
876 qemu_sem_post(&multifd_send_state
->channels_ready
);
877 qemu_sem_post(&p
->sem_sync
);
879 * Although multifd_send_thread is not created, but main migration
880 * thread neet to judge whether it is running, so we need to mark
884 object_unref(OBJECT(ioc
));
888 static void multifd_new_send_channel_async(QIOTask
*task
, gpointer opaque
)
890 MultiFDSendParams
*p
= opaque
;
891 QIOChannel
*sioc
= QIO_CHANNEL(qio_task_get_source(task
));
892 Error
*local_err
= NULL
;
894 trace_multifd_new_send_channel_async(p
->id
);
895 if (!qio_task_propagate_error(task
, &local_err
)) {
896 p
->c
= QIO_CHANNEL(sioc
);
897 qio_channel_set_delay(p
->c
, false);
899 if (multifd_channel_connect(p
, sioc
, local_err
)) {
904 multifd_new_send_channel_cleanup(p
, sioc
, local_err
);
907 int multifd_save_setup(Error
**errp
)
910 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
913 if (!migrate_multifd()) {
917 thread_count
= migrate_multifd_channels();
918 multifd_send_state
= g_malloc0(sizeof(*multifd_send_state
));
919 multifd_send_state
->params
= g_new0(MultiFDSendParams
, thread_count
);
920 multifd_send_state
->pages
= multifd_pages_init(page_count
);
921 qemu_sem_init(&multifd_send_state
->channels_ready
, 0);
922 qatomic_set(&multifd_send_state
->exiting
, 0);
923 multifd_send_state
->ops
= multifd_ops
[migrate_multifd_compression()];
925 for (i
= 0; i
< thread_count
; i
++) {
926 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
928 qemu_mutex_init(&p
->mutex
);
929 qemu_sem_init(&p
->sem
, 0);
930 qemu_sem_init(&p
->sem_sync
, 0);
934 p
->pages
= multifd_pages_init(page_count
);
935 p
->packet_len
= sizeof(MultiFDPacket_t
)
936 + sizeof(uint64_t) * page_count
;
937 p
->packet
= g_malloc0(p
->packet_len
);
938 p
->packet
->magic
= cpu_to_be32(MULTIFD_MAGIC
);
939 p
->packet
->version
= cpu_to_be32(MULTIFD_VERSION
);
940 p
->name
= g_strdup_printf("multifdsend_%d", i
);
941 /* We need one extra place for the packet header */
942 p
->iov
= g_new0(struct iovec
, page_count
+ 1);
943 p
->normal
= g_new0(ram_addr_t
, page_count
);
944 p
->page_size
= qemu_target_page_size();
945 p
->page_count
= page_count
;
947 if (migrate_zero_copy_send()) {
948 p
->write_flags
= QIO_CHANNEL_WRITE_FLAG_ZERO_COPY
;
953 socket_send_channel_create(multifd_new_send_channel_async
, p
);
956 for (i
= 0; i
< thread_count
; i
++) {
957 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
958 Error
*local_err
= NULL
;
961 ret
= multifd_send_state
->ops
->send_setup(p
, &local_err
);
963 error_propagate(errp
, local_err
);
971 MultiFDRecvParams
*params
;
972 /* number of created threads */
974 /* syncs main thread and channels */
975 QemuSemaphore sem_sync
;
976 /* global number of generated multifd packets */
980 } *multifd_recv_state
;
982 static void multifd_recv_terminate_threads(Error
*err
)
986 trace_multifd_recv_terminate_threads(err
!= NULL
);
989 MigrationState
*s
= migrate_get_current();
990 migrate_set_error(s
, err
);
991 if (s
->state
== MIGRATION_STATUS_SETUP
||
992 s
->state
== MIGRATION_STATUS_ACTIVE
) {
993 migrate_set_state(&s
->state
, s
->state
,
994 MIGRATION_STATUS_FAILED
);
998 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
999 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1001 qemu_mutex_lock(&p
->mutex
);
1004 * We could arrive here for two reasons:
1005 * - normal quit, i.e. everything went fine, just finished
1006 * - error quit: We close the channels so the channel threads
1007 * finish the qio_channel_read_all_eof()
1010 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
1012 qemu_mutex_unlock(&p
->mutex
);
1016 void multifd_load_shutdown(void)
1018 if (migrate_multifd()) {
1019 multifd_recv_terminate_threads(NULL
);
1023 void multifd_load_cleanup(void)
1027 if (!migrate_multifd()) {
1030 multifd_recv_terminate_threads(NULL
);
1031 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1032 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1036 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
1037 * however try to wakeup it without harm in cleanup phase.
1039 qemu_sem_post(&p
->sem_sync
);
1042 qemu_thread_join(&p
->thread
);
1044 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1045 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1047 migration_ioc_unregister_yank(p
->c
);
1048 object_unref(OBJECT(p
->c
));
1050 qemu_mutex_destroy(&p
->mutex
);
1051 qemu_sem_destroy(&p
->sem_sync
);
1061 multifd_recv_state
->ops
->recv_cleanup(p
);
1063 qemu_sem_destroy(&multifd_recv_state
->sem_sync
);
1064 g_free(multifd_recv_state
->params
);
1065 multifd_recv_state
->params
= NULL
;
1066 g_free(multifd_recv_state
);
1067 multifd_recv_state
= NULL
;
1070 void multifd_recv_sync_main(void)
1074 if (!migrate_multifd()) {
1077 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1078 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1080 trace_multifd_recv_sync_main_wait(p
->id
);
1081 qemu_sem_wait(&multifd_recv_state
->sem_sync
);
1083 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1084 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1086 WITH_QEMU_LOCK_GUARD(&p
->mutex
) {
1087 if (multifd_recv_state
->packet_num
< p
->packet_num
) {
1088 multifd_recv_state
->packet_num
= p
->packet_num
;
1091 trace_multifd_recv_sync_main_signal(p
->id
);
1092 qemu_sem_post(&p
->sem_sync
);
1094 trace_multifd_recv_sync_main(multifd_recv_state
->packet_num
);
1097 static void *multifd_recv_thread(void *opaque
)
1099 MultiFDRecvParams
*p
= opaque
;
1100 Error
*local_err
= NULL
;
1103 trace_multifd_recv_thread_start(p
->id
);
1104 rcu_register_thread();
1113 ret
= qio_channel_read_all_eof(p
->c
, (void *)p
->packet
,
1114 p
->packet_len
, &local_err
);
1115 if (ret
== 0 || ret
== -1) { /* 0: EOF -1: Error */
1119 qemu_mutex_lock(&p
->mutex
);
1120 ret
= multifd_recv_unfill_packet(p
, &local_err
);
1122 qemu_mutex_unlock(&p
->mutex
);
1127 /* recv methods don't know how to handle the SYNC flag */
1128 p
->flags
&= ~MULTIFD_FLAG_SYNC
;
1129 trace_multifd_recv(p
->id
, p
->packet_num
, p
->normal_num
, flags
,
1130 p
->next_packet_size
);
1132 p
->total_normal_pages
+= p
->normal_num
;
1133 qemu_mutex_unlock(&p
->mutex
);
1135 if (p
->normal_num
) {
1136 ret
= multifd_recv_state
->ops
->recv_pages(p
, &local_err
);
1142 if (flags
& MULTIFD_FLAG_SYNC
) {
1143 qemu_sem_post(&multifd_recv_state
->sem_sync
);
1144 qemu_sem_wait(&p
->sem_sync
);
1149 multifd_recv_terminate_threads(local_err
);
1150 error_free(local_err
);
1152 qemu_mutex_lock(&p
->mutex
);
1154 qemu_mutex_unlock(&p
->mutex
);
1156 rcu_unregister_thread();
1157 trace_multifd_recv_thread_end(p
->id
, p
->num_packets
, p
->total_normal_pages
);
1162 int multifd_load_setup(Error
**errp
)
1165 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
1169 * Return successfully if multiFD recv state is already initialised
1170 * or multiFD is not enabled.
1172 if (multifd_recv_state
|| !migrate_multifd()) {
1176 thread_count
= migrate_multifd_channels();
1177 multifd_recv_state
= g_malloc0(sizeof(*multifd_recv_state
));
1178 multifd_recv_state
->params
= g_new0(MultiFDRecvParams
, thread_count
);
1179 qatomic_set(&multifd_recv_state
->count
, 0);
1180 qemu_sem_init(&multifd_recv_state
->sem_sync
, 0);
1181 multifd_recv_state
->ops
= multifd_ops
[migrate_multifd_compression()];
1183 for (i
= 0; i
< thread_count
; i
++) {
1184 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1186 qemu_mutex_init(&p
->mutex
);
1187 qemu_sem_init(&p
->sem_sync
, 0);
1190 p
->packet_len
= sizeof(MultiFDPacket_t
)
1191 + sizeof(uint64_t) * page_count
;
1192 p
->packet
= g_malloc0(p
->packet_len
);
1193 p
->name
= g_strdup_printf("multifdrecv_%d", i
);
1194 p
->iov
= g_new0(struct iovec
, page_count
);
1195 p
->normal
= g_new0(ram_addr_t
, page_count
);
1196 p
->page_count
= page_count
;
1197 p
->page_size
= qemu_target_page_size();
1200 for (i
= 0; i
< thread_count
; i
++) {
1201 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1202 Error
*local_err
= NULL
;
1205 ret
= multifd_recv_state
->ops
->recv_setup(p
, &local_err
);
1207 error_propagate(errp
, local_err
);
1214 bool multifd_recv_all_channels_created(void)
1216 int thread_count
= migrate_multifd_channels();
1218 if (!migrate_multifd()) {
1222 if (!multifd_recv_state
) {
1223 /* Called before any connections created */
1227 return thread_count
== qatomic_read(&multifd_recv_state
->count
);
1231 * Try to receive all multifd channels to get ready for the migration.
1232 * Sets @errp when failing to receive the current channel.
1234 void multifd_recv_new_channel(QIOChannel
*ioc
, Error
**errp
)
1236 MultiFDRecvParams
*p
;
1237 Error
*local_err
= NULL
;
1240 id
= multifd_recv_initial_packet(ioc
, &local_err
);
1242 multifd_recv_terminate_threads(local_err
);
1243 error_propagate_prepend(errp
, local_err
,
1244 "failed to receive packet"
1245 " via multifd channel %d: ",
1246 qatomic_read(&multifd_recv_state
->count
));
1249 trace_multifd_recv_new_channel(id
);
1251 p
= &multifd_recv_state
->params
[id
];
1253 error_setg(&local_err
, "multifd: received id '%d' already setup'",
1255 multifd_recv_terminate_threads(local_err
);
1256 error_propagate(errp
, local_err
);
1260 object_ref(OBJECT(ioc
));
1261 /* initial packet */
1265 qemu_thread_create(&p
->thread
, p
->name
, multifd_recv_thread
, p
,
1266 QEMU_THREAD_JOINABLE
);
1267 qatomic_inc(&multifd_recv_state
->count
);