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 static int multifd_send_initial_packet(MultiFDSendParams
*p
, Error
**errp
)
171 MultiFDInit_t msg
= {};
174 msg
.magic
= cpu_to_be32(MULTIFD_MAGIC
);
175 msg
.version
= cpu_to_be32(MULTIFD_VERSION
);
177 memcpy(msg
.uuid
, &qemu_uuid
.data
, sizeof(msg
.uuid
));
179 ret
= qio_channel_write_all(p
->c
, (char *)&msg
, sizeof(msg
), errp
);
186 static int multifd_recv_initial_packet(QIOChannel
*c
, Error
**errp
)
191 ret
= qio_channel_read_all(c
, (char *)&msg
, sizeof(msg
), errp
);
196 msg
.magic
= be32_to_cpu(msg
.magic
);
197 msg
.version
= be32_to_cpu(msg
.version
);
199 if (msg
.magic
!= MULTIFD_MAGIC
) {
200 error_setg(errp
, "multifd: received packet magic %x "
201 "expected %x", msg
.magic
, MULTIFD_MAGIC
);
205 if (msg
.version
!= MULTIFD_VERSION
) {
206 error_setg(errp
, "multifd: received packet version %d "
207 "expected %d", msg
.version
, MULTIFD_VERSION
);
211 if (memcmp(msg
.uuid
, &qemu_uuid
, sizeof(qemu_uuid
))) {
212 char *uuid
= qemu_uuid_unparse_strdup(&qemu_uuid
);
213 char *msg_uuid
= qemu_uuid_unparse_strdup((const QemuUUID
*)msg
.uuid
);
215 error_setg(errp
, "multifd: received uuid '%s' and expected "
216 "uuid '%s' for channel %hhd", msg_uuid
, uuid
, msg
.id
);
222 if (msg
.id
> migrate_multifd_channels()) {
223 error_setg(errp
, "multifd: received channel version %d "
224 "expected %d", msg
.version
, MULTIFD_VERSION
);
231 static MultiFDPages_t
*multifd_pages_init(size_t size
)
233 MultiFDPages_t
*pages
= g_new0(MultiFDPages_t
, 1);
235 pages
->allocated
= size
;
236 pages
->iov
= g_new0(struct iovec
, size
);
237 pages
->offset
= g_new0(ram_addr_t
, size
);
242 static void multifd_pages_clear(MultiFDPages_t
*pages
)
245 pages
->allocated
= 0;
246 pages
->packet_num
= 0;
250 g_free(pages
->offset
);
251 pages
->offset
= NULL
;
255 static void multifd_send_fill_packet(MultiFDSendParams
*p
)
257 MultiFDPacket_t
*packet
= p
->packet
;
260 packet
->flags
= cpu_to_be32(p
->flags
);
261 packet
->pages_alloc
= cpu_to_be32(p
->pages
->allocated
);
262 packet
->pages_used
= cpu_to_be32(p
->pages
->used
);
263 packet
->next_packet_size
= cpu_to_be32(p
->next_packet_size
);
264 packet
->packet_num
= cpu_to_be64(p
->packet_num
);
266 if (p
->pages
->block
) {
267 strncpy(packet
->ramblock
, p
->pages
->block
->idstr
, 256);
270 for (i
= 0; i
< p
->pages
->used
; i
++) {
271 /* there are architectures where ram_addr_t is 32 bit */
272 uint64_t temp
= p
->pages
->offset
[i
];
274 packet
->offset
[i
] = cpu_to_be64(temp
);
278 static int multifd_recv_unfill_packet(MultiFDRecvParams
*p
, Error
**errp
)
280 MultiFDPacket_t
*packet
= p
->packet
;
281 uint32_t pages_max
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
285 packet
->magic
= be32_to_cpu(packet
->magic
);
286 if (packet
->magic
!= MULTIFD_MAGIC
) {
287 error_setg(errp
, "multifd: received packet "
288 "magic %x and expected magic %x",
289 packet
->magic
, MULTIFD_MAGIC
);
293 packet
->version
= be32_to_cpu(packet
->version
);
294 if (packet
->version
!= MULTIFD_VERSION
) {
295 error_setg(errp
, "multifd: received packet "
296 "version %d and expected version %d",
297 packet
->version
, MULTIFD_VERSION
);
301 p
->flags
= be32_to_cpu(packet
->flags
);
303 packet
->pages_alloc
= be32_to_cpu(packet
->pages_alloc
);
305 * If we received a packet that is 100 times bigger than expected
306 * just stop migration. It is a magic number.
308 if (packet
->pages_alloc
> pages_max
* 100) {
309 error_setg(errp
, "multifd: received packet "
310 "with size %d and expected a maximum size of %d",
311 packet
->pages_alloc
, pages_max
* 100) ;
315 * We received a packet that is bigger than expected but inside
316 * reasonable limits (see previous comment). Just reallocate.
318 if (packet
->pages_alloc
> p
->pages
->allocated
) {
319 multifd_pages_clear(p
->pages
);
320 p
->pages
= multifd_pages_init(packet
->pages_alloc
);
323 p
->pages
->used
= be32_to_cpu(packet
->pages_used
);
324 if (p
->pages
->used
> packet
->pages_alloc
) {
325 error_setg(errp
, "multifd: received packet "
326 "with %d pages and expected maximum pages are %d",
327 p
->pages
->used
, packet
->pages_alloc
) ;
331 p
->next_packet_size
= be32_to_cpu(packet
->next_packet_size
);
332 p
->packet_num
= be64_to_cpu(packet
->packet_num
);
334 if (p
->pages
->used
== 0) {
338 /* make sure that ramblock is 0 terminated */
339 packet
->ramblock
[255] = 0;
340 block
= qemu_ram_block_by_name(packet
->ramblock
);
342 error_setg(errp
, "multifd: unknown ram block %s",
347 for (i
= 0; i
< p
->pages
->used
; i
++) {
348 uint64_t offset
= be64_to_cpu(packet
->offset
[i
]);
350 if (offset
> (block
->used_length
- qemu_target_page_size())) {
351 error_setg(errp
, "multifd: offset too long %" PRIu64
352 " (max " RAM_ADDR_FMT
")",
353 offset
, block
->max_length
);
356 p
->pages
->iov
[i
].iov_base
= block
->host
+ offset
;
357 p
->pages
->iov
[i
].iov_len
= qemu_target_page_size();
364 MultiFDSendParams
*params
;
365 /* array of pages to sent */
366 MultiFDPages_t
*pages
;
367 /* global number of generated multifd packets */
369 /* send channels ready */
370 QemuSemaphore channels_ready
;
372 * Have we already run terminate threads. There is a race when it
373 * happens that we got one error while we are exiting.
374 * We will use atomic operations. Only valid values are 0 and 1.
379 } *multifd_send_state
;
382 * How we use multifd_send_state->pages and channel->pages?
384 * We create a pages for each channel, and a main one. Each time that
385 * we need to send a batch of pages we interchange the ones between
386 * multifd_send_state and the channel that is sending it. There are
387 * two reasons for that:
388 * - to not have to do so many mallocs during migration
389 * - to make easier to know what to free at the end of migration
391 * This way we always know who is the owner of each "pages" struct,
392 * and we don't need any locking. It belongs to the migration thread
393 * or to the channel thread. Switching is safe because the migration
394 * thread is using the channel mutex when changing it, and the channel
395 * have to had finish with its own, otherwise pending_job can't be
399 static int multifd_send_pages(QEMUFile
*f
)
402 static int next_channel
;
403 MultiFDSendParams
*p
= NULL
; /* make happy gcc */
404 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
405 uint64_t transferred
;
407 if (atomic_read(&multifd_send_state
->exiting
)) {
411 qemu_sem_wait(&multifd_send_state
->channels_ready
);
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
->used
);
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
->used
) * qemu_target_page_size()
436 qemu_file_update_transfer(f
, transferred
);
437 ram_counters
.multifd_bytes
+= transferred
;
438 ram_counters
.transferred
+= transferred
;;
439 qemu_mutex_unlock(&p
->mutex
);
440 qemu_sem_post(&p
->sem
);
445 int multifd_queue_page(QEMUFile
*f
, RAMBlock
*block
, ram_addr_t offset
)
447 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
450 pages
->block
= block
;
453 if (pages
->block
== block
) {
454 pages
->offset
[pages
->used
] = offset
;
455 pages
->iov
[pages
->used
].iov_base
= block
->host
+ offset
;
456 pages
->iov
[pages
->used
].iov_len
= qemu_target_page_size();
459 if (pages
->used
< pages
->allocated
) {
464 if (multifd_send_pages(f
) < 0) {
468 if (pages
->block
!= block
) {
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 (atomic_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
);
509 qemu_mutex_unlock(&p
->mutex
);
513 void multifd_save_cleanup(void)
517 if (!migrate_use_multifd()) {
520 multifd_send_terminate_threads(NULL
);
521 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
522 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
525 qemu_thread_join(&p
->thread
);
528 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
529 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
530 Error
*local_err
= NULL
;
532 socket_send_channel_destroy(p
->c
);
534 qemu_mutex_destroy(&p
->mutex
);
535 qemu_sem_destroy(&p
->sem
);
536 qemu_sem_destroy(&p
->sem_sync
);
539 multifd_pages_clear(p
->pages
);
544 multifd_send_state
->ops
->send_cleanup(p
, &local_err
);
546 migrate_set_error(migrate_get_current(), local_err
);
549 qemu_sem_destroy(&multifd_send_state
->channels_ready
);
550 g_free(multifd_send_state
->params
);
551 multifd_send_state
->params
= NULL
;
552 multifd_pages_clear(multifd_send_state
->pages
);
553 multifd_send_state
->pages
= NULL
;
554 g_free(multifd_send_state
);
555 multifd_send_state
= NULL
;
558 void multifd_send_sync_main(QEMUFile
*f
)
562 if (!migrate_use_multifd()) {
565 if (multifd_send_state
->pages
->used
) {
566 if (multifd_send_pages(f
) < 0) {
567 error_report("%s: multifd_send_pages fail", __func__
);
571 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
572 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
574 trace_multifd_send_sync_main_signal(p
->id
);
576 qemu_mutex_lock(&p
->mutex
);
579 error_report("%s: channel %d has already quit", __func__
, i
);
580 qemu_mutex_unlock(&p
->mutex
);
584 p
->packet_num
= multifd_send_state
->packet_num
++;
585 p
->flags
|= MULTIFD_FLAG_SYNC
;
587 qemu_file_update_transfer(f
, p
->packet_len
);
588 ram_counters
.multifd_bytes
+= p
->packet_len
;
589 ram_counters
.transferred
+= p
->packet_len
;
590 qemu_mutex_unlock(&p
->mutex
);
591 qemu_sem_post(&p
->sem
);
593 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
594 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
596 trace_multifd_send_sync_main_wait(p
->id
);
597 qemu_sem_wait(&p
->sem_sync
);
599 trace_multifd_send_sync_main(multifd_send_state
->packet_num
);
602 static void *multifd_send_thread(void *opaque
)
604 MultiFDSendParams
*p
= opaque
;
605 Error
*local_err
= NULL
;
609 trace_multifd_send_thread_start(p
->id
);
610 rcu_register_thread();
612 if (multifd_send_initial_packet(p
, &local_err
) < 0) {
620 qemu_sem_wait(&p
->sem
);
622 if (atomic_read(&multifd_send_state
->exiting
)) {
625 qemu_mutex_lock(&p
->mutex
);
627 if (p
->pending_job
) {
628 uint32_t used
= p
->pages
->used
;
629 uint64_t packet_num
= p
->packet_num
;
633 ret
= multifd_send_state
->ops
->send_prepare(p
, used
,
636 qemu_mutex_unlock(&p
->mutex
);
640 multifd_send_fill_packet(p
);
643 p
->num_pages
+= used
;
645 p
->pages
->block
= NULL
;
646 qemu_mutex_unlock(&p
->mutex
);
648 trace_multifd_send(p
->id
, packet_num
, used
, flags
,
649 p
->next_packet_size
);
651 ret
= qio_channel_write_all(p
->c
, (void *)p
->packet
,
652 p
->packet_len
, &local_err
);
658 ret
= multifd_send_state
->ops
->send_write(p
, used
, &local_err
);
664 qemu_mutex_lock(&p
->mutex
);
666 qemu_mutex_unlock(&p
->mutex
);
668 if (flags
& MULTIFD_FLAG_SYNC
) {
669 qemu_sem_post(&p
->sem_sync
);
671 qemu_sem_post(&multifd_send_state
->channels_ready
);
672 } else if (p
->quit
) {
673 qemu_mutex_unlock(&p
->mutex
);
676 qemu_mutex_unlock(&p
->mutex
);
677 /* sometimes there are spurious wakeups */
683 trace_multifd_send_error(p
->id
);
684 multifd_send_terminate_threads(local_err
);
688 * Error happen, I will exit, but I can't just leave, tell
689 * who pay attention to me.
692 qemu_sem_post(&p
->sem_sync
);
693 qemu_sem_post(&multifd_send_state
->channels_ready
);
696 qemu_mutex_lock(&p
->mutex
);
698 qemu_mutex_unlock(&p
->mutex
);
700 rcu_unregister_thread();
701 trace_multifd_send_thread_end(p
->id
, p
->num_packets
, p
->num_pages
);
706 static void multifd_new_send_channel_async(QIOTask
*task
, gpointer opaque
)
708 MultiFDSendParams
*p
= opaque
;
709 QIOChannel
*sioc
= QIO_CHANNEL(qio_task_get_source(task
));
710 Error
*local_err
= NULL
;
712 trace_multifd_new_send_channel_async(p
->id
);
713 if (qio_task_propagate_error(task
, &local_err
)) {
714 migrate_set_error(migrate_get_current(), local_err
);
715 /* Error happen, we need to tell who pay attention to me */
716 qemu_sem_post(&multifd_send_state
->channels_ready
);
717 qemu_sem_post(&p
->sem_sync
);
719 * Although multifd_send_thread is not created, but main migration
720 * thread neet to judge whether it is running, so we need to mark
725 p
->c
= QIO_CHANNEL(sioc
);
726 qio_channel_set_delay(p
->c
, false);
728 qemu_thread_create(&p
->thread
, p
->name
, multifd_send_thread
, p
,
729 QEMU_THREAD_JOINABLE
);
733 int multifd_save_setup(Error
**errp
)
736 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
739 if (!migrate_use_multifd()) {
742 thread_count
= migrate_multifd_channels();
743 multifd_send_state
= g_malloc0(sizeof(*multifd_send_state
));
744 multifd_send_state
->params
= g_new0(MultiFDSendParams
, thread_count
);
745 multifd_send_state
->pages
= multifd_pages_init(page_count
);
746 qemu_sem_init(&multifd_send_state
->channels_ready
, 0);
747 atomic_set(&multifd_send_state
->exiting
, 0);
748 multifd_send_state
->ops
= multifd_ops
[migrate_multifd_compression()];
750 for (i
= 0; i
< thread_count
; i
++) {
751 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
753 qemu_mutex_init(&p
->mutex
);
754 qemu_sem_init(&p
->sem
, 0);
755 qemu_sem_init(&p
->sem_sync
, 0);
759 p
->pages
= multifd_pages_init(page_count
);
760 p
->packet_len
= sizeof(MultiFDPacket_t
)
761 + sizeof(uint64_t) * page_count
;
762 p
->packet
= g_malloc0(p
->packet_len
);
763 p
->packet
->magic
= cpu_to_be32(MULTIFD_MAGIC
);
764 p
->packet
->version
= cpu_to_be32(MULTIFD_VERSION
);
765 p
->name
= g_strdup_printf("multifdsend_%d", i
);
766 socket_send_channel_create(multifd_new_send_channel_async
, p
);
769 for (i
= 0; i
< thread_count
; i
++) {
770 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
771 Error
*local_err
= NULL
;
774 ret
= multifd_send_state
->ops
->send_setup(p
, &local_err
);
776 error_propagate(errp
, local_err
);
784 MultiFDRecvParams
*params
;
785 /* number of created threads */
787 /* syncs main thread and channels */
788 QemuSemaphore sem_sync
;
789 /* global number of generated multifd packets */
793 } *multifd_recv_state
;
795 static void multifd_recv_terminate_threads(Error
*err
)
799 trace_multifd_recv_terminate_threads(err
!= NULL
);
802 MigrationState
*s
= migrate_get_current();
803 migrate_set_error(s
, err
);
804 if (s
->state
== MIGRATION_STATUS_SETUP
||
805 s
->state
== MIGRATION_STATUS_ACTIVE
) {
806 migrate_set_state(&s
->state
, s
->state
,
807 MIGRATION_STATUS_FAILED
);
811 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
812 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
814 qemu_mutex_lock(&p
->mutex
);
817 * We could arrive here for two reasons:
818 * - normal quit, i.e. everything went fine, just finished
819 * - error quit: We close the channels so the channel threads
820 * finish the qio_channel_read_all_eof()
823 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
825 qemu_mutex_unlock(&p
->mutex
);
829 int multifd_load_cleanup(Error
**errp
)
833 if (!migrate_use_multifd()) {
836 multifd_recv_terminate_threads(NULL
);
837 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
838 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
843 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
844 * however try to wakeup it without harm in cleanup phase.
846 qemu_sem_post(&p
->sem_sync
);
847 qemu_thread_join(&p
->thread
);
850 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
851 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
853 object_unref(OBJECT(p
->c
));
855 qemu_mutex_destroy(&p
->mutex
);
856 qemu_sem_destroy(&p
->sem_sync
);
859 multifd_pages_clear(p
->pages
);
864 multifd_recv_state
->ops
->recv_cleanup(p
);
866 qemu_sem_destroy(&multifd_recv_state
->sem_sync
);
867 g_free(multifd_recv_state
->params
);
868 multifd_recv_state
->params
= NULL
;
869 g_free(multifd_recv_state
);
870 multifd_recv_state
= NULL
;
875 void multifd_recv_sync_main(void)
879 if (!migrate_use_multifd()) {
882 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
883 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
885 trace_multifd_recv_sync_main_wait(p
->id
);
886 qemu_sem_wait(&multifd_recv_state
->sem_sync
);
888 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
889 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
891 qemu_mutex_lock(&p
->mutex
);
892 if (multifd_recv_state
->packet_num
< p
->packet_num
) {
893 multifd_recv_state
->packet_num
= p
->packet_num
;
895 qemu_mutex_unlock(&p
->mutex
);
896 trace_multifd_recv_sync_main_signal(p
->id
);
897 qemu_sem_post(&p
->sem_sync
);
899 trace_multifd_recv_sync_main(multifd_recv_state
->packet_num
);
902 static void *multifd_recv_thread(void *opaque
)
904 MultiFDRecvParams
*p
= opaque
;
905 Error
*local_err
= NULL
;
908 trace_multifd_recv_thread_start(p
->id
);
909 rcu_register_thread();
919 ret
= qio_channel_read_all_eof(p
->c
, (void *)p
->packet
,
920 p
->packet_len
, &local_err
);
921 if (ret
== 0) { /* EOF */
924 if (ret
== -1) { /* Error */
928 qemu_mutex_lock(&p
->mutex
);
929 ret
= multifd_recv_unfill_packet(p
, &local_err
);
931 qemu_mutex_unlock(&p
->mutex
);
935 used
= p
->pages
->used
;
937 /* recv methods don't know how to handle the SYNC flag */
938 p
->flags
&= ~MULTIFD_FLAG_SYNC
;
939 trace_multifd_recv(p
->id
, p
->packet_num
, used
, flags
,
940 p
->next_packet_size
);
942 p
->num_pages
+= used
;
943 qemu_mutex_unlock(&p
->mutex
);
946 ret
= multifd_recv_state
->ops
->recv_pages(p
, used
, &local_err
);
952 if (flags
& MULTIFD_FLAG_SYNC
) {
953 qemu_sem_post(&multifd_recv_state
->sem_sync
);
954 qemu_sem_wait(&p
->sem_sync
);
959 multifd_recv_terminate_threads(local_err
);
961 qemu_mutex_lock(&p
->mutex
);
963 qemu_mutex_unlock(&p
->mutex
);
965 rcu_unregister_thread();
966 trace_multifd_recv_thread_end(p
->id
, p
->num_packets
, p
->num_pages
);
971 int multifd_load_setup(Error
**errp
)
974 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
977 if (!migrate_use_multifd()) {
980 thread_count
= migrate_multifd_channels();
981 multifd_recv_state
= g_malloc0(sizeof(*multifd_recv_state
));
982 multifd_recv_state
->params
= g_new0(MultiFDRecvParams
, thread_count
);
983 atomic_set(&multifd_recv_state
->count
, 0);
984 qemu_sem_init(&multifd_recv_state
->sem_sync
, 0);
985 multifd_recv_state
->ops
= multifd_ops
[migrate_multifd_compression()];
987 for (i
= 0; i
< thread_count
; i
++) {
988 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
990 qemu_mutex_init(&p
->mutex
);
991 qemu_sem_init(&p
->sem_sync
, 0);
994 p
->pages
= multifd_pages_init(page_count
);
995 p
->packet_len
= sizeof(MultiFDPacket_t
)
996 + sizeof(uint64_t) * page_count
;
997 p
->packet
= g_malloc0(p
->packet_len
);
998 p
->name
= g_strdup_printf("multifdrecv_%d", i
);
1001 for (i
= 0; i
< thread_count
; i
++) {
1002 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1003 Error
*local_err
= NULL
;
1006 ret
= multifd_recv_state
->ops
->recv_setup(p
, &local_err
);
1008 error_propagate(errp
, local_err
);
1015 bool multifd_recv_all_channels_created(void)
1017 int thread_count
= migrate_multifd_channels();
1019 if (!migrate_use_multifd()) {
1023 return thread_count
== atomic_read(&multifd_recv_state
->count
);
1027 * Try to receive all multifd channels to get ready for the migration.
1028 * - Return true and do not set @errp when correctly receving all channels;
1029 * - Return false and do not set @errp when correctly receiving the current one;
1030 * - Return false and set @errp when failing to receive the current channel.
1032 bool multifd_recv_new_channel(QIOChannel
*ioc
, Error
**errp
)
1034 MultiFDRecvParams
*p
;
1035 Error
*local_err
= NULL
;
1038 id
= multifd_recv_initial_packet(ioc
, &local_err
);
1040 multifd_recv_terminate_threads(local_err
);
1041 error_propagate_prepend(errp
, local_err
,
1042 "failed to receive packet"
1043 " via multifd channel %d: ",
1044 atomic_read(&multifd_recv_state
->count
));
1047 trace_multifd_recv_new_channel(id
);
1049 p
= &multifd_recv_state
->params
[id
];
1051 error_setg(&local_err
, "multifd: received id '%d' already setup'",
1053 multifd_recv_terminate_threads(local_err
);
1054 error_propagate(errp
, local_err
);
1058 object_ref(OBJECT(ioc
));
1059 /* initial packet */
1063 qemu_thread_create(&p
->thread
, p
->name
, multifd_recv_thread
, p
,
1064 QEMU_THREAD_JOINABLE
);
1065 atomic_inc(&multifd_recv_state
->count
);
1066 return atomic_read(&multifd_recv_state
->count
) ==
1067 migrate_multifd_channels();