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 MultiFDPages_t
*pages
= p
->pages
;
90 size_t page_size
= qemu_target_page_size();
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
= page_size
;
98 p
->next_packet_size
= p
->normal_num
* 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
;
142 size_t page_size
= qemu_target_page_size();
144 if (flags
!= MULTIFD_FLAG_NOCOMP
) {
145 error_setg(errp
, "multifd %u: flags received %x flags expected %x",
146 p
->id
, flags
, MULTIFD_FLAG_NOCOMP
);
149 for (int i
= 0; i
< p
->normal_num
; i
++) {
150 p
->iov
[i
].iov_base
= p
->host
+ p
->normal
[i
];
151 p
->iov
[i
].iov_len
= page_size
;
153 return qio_channel_readv_all(p
->c
, p
->iov
, p
->normal_num
, errp
);
156 static MultiFDMethods multifd_nocomp_ops
= {
157 .send_setup
= nocomp_send_setup
,
158 .send_cleanup
= nocomp_send_cleanup
,
159 .send_prepare
= nocomp_send_prepare
,
160 .recv_setup
= nocomp_recv_setup
,
161 .recv_cleanup
= nocomp_recv_cleanup
,
162 .recv_pages
= nocomp_recv_pages
165 static MultiFDMethods
*multifd_ops
[MULTIFD_COMPRESSION__MAX
] = {
166 [MULTIFD_COMPRESSION_NONE
] = &multifd_nocomp_ops
,
169 void multifd_register_ops(int method
, MultiFDMethods
*ops
)
171 assert(0 < method
&& method
< MULTIFD_COMPRESSION__MAX
);
172 multifd_ops
[method
] = ops
;
175 static int multifd_send_initial_packet(MultiFDSendParams
*p
, Error
**errp
)
177 MultiFDInit_t msg
= {};
180 msg
.magic
= cpu_to_be32(MULTIFD_MAGIC
);
181 msg
.version
= cpu_to_be32(MULTIFD_VERSION
);
183 memcpy(msg
.uuid
, &qemu_uuid
.data
, sizeof(msg
.uuid
));
185 ret
= qio_channel_write_all(p
->c
, (char *)&msg
, sizeof(msg
), errp
);
192 static int multifd_recv_initial_packet(QIOChannel
*c
, Error
**errp
)
197 ret
= qio_channel_read_all(c
, (char *)&msg
, sizeof(msg
), errp
);
202 msg
.magic
= be32_to_cpu(msg
.magic
);
203 msg
.version
= be32_to_cpu(msg
.version
);
205 if (msg
.magic
!= MULTIFD_MAGIC
) {
206 error_setg(errp
, "multifd: received packet magic %x "
207 "expected %x", msg
.magic
, MULTIFD_MAGIC
);
211 if (msg
.version
!= MULTIFD_VERSION
) {
212 error_setg(errp
, "multifd: received packet version %u "
213 "expected %u", msg
.version
, MULTIFD_VERSION
);
217 if (memcmp(msg
.uuid
, &qemu_uuid
, sizeof(qemu_uuid
))) {
218 char *uuid
= qemu_uuid_unparse_strdup(&qemu_uuid
);
219 char *msg_uuid
= qemu_uuid_unparse_strdup((const QemuUUID
*)msg
.uuid
);
221 error_setg(errp
, "multifd: received uuid '%s' and expected "
222 "uuid '%s' for channel %hhd", msg_uuid
, uuid
, msg
.id
);
228 if (msg
.id
> migrate_multifd_channels()) {
229 error_setg(errp
, "multifd: received channel version %u "
230 "expected %u", msg
.version
, MULTIFD_VERSION
);
237 static MultiFDPages_t
*multifd_pages_init(size_t size
)
239 MultiFDPages_t
*pages
= g_new0(MultiFDPages_t
, 1);
241 pages
->allocated
= size
;
242 pages
->offset
= g_new0(ram_addr_t
, size
);
247 static void multifd_pages_clear(MultiFDPages_t
*pages
)
250 pages
->allocated
= 0;
251 pages
->packet_num
= 0;
253 g_free(pages
->offset
);
254 pages
->offset
= NULL
;
258 static void multifd_send_fill_packet(MultiFDSendParams
*p
)
260 MultiFDPacket_t
*packet
= p
->packet
;
263 packet
->flags
= cpu_to_be32(p
->flags
);
264 packet
->pages_alloc
= cpu_to_be32(p
->pages
->allocated
);
265 packet
->normal_pages
= cpu_to_be32(p
->normal_num
);
266 packet
->next_packet_size
= cpu_to_be32(p
->next_packet_size
);
267 packet
->packet_num
= cpu_to_be64(p
->packet_num
);
269 if (p
->pages
->block
) {
270 strncpy(packet
->ramblock
, p
->pages
->block
->idstr
, 256);
273 for (i
= 0; i
< p
->normal_num
; i
++) {
274 /* there are architectures where ram_addr_t is 32 bit */
275 uint64_t temp
= p
->normal
[i
];
277 packet
->offset
[i
] = cpu_to_be64(temp
);
281 static int multifd_recv_unfill_packet(MultiFDRecvParams
*p
, Error
**errp
)
283 MultiFDPacket_t
*packet
= p
->packet
;
284 size_t page_size
= qemu_target_page_size();
285 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ page_size
;
289 packet
->magic
= be32_to_cpu(packet
->magic
);
290 if (packet
->magic
!= MULTIFD_MAGIC
) {
291 error_setg(errp
, "multifd: received packet "
292 "magic %x and expected magic %x",
293 packet
->magic
, MULTIFD_MAGIC
);
297 packet
->version
= be32_to_cpu(packet
->version
);
298 if (packet
->version
!= MULTIFD_VERSION
) {
299 error_setg(errp
, "multifd: received packet "
300 "version %u and expected version %u",
301 packet
->version
, MULTIFD_VERSION
);
305 p
->flags
= be32_to_cpu(packet
->flags
);
307 packet
->pages_alloc
= be32_to_cpu(packet
->pages_alloc
);
309 * If we received a packet that is 100 times bigger than expected
310 * just stop migration. It is a magic number.
312 if (packet
->pages_alloc
> page_count
) {
313 error_setg(errp
, "multifd: received packet "
314 "with size %u and expected a size of %u",
315 packet
->pages_alloc
, page_count
) ;
319 p
->normal_num
= be32_to_cpu(packet
->normal_pages
);
320 if (p
->normal_num
> packet
->pages_alloc
) {
321 error_setg(errp
, "multifd: received packet "
322 "with %u pages and expected maximum pages are %u",
323 p
->normal_num
, packet
->pages_alloc
) ;
327 p
->next_packet_size
= be32_to_cpu(packet
->next_packet_size
);
328 p
->packet_num
= be64_to_cpu(packet
->packet_num
);
330 if (p
->normal_num
== 0) {
334 /* make sure that ramblock is 0 terminated */
335 packet
->ramblock
[255] = 0;
336 block
= qemu_ram_block_by_name(packet
->ramblock
);
338 error_setg(errp
, "multifd: unknown ram block %s",
343 p
->host
= block
->host
;
344 for (i
= 0; i
< p
->normal_num
; i
++) {
345 uint64_t offset
= be64_to_cpu(packet
->offset
[i
]);
347 if (offset
> (block
->used_length
- page_size
)) {
348 error_setg(errp
, "multifd: offset too long %" PRIu64
349 " (max " RAM_ADDR_FMT
")",
350 offset
, block
->used_length
);
353 p
->normal
[i
] = offset
;
360 MultiFDSendParams
*params
;
361 /* array of pages to sent */
362 MultiFDPages_t
*pages
;
363 /* global number of generated multifd packets */
365 /* send channels ready */
366 QemuSemaphore channels_ready
;
368 * Have we already run terminate threads. There is a race when it
369 * happens that we got one error while we are exiting.
370 * We will use atomic operations. Only valid values are 0 and 1.
375 } *multifd_send_state
;
378 * How we use multifd_send_state->pages and channel->pages?
380 * We create a pages for each channel, and a main one. Each time that
381 * we need to send a batch of pages we interchange the ones between
382 * multifd_send_state and the channel that is sending it. There are
383 * two reasons for that:
384 * - to not have to do so many mallocs during migration
385 * - to make easier to know what to free at the end of migration
387 * This way we always know who is the owner of each "pages" struct,
388 * and we don't need any locking. It belongs to the migration thread
389 * or to the channel thread. Switching is safe because the migration
390 * thread is using the channel mutex when changing it, and the channel
391 * have to had finish with its own, otherwise pending_job can't be
395 static int multifd_send_pages(QEMUFile
*f
)
398 static int next_channel
;
399 MultiFDSendParams
*p
= NULL
; /* make happy gcc */
400 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
401 uint64_t transferred
;
403 if (qatomic_read(&multifd_send_state
->exiting
)) {
407 qemu_sem_wait(&multifd_send_state
->channels_ready
);
409 * next_channel can remain from a previous migration that was
410 * using more channels, so ensure it doesn't overflow if the
411 * limit is lower now.
413 next_channel
%= migrate_multifd_channels();
414 for (i
= next_channel
;; i
= (i
+ 1) % migrate_multifd_channels()) {
415 p
= &multifd_send_state
->params
[i
];
417 qemu_mutex_lock(&p
->mutex
);
419 error_report("%s: channel %d has already quit!", __func__
, i
);
420 qemu_mutex_unlock(&p
->mutex
);
423 if (!p
->pending_job
) {
425 next_channel
= (i
+ 1) % migrate_multifd_channels();
428 qemu_mutex_unlock(&p
->mutex
);
430 assert(!p
->pages
->num
);
431 assert(!p
->pages
->block
);
433 p
->packet_num
= multifd_send_state
->packet_num
++;
434 multifd_send_state
->pages
= p
->pages
;
436 transferred
= ((uint64_t) pages
->num
) * qemu_target_page_size()
438 qemu_file_update_transfer(f
, transferred
);
439 ram_counters
.multifd_bytes
+= transferred
;
440 ram_counters
.transferred
+= transferred
;
441 qemu_mutex_unlock(&p
->mutex
);
442 qemu_sem_post(&p
->sem
);
447 int multifd_queue_page(QEMUFile
*f
, RAMBlock
*block
, ram_addr_t offset
)
449 MultiFDPages_t
*pages
= multifd_send_state
->pages
;
452 pages
->block
= block
;
455 if (pages
->block
== block
) {
456 pages
->offset
[pages
->num
] = offset
;
459 if (pages
->num
< 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 (qatomic_xchg(&multifd_send_state
->exiting
, 1)) {
503 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
504 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
506 qemu_mutex_lock(&p
->mutex
);
508 qemu_sem_post(&p
->sem
);
510 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
512 qemu_mutex_unlock(&p
->mutex
);
516 void multifd_save_cleanup(void)
520 if (!migrate_use_multifd() || !migrate_multifd_is_allowed()) {
523 multifd_send_terminate_threads(NULL
);
524 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
525 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
528 qemu_thread_join(&p
->thread
);
531 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
532 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
533 Error
*local_err
= NULL
;
535 if (p
->registered_yank
) {
536 migration_ioc_unregister_yank(p
->c
);
538 socket_send_channel_destroy(p
->c
);
540 qemu_mutex_destroy(&p
->mutex
);
541 qemu_sem_destroy(&p
->sem
);
542 qemu_sem_destroy(&p
->sem_sync
);
545 multifd_pages_clear(p
->pages
);
554 multifd_send_state
->ops
->send_cleanup(p
, &local_err
);
556 migrate_set_error(migrate_get_current(), local_err
);
557 error_free(local_err
);
560 qemu_sem_destroy(&multifd_send_state
->channels_ready
);
561 g_free(multifd_send_state
->params
);
562 multifd_send_state
->params
= NULL
;
563 multifd_pages_clear(multifd_send_state
->pages
);
564 multifd_send_state
->pages
= NULL
;
565 g_free(multifd_send_state
);
566 multifd_send_state
= NULL
;
569 void multifd_send_sync_main(QEMUFile
*f
)
573 if (!migrate_use_multifd()) {
576 if (multifd_send_state
->pages
->num
) {
577 if (multifd_send_pages(f
) < 0) {
578 error_report("%s: multifd_send_pages fail", __func__
);
582 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
583 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
585 trace_multifd_send_sync_main_signal(p
->id
);
587 qemu_mutex_lock(&p
->mutex
);
590 error_report("%s: channel %d has already quit", __func__
, i
);
591 qemu_mutex_unlock(&p
->mutex
);
595 p
->packet_num
= multifd_send_state
->packet_num
++;
596 p
->flags
|= MULTIFD_FLAG_SYNC
;
598 qemu_file_update_transfer(f
, p
->packet_len
);
599 ram_counters
.multifd_bytes
+= p
->packet_len
;
600 ram_counters
.transferred
+= p
->packet_len
;
601 qemu_mutex_unlock(&p
->mutex
);
602 qemu_sem_post(&p
->sem
);
604 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
605 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
607 trace_multifd_send_sync_main_wait(p
->id
);
608 qemu_sem_wait(&p
->sem_sync
);
610 trace_multifd_send_sync_main(multifd_send_state
->packet_num
);
613 static void *multifd_send_thread(void *opaque
)
615 MultiFDSendParams
*p
= opaque
;
616 Error
*local_err
= NULL
;
619 trace_multifd_send_thread_start(p
->id
);
620 rcu_register_thread();
622 if (multifd_send_initial_packet(p
, &local_err
) < 0) {
630 qemu_sem_wait(&p
->sem
);
632 if (qatomic_read(&multifd_send_state
->exiting
)) {
635 qemu_mutex_lock(&p
->mutex
);
637 if (p
->pending_job
) {
638 uint64_t packet_num
= p
->packet_num
;
639 uint32_t flags
= p
->flags
;
643 for (int i
= 0; i
< p
->pages
->num
; i
++) {
644 p
->normal
[p
->normal_num
] = p
->pages
->offset
[i
];
649 ret
= multifd_send_state
->ops
->send_prepare(p
, &local_err
);
651 qemu_mutex_unlock(&p
->mutex
);
655 multifd_send_fill_packet(p
);
658 p
->total_normal_pages
+= p
->normal_num
;
660 p
->pages
->block
= NULL
;
661 qemu_mutex_unlock(&p
->mutex
);
663 trace_multifd_send(p
->id
, packet_num
, p
->normal_num
, flags
,
664 p
->next_packet_size
);
666 p
->iov
[0].iov_len
= p
->packet_len
;
667 p
->iov
[0].iov_base
= p
->packet
;
669 ret
= qio_channel_writev_all(p
->c
, p
->iov
, p
->iovs_num
,
675 qemu_mutex_lock(&p
->mutex
);
677 qemu_mutex_unlock(&p
->mutex
);
679 if (flags
& MULTIFD_FLAG_SYNC
) {
680 qemu_sem_post(&p
->sem_sync
);
682 qemu_sem_post(&multifd_send_state
->channels_ready
);
683 } else if (p
->quit
) {
684 qemu_mutex_unlock(&p
->mutex
);
687 qemu_mutex_unlock(&p
->mutex
);
688 /* sometimes there are spurious wakeups */
694 trace_multifd_send_error(p
->id
);
695 multifd_send_terminate_threads(local_err
);
696 error_free(local_err
);
700 * Error happen, I will exit, but I can't just leave, tell
701 * who pay attention to me.
704 qemu_sem_post(&p
->sem_sync
);
705 qemu_sem_post(&multifd_send_state
->channels_ready
);
708 qemu_mutex_lock(&p
->mutex
);
710 qemu_mutex_unlock(&p
->mutex
);
712 rcu_unregister_thread();
713 trace_multifd_send_thread_end(p
->id
, p
->num_packets
, p
->total_normal_pages
);
718 static bool multifd_channel_connect(MultiFDSendParams
*p
,
722 static void multifd_tls_outgoing_handshake(QIOTask
*task
,
725 MultiFDSendParams
*p
= opaque
;
726 QIOChannel
*ioc
= QIO_CHANNEL(qio_task_get_source(task
));
729 if (qio_task_propagate_error(task
, &err
)) {
730 trace_multifd_tls_outgoing_handshake_error(ioc
, error_get_pretty(err
));
732 trace_multifd_tls_outgoing_handshake_complete(ioc
);
735 if (!multifd_channel_connect(p
, ioc
, err
)) {
737 * Error happen, mark multifd_send_thread status as 'quit' although it
738 * is not created, and then tell who pay attention to me.
741 qemu_sem_post(&multifd_send_state
->channels_ready
);
742 qemu_sem_post(&p
->sem_sync
);
746 static void *multifd_tls_handshake_thread(void *opaque
)
748 MultiFDSendParams
*p
= opaque
;
749 QIOChannelTLS
*tioc
= QIO_CHANNEL_TLS(p
->c
);
751 qio_channel_tls_handshake(tioc
,
752 multifd_tls_outgoing_handshake
,
759 static void multifd_tls_channel_connect(MultiFDSendParams
*p
,
763 MigrationState
*s
= migrate_get_current();
764 const char *hostname
= s
->hostname
;
767 tioc
= migration_tls_client_create(s
, ioc
, hostname
, errp
);
772 object_unref(OBJECT(ioc
));
773 trace_multifd_tls_outgoing_handshake_start(ioc
, tioc
, hostname
);
774 qio_channel_set_name(QIO_CHANNEL(tioc
), "multifd-tls-outgoing");
775 p
->c
= QIO_CHANNEL(tioc
);
776 qemu_thread_create(&p
->thread
, "multifd-tls-handshake-worker",
777 multifd_tls_handshake_thread
, p
,
778 QEMU_THREAD_JOINABLE
);
781 static bool multifd_channel_connect(MultiFDSendParams
*p
,
785 MigrationState
*s
= migrate_get_current();
787 trace_multifd_set_outgoing_channel(
788 ioc
, object_get_typename(OBJECT(ioc
)),
789 migrate_get_current()->hostname
, error
);
792 if (s
->parameters
.tls_creds
&&
793 *s
->parameters
.tls_creds
&&
794 !object_dynamic_cast(OBJECT(ioc
),
795 TYPE_QIO_CHANNEL_TLS
)) {
796 multifd_tls_channel_connect(p
, ioc
, &error
);
799 * tls_channel_connect will call back to this
800 * function after the TLS handshake,
801 * so we mustn't call multifd_send_thread until then
808 migration_ioc_register_yank(ioc
);
809 p
->registered_yank
= true;
811 qemu_thread_create(&p
->thread
, p
->name
, multifd_send_thread
, p
,
812 QEMU_THREAD_JOINABLE
);
820 static void multifd_new_send_channel_cleanup(MultiFDSendParams
*p
,
821 QIOChannel
*ioc
, Error
*err
)
823 migrate_set_error(migrate_get_current(), err
);
824 /* Error happen, we need to tell who pay attention to me */
825 qemu_sem_post(&multifd_send_state
->channels_ready
);
826 qemu_sem_post(&p
->sem_sync
);
828 * Although multifd_send_thread is not created, but main migration
829 * thread neet to judge whether it is running, so we need to mark
833 object_unref(OBJECT(ioc
));
837 static void multifd_new_send_channel_async(QIOTask
*task
, gpointer opaque
)
839 MultiFDSendParams
*p
= opaque
;
840 QIOChannel
*sioc
= QIO_CHANNEL(qio_task_get_source(task
));
841 Error
*local_err
= NULL
;
843 trace_multifd_new_send_channel_async(p
->id
);
844 if (qio_task_propagate_error(task
, &local_err
)) {
847 p
->c
= QIO_CHANNEL(sioc
);
848 qio_channel_set_delay(p
->c
, false);
850 if (!multifd_channel_connect(p
, sioc
, local_err
)) {
857 multifd_new_send_channel_cleanup(p
, sioc
, local_err
);
860 static bool migrate_allow_multifd
= true;
861 void migrate_protocol_allow_multifd(bool allow
)
863 migrate_allow_multifd
= allow
;
866 bool migrate_multifd_is_allowed(void)
868 return migrate_allow_multifd
;
871 int multifd_save_setup(Error
**errp
)
874 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
877 if (!migrate_use_multifd()) {
880 if (!migrate_multifd_is_allowed()) {
881 error_setg(errp
, "multifd is not supported by current protocol");
885 thread_count
= migrate_multifd_channels();
886 multifd_send_state
= g_malloc0(sizeof(*multifd_send_state
));
887 multifd_send_state
->params
= g_new0(MultiFDSendParams
, thread_count
);
888 multifd_send_state
->pages
= multifd_pages_init(page_count
);
889 qemu_sem_init(&multifd_send_state
->channels_ready
, 0);
890 qatomic_set(&multifd_send_state
->exiting
, 0);
891 multifd_send_state
->ops
= multifd_ops
[migrate_multifd_compression()];
893 for (i
= 0; i
< thread_count
; i
++) {
894 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
896 qemu_mutex_init(&p
->mutex
);
897 qemu_sem_init(&p
->sem
, 0);
898 qemu_sem_init(&p
->sem_sync
, 0);
902 p
->pages
= multifd_pages_init(page_count
);
903 p
->packet_len
= sizeof(MultiFDPacket_t
)
904 + sizeof(uint64_t) * page_count
;
905 p
->packet
= g_malloc0(p
->packet_len
);
906 p
->packet
->magic
= cpu_to_be32(MULTIFD_MAGIC
);
907 p
->packet
->version
= cpu_to_be32(MULTIFD_VERSION
);
908 p
->name
= g_strdup_printf("multifdsend_%d", i
);
909 /* We need one extra place for the packet header */
910 p
->iov
= g_new0(struct iovec
, page_count
+ 1);
911 p
->normal
= g_new0(ram_addr_t
, page_count
);
912 socket_send_channel_create(multifd_new_send_channel_async
, p
);
915 for (i
= 0; i
< thread_count
; i
++) {
916 MultiFDSendParams
*p
= &multifd_send_state
->params
[i
];
917 Error
*local_err
= NULL
;
920 ret
= multifd_send_state
->ops
->send_setup(p
, &local_err
);
922 error_propagate(errp
, local_err
);
930 MultiFDRecvParams
*params
;
931 /* number of created threads */
933 /* syncs main thread and channels */
934 QemuSemaphore sem_sync
;
935 /* global number of generated multifd packets */
939 } *multifd_recv_state
;
941 static void multifd_recv_terminate_threads(Error
*err
)
945 trace_multifd_recv_terminate_threads(err
!= NULL
);
948 MigrationState
*s
= migrate_get_current();
949 migrate_set_error(s
, err
);
950 if (s
->state
== MIGRATION_STATUS_SETUP
||
951 s
->state
== MIGRATION_STATUS_ACTIVE
) {
952 migrate_set_state(&s
->state
, s
->state
,
953 MIGRATION_STATUS_FAILED
);
957 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
958 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
960 qemu_mutex_lock(&p
->mutex
);
963 * We could arrive here for two reasons:
964 * - normal quit, i.e. everything went fine, just finished
965 * - error quit: We close the channels so the channel threads
966 * finish the qio_channel_read_all_eof()
969 qio_channel_shutdown(p
->c
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
971 qemu_mutex_unlock(&p
->mutex
);
975 int multifd_load_cleanup(Error
**errp
)
979 if (!migrate_use_multifd() || !migrate_multifd_is_allowed()) {
982 multifd_recv_terminate_threads(NULL
);
983 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
984 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
989 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
990 * however try to wakeup it without harm in cleanup phase.
992 qemu_sem_post(&p
->sem_sync
);
993 qemu_thread_join(&p
->thread
);
996 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
997 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
999 migration_ioc_unregister_yank(p
->c
);
1000 object_unref(OBJECT(p
->c
));
1002 qemu_mutex_destroy(&p
->mutex
);
1003 qemu_sem_destroy(&p
->sem_sync
);
1013 multifd_recv_state
->ops
->recv_cleanup(p
);
1015 qemu_sem_destroy(&multifd_recv_state
->sem_sync
);
1016 g_free(multifd_recv_state
->params
);
1017 multifd_recv_state
->params
= NULL
;
1018 g_free(multifd_recv_state
);
1019 multifd_recv_state
= NULL
;
1024 void multifd_recv_sync_main(void)
1028 if (!migrate_use_multifd()) {
1031 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1032 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1034 trace_multifd_recv_sync_main_wait(p
->id
);
1035 qemu_sem_wait(&multifd_recv_state
->sem_sync
);
1037 for (i
= 0; i
< migrate_multifd_channels(); i
++) {
1038 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1040 WITH_QEMU_LOCK_GUARD(&p
->mutex
) {
1041 if (multifd_recv_state
->packet_num
< p
->packet_num
) {
1042 multifd_recv_state
->packet_num
= p
->packet_num
;
1045 trace_multifd_recv_sync_main_signal(p
->id
);
1046 qemu_sem_post(&p
->sem_sync
);
1048 trace_multifd_recv_sync_main(multifd_recv_state
->packet_num
);
1051 static void *multifd_recv_thread(void *opaque
)
1053 MultiFDRecvParams
*p
= opaque
;
1054 Error
*local_err
= NULL
;
1057 trace_multifd_recv_thread_start(p
->id
);
1058 rcu_register_thread();
1067 ret
= qio_channel_read_all_eof(p
->c
, (void *)p
->packet
,
1068 p
->packet_len
, &local_err
);
1069 if (ret
== 0) { /* EOF */
1072 if (ret
== -1) { /* Error */
1076 qemu_mutex_lock(&p
->mutex
);
1077 ret
= multifd_recv_unfill_packet(p
, &local_err
);
1079 qemu_mutex_unlock(&p
->mutex
);
1084 /* recv methods don't know how to handle the SYNC flag */
1085 p
->flags
&= ~MULTIFD_FLAG_SYNC
;
1086 trace_multifd_recv(p
->id
, p
->packet_num
, p
->normal_num
, flags
,
1087 p
->next_packet_size
);
1089 p
->total_normal_pages
+= p
->normal_num
;
1090 qemu_mutex_unlock(&p
->mutex
);
1092 if (p
->normal_num
) {
1093 ret
= multifd_recv_state
->ops
->recv_pages(p
, &local_err
);
1099 if (flags
& MULTIFD_FLAG_SYNC
) {
1100 qemu_sem_post(&multifd_recv_state
->sem_sync
);
1101 qemu_sem_wait(&p
->sem_sync
);
1106 multifd_recv_terminate_threads(local_err
);
1107 error_free(local_err
);
1109 qemu_mutex_lock(&p
->mutex
);
1111 qemu_mutex_unlock(&p
->mutex
);
1113 rcu_unregister_thread();
1114 trace_multifd_recv_thread_end(p
->id
, p
->num_packets
, p
->total_normal_pages
);
1119 int multifd_load_setup(Error
**errp
)
1122 uint32_t page_count
= MULTIFD_PACKET_SIZE
/ qemu_target_page_size();
1125 if (!migrate_use_multifd()) {
1128 if (!migrate_multifd_is_allowed()) {
1129 error_setg(errp
, "multifd is not supported by current protocol");
1132 thread_count
= migrate_multifd_channels();
1133 multifd_recv_state
= g_malloc0(sizeof(*multifd_recv_state
));
1134 multifd_recv_state
->params
= g_new0(MultiFDRecvParams
, thread_count
);
1135 qatomic_set(&multifd_recv_state
->count
, 0);
1136 qemu_sem_init(&multifd_recv_state
->sem_sync
, 0);
1137 multifd_recv_state
->ops
= multifd_ops
[migrate_multifd_compression()];
1139 for (i
= 0; i
< thread_count
; i
++) {
1140 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1142 qemu_mutex_init(&p
->mutex
);
1143 qemu_sem_init(&p
->sem_sync
, 0);
1146 p
->packet_len
= sizeof(MultiFDPacket_t
)
1147 + sizeof(uint64_t) * page_count
;
1148 p
->packet
= g_malloc0(p
->packet_len
);
1149 p
->name
= g_strdup_printf("multifdrecv_%d", i
);
1150 p
->iov
= g_new0(struct iovec
, page_count
);
1151 p
->normal
= g_new0(ram_addr_t
, page_count
);
1154 for (i
= 0; i
< thread_count
; i
++) {
1155 MultiFDRecvParams
*p
= &multifd_recv_state
->params
[i
];
1156 Error
*local_err
= NULL
;
1159 ret
= multifd_recv_state
->ops
->recv_setup(p
, &local_err
);
1161 error_propagate(errp
, local_err
);
1168 bool multifd_recv_all_channels_created(void)
1170 int thread_count
= migrate_multifd_channels();
1172 if (!migrate_use_multifd()) {
1176 if (!multifd_recv_state
) {
1177 /* Called before any connections created */
1181 return thread_count
== qatomic_read(&multifd_recv_state
->count
);
1185 * Try to receive all multifd channels to get ready for the migration.
1186 * - Return true and do not set @errp when correctly receiving all channels;
1187 * - Return false and do not set @errp when correctly receiving the current one;
1188 * - Return false and set @errp when failing to receive the current channel.
1190 bool multifd_recv_new_channel(QIOChannel
*ioc
, Error
**errp
)
1192 MultiFDRecvParams
*p
;
1193 Error
*local_err
= NULL
;
1196 id
= multifd_recv_initial_packet(ioc
, &local_err
);
1198 multifd_recv_terminate_threads(local_err
);
1199 error_propagate_prepend(errp
, local_err
,
1200 "failed to receive packet"
1201 " via multifd channel %d: ",
1202 qatomic_read(&multifd_recv_state
->count
));
1205 trace_multifd_recv_new_channel(id
);
1207 p
= &multifd_recv_state
->params
[id
];
1209 error_setg(&local_err
, "multifd: received id '%d' already setup'",
1211 multifd_recv_terminate_threads(local_err
);
1212 error_propagate(errp
, local_err
);
1216 object_ref(OBJECT(ioc
));
1217 /* initial packet */
1221 qemu_thread_create(&p
->thread
, p
->name
, multifd_recv_thread
, p
,
1222 QEMU_THREAD_JOINABLE
);
1223 qatomic_inc(&multifd_recv_state
->count
);
1224 return qatomic_read(&multifd_recv_state
->count
) ==
1225 migrate_multifd_channels();