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"
27 #include "threadinfo.h"
29 #include "qemu/yank.h"
30 #include "io/channel-socket.h"
31 #include "yank_functions.h"
35 #define MULTIFD_MAGIC 0x11223344U
36 #define MULTIFD_VERSION 1
41 unsigned char uuid
[16]; /* QemuUUID */
43 uint8_t unused1
[7]; /* Reserved for future use */
44 uint64_t unused2
[4]; /* Reserved for future use */
45 } __attribute__((packed
)) MultiFDInit_t
;
47 /* Multifd without compression */
50 * nocomp_send_setup: setup send side
52 * For no compression this function does nothing.
54 * Returns 0 for success or -1 for error
56 * @p: Params for the channel that we are using
57 * @errp: pointer to an error
59 static int nocomp_send_setup(MultiFDSendParams
*p
, Error
**errp
)
65 * nocomp_send_cleanup: cleanup send side
67 * For no compression this function does nothing.
69 * @p: Params for the channel that we are using
70 * @errp: pointer to an error
72 static void nocomp_send_cleanup(MultiFDSendParams
*p
, Error
**errp
)
78 * nocomp_send_prepare: prepare date to be able to send
80 * For no compression we just have to calculate the size of the
83 * Returns 0 for success or -1 for error
85 * @p: Params for the channel that we are using
86 * @errp: pointer to an error
88 static int nocomp_send_prepare(MultiFDSendParams
*p
, Error
**errp
)
90 MultiFDPages_t
*pages
= p
->pages
;
92 for (int i
= 0; i
< p
->normal_num
; i
++) {
93 p
->iov
[p
->iovs_num
].iov_base
= pages
->block
->host
+ p
->normal
[i
];
94 p
->iov
[p
->iovs_num
].iov_len
= p
->page_size
;
98 p
->next_packet_size
= p
->normal_num
* p
->page_size
;
99 p
->flags
|= MULTIFD_FLAG_NOCOMP
;
104 * nocomp_recv_setup: setup receive side
106 * For no compression this function does nothing.
108 * Returns 0 for success or -1 for error
110 * @p: Params for the channel that we are using
111 * @errp: pointer to an error
113 static int nocomp_recv_setup(MultiFDRecvParams
*p
, Error
**errp
)
119 * nocomp_recv_cleanup: setup receive side
121 * For no compression this function does nothing.
123 * @p: Params for the channel that we are using
125 static void nocomp_recv_cleanup(MultiFDRecvParams
*p
)
130 * nocomp_recv_pages: read the data from the channel into actual pages
132 * For no compression we just need to read things into the correct place.
134 * Returns 0 for success or -1 for error
136 * @p: Params for the channel that we are using
137 * @errp: pointer to an error
139 static int nocomp_recv_pages(MultiFDRecvParams
*p
, Error
**errp
)
141 uint32_t flags
= p
->flags
& MULTIFD_FLAG_COMPRESSION_MASK
;
143 if (flags
!= MULTIFD_FLAG_NOCOMP
) {
144 error_setg(errp
, "multifd %u: flags received %x flags expected %x",
145 p
->id
, flags
, MULTIFD_FLAG_NOCOMP
);
148 for (int i
= 0; i
< p
->normal_num
; i
++) {
149 p
->iov
[i
].iov_base
= p
->host
+ p
->normal
[i
];
150 p
->iov
[i
].iov_len
= p
->page_size
;
152 return qio_channel_readv_all(p
->c
, p
->iov
, p
->normal_num
, 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 .recv_setup
= nocomp_recv_setup
,
160 .recv_cleanup
= nocomp_recv_cleanup
,
161 .recv_pages
= nocomp_recv_pages
164 static MultiFDMethods
*multifd_ops
[MULTIFD_COMPRESSION__MAX
] = {
165 [MULTIFD_COMPRESSION_NONE
] = &multifd_nocomp_ops
,
168 void multifd_register_ops(int method
, MultiFDMethods
*ops
)
170 assert(0 < method
&& method
< MULTIFD_COMPRESSION__MAX
);
171 multifd_ops
[method
] = ops
;
174 static int multifd_send_initial_packet(MultiFDSendParams
*p
, Error
**errp
)
176 MultiFDInit_t msg
= {};
179 msg
.magic
= cpu_to_be32(MULTIFD_MAGIC
);
180 msg
.version
= cpu_to_be32(MULTIFD_VERSION
);
182 memcpy(msg
.uuid
, &qemu_uuid
.data
, sizeof(msg
.uuid
));
184 ret
= qio_channel_write_all(p
->c
, (char *)&msg
, sizeof(msg
), errp
);
191 static int multifd_recv_initial_packet(QIOChannel
*c
, Error
**errp
)
196 ret
= qio_channel_read_all(c
, (char *)&msg
, sizeof(msg
), errp
);
201 msg
.magic
= be32_to_cpu(msg
.magic
);
202 msg
.version
= be32_to_cpu(msg
.version
);
204 if (msg
.magic
!= MULTIFD_MAGIC
) {
205 error_setg(errp
, "multifd: received packet magic %x "
206 "expected %x", msg
.magic
, MULTIFD_MAGIC
);
210 if (msg
.version
!= MULTIFD_VERSION
) {
211 error_setg(errp
, "multifd: received packet version %u "
212 "expected %u", msg
.version
, MULTIFD_VERSION
);
216 if (memcmp(msg
.uuid
, &qemu_uuid
, sizeof(qemu_uuid
))) {
217 char *uuid
= qemu_uuid_unparse_strdup(&qemu_uuid
);
218 char *msg_uuid
= qemu_uuid_unparse_strdup((const QemuUUID
*)msg
.uuid
);
220 error_setg(errp
, "multifd: received uuid '%s' and expected "
221 "uuid '%s' for channel %hhd", msg_uuid
, uuid
, msg
.id
);
227 if (msg
.id
> migrate_multifd_channels()) {
228 error_setg(errp
, "multifd: received channel version %u "
229 "expected %u", msg
.version
, MULTIFD_VERSION
);
236 static MultiFDPages_t
*multifd_pages_init(size_t size
)
238 MultiFDPages_t
*pages
= g_new0(MultiFDPages_t
, 1);
240 pages
->allocated
= size
;
241 pages
->offset
= g_new0(ram_addr_t
, size
);
246 static void multifd_pages_clear(MultiFDPages_t
*pages
)
249 pages
->allocated
= 0;
250 pages
->packet_num
= 0;
252 g_free(pages
->offset
);
253 pages
->offset
= NULL
;
257 static void multifd_send_fill_packet(MultiFDSendParams
*p
)
259 MultiFDPacket_t
*packet
= p
->packet
;
262 packet
->flags
= cpu_to_be32(p
->flags
);
263 packet
->pages_alloc
= cpu_to_be32(p
->pages
->allocated
);
264 packet
->normal_pages
= cpu_to_be32(p
->normal_num
);
265 packet
->next_packet_size
= cpu_to_be32(p
->next_packet_size
);
266 packet
->packet_num
= cpu_to_be64(p
->packet_num
);
268 if (p
->pages
->block
) {
269 strncpy(packet
->ramblock
, p
->pages
->block
->idstr
, 256);
272 for (i
= 0; i
< p
->normal_num
; i
++) {
273 /* there are architectures where ram_addr_t is 32 bit */
274 uint64_t temp
= p
->normal
[i
];
276 packet
->offset
[i
] = cpu_to_be64(temp
);
280 static int multifd_recv_unfill_packet(MultiFDRecvParams
*p
, Error
**errp
)
282 MultiFDPacket_t
*packet
= p
->packet
;
286 packet
->magic
= be32_to_cpu(packet
->magic
);
287 if (packet
->magic
!= MULTIFD_MAGIC
) {
288 error_setg(errp
, "multifd: received packet "
289 "magic %x and expected magic %x",
290 packet
->magic
, MULTIFD_MAGIC
);
294 packet
->version
= be32_to_cpu(packet
->version
);
295 if (packet
->version
!= MULTIFD_VERSION
) {
296 error_setg(errp
, "multifd: received packet "
297 "version %u and expected version %u",
298 packet
->version
, MULTIFD_VERSION
);
302 p
->flags
= be32_to_cpu(packet
->flags
);
304 packet
->pages_alloc
= be32_to_cpu(packet
->pages_alloc
);
306 * If we received a packet that is 100 times bigger than expected
307 * just stop migration. It is a magic number.
309 if (packet
->pages_alloc
> p
->page_count
) {
310 error_setg(errp
, "multifd: received packet "
311 "with size %u and expected a size of %u",
312 packet
->pages_alloc
, p
->page_count
) ;
316 p
->normal_num
= be32_to_cpu(packet
->normal_pages
);
317 if (p
->normal_num
> packet
->pages_alloc
) {
318 error_setg(errp
, "multifd: received packet "
319 "with %u pages and expected maximum pages are %u",
320 p
->normal_num
, packet
->pages_alloc
) ;
324 p
->next_packet_size
= be32_to_cpu(packet
->next_packet_size
);
325 p
->packet_num
= be64_to_cpu(packet
->packet_num
);
327 if (p
->normal_num
== 0) {
331 /* make sure that ramblock is 0 terminated */
332 packet
->ramblock
[255] = 0;
333 block
= qemu_ram_block_by_name(packet
->ramblock
);
335 error_setg(errp
, "multifd: unknown ram block %s",
340 p
->host
= block
->host
;
341 for (i
= 0; i
< p
->normal_num
; i
++) {
342 uint64_t offset
= be64_to_cpu(packet
->offset
[i
]);
344 if (offset
> (block
->used_length
- p
->page_size
)) {
345 error_setg(errp
, "multifd: offset too long %" PRIu64
346 " (max " RAM_ADDR_FMT
")",
347 offset
, block
->used_length
);
350 p
->normal
[i
] = offset
;
357 MultiFDSendParams
*params
;
358 /* array of pages to sent */
359 MultiFDPages_t
*pages
;
360 /* global number of generated multifd packets */
362 /* send channels ready */
363 QemuSemaphore channels_ready
;
365 * Have we already run terminate threads. There is a race when it
366 * happens that we got one error while we are exiting.
367 * We will use atomic operations. Only valid values are 0 and 1.
372 } *multifd_send_state
;
375 * How we use multifd_send_state->pages and channel->pages?
377 * We create a pages for each channel, and a main one. Each time that
378 * we need to send a batch of pages we interchange the ones between
379 * multifd_send_state and the channel that is sending it. There are
380 * two reasons for that:
381 * - to not have to do so many mallocs during migration
382 * - to make easier to know what to free at the end of migration
384 * This way we always know who is the owner of each "pages" struct,
385 * and we don't need any locking. It belongs to the migration thread
386 * or to the channel thread. Switching is safe because the migration
387 * thread is using the channel mutex when changing it, and the channel
388 * have to had finish with its own, otherwise pending_job can't be
392 static int multifd_send_pages(QEMUFile
*f
)
395 static int next_channel
;
396 MultiFDSendParams
*p
= NULL
; /* make happy gcc */
397 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
398 uint64_t transferred
;
400 if (qatomic_read(&multifd_send_state
->exiting
)) {
404 qemu_sem_wait(&multifd_send_state
->channels_ready
);
406 * next_channel can remain from a previous migration that was
407 * using more channels, so ensure it doesn't overflow if the
408 * limit is lower now.
410 next_channel
%= migrate_multifd_channels();
411 for (i
= next_channel
;; i
= (i
+ 1) % migrate_multifd_channels()) {
412 p
= &multifd_send_state
->params
[i
];
414 qemu_mutex_lock(&p
->mutex
);
416 error_report("%s: channel %d has already quit!", __func__
, i
);
417 qemu_mutex_unlock(&p
->mutex
);
420 if (!p
->pending_job
) {
422 next_channel
= (i
+ 1) % migrate_multifd_channels();
425 qemu_mutex_unlock(&p
->mutex
);
427 assert(!p
->pages
->num
);
428 assert(!p
->pages
->block
);
430 p
->packet_num
= multifd_send_state
->packet_num
++;
431 multifd_send_state
->pages
= p
->pages
;
433 transferred
= ((uint64_t) pages
->num
) * p
->page_size
+ p
->packet_len
;
434 qemu_file_acct_rate_limit(f
, transferred
);
435 qemu_mutex_unlock(&p
->mutex
);
436 stat64_add(&ram_counters
.transferred
, transferred
);
437 stat64_add(&ram_counters
.multifd_bytes
, transferred
);
438 qemu_sem_post(&p
->sem
);
443 int multifd_queue_page(QEMUFile
*f
, RAMBlock
*block
, ram_addr_t offset
)
445 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
446 bool changed
= false;
449 pages
->block
= block
;
452 if (pages
->block
== block
) {
453 pages
->offset
[pages
->num
] = offset
;
456 if (pages
->num
< pages
->allocated
) {
463 if (multifd_send_pages(f
) < 0) {
468 return multifd_queue_page(f
, block
, offset
);
474 static void multifd_send_terminate_threads(Error
*err
)
478 trace_multifd_send_terminate_threads(err
!= NULL
);
481 MigrationState
*s
= migrate_get_current();
482 migrate_set_error(s
, err
);
483 if (s
->state
== MIGRATION_STATUS_SETUP
||
484 s
->state
== MIGRATION_STATUS_PRE_SWITCHOVER
||
485 s
->state
== MIGRATION_STATUS_DEVICE
||
486 s
->state
== MIGRATION_STATUS_ACTIVE
) {
487 migrate_set_state(&s
->state
, s
->state
,
488 MIGRATION_STATUS_FAILED
);
493 * We don't want to exit each threads twice. Depending on where
494 * we get the error, or if there are two independent errors in two
495 * threads at the same time, we can end calling this function
498 if (qatomic_xchg(&multifd_send_state
->exiting
, 1)) {
502 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
503 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
505 qemu_mutex_lock(&p
->mutex
);
507 qemu_sem_post(&p
->sem
);
509 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
511 qemu_mutex_unlock(&p
->mutex
);
515 void multifd_save_cleanup(void)
519 if (!migrate_multifd()) {
522 multifd_send_terminate_threads(NULL
);
523 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
524 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
527 qemu_thread_join(&p
->thread
);
530 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
531 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
532 Error
*local_err
= NULL
;
534 if (p
->registered_yank
) {
535 migration_ioc_unregister_yank(p
->c
);
537 socket_send_channel_destroy(p
->c
);
539 qemu_mutex_destroy(&p
->mutex
);
540 qemu_sem_destroy(&p
->sem
);
541 qemu_sem_destroy(&p
->sem_sync
);
544 multifd_pages_clear(p
->pages
);
553 multifd_send_state
->ops
->send_cleanup(p
, &local_err
);
555 migrate_set_error(migrate_get_current(), local_err
);
556 error_free(local_err
);
559 qemu_sem_destroy(&multifd_send_state
->channels_ready
);
560 g_free(multifd_send_state
->params
);
561 multifd_send_state
->params
= NULL
;
562 multifd_pages_clear(multifd_send_state
->pages
);
563 multifd_send_state
->pages
= NULL
;
564 g_free(multifd_send_state
);
565 multifd_send_state
= NULL
;
568 static int multifd_zero_copy_flush(QIOChannel
*c
)
573 ret
= qio_channel_flush(c
, &err
);
575 error_report_err(err
);
579 stat64_add(&ram_counters
.dirty_sync_missed_zero_copy
, 1);
585 int multifd_send_sync_main(QEMUFile
*f
)
588 bool flush_zero_copy
;
590 if (!migrate_multifd()) {
593 if (multifd_send_state
->pages
->num
) {
594 if (multifd_send_pages(f
) < 0) {
595 error_report("%s: multifd_send_pages fail", __func__
);
601 * When using zero-copy, it's necessary to flush the pages before any of
602 * the pages can be sent again, so we'll make sure the new version of the
603 * pages will always arrive _later_ than the old pages.
605 * Currently we achieve this by flushing the zero-page requested writes
606 * per ram iteration, but in the future we could potentially optimize it
607 * to be less frequent, e.g. only after we finished one whole scanning of
608 * all the dirty bitmaps.
611 flush_zero_copy
= migrate_zero_copy_send();
613 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
614 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
616 trace_multifd_send_sync_main_signal(p
->id
);
618 qemu_mutex_lock(&p
->mutex
);
621 error_report("%s: channel %d has already quit", __func__
, i
);
622 qemu_mutex_unlock(&p
->mutex
);
626 p
->packet_num
= multifd_send_state
->packet_num
++;
627 p
->flags
|= MULTIFD_FLAG_SYNC
;
629 qemu_file_acct_rate_limit(f
, p
->packet_len
);
630 qemu_mutex_unlock(&p
->mutex
);
631 stat64_add(&ram_counters
.transferred
, p
->packet_len
);
632 stat64_add(&ram_counters
.multifd_bytes
, p
->packet_len
);
633 qemu_sem_post(&p
->sem
);
635 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
636 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
638 trace_multifd_send_sync_main_wait(p
->id
);
639 qemu_sem_wait(&p
->sem_sync
);
641 if (flush_zero_copy
&& p
->c
&& (multifd_zero_copy_flush(p
->c
) < 0)) {
645 trace_multifd_send_sync_main(multifd_send_state
->packet_num
);
650 static void *multifd_send_thread(void *opaque
)
652 MultiFDSendParams
*p
= opaque
;
653 MigrationThread
*thread
= NULL
;
654 Error
*local_err
= NULL
;
656 bool use_zero_copy_send
= migrate_zero_copy_send();
658 thread
= MigrationThreadAdd(p
->name
, qemu_get_thread_id());
660 trace_multifd_send_thread_start(p
->id
);
661 rcu_register_thread();
663 if (multifd_send_initial_packet(p
, &local_err
) < 0) {
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 qemu_sem_post(&multifd_send_state
->channels_ready
);
740 } else if (p
->quit
) {
741 qemu_mutex_unlock(&p
->mutex
);
744 qemu_mutex_unlock(&p
->mutex
);
745 /* sometimes there are spurious wakeups */
751 trace_multifd_send_error(p
->id
);
752 multifd_send_terminate_threads(local_err
);
753 error_free(local_err
);
757 * Error happen, I will exit, but I can't just leave, tell
758 * who pay attention to me.
761 qemu_sem_post(&p
->sem_sync
);
762 qemu_sem_post(&multifd_send_state
->channels_ready
);
765 qemu_mutex_lock(&p
->mutex
);
767 qemu_mutex_unlock(&p
->mutex
);
769 rcu_unregister_thread();
770 MigrationThreadDel(thread
);
771 trace_multifd_send_thread_end(p
->id
, p
->num_packets
, p
->total_normal_pages
);
776 static bool multifd_channel_connect(MultiFDSendParams
*p
,
780 static void multifd_tls_outgoing_handshake(QIOTask
*task
,
783 MultiFDSendParams
*p
= opaque
;
784 QIOChannel
*ioc
= QIO_CHANNEL(qio_task_get_source(task
));
787 if (qio_task_propagate_error(task
, &err
)) {
788 trace_multifd_tls_outgoing_handshake_error(ioc
, error_get_pretty(err
));
790 trace_multifd_tls_outgoing_handshake_complete(ioc
);
793 if (!multifd_channel_connect(p
, ioc
, err
)) {
795 * Error happen, mark multifd_send_thread status as 'quit' although it
796 * is not created, and then tell who pay attention to me.
799 qemu_sem_post(&multifd_send_state
->channels_ready
);
800 qemu_sem_post(&p
->sem_sync
);
804 static void *multifd_tls_handshake_thread(void *opaque
)
806 MultiFDSendParams
*p
= opaque
;
807 QIOChannelTLS
*tioc
= QIO_CHANNEL_TLS(p
->c
);
809 qio_channel_tls_handshake(tioc
,
810 multifd_tls_outgoing_handshake
,
817 static void multifd_tls_channel_connect(MultiFDSendParams
*p
,
821 MigrationState
*s
= migrate_get_current();
822 const char *hostname
= s
->hostname
;
825 tioc
= migration_tls_client_create(s
, ioc
, hostname
, errp
);
830 object_unref(OBJECT(ioc
));
831 trace_multifd_tls_outgoing_handshake_start(ioc
, tioc
, hostname
);
832 qio_channel_set_name(QIO_CHANNEL(tioc
), "multifd-tls-outgoing");
833 p
->c
= QIO_CHANNEL(tioc
);
834 qemu_thread_create(&p
->thread
, "multifd-tls-handshake-worker",
835 multifd_tls_handshake_thread
, p
,
836 QEMU_THREAD_JOINABLE
);
839 static bool multifd_channel_connect(MultiFDSendParams
*p
,
843 trace_multifd_set_outgoing_channel(
844 ioc
, object_get_typename(OBJECT(ioc
)),
845 migrate_get_current()->hostname
, error
);
850 if (migrate_channel_requires_tls_upgrade(ioc
)) {
851 multifd_tls_channel_connect(p
, ioc
, &error
);
854 * tls_channel_connect will call back to this
855 * function after the TLS handshake,
856 * so we mustn't call multifd_send_thread until then
863 migration_ioc_register_yank(ioc
);
864 p
->registered_yank
= true;
866 qemu_thread_create(&p
->thread
, p
->name
, multifd_send_thread
, p
,
867 QEMU_THREAD_JOINABLE
);
872 static void multifd_new_send_channel_cleanup(MultiFDSendParams
*p
,
873 QIOChannel
*ioc
, Error
*err
)
875 migrate_set_error(migrate_get_current(), err
);
876 /* Error happen, we need to tell who pay attention to me */
877 qemu_sem_post(&multifd_send_state
->channels_ready
);
878 qemu_sem_post(&p
->sem_sync
);
880 * Although multifd_send_thread is not created, but main migration
881 * thread neet to judge whether it is running, so we need to mark
885 object_unref(OBJECT(ioc
));
889 static void multifd_new_send_channel_async(QIOTask
*task
, gpointer opaque
)
891 MultiFDSendParams
*p
= opaque
;
892 QIOChannel
*sioc
= QIO_CHANNEL(qio_task_get_source(task
));
893 Error
*local_err
= NULL
;
895 trace_multifd_new_send_channel_async(p
->id
);
896 if (!qio_task_propagate_error(task
, &local_err
)) {
897 p
->c
= QIO_CHANNEL(sioc
);
898 qio_channel_set_delay(p
->c
, false);
900 if (multifd_channel_connect(p
, sioc
, local_err
)) {
905 multifd_new_send_channel_cleanup(p
, sioc
, local_err
);
908 int multifd_save_setup(Error
**errp
)
911 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
914 if (!migrate_multifd()) {
918 thread_count
= migrate_multifd_channels();
919 multifd_send_state
= g_malloc0(sizeof(*multifd_send_state
));
920 multifd_send_state
->params
= g_new0(MultiFDSendParams
, thread_count
);
921 multifd_send_state
->pages
= multifd_pages_init(page_count
);
922 qemu_sem_init(&multifd_send_state
->channels_ready
, 0);
923 qatomic_set(&multifd_send_state
->exiting
, 0);
924 multifd_send_state
->ops
= multifd_ops
[migrate_multifd_compression()];
926 for (i
= 0; i
< thread_count
; i
++) {
927 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
929 qemu_mutex_init(&p
->mutex
);
930 qemu_sem_init(&p
->sem
, 0);
931 qemu_sem_init(&p
->sem_sync
, 0);
935 p
->pages
= multifd_pages_init(page_count
);
936 p
->packet_len
= sizeof(MultiFDPacket_t
)
937 + sizeof(uint64_t) * page_count
;
938 p
->packet
= g_malloc0(p
->packet_len
);
939 p
->packet
->magic
= cpu_to_be32(MULTIFD_MAGIC
);
940 p
->packet
->version
= cpu_to_be32(MULTIFD_VERSION
);
941 p
->name
= g_strdup_printf("multifdsend_%d", i
);
942 /* We need one extra place for the packet header */
943 p
->iov
= g_new0(struct iovec
, page_count
+ 1);
944 p
->normal
= g_new0(ram_addr_t
, page_count
);
945 p
->page_size
= qemu_target_page_size();
946 p
->page_count
= page_count
;
948 if (migrate_zero_copy_send()) {
949 p
->write_flags
= QIO_CHANNEL_WRITE_FLAG_ZERO_COPY
;
954 socket_send_channel_create(multifd_new_send_channel_async
, p
);
957 for (i
= 0; i
< thread_count
; i
++) {
958 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
959 Error
*local_err
= NULL
;
962 ret
= multifd_send_state
->ops
->send_setup(p
, &local_err
);
964 error_propagate(errp
, local_err
);
972 MultiFDRecvParams
*params
;
973 /* number of created threads */
975 /* syncs main thread and channels */
976 QemuSemaphore sem_sync
;
977 /* global number of generated multifd packets */
981 } *multifd_recv_state
;
983 static void multifd_recv_terminate_threads(Error
*err
)
987 trace_multifd_recv_terminate_threads(err
!= NULL
);
990 MigrationState
*s
= migrate_get_current();
991 migrate_set_error(s
, err
);
992 if (s
->state
== MIGRATION_STATUS_SETUP
||
993 s
->state
== MIGRATION_STATUS_ACTIVE
) {
994 migrate_set_state(&s
->state
, s
->state
,
995 MIGRATION_STATUS_FAILED
);
999 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1000 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1002 qemu_mutex_lock(&p
->mutex
);
1005 * We could arrive here for two reasons:
1006 * - normal quit, i.e. everything went fine, just finished
1007 * - error quit: We close the channels so the channel threads
1008 * finish the qio_channel_read_all_eof()
1011 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
1013 qemu_mutex_unlock(&p
->mutex
);
1017 void multifd_load_shutdown(void)
1019 if (migrate_multifd()) {
1020 multifd_recv_terminate_threads(NULL
);
1024 void multifd_load_cleanup(void)
1028 if (!migrate_multifd()) {
1031 multifd_recv_terminate_threads(NULL
);
1032 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1033 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1037 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
1038 * however try to wakeup it without harm in cleanup phase.
1040 qemu_sem_post(&p
->sem_sync
);
1043 qemu_thread_join(&p
->thread
);
1045 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1046 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1048 migration_ioc_unregister_yank(p
->c
);
1049 object_unref(OBJECT(p
->c
));
1051 qemu_mutex_destroy(&p
->mutex
);
1052 qemu_sem_destroy(&p
->sem_sync
);
1062 multifd_recv_state
->ops
->recv_cleanup(p
);
1064 qemu_sem_destroy(&multifd_recv_state
->sem_sync
);
1065 g_free(multifd_recv_state
->params
);
1066 multifd_recv_state
->params
= NULL
;
1067 g_free(multifd_recv_state
);
1068 multifd_recv_state
= NULL
;
1071 void multifd_recv_sync_main(void)
1075 if (!migrate_multifd()) {
1078 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1079 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1081 trace_multifd_recv_sync_main_wait(p
->id
);
1082 qemu_sem_wait(&multifd_recv_state
->sem_sync
);
1084 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1085 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1087 WITH_QEMU_LOCK_GUARD(&p
->mutex
) {
1088 if (multifd_recv_state
->packet_num
< p
->packet_num
) {
1089 multifd_recv_state
->packet_num
= p
->packet_num
;
1092 trace_multifd_recv_sync_main_signal(p
->id
);
1093 qemu_sem_post(&p
->sem_sync
);
1095 trace_multifd_recv_sync_main(multifd_recv_state
->packet_num
);
1098 static void *multifd_recv_thread(void *opaque
)
1100 MultiFDRecvParams
*p
= opaque
;
1101 Error
*local_err
= NULL
;
1104 trace_multifd_recv_thread_start(p
->id
);
1105 rcu_register_thread();
1114 ret
= qio_channel_read_all_eof(p
->c
, (void *)p
->packet
,
1115 p
->packet_len
, &local_err
);
1116 if (ret
== 0 || ret
== -1) { /* 0: EOF -1: Error */
1120 qemu_mutex_lock(&p
->mutex
);
1121 ret
= multifd_recv_unfill_packet(p
, &local_err
);
1123 qemu_mutex_unlock(&p
->mutex
);
1128 /* recv methods don't know how to handle the SYNC flag */
1129 p
->flags
&= ~MULTIFD_FLAG_SYNC
;
1130 trace_multifd_recv(p
->id
, p
->packet_num
, p
->normal_num
, flags
,
1131 p
->next_packet_size
);
1133 p
->total_normal_pages
+= p
->normal_num
;
1134 qemu_mutex_unlock(&p
->mutex
);
1136 if (p
->normal_num
) {
1137 ret
= multifd_recv_state
->ops
->recv_pages(p
, &local_err
);
1143 if (flags
& MULTIFD_FLAG_SYNC
) {
1144 qemu_sem_post(&multifd_recv_state
->sem_sync
);
1145 qemu_sem_wait(&p
->sem_sync
);
1150 multifd_recv_terminate_threads(local_err
);
1151 error_free(local_err
);
1153 qemu_mutex_lock(&p
->mutex
);
1155 qemu_mutex_unlock(&p
->mutex
);
1157 rcu_unregister_thread();
1158 trace_multifd_recv_thread_end(p
->id
, p
->num_packets
, p
->total_normal_pages
);
1163 int multifd_load_setup(Error
**errp
)
1166 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
1170 * Return successfully if multiFD recv state is already initialised
1171 * or multiFD is not enabled.
1173 if (multifd_recv_state
|| !migrate_multifd()) {
1177 thread_count
= migrate_multifd_channels();
1178 multifd_recv_state
= g_malloc0(sizeof(*multifd_recv_state
));
1179 multifd_recv_state
->params
= g_new0(MultiFDRecvParams
, thread_count
);
1180 qatomic_set(&multifd_recv_state
->count
, 0);
1181 qemu_sem_init(&multifd_recv_state
->sem_sync
, 0);
1182 multifd_recv_state
->ops
= multifd_ops
[migrate_multifd_compression()];
1184 for (i
= 0; i
< thread_count
; i
++) {
1185 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1187 qemu_mutex_init(&p
->mutex
);
1188 qemu_sem_init(&p
->sem_sync
, 0);
1191 p
->packet_len
= sizeof(MultiFDPacket_t
)
1192 + sizeof(uint64_t) * page_count
;
1193 p
->packet
= g_malloc0(p
->packet_len
);
1194 p
->name
= g_strdup_printf("multifdrecv_%d", i
);
1195 p
->iov
= g_new0(struct iovec
, page_count
);
1196 p
->normal
= g_new0(ram_addr_t
, page_count
);
1197 p
->page_count
= page_count
;
1198 p
->page_size
= qemu_target_page_size();
1201 for (i
= 0; i
< thread_count
; i
++) {
1202 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1203 Error
*local_err
= NULL
;
1206 ret
= multifd_recv_state
->ops
->recv_setup(p
, &local_err
);
1208 error_propagate(errp
, local_err
);
1215 bool multifd_recv_all_channels_created(void)
1217 int thread_count
= migrate_multifd_channels();
1219 if (!migrate_multifd()) {
1223 if (!multifd_recv_state
) {
1224 /* Called before any connections created */
1228 return thread_count
== qatomic_read(&multifd_recv_state
->count
);
1232 * Try to receive all multifd channels to get ready for the migration.
1233 * Sets @errp when failing to receive the current channel.
1235 void multifd_recv_new_channel(QIOChannel
*ioc
, Error
**errp
)
1237 MultiFDRecvParams
*p
;
1238 Error
*local_err
= NULL
;
1241 id
= multifd_recv_initial_packet(ioc
, &local_err
);
1243 multifd_recv_terminate_threads(local_err
);
1244 error_propagate_prepend(errp
, local_err
,
1245 "failed to receive packet"
1246 " via multifd channel %d: ",
1247 qatomic_read(&multifd_recv_state
->count
));
1250 trace_multifd_recv_new_channel(id
);
1252 p
= &multifd_recv_state
->params
[id
];
1254 error_setg(&local_err
, "multifd: received id '%d' already setup'",
1256 multifd_recv_terminate_threads(local_err
);
1257 error_propagate(errp
, local_err
);
1261 object_ref(OBJECT(ioc
));
1262 /* initial packet */
1266 qemu_thread_create(&p
->thread
, p
->name
, multifd_recv_thread
, p
,
1267 QEMU_THREAD_JOINABLE
);
1268 qatomic_inc(&multifd_recv_state
->count
);