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"
28 #include "qemu/yank.h"
29 #include "io/channel-socket.h"
30 #include "yank_functions.h"
34 #define MULTIFD_MAGIC 0x11223344U
35 #define MULTIFD_VERSION 1
40 unsigned char uuid
[16]; /* QemuUUID */
42 uint8_t unused1
[7]; /* Reserved for future use */
43 uint64_t unused2
[4]; /* Reserved for future use */
44 } __attribute__((packed
)) MultiFDInit_t
;
46 /* Multifd without compression */
49 * nocomp_send_setup: setup send side
51 * For no compression this function does nothing.
53 * Returns 0 for success or -1 for error
55 * @p: Params for the channel that we are using
56 * @errp: pointer to an error
58 static int nocomp_send_setup(MultiFDSendParams
*p
, Error
**errp
)
64 * nocomp_send_cleanup: cleanup send side
66 * For no compression this function does nothing.
68 * @p: Params for the channel that we are using
69 * @errp: pointer to an error
71 static void nocomp_send_cleanup(MultiFDSendParams
*p
, Error
**errp
)
77 * nocomp_send_prepare: prepare date to be able to send
79 * For no compression we just have to calculate the size of the
82 * Returns 0 for success or -1 for error
84 * @p: Params for the channel that we are using
85 * @errp: pointer to an error
87 static int nocomp_send_prepare(MultiFDSendParams
*p
, Error
**errp
)
89 p
->next_packet_size
= p
->pages
->num
* qemu_target_page_size();
90 p
->flags
|= MULTIFD_FLAG_NOCOMP
;
95 * nocomp_send_write: do the actual write of the data
97 * For no compression we just have to write the data.
99 * Returns 0 for success or -1 for error
101 * @p: Params for the channel that we are using
102 * @used: number of pages used
103 * @errp: pointer to an error
105 static int nocomp_send_write(MultiFDSendParams
*p
, uint32_t used
, Error
**errp
)
107 return qio_channel_writev_all(p
->c
, p
->pages
->iov
, used
, errp
);
111 * nocomp_recv_setup: setup receive side
113 * For no compression this function does nothing.
115 * Returns 0 for success or -1 for error
117 * @p: Params for the channel that we are using
118 * @errp: pointer to an error
120 static int nocomp_recv_setup(MultiFDRecvParams
*p
, Error
**errp
)
126 * nocomp_recv_cleanup: setup receive side
128 * For no compression this function does nothing.
130 * @p: Params for the channel that we are using
132 static void nocomp_recv_cleanup(MultiFDRecvParams
*p
)
137 * nocomp_recv_pages: read the data from the channel into actual pages
139 * For no compression we just need to read things into the correct place.
141 * Returns 0 for success or -1 for error
143 * @p: Params for the channel that we are using
144 * @errp: pointer to an error
146 static int nocomp_recv_pages(MultiFDRecvParams
*p
, Error
**errp
)
148 uint32_t flags
= p
->flags
& MULTIFD_FLAG_COMPRESSION_MASK
;
150 if (flags
!= MULTIFD_FLAG_NOCOMP
) {
151 error_setg(errp
, "multifd %d: flags received %x flags expected %x",
152 p
->id
, flags
, MULTIFD_FLAG_NOCOMP
);
155 return qio_channel_readv_all(p
->c
, p
->pages
->iov
, p
->pages
->num
, errp
);
158 static MultiFDMethods multifd_nocomp_ops
= {
159 .send_setup
= nocomp_send_setup
,
160 .send_cleanup
= nocomp_send_cleanup
,
161 .send_prepare
= nocomp_send_prepare
,
162 .send_write
= nocomp_send_write
,
163 .recv_setup
= nocomp_recv_setup
,
164 .recv_cleanup
= nocomp_recv_cleanup
,
165 .recv_pages
= nocomp_recv_pages
168 static MultiFDMethods
*multifd_ops
[MULTIFD_COMPRESSION__MAX
] = {
169 [MULTIFD_COMPRESSION_NONE
] = &multifd_nocomp_ops
,
172 void multifd_register_ops(int method
, MultiFDMethods
*ops
)
174 assert(0 < method
&& method
< MULTIFD_COMPRESSION__MAX
);
175 multifd_ops
[method
] = ops
;
178 static int multifd_send_initial_packet(MultiFDSendParams
*p
, Error
**errp
)
180 MultiFDInit_t msg
= {};
183 msg
.magic
= cpu_to_be32(MULTIFD_MAGIC
);
184 msg
.version
= cpu_to_be32(MULTIFD_VERSION
);
186 memcpy(msg
.uuid
, &qemu_uuid
.data
, sizeof(msg
.uuid
));
188 ret
= qio_channel_write_all(p
->c
, (char *)&msg
, sizeof(msg
), errp
);
195 static int multifd_recv_initial_packet(QIOChannel
*c
, Error
**errp
)
200 ret
= qio_channel_read_all(c
, (char *)&msg
, sizeof(msg
), errp
);
205 msg
.magic
= be32_to_cpu(msg
.magic
);
206 msg
.version
= be32_to_cpu(msg
.version
);
208 if (msg
.magic
!= MULTIFD_MAGIC
) {
209 error_setg(errp
, "multifd: received packet magic %x "
210 "expected %x", msg
.magic
, MULTIFD_MAGIC
);
214 if (msg
.version
!= MULTIFD_VERSION
) {
215 error_setg(errp
, "multifd: received packet version %d "
216 "expected %d", msg
.version
, MULTIFD_VERSION
);
220 if (memcmp(msg
.uuid
, &qemu_uuid
, sizeof(qemu_uuid
))) {
221 char *uuid
= qemu_uuid_unparse_strdup(&qemu_uuid
);
222 char *msg_uuid
= qemu_uuid_unparse_strdup((const QemuUUID
*)msg
.uuid
);
224 error_setg(errp
, "multifd: received uuid '%s' and expected "
225 "uuid '%s' for channel %hhd", msg_uuid
, uuid
, msg
.id
);
231 if (msg
.id
> migrate_multifd_channels()) {
232 error_setg(errp
, "multifd: received channel version %d "
233 "expected %d", msg
.version
, MULTIFD_VERSION
);
240 static MultiFDPages_t
*multifd_pages_init(size_t size
)
242 MultiFDPages_t
*pages
= g_new0(MultiFDPages_t
, 1);
244 pages
->allocated
= size
;
245 pages
->iov
= g_new0(struct iovec
, size
);
246 pages
->offset
= g_new0(ram_addr_t
, size
);
251 static void multifd_pages_clear(MultiFDPages_t
*pages
)
254 pages
->allocated
= 0;
255 pages
->packet_num
= 0;
259 g_free(pages
->offset
);
260 pages
->offset
= NULL
;
264 static void multifd_send_fill_packet(MultiFDSendParams
*p
)
266 MultiFDPacket_t
*packet
= p
->packet
;
269 packet
->flags
= cpu_to_be32(p
->flags
);
270 packet
->pages_alloc
= cpu_to_be32(p
->pages
->allocated
);
271 packet
->pages_used
= cpu_to_be32(p
->pages
->num
);
272 packet
->next_packet_size
= cpu_to_be32(p
->next_packet_size
);
273 packet
->packet_num
= cpu_to_be64(p
->packet_num
);
275 if (p
->pages
->block
) {
276 strncpy(packet
->ramblock
, p
->pages
->block
->idstr
, 256);
279 for (i
= 0; i
< p
->pages
->num
; i
++) {
280 /* there are architectures where ram_addr_t is 32 bit */
281 uint64_t temp
= p
->pages
->offset
[i
];
283 packet
->offset
[i
] = cpu_to_be64(temp
);
287 static int multifd_recv_unfill_packet(MultiFDRecvParams
*p
, Error
**errp
)
289 MultiFDPacket_t
*packet
= p
->packet
;
290 size_t page_size
= qemu_target_page_size();
291 uint32_t pages_max
= MULTIFD_PACKET_SIZE
/ page_size
;
295 packet
->magic
= be32_to_cpu(packet
->magic
);
296 if (packet
->magic
!= MULTIFD_MAGIC
) {
297 error_setg(errp
, "multifd: received packet "
298 "magic %x and expected magic %x",
299 packet
->magic
, MULTIFD_MAGIC
);
303 packet
->version
= be32_to_cpu(packet
->version
);
304 if (packet
->version
!= MULTIFD_VERSION
) {
305 error_setg(errp
, "multifd: received packet "
306 "version %d and expected version %d",
307 packet
->version
, MULTIFD_VERSION
);
311 p
->flags
= be32_to_cpu(packet
->flags
);
313 packet
->pages_alloc
= be32_to_cpu(packet
->pages_alloc
);
315 * If we received a packet that is 100 times bigger than expected
316 * just stop migration. It is a magic number.
318 if (packet
->pages_alloc
> pages_max
* 100) {
319 error_setg(errp
, "multifd: received packet "
320 "with size %d and expected a maximum size of %d",
321 packet
->pages_alloc
, pages_max
* 100) ;
325 * We received a packet that is bigger than expected but inside
326 * reasonable limits (see previous comment). Just reallocate.
328 if (packet
->pages_alloc
> p
->pages
->allocated
) {
329 multifd_pages_clear(p
->pages
);
330 p
->pages
= multifd_pages_init(packet
->pages_alloc
);
333 p
->pages
->num
= be32_to_cpu(packet
->pages_used
);
334 if (p
->pages
->num
> packet
->pages_alloc
) {
335 error_setg(errp
, "multifd: received packet "
336 "with %d pages and expected maximum pages are %d",
337 p
->pages
->num
, packet
->pages_alloc
) ;
341 p
->next_packet_size
= be32_to_cpu(packet
->next_packet_size
);
342 p
->packet_num
= be64_to_cpu(packet
->packet_num
);
344 if (p
->pages
->num
== 0) {
348 /* make sure that ramblock is 0 terminated */
349 packet
->ramblock
[255] = 0;
350 block
= qemu_ram_block_by_name(packet
->ramblock
);
352 error_setg(errp
, "multifd: unknown ram block %s",
357 p
->pages
->block
= block
;
358 for (i
= 0; i
< p
->pages
->num
; i
++) {
359 uint64_t offset
= be64_to_cpu(packet
->offset
[i
]);
361 if (offset
> (block
->used_length
- page_size
)) {
362 error_setg(errp
, "multifd: offset too long %" PRIu64
363 " (max " RAM_ADDR_FMT
")",
364 offset
, block
->used_length
);
367 p
->pages
->offset
[i
] = offset
;
368 p
->pages
->iov
[i
].iov_base
= block
->host
+ offset
;
369 p
->pages
->iov
[i
].iov_len
= page_size
;
376 MultiFDSendParams
*params
;
377 /* array of pages to sent */
378 MultiFDPages_t
*pages
;
379 /* global number of generated multifd packets */
381 /* send channels ready */
382 QemuSemaphore channels_ready
;
384 * Have we already run terminate threads. There is a race when it
385 * happens that we got one error while we are exiting.
386 * We will use atomic operations. Only valid values are 0 and 1.
391 } *multifd_send_state
;
394 * How we use multifd_send_state->pages and channel->pages?
396 * We create a pages for each channel, and a main one. Each time that
397 * we need to send a batch of pages we interchange the ones between
398 * multifd_send_state and the channel that is sending it. There are
399 * two reasons for that:
400 * - to not have to do so many mallocs during migration
401 * - to make easier to know what to free at the end of migration
403 * This way we always know who is the owner of each "pages" struct,
404 * and we don't need any locking. It belongs to the migration thread
405 * or to the channel thread. Switching is safe because the migration
406 * thread is using the channel mutex when changing it, and the channel
407 * have to had finish with its own, otherwise pending_job can't be
411 static int multifd_send_pages(QEMUFile
*f
)
414 static int next_channel
;
415 MultiFDSendParams
*p
= NULL
; /* make happy gcc */
416 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
417 uint64_t transferred
;
419 if (qatomic_read(&multifd_send_state
->exiting
)) {
423 qemu_sem_wait(&multifd_send_state
->channels_ready
);
425 * next_channel can remain from a previous migration that was
426 * using more channels, so ensure it doesn't overflow if the
427 * limit is lower now.
429 next_channel
%= migrate_multifd_channels();
430 for (i
= next_channel
;; i
= (i
+ 1) % migrate_multifd_channels()) {
431 p
= &multifd_send_state
->params
[i
];
433 qemu_mutex_lock(&p
->mutex
);
435 error_report("%s: channel %d has already quit!", __func__
, i
);
436 qemu_mutex_unlock(&p
->mutex
);
439 if (!p
->pending_job
) {
441 next_channel
= (i
+ 1) % migrate_multifd_channels();
444 qemu_mutex_unlock(&p
->mutex
);
446 assert(!p
->pages
->num
);
447 assert(!p
->pages
->block
);
449 p
->packet_num
= multifd_send_state
->packet_num
++;
450 multifd_send_state
->pages
= p
->pages
;
452 transferred
= ((uint64_t) pages
->num
) * qemu_target_page_size()
454 qemu_file_update_transfer(f
, transferred
);
455 ram_counters
.multifd_bytes
+= transferred
;
456 ram_counters
.transferred
+= transferred
;
457 qemu_mutex_unlock(&p
->mutex
);
458 qemu_sem_post(&p
->sem
);
463 int multifd_queue_page(QEMUFile
*f
, RAMBlock
*block
, ram_addr_t offset
)
465 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
468 pages
->block
= block
;
471 if (pages
->block
== block
) {
472 pages
->offset
[pages
->num
] = offset
;
473 pages
->iov
[pages
->num
].iov_base
= block
->host
+ offset
;
474 pages
->iov
[pages
->num
].iov_len
= qemu_target_page_size();
477 if (pages
->num
< pages
->allocated
) {
482 if (multifd_send_pages(f
) < 0) {
486 if (pages
->block
!= block
) {
487 return multifd_queue_page(f
, block
, offset
);
493 static void multifd_send_terminate_threads(Error
*err
)
497 trace_multifd_send_terminate_threads(err
!= NULL
);
500 MigrationState
*s
= migrate_get_current();
501 migrate_set_error(s
, err
);
502 if (s
->state
== MIGRATION_STATUS_SETUP
||
503 s
->state
== MIGRATION_STATUS_PRE_SWITCHOVER
||
504 s
->state
== MIGRATION_STATUS_DEVICE
||
505 s
->state
== MIGRATION_STATUS_ACTIVE
) {
506 migrate_set_state(&s
->state
, s
->state
,
507 MIGRATION_STATUS_FAILED
);
512 * We don't want to exit each threads twice. Depending on where
513 * we get the error, or if there are two independent errors in two
514 * threads at the same time, we can end calling this function
517 if (qatomic_xchg(&multifd_send_state
->exiting
, 1)) {
521 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
522 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
524 qemu_mutex_lock(&p
->mutex
);
526 qemu_sem_post(&p
->sem
);
528 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
530 qemu_mutex_unlock(&p
->mutex
);
534 void multifd_save_cleanup(void)
538 if (!migrate_use_multifd() || !migrate_multifd_is_allowed()) {
541 multifd_send_terminate_threads(NULL
);
542 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
543 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
546 qemu_thread_join(&p
->thread
);
549 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
550 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
551 Error
*local_err
= NULL
;
553 if (p
->registered_yank
) {
554 migration_ioc_unregister_yank(p
->c
);
556 socket_send_channel_destroy(p
->c
);
558 qemu_mutex_destroy(&p
->mutex
);
559 qemu_sem_destroy(&p
->sem
);
560 qemu_sem_destroy(&p
->sem_sync
);
563 g_free(p
->tls_hostname
);
564 p
->tls_hostname
= NULL
;
565 multifd_pages_clear(p
->pages
);
570 multifd_send_state
->ops
->send_cleanup(p
, &local_err
);
572 migrate_set_error(migrate_get_current(), local_err
);
573 error_free(local_err
);
576 qemu_sem_destroy(&multifd_send_state
->channels_ready
);
577 g_free(multifd_send_state
->params
);
578 multifd_send_state
->params
= NULL
;
579 multifd_pages_clear(multifd_send_state
->pages
);
580 multifd_send_state
->pages
= NULL
;
581 g_free(multifd_send_state
);
582 multifd_send_state
= NULL
;
585 void multifd_send_sync_main(QEMUFile
*f
)
589 if (!migrate_use_multifd()) {
592 if (multifd_send_state
->pages
->num
) {
593 if (multifd_send_pages(f
) < 0) {
594 error_report("%s: multifd_send_pages fail", __func__
);
598 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
599 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
601 trace_multifd_send_sync_main_signal(p
->id
);
603 qemu_mutex_lock(&p
->mutex
);
606 error_report("%s: channel %d has already quit", __func__
, i
);
607 qemu_mutex_unlock(&p
->mutex
);
611 p
->packet_num
= multifd_send_state
->packet_num
++;
612 p
->flags
|= MULTIFD_FLAG_SYNC
;
614 qemu_file_update_transfer(f
, p
->packet_len
);
615 ram_counters
.multifd_bytes
+= p
->packet_len
;
616 ram_counters
.transferred
+= p
->packet_len
;
617 qemu_mutex_unlock(&p
->mutex
);
618 qemu_sem_post(&p
->sem
);
620 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
621 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
623 trace_multifd_send_sync_main_wait(p
->id
);
624 qemu_sem_wait(&p
->sem_sync
);
626 trace_multifd_send_sync_main(multifd_send_state
->packet_num
);
629 static void *multifd_send_thread(void *opaque
)
631 MultiFDSendParams
*p
= opaque
;
632 Error
*local_err
= NULL
;
635 trace_multifd_send_thread_start(p
->id
);
636 rcu_register_thread();
638 if (multifd_send_initial_packet(p
, &local_err
) < 0) {
646 qemu_sem_wait(&p
->sem
);
648 if (qatomic_read(&multifd_send_state
->exiting
)) {
651 qemu_mutex_lock(&p
->mutex
);
653 if (p
->pending_job
) {
654 uint32_t used
= p
->pages
->num
;
655 uint64_t packet_num
= p
->packet_num
;
656 uint32_t flags
= p
->flags
;
659 ret
= multifd_send_state
->ops
->send_prepare(p
, &local_err
);
661 qemu_mutex_unlock(&p
->mutex
);
665 multifd_send_fill_packet(p
);
668 p
->num_pages
+= used
;
670 p
->pages
->block
= NULL
;
671 qemu_mutex_unlock(&p
->mutex
);
673 trace_multifd_send(p
->id
, packet_num
, used
, flags
,
674 p
->next_packet_size
);
676 ret
= qio_channel_write_all(p
->c
, (void *)p
->packet
,
677 p
->packet_len
, &local_err
);
683 ret
= multifd_send_state
->ops
->send_write(p
, used
, &local_err
);
689 qemu_mutex_lock(&p
->mutex
);
691 qemu_mutex_unlock(&p
->mutex
);
693 if (flags
& MULTIFD_FLAG_SYNC
) {
694 qemu_sem_post(&p
->sem_sync
);
696 qemu_sem_post(&multifd_send_state
->channels_ready
);
697 } else if (p
->quit
) {
698 qemu_mutex_unlock(&p
->mutex
);
701 qemu_mutex_unlock(&p
->mutex
);
702 /* sometimes there are spurious wakeups */
708 trace_multifd_send_error(p
->id
);
709 multifd_send_terminate_threads(local_err
);
710 error_free(local_err
);
714 * Error happen, I will exit, but I can't just leave, tell
715 * who pay attention to me.
718 qemu_sem_post(&p
->sem_sync
);
719 qemu_sem_post(&multifd_send_state
->channels_ready
);
722 qemu_mutex_lock(&p
->mutex
);
724 qemu_mutex_unlock(&p
->mutex
);
726 rcu_unregister_thread();
727 trace_multifd_send_thread_end(p
->id
, p
->num_packets
, p
->num_pages
);
732 static bool multifd_channel_connect(MultiFDSendParams
*p
,
736 static void multifd_tls_outgoing_handshake(QIOTask
*task
,
739 MultiFDSendParams
*p
= opaque
;
740 QIOChannel
*ioc
= QIO_CHANNEL(qio_task_get_source(task
));
743 if (qio_task_propagate_error(task
, &err
)) {
744 trace_multifd_tls_outgoing_handshake_error(ioc
, error_get_pretty(err
));
746 trace_multifd_tls_outgoing_handshake_complete(ioc
);
749 if (!multifd_channel_connect(p
, ioc
, err
)) {
751 * Error happen, mark multifd_send_thread status as 'quit' although it
752 * is not created, and then tell who pay attention to me.
755 qemu_sem_post(&multifd_send_state
->channels_ready
);
756 qemu_sem_post(&p
->sem_sync
);
760 static void *multifd_tls_handshake_thread(void *opaque
)
762 MultiFDSendParams
*p
= opaque
;
763 QIOChannelTLS
*tioc
= QIO_CHANNEL_TLS(p
->c
);
765 qio_channel_tls_handshake(tioc
,
766 multifd_tls_outgoing_handshake
,
773 static void multifd_tls_channel_connect(MultiFDSendParams
*p
,
777 MigrationState
*s
= migrate_get_current();
778 const char *hostname
= p
->tls_hostname
;
781 tioc
= migration_tls_client_create(s
, ioc
, hostname
, errp
);
786 object_unref(OBJECT(ioc
));
787 trace_multifd_tls_outgoing_handshake_start(ioc
, tioc
, hostname
);
788 qio_channel_set_name(QIO_CHANNEL(tioc
), "multifd-tls-outgoing");
789 p
->c
= QIO_CHANNEL(tioc
);
790 qemu_thread_create(&p
->thread
, "multifd-tls-handshake-worker",
791 multifd_tls_handshake_thread
, p
,
792 QEMU_THREAD_JOINABLE
);
795 static bool multifd_channel_connect(MultiFDSendParams
*p
,
799 MigrationState
*s
= migrate_get_current();
801 trace_multifd_set_outgoing_channel(
802 ioc
, object_get_typename(OBJECT(ioc
)), p
->tls_hostname
, error
);
805 if (s
->parameters
.tls_creds
&&
806 *s
->parameters
.tls_creds
&&
807 !object_dynamic_cast(OBJECT(ioc
),
808 TYPE_QIO_CHANNEL_TLS
)) {
809 multifd_tls_channel_connect(p
, ioc
, &error
);
812 * tls_channel_connect will call back to this
813 * function after the TLS handshake,
814 * so we mustn't call multifd_send_thread until then
821 migration_ioc_register_yank(ioc
);
822 p
->registered_yank
= true;
824 qemu_thread_create(&p
->thread
, p
->name
, multifd_send_thread
, p
,
825 QEMU_THREAD_JOINABLE
);
833 static void multifd_new_send_channel_cleanup(MultiFDSendParams
*p
,
834 QIOChannel
*ioc
, Error
*err
)
836 migrate_set_error(migrate_get_current(), err
);
837 /* Error happen, we need to tell who pay attention to me */
838 qemu_sem_post(&multifd_send_state
->channels_ready
);
839 qemu_sem_post(&p
->sem_sync
);
841 * Although multifd_send_thread is not created, but main migration
842 * thread neet to judge whether it is running, so we need to mark
846 object_unref(OBJECT(ioc
));
850 static void multifd_new_send_channel_async(QIOTask
*task
, gpointer opaque
)
852 MultiFDSendParams
*p
= opaque
;
853 QIOChannel
*sioc
= QIO_CHANNEL(qio_task_get_source(task
));
854 Error
*local_err
= NULL
;
856 trace_multifd_new_send_channel_async(p
->id
);
857 if (qio_task_propagate_error(task
, &local_err
)) {
860 p
->c
= QIO_CHANNEL(sioc
);
861 qio_channel_set_delay(p
->c
, false);
863 if (!multifd_channel_connect(p
, sioc
, local_err
)) {
870 multifd_new_send_channel_cleanup(p
, sioc
, local_err
);
873 static bool migrate_allow_multifd
= true;
874 void migrate_protocol_allow_multifd(bool allow
)
876 migrate_allow_multifd
= allow
;
879 bool migrate_multifd_is_allowed(void)
881 return migrate_allow_multifd
;
884 int multifd_save_setup(Error
**errp
)
887 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
891 if (!migrate_use_multifd()) {
894 if (!migrate_multifd_is_allowed()) {
895 error_setg(errp
, "multifd is not supported by current protocol");
899 s
= migrate_get_current();
900 thread_count
= migrate_multifd_channels();
901 multifd_send_state
= g_malloc0(sizeof(*multifd_send_state
));
902 multifd_send_state
->params
= g_new0(MultiFDSendParams
, thread_count
);
903 multifd_send_state
->pages
= multifd_pages_init(page_count
);
904 qemu_sem_init(&multifd_send_state
->channels_ready
, 0);
905 qatomic_set(&multifd_send_state
->exiting
, 0);
906 multifd_send_state
->ops
= multifd_ops
[migrate_multifd_compression()];
908 for (i
= 0; i
< thread_count
; i
++) {
909 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
911 qemu_mutex_init(&p
->mutex
);
912 qemu_sem_init(&p
->sem
, 0);
913 qemu_sem_init(&p
->sem_sync
, 0);
917 p
->pages
= multifd_pages_init(page_count
);
918 p
->packet_len
= sizeof(MultiFDPacket_t
)
919 + sizeof(uint64_t) * page_count
;
920 p
->packet
= g_malloc0(p
->packet_len
);
921 p
->packet
->magic
= cpu_to_be32(MULTIFD_MAGIC
);
922 p
->packet
->version
= cpu_to_be32(MULTIFD_VERSION
);
923 p
->name
= g_strdup_printf("multifdsend_%d", i
);
924 p
->tls_hostname
= g_strdup(s
->hostname
);
925 socket_send_channel_create(multifd_new_send_channel_async
, p
);
928 for (i
= 0; i
< thread_count
; i
++) {
929 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
930 Error
*local_err
= NULL
;
933 ret
= multifd_send_state
->ops
->send_setup(p
, &local_err
);
935 error_propagate(errp
, local_err
);
943 MultiFDRecvParams
*params
;
944 /* number of created threads */
946 /* syncs main thread and channels */
947 QemuSemaphore sem_sync
;
948 /* global number of generated multifd packets */
952 } *multifd_recv_state
;
954 static void multifd_recv_terminate_threads(Error
*err
)
958 trace_multifd_recv_terminate_threads(err
!= NULL
);
961 MigrationState
*s
= migrate_get_current();
962 migrate_set_error(s
, err
);
963 if (s
->state
== MIGRATION_STATUS_SETUP
||
964 s
->state
== MIGRATION_STATUS_ACTIVE
) {
965 migrate_set_state(&s
->state
, s
->state
,
966 MIGRATION_STATUS_FAILED
);
970 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
971 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
973 qemu_mutex_lock(&p
->mutex
);
976 * We could arrive here for two reasons:
977 * - normal quit, i.e. everything went fine, just finished
978 * - error quit: We close the channels so the channel threads
979 * finish the qio_channel_read_all_eof()
982 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
984 qemu_mutex_unlock(&p
->mutex
);
988 int multifd_load_cleanup(Error
**errp
)
992 if (!migrate_use_multifd() || !migrate_multifd_is_allowed()) {
995 multifd_recv_terminate_threads(NULL
);
996 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
997 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1002 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
1003 * however try to wakeup it without harm in cleanup phase.
1005 qemu_sem_post(&p
->sem_sync
);
1006 qemu_thread_join(&p
->thread
);
1009 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1010 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1012 migration_ioc_unregister_yank(p
->c
);
1013 object_unref(OBJECT(p
->c
));
1015 qemu_mutex_destroy(&p
->mutex
);
1016 qemu_sem_destroy(&p
->sem_sync
);
1019 multifd_pages_clear(p
->pages
);
1024 multifd_recv_state
->ops
->recv_cleanup(p
);
1026 qemu_sem_destroy(&multifd_recv_state
->sem_sync
);
1027 g_free(multifd_recv_state
->params
);
1028 multifd_recv_state
->params
= NULL
;
1029 g_free(multifd_recv_state
);
1030 multifd_recv_state
= NULL
;
1035 void multifd_recv_sync_main(void)
1039 if (!migrate_use_multifd()) {
1042 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1043 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1045 trace_multifd_recv_sync_main_wait(p
->id
);
1046 qemu_sem_wait(&multifd_recv_state
->sem_sync
);
1048 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1049 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1051 WITH_QEMU_LOCK_GUARD(&p
->mutex
) {
1052 if (multifd_recv_state
->packet_num
< p
->packet_num
) {
1053 multifd_recv_state
->packet_num
= p
->packet_num
;
1056 trace_multifd_recv_sync_main_signal(p
->id
);
1057 qemu_sem_post(&p
->sem_sync
);
1059 trace_multifd_recv_sync_main(multifd_recv_state
->packet_num
);
1062 static void *multifd_recv_thread(void *opaque
)
1064 MultiFDRecvParams
*p
= opaque
;
1065 Error
*local_err
= NULL
;
1068 trace_multifd_recv_thread_start(p
->id
);
1069 rcu_register_thread();
1079 ret
= qio_channel_read_all_eof(p
->c
, (void *)p
->packet
,
1080 p
->packet_len
, &local_err
);
1081 if (ret
== 0) { /* EOF */
1084 if (ret
== -1) { /* Error */
1088 qemu_mutex_lock(&p
->mutex
);
1089 ret
= multifd_recv_unfill_packet(p
, &local_err
);
1091 qemu_mutex_unlock(&p
->mutex
);
1095 used
= p
->pages
->num
;
1097 /* recv methods don't know how to handle the SYNC flag */
1098 p
->flags
&= ~MULTIFD_FLAG_SYNC
;
1099 trace_multifd_recv(p
->id
, p
->packet_num
, used
, flags
,
1100 p
->next_packet_size
);
1102 p
->num_pages
+= used
;
1103 qemu_mutex_unlock(&p
->mutex
);
1106 ret
= multifd_recv_state
->ops
->recv_pages(p
, &local_err
);
1112 if (flags
& MULTIFD_FLAG_SYNC
) {
1113 qemu_sem_post(&multifd_recv_state
->sem_sync
);
1114 qemu_sem_wait(&p
->sem_sync
);
1119 multifd_recv_terminate_threads(local_err
);
1120 error_free(local_err
);
1122 qemu_mutex_lock(&p
->mutex
);
1124 qemu_mutex_unlock(&p
->mutex
);
1126 rcu_unregister_thread();
1127 trace_multifd_recv_thread_end(p
->id
, p
->num_packets
, p
->num_pages
);
1132 int multifd_load_setup(Error
**errp
)
1135 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
1138 if (!migrate_use_multifd()) {
1141 if (!migrate_multifd_is_allowed()) {
1142 error_setg(errp
, "multifd is not supported by current protocol");
1145 thread_count
= migrate_multifd_channels();
1146 multifd_recv_state
= g_malloc0(sizeof(*multifd_recv_state
));
1147 multifd_recv_state
->params
= g_new0(MultiFDRecvParams
, thread_count
);
1148 qatomic_set(&multifd_recv_state
->count
, 0);
1149 qemu_sem_init(&multifd_recv_state
->sem_sync
, 0);
1150 multifd_recv_state
->ops
= multifd_ops
[migrate_multifd_compression()];
1152 for (i
= 0; i
< thread_count
; i
++) {
1153 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1155 qemu_mutex_init(&p
->mutex
);
1156 qemu_sem_init(&p
->sem_sync
, 0);
1159 p
->pages
= multifd_pages_init(page_count
);
1160 p
->packet_len
= sizeof(MultiFDPacket_t
)
1161 + sizeof(uint64_t) * page_count
;
1162 p
->packet
= g_malloc0(p
->packet_len
);
1163 p
->name
= g_strdup_printf("multifdrecv_%d", i
);
1166 for (i
= 0; i
< thread_count
; i
++) {
1167 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1168 Error
*local_err
= NULL
;
1171 ret
= multifd_recv_state
->ops
->recv_setup(p
, &local_err
);
1173 error_propagate(errp
, local_err
);
1180 bool multifd_recv_all_channels_created(void)
1182 int thread_count
= migrate_multifd_channels();
1184 if (!migrate_use_multifd()) {
1188 if (!multifd_recv_state
) {
1189 /* Called before any connections created */
1193 return thread_count
== qatomic_read(&multifd_recv_state
->count
);
1197 * Try to receive all multifd channels to get ready for the migration.
1198 * - Return true and do not set @errp when correctly receiving all channels;
1199 * - Return false and do not set @errp when correctly receiving the current one;
1200 * - Return false and set @errp when failing to receive the current channel.
1202 bool multifd_recv_new_channel(QIOChannel
*ioc
, Error
**errp
)
1204 MultiFDRecvParams
*p
;
1205 Error
*local_err
= NULL
;
1208 id
= multifd_recv_initial_packet(ioc
, &local_err
);
1210 multifd_recv_terminate_threads(local_err
);
1211 error_propagate_prepend(errp
, local_err
,
1212 "failed to receive packet"
1213 " via multifd channel %d: ",
1214 qatomic_read(&multifd_recv_state
->count
));
1217 trace_multifd_recv_new_channel(id
);
1219 p
= &multifd_recv_state
->params
[id
];
1221 error_setg(&local_err
, "multifd: received id '%d' already setup'",
1223 multifd_recv_terminate_threads(local_err
);
1224 error_propagate(errp
, local_err
);
1228 object_ref(OBJECT(ioc
));
1229 /* initial packet */
1233 qemu_thread_create(&p
->thread
, p
->name
, multifd_recv_thread
, p
,
1234 QEMU_THREAD_JOINABLE
);
1235 qatomic_inc(&multifd_recv_state
->count
);
1236 return qatomic_read(&multifd_recv_state
->count
) ==
1237 migrate_multifd_channels();