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"
23 #include "qemu-file.h"
29 #define MULTIFD_MAGIC 0x11223344U
30 #define MULTIFD_VERSION 1
35 unsigned char uuid
[16]; /* QemuUUID */
37 uint8_t unused1
[7]; /* Reserved for future use */
38 uint64_t unused2
[4]; /* Reserved for future use */
39 } __attribute__((packed
)) MultiFDInit_t
;
41 /* Multifd without compression */
44 * nocomp_send_setup: setup send side
46 * For no compression this function does nothing.
48 * Returns 0 for success or -1 for error
50 * @p: Params for the channel that we are using
51 * @errp: pointer to an error
53 static int nocomp_send_setup(MultiFDSendParams
*p
, Error
**errp
)
59 * nocomp_send_cleanup: cleanup send side
61 * For no compression this function does nothing.
63 * @p: Params for the channel that we are using
65 static void nocomp_send_cleanup(MultiFDSendParams
*p
, Error
**errp
)
71 * nocomp_send_prepare: prepare date to be able to send
73 * For no compression we just have to calculate the size of the
76 * Returns 0 for success or -1 for error
78 * @p: Params for the channel that we are using
79 * @used: number of pages used
80 * @errp: pointer to an error
82 static int nocomp_send_prepare(MultiFDSendParams
*p
, uint32_t used
,
85 p
->next_packet_size
= used
* qemu_target_page_size();
86 p
->flags
|= MULTIFD_FLAG_NOCOMP
;
91 * nocomp_send_write: do the actual write of the data
93 * For no compression we just have to write the data.
95 * Returns 0 for success or -1 for error
97 * @p: Params for the channel that we are using
98 * @used: number of pages used
99 * @errp: pointer to an error
101 static int nocomp_send_write(MultiFDSendParams
*p
, uint32_t used
, Error
**errp
)
103 return qio_channel_writev_all(p
->c
, p
->pages
->iov
, used
, errp
);
107 * nocomp_recv_setup: setup receive side
109 * For no compression this function does nothing.
111 * Returns 0 for success or -1 for error
113 * @p: Params for the channel that we are using
114 * @errp: pointer to an error
116 static int nocomp_recv_setup(MultiFDRecvParams
*p
, Error
**errp
)
122 * nocomp_recv_cleanup: setup receive side
124 * For no compression this function does nothing.
126 * @p: Params for the channel that we are using
128 static void nocomp_recv_cleanup(MultiFDRecvParams
*p
)
133 * nocomp_recv_pages: read the data from the channel into actual pages
135 * For no compression we just need to read things into the correct place.
137 * Returns 0 for success or -1 for error
139 * @p: Params for the channel that we are using
140 * @used: number of pages used
141 * @errp: pointer to an error
143 static int nocomp_recv_pages(MultiFDRecvParams
*p
, uint32_t used
, Error
**errp
)
145 uint32_t flags
= p
->flags
& MULTIFD_FLAG_COMPRESSION_MASK
;
147 if (flags
!= MULTIFD_FLAG_NOCOMP
) {
148 error_setg(errp
, "multifd %d: flags received %x flags expected %x",
149 p
->id
, flags
, MULTIFD_FLAG_NOCOMP
);
152 return qio_channel_readv_all(p
->c
, p
->pages
->iov
, used
, 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 .send_write
= nocomp_send_write
,
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 %d "
213 "expected %d", 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 %d "
230 "expected %d", 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
->iov
= g_new0(struct iovec
, size
);
243 pages
->offset
= g_new0(ram_addr_t
, size
);
248 static void multifd_pages_clear(MultiFDPages_t
*pages
)
251 pages
->allocated
= 0;
252 pages
->packet_num
= 0;
256 g_free(pages
->offset
);
257 pages
->offset
= NULL
;
261 static void multifd_send_fill_packet(MultiFDSendParams
*p
)
263 MultiFDPacket_t
*packet
= p
->packet
;
266 packet
->flags
= cpu_to_be32(p
->flags
);
267 packet
->pages_alloc
= cpu_to_be32(p
->pages
->allocated
);
268 packet
->pages_used
= cpu_to_be32(p
->pages
->used
);
269 packet
->next_packet_size
= cpu_to_be32(p
->next_packet_size
);
270 packet
->packet_num
= cpu_to_be64(p
->packet_num
);
272 if (p
->pages
->block
) {
273 strncpy(packet
->ramblock
, p
->pages
->block
->idstr
, 256);
276 for (i
= 0; i
< p
->pages
->used
; i
++) {
277 /* there are architectures where ram_addr_t is 32 bit */
278 uint64_t temp
= p
->pages
->offset
[i
];
280 packet
->offset
[i
] = cpu_to_be64(temp
);
284 static int multifd_recv_unfill_packet(MultiFDRecvParams
*p
, Error
**errp
)
286 MultiFDPacket_t
*packet
= p
->packet
;
287 uint32_t pages_max
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
291 packet
->magic
= be32_to_cpu(packet
->magic
);
292 if (packet
->magic
!= MULTIFD_MAGIC
) {
293 error_setg(errp
, "multifd: received packet "
294 "magic %x and expected magic %x",
295 packet
->magic
, MULTIFD_MAGIC
);
299 packet
->version
= be32_to_cpu(packet
->version
);
300 if (packet
->version
!= MULTIFD_VERSION
) {
301 error_setg(errp
, "multifd: received packet "
302 "version %d and expected version %d",
303 packet
->version
, MULTIFD_VERSION
);
307 p
->flags
= be32_to_cpu(packet
->flags
);
309 packet
->pages_alloc
= be32_to_cpu(packet
->pages_alloc
);
311 * If we received a packet that is 100 times bigger than expected
312 * just stop migration. It is a magic number.
314 if (packet
->pages_alloc
> pages_max
* 100) {
315 error_setg(errp
, "multifd: received packet "
316 "with size %d and expected a maximum size of %d",
317 packet
->pages_alloc
, pages_max
* 100) ;
321 * We received a packet that is bigger than expected but inside
322 * reasonable limits (see previous comment). Just reallocate.
324 if (packet
->pages_alloc
> p
->pages
->allocated
) {
325 multifd_pages_clear(p
->pages
);
326 p
->pages
= multifd_pages_init(packet
->pages_alloc
);
329 p
->pages
->used
= be32_to_cpu(packet
->pages_used
);
330 if (p
->pages
->used
> packet
->pages_alloc
) {
331 error_setg(errp
, "multifd: received packet "
332 "with %d pages and expected maximum pages are %d",
333 p
->pages
->used
, packet
->pages_alloc
) ;
337 p
->next_packet_size
= be32_to_cpu(packet
->next_packet_size
);
338 p
->packet_num
= be64_to_cpu(packet
->packet_num
);
340 if (p
->pages
->used
== 0) {
344 /* make sure that ramblock is 0 terminated */
345 packet
->ramblock
[255] = 0;
346 block
= qemu_ram_block_by_name(packet
->ramblock
);
348 error_setg(errp
, "multifd: unknown ram block %s",
353 for (i
= 0; i
< p
->pages
->used
; i
++) {
354 uint64_t offset
= be64_to_cpu(packet
->offset
[i
]);
356 if (offset
> (block
->used_length
- qemu_target_page_size())) {
357 error_setg(errp
, "multifd: offset too long %" PRIu64
358 " (max " RAM_ADDR_FMT
")",
359 offset
, block
->max_length
);
362 p
->pages
->iov
[i
].iov_base
= block
->host
+ offset
;
363 p
->pages
->iov
[i
].iov_len
= qemu_target_page_size();
370 MultiFDSendParams
*params
;
371 /* array of pages to sent */
372 MultiFDPages_t
*pages
;
373 /* global number of generated multifd packets */
375 /* send channels ready */
376 QemuSemaphore channels_ready
;
378 * Have we already run terminate threads. There is a race when it
379 * happens that we got one error while we are exiting.
380 * We will use atomic operations. Only valid values are 0 and 1.
385 } *multifd_send_state
;
388 * How we use multifd_send_state->pages and channel->pages?
390 * We create a pages for each channel, and a main one. Each time that
391 * we need to send a batch of pages we interchange the ones between
392 * multifd_send_state and the channel that is sending it. There are
393 * two reasons for that:
394 * - to not have to do so many mallocs during migration
395 * - to make easier to know what to free at the end of migration
397 * This way we always know who is the owner of each "pages" struct,
398 * and we don't need any locking. It belongs to the migration thread
399 * or to the channel thread. Switching is safe because the migration
400 * thread is using the channel mutex when changing it, and the channel
401 * have to had finish with its own, otherwise pending_job can't be
405 static int multifd_send_pages(QEMUFile
*f
)
408 static int next_channel
;
409 MultiFDSendParams
*p
= NULL
; /* make happy gcc */
410 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
411 uint64_t transferred
;
413 if (atomic_read(&multifd_send_state
->exiting
)) {
417 qemu_sem_wait(&multifd_send_state
->channels_ready
);
418 for (i
= next_channel
;; i
= (i
+ 1) % migrate_multifd_channels()) {
419 p
= &multifd_send_state
->params
[i
];
421 qemu_mutex_lock(&p
->mutex
);
423 error_report("%s: channel %d has already quit!", __func__
, i
);
424 qemu_mutex_unlock(&p
->mutex
);
427 if (!p
->pending_job
) {
429 next_channel
= (i
+ 1) % migrate_multifd_channels();
432 qemu_mutex_unlock(&p
->mutex
);
434 assert(!p
->pages
->used
);
435 assert(!p
->pages
->block
);
437 p
->packet_num
= multifd_send_state
->packet_num
++;
438 multifd_send_state
->pages
= p
->pages
;
440 transferred
= ((uint64_t) pages
->used
) * qemu_target_page_size()
442 qemu_file_update_transfer(f
, transferred
);
443 ram_counters
.multifd_bytes
+= transferred
;
444 ram_counters
.transferred
+= transferred
;;
445 qemu_mutex_unlock(&p
->mutex
);
446 qemu_sem_post(&p
->sem
);
451 int multifd_queue_page(QEMUFile
*f
, RAMBlock
*block
, ram_addr_t offset
)
453 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
456 pages
->block
= block
;
459 if (pages
->block
== block
) {
460 pages
->offset
[pages
->used
] = offset
;
461 pages
->iov
[pages
->used
].iov_base
= block
->host
+ offset
;
462 pages
->iov
[pages
->used
].iov_len
= qemu_target_page_size();
465 if (pages
->used
< pages
->allocated
) {
470 if (multifd_send_pages(f
) < 0) {
474 if (pages
->block
!= block
) {
475 return multifd_queue_page(f
, block
, offset
);
481 static void multifd_send_terminate_threads(Error
*err
)
485 trace_multifd_send_terminate_threads(err
!= NULL
);
488 MigrationState
*s
= migrate_get_current();
489 migrate_set_error(s
, err
);
490 if (s
->state
== MIGRATION_STATUS_SETUP
||
491 s
->state
== MIGRATION_STATUS_PRE_SWITCHOVER
||
492 s
->state
== MIGRATION_STATUS_DEVICE
||
493 s
->state
== MIGRATION_STATUS_ACTIVE
) {
494 migrate_set_state(&s
->state
, s
->state
,
495 MIGRATION_STATUS_FAILED
);
500 * We don't want to exit each threads twice. Depending on where
501 * we get the error, or if there are two independent errors in two
502 * threads at the same time, we can end calling this function
505 if (atomic_xchg(&multifd_send_state
->exiting
, 1)) {
509 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
510 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
512 qemu_mutex_lock(&p
->mutex
);
514 qemu_sem_post(&p
->sem
);
515 qemu_mutex_unlock(&p
->mutex
);
519 void multifd_save_cleanup(void)
523 if (!migrate_use_multifd()) {
526 multifd_send_terminate_threads(NULL
);
527 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
528 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
531 qemu_thread_join(&p
->thread
);
534 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
535 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
536 Error
*local_err
= NULL
;
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
);
550 multifd_send_state
->ops
->send_cleanup(p
, &local_err
);
552 migrate_set_error(migrate_get_current(), local_err
);
553 error_free(local_err
);
556 qemu_sem_destroy(&multifd_send_state
->channels_ready
);
557 g_free(multifd_send_state
->params
);
558 multifd_send_state
->params
= NULL
;
559 multifd_pages_clear(multifd_send_state
->pages
);
560 multifd_send_state
->pages
= NULL
;
561 g_free(multifd_send_state
);
562 multifd_send_state
= NULL
;
565 void multifd_send_sync_main(QEMUFile
*f
)
569 if (!migrate_use_multifd()) {
572 if (multifd_send_state
->pages
->used
) {
573 if (multifd_send_pages(f
) < 0) {
574 error_report("%s: multifd_send_pages fail", __func__
);
578 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
579 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
581 trace_multifd_send_sync_main_signal(p
->id
);
583 qemu_mutex_lock(&p
->mutex
);
586 error_report("%s: channel %d has already quit", __func__
, i
);
587 qemu_mutex_unlock(&p
->mutex
);
591 p
->packet_num
= multifd_send_state
->packet_num
++;
592 p
->flags
|= MULTIFD_FLAG_SYNC
;
594 qemu_file_update_transfer(f
, p
->packet_len
);
595 ram_counters
.multifd_bytes
+= p
->packet_len
;
596 ram_counters
.transferred
+= p
->packet_len
;
597 qemu_mutex_unlock(&p
->mutex
);
598 qemu_sem_post(&p
->sem
);
600 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
601 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
603 trace_multifd_send_sync_main_wait(p
->id
);
604 qemu_sem_wait(&p
->sem_sync
);
606 trace_multifd_send_sync_main(multifd_send_state
->packet_num
);
609 static void *multifd_send_thread(void *opaque
)
611 MultiFDSendParams
*p
= opaque
;
612 Error
*local_err
= NULL
;
616 trace_multifd_send_thread_start(p
->id
);
617 rcu_register_thread();
619 if (multifd_send_initial_packet(p
, &local_err
) < 0) {
627 qemu_sem_wait(&p
->sem
);
629 if (atomic_read(&multifd_send_state
->exiting
)) {
632 qemu_mutex_lock(&p
->mutex
);
634 if (p
->pending_job
) {
635 uint32_t used
= p
->pages
->used
;
636 uint64_t packet_num
= p
->packet_num
;
640 ret
= multifd_send_state
->ops
->send_prepare(p
, used
,
643 qemu_mutex_unlock(&p
->mutex
);
647 multifd_send_fill_packet(p
);
650 p
->num_pages
+= used
;
652 p
->pages
->block
= NULL
;
653 qemu_mutex_unlock(&p
->mutex
);
655 trace_multifd_send(p
->id
, packet_num
, used
, flags
,
656 p
->next_packet_size
);
658 ret
= qio_channel_write_all(p
->c
, (void *)p
->packet
,
659 p
->packet_len
, &local_err
);
665 ret
= multifd_send_state
->ops
->send_write(p
, used
, &local_err
);
671 qemu_mutex_lock(&p
->mutex
);
673 qemu_mutex_unlock(&p
->mutex
);
675 if (flags
& MULTIFD_FLAG_SYNC
) {
676 qemu_sem_post(&p
->sem_sync
);
678 qemu_sem_post(&multifd_send_state
->channels_ready
);
679 } else if (p
->quit
) {
680 qemu_mutex_unlock(&p
->mutex
);
683 qemu_mutex_unlock(&p
->mutex
);
684 /* sometimes there are spurious wakeups */
690 trace_multifd_send_error(p
->id
);
691 multifd_send_terminate_threads(local_err
);
692 error_free(local_err
);
696 * Error happen, I will exit, but I can't just leave, tell
697 * who pay attention to me.
700 qemu_sem_post(&p
->sem_sync
);
701 qemu_sem_post(&multifd_send_state
->channels_ready
);
704 qemu_mutex_lock(&p
->mutex
);
706 qemu_mutex_unlock(&p
->mutex
);
708 rcu_unregister_thread();
709 trace_multifd_send_thread_end(p
->id
, p
->num_packets
, p
->num_pages
);
714 static void multifd_new_send_channel_async(QIOTask
*task
, gpointer opaque
)
716 MultiFDSendParams
*p
= opaque
;
717 QIOChannel
*sioc
= QIO_CHANNEL(qio_task_get_source(task
));
718 Error
*local_err
= NULL
;
720 trace_multifd_new_send_channel_async(p
->id
);
721 if (qio_task_propagate_error(task
, &local_err
)) {
722 migrate_set_error(migrate_get_current(), local_err
);
723 /* Error happen, we need to tell who pay attention to me */
724 qemu_sem_post(&multifd_send_state
->channels_ready
);
725 qemu_sem_post(&p
->sem_sync
);
727 * Although multifd_send_thread is not created, but main migration
728 * thread neet to judge whether it is running, so we need to mark
732 object_unref(OBJECT(sioc
));
733 error_free(local_err
);
735 p
->c
= QIO_CHANNEL(sioc
);
736 qio_channel_set_delay(p
->c
, false);
738 qemu_thread_create(&p
->thread
, p
->name
, multifd_send_thread
, p
,
739 QEMU_THREAD_JOINABLE
);
743 int multifd_save_setup(Error
**errp
)
746 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
749 if (!migrate_use_multifd()) {
752 thread_count
= migrate_multifd_channels();
753 multifd_send_state
= g_malloc0(sizeof(*multifd_send_state
));
754 multifd_send_state
->params
= g_new0(MultiFDSendParams
, thread_count
);
755 multifd_send_state
->pages
= multifd_pages_init(page_count
);
756 qemu_sem_init(&multifd_send_state
->channels_ready
, 0);
757 atomic_set(&multifd_send_state
->exiting
, 0);
758 multifd_send_state
->ops
= multifd_ops
[migrate_multifd_compression()];
760 for (i
= 0; i
< thread_count
; i
++) {
761 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
763 qemu_mutex_init(&p
->mutex
);
764 qemu_sem_init(&p
->sem
, 0);
765 qemu_sem_init(&p
->sem_sync
, 0);
769 p
->pages
= multifd_pages_init(page_count
);
770 p
->packet_len
= sizeof(MultiFDPacket_t
)
771 + sizeof(uint64_t) * page_count
;
772 p
->packet
= g_malloc0(p
->packet_len
);
773 p
->packet
->magic
= cpu_to_be32(MULTIFD_MAGIC
);
774 p
->packet
->version
= cpu_to_be32(MULTIFD_VERSION
);
775 p
->name
= g_strdup_printf("multifdsend_%d", i
);
776 socket_send_channel_create(multifd_new_send_channel_async
, p
);
779 for (i
= 0; i
< thread_count
; i
++) {
780 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
781 Error
*local_err
= NULL
;
784 ret
= multifd_send_state
->ops
->send_setup(p
, &local_err
);
786 error_propagate(errp
, local_err
);
794 MultiFDRecvParams
*params
;
795 /* number of created threads */
797 /* syncs main thread and channels */
798 QemuSemaphore sem_sync
;
799 /* global number of generated multifd packets */
803 } *multifd_recv_state
;
805 static void multifd_recv_terminate_threads(Error
*err
)
809 trace_multifd_recv_terminate_threads(err
!= NULL
);
812 MigrationState
*s
= migrate_get_current();
813 migrate_set_error(s
, err
);
814 if (s
->state
== MIGRATION_STATUS_SETUP
||
815 s
->state
== MIGRATION_STATUS_ACTIVE
) {
816 migrate_set_state(&s
->state
, s
->state
,
817 MIGRATION_STATUS_FAILED
);
821 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
822 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
824 qemu_mutex_lock(&p
->mutex
);
827 * We could arrive here for two reasons:
828 * - normal quit, i.e. everything went fine, just finished
829 * - error quit: We close the channels so the channel threads
830 * finish the qio_channel_read_all_eof()
833 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
835 qemu_mutex_unlock(&p
->mutex
);
839 int multifd_load_cleanup(Error
**errp
)
843 if (!migrate_use_multifd()) {
846 multifd_recv_terminate_threads(NULL
);
847 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
848 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
853 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
854 * however try to wakeup it without harm in cleanup phase.
856 qemu_sem_post(&p
->sem_sync
);
857 qemu_thread_join(&p
->thread
);
860 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
861 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
863 object_unref(OBJECT(p
->c
));
865 qemu_mutex_destroy(&p
->mutex
);
866 qemu_sem_destroy(&p
->sem_sync
);
869 multifd_pages_clear(p
->pages
);
874 multifd_recv_state
->ops
->recv_cleanup(p
);
876 qemu_sem_destroy(&multifd_recv_state
->sem_sync
);
877 g_free(multifd_recv_state
->params
);
878 multifd_recv_state
->params
= NULL
;
879 g_free(multifd_recv_state
);
880 multifd_recv_state
= NULL
;
885 void multifd_recv_sync_main(void)
889 if (!migrate_use_multifd()) {
892 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
893 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
895 trace_multifd_recv_sync_main_wait(p
->id
);
896 qemu_sem_wait(&multifd_recv_state
->sem_sync
);
898 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
899 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
901 WITH_QEMU_LOCK_GUARD(&p
->mutex
) {
902 if (multifd_recv_state
->packet_num
< p
->packet_num
) {
903 multifd_recv_state
->packet_num
= p
->packet_num
;
906 trace_multifd_recv_sync_main_signal(p
->id
);
907 qemu_sem_post(&p
->sem_sync
);
909 trace_multifd_recv_sync_main(multifd_recv_state
->packet_num
);
912 static void *multifd_recv_thread(void *opaque
)
914 MultiFDRecvParams
*p
= opaque
;
915 Error
*local_err
= NULL
;
918 trace_multifd_recv_thread_start(p
->id
);
919 rcu_register_thread();
929 ret
= qio_channel_read_all_eof(p
->c
, (void *)p
->packet
,
930 p
->packet_len
, &local_err
);
931 if (ret
== 0) { /* EOF */
934 if (ret
== -1) { /* Error */
938 qemu_mutex_lock(&p
->mutex
);
939 ret
= multifd_recv_unfill_packet(p
, &local_err
);
941 qemu_mutex_unlock(&p
->mutex
);
945 used
= p
->pages
->used
;
947 /* recv methods don't know how to handle the SYNC flag */
948 p
->flags
&= ~MULTIFD_FLAG_SYNC
;
949 trace_multifd_recv(p
->id
, p
->packet_num
, used
, flags
,
950 p
->next_packet_size
);
952 p
->num_pages
+= used
;
953 qemu_mutex_unlock(&p
->mutex
);
956 ret
= multifd_recv_state
->ops
->recv_pages(p
, used
, &local_err
);
962 if (flags
& MULTIFD_FLAG_SYNC
) {
963 qemu_sem_post(&multifd_recv_state
->sem_sync
);
964 qemu_sem_wait(&p
->sem_sync
);
969 multifd_recv_terminate_threads(local_err
);
970 error_free(local_err
);
972 qemu_mutex_lock(&p
->mutex
);
974 qemu_mutex_unlock(&p
->mutex
);
976 rcu_unregister_thread();
977 trace_multifd_recv_thread_end(p
->id
, p
->num_packets
, p
->num_pages
);
982 int multifd_load_setup(Error
**errp
)
985 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
988 if (!migrate_use_multifd()) {
991 thread_count
= migrate_multifd_channels();
992 multifd_recv_state
= g_malloc0(sizeof(*multifd_recv_state
));
993 multifd_recv_state
->params
= g_new0(MultiFDRecvParams
, thread_count
);
994 atomic_set(&multifd_recv_state
->count
, 0);
995 qemu_sem_init(&multifd_recv_state
->sem_sync
, 0);
996 multifd_recv_state
->ops
= multifd_ops
[migrate_multifd_compression()];
998 for (i
= 0; i
< thread_count
; i
++) {
999 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1001 qemu_mutex_init(&p
->mutex
);
1002 qemu_sem_init(&p
->sem_sync
, 0);
1005 p
->pages
= multifd_pages_init(page_count
);
1006 p
->packet_len
= sizeof(MultiFDPacket_t
)
1007 + sizeof(uint64_t) * page_count
;
1008 p
->packet
= g_malloc0(p
->packet_len
);
1009 p
->name
= g_strdup_printf("multifdrecv_%d", i
);
1012 for (i
= 0; i
< thread_count
; i
++) {
1013 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1014 Error
*local_err
= NULL
;
1017 ret
= multifd_recv_state
->ops
->recv_setup(p
, &local_err
);
1019 error_propagate(errp
, local_err
);
1026 bool multifd_recv_all_channels_created(void)
1028 int thread_count
= migrate_multifd_channels();
1030 if (!migrate_use_multifd()) {
1034 return thread_count
== atomic_read(&multifd_recv_state
->count
);
1038 * Try to receive all multifd channels to get ready for the migration.
1039 * - Return true and do not set @errp when correctly receving all channels;
1040 * - Return false and do not set @errp when correctly receiving the current one;
1041 * - Return false and set @errp when failing to receive the current channel.
1043 bool multifd_recv_new_channel(QIOChannel
*ioc
, Error
**errp
)
1045 MultiFDRecvParams
*p
;
1046 Error
*local_err
= NULL
;
1049 id
= multifd_recv_initial_packet(ioc
, &local_err
);
1051 multifd_recv_terminate_threads(local_err
);
1052 error_propagate_prepend(errp
, local_err
,
1053 "failed to receive packet"
1054 " via multifd channel %d: ",
1055 atomic_read(&multifd_recv_state
->count
));
1058 trace_multifd_recv_new_channel(id
);
1060 p
= &multifd_recv_state
->params
[id
];
1062 error_setg(&local_err
, "multifd: received id '%d' already setup'",
1064 multifd_recv_terminate_threads(local_err
);
1065 error_propagate(errp
, local_err
);
1069 object_ref(OBJECT(ioc
));
1070 /* initial packet */
1074 qemu_thread_create(&p
->thread
, p
->name
, multifd_recv_thread
, p
,
1075 QEMU_THREAD_JOINABLE
);
1076 atomic_inc(&multifd_recv_state
->count
);
1077 return atomic_read(&multifd_recv_state
->count
) ==
1078 migrate_multifd_channels();