target/mips/rel6_translate: Change license to GNU LGPL v2.1 (or later)
[qemu/ar7.git] / migration / multifd.c
bloba6677c45c86349ec09809c942b8c358ea5f4c533
1 /*
2 * Multifd common code
4 * Copyright (c) 2019-2020 Red Hat Inc
6 * Authors:
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"
14 #include "qemu/rcu.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"
20 #include "ram.h"
21 #include "migration.h"
22 #include "socket.h"
23 #include "tls.h"
24 #include "qemu-file.h"
25 #include "trace.h"
26 #include "multifd.h"
28 #include "qemu/yank.h"
29 #include "io/channel-socket.h"
30 #include "yank_functions.h"
32 /* Multiple fd's */
34 #define MULTIFD_MAGIC 0x11223344U
35 #define MULTIFD_VERSION 1
37 typedef struct {
38 uint32_t magic;
39 uint32_t version;
40 unsigned char uuid[16]; /* QemuUUID */
41 uint8_t id;
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 */
48 /**
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)
60 return 0;
63 /**
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
70 static void nocomp_send_cleanup(MultiFDSendParams *p, Error **errp)
72 return;
75 /**
76 * nocomp_send_prepare: prepare date to be able to send
78 * For no compression we just have to calculate the size of the
79 * packet.
81 * Returns 0 for success or -1 for error
83 * @p: Params for the channel that we are using
84 * @used: number of pages used
85 * @errp: pointer to an error
87 static int nocomp_send_prepare(MultiFDSendParams *p, uint32_t used,
88 Error **errp)
90 p->next_packet_size = used * qemu_target_page_size();
91 p->flags |= MULTIFD_FLAG_NOCOMP;
92 return 0;
95 /**
96 * nocomp_send_write: do the actual write of the data
98 * For no compression we just have to write the data.
100 * Returns 0 for success or -1 for error
102 * @p: Params for the channel that we are using
103 * @used: number of pages used
104 * @errp: pointer to an error
106 static int nocomp_send_write(MultiFDSendParams *p, uint32_t used, Error **errp)
108 return qio_channel_writev_all(p->c, p->pages->iov, used, errp);
112 * nocomp_recv_setup: setup receive side
114 * For no compression this function does nothing.
116 * Returns 0 for success or -1 for error
118 * @p: Params for the channel that we are using
119 * @errp: pointer to an error
121 static int nocomp_recv_setup(MultiFDRecvParams *p, Error **errp)
123 return 0;
127 * nocomp_recv_cleanup: setup receive side
129 * For no compression this function does nothing.
131 * @p: Params for the channel that we are using
133 static void nocomp_recv_cleanup(MultiFDRecvParams *p)
138 * nocomp_recv_pages: read the data from the channel into actual pages
140 * For no compression we just need to read things into the correct place.
142 * Returns 0 for success or -1 for error
144 * @p: Params for the channel that we are using
145 * @used: number of pages used
146 * @errp: pointer to an error
148 static int nocomp_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
150 uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
152 if (flags != MULTIFD_FLAG_NOCOMP) {
153 error_setg(errp, "multifd %d: flags received %x flags expected %x",
154 p->id, flags, MULTIFD_FLAG_NOCOMP);
155 return -1;
157 return qio_channel_readv_all(p->c, p->pages->iov, used, errp);
160 static MultiFDMethods multifd_nocomp_ops = {
161 .send_setup = nocomp_send_setup,
162 .send_cleanup = nocomp_send_cleanup,
163 .send_prepare = nocomp_send_prepare,
164 .send_write = nocomp_send_write,
165 .recv_setup = nocomp_recv_setup,
166 .recv_cleanup = nocomp_recv_cleanup,
167 .recv_pages = nocomp_recv_pages
170 static MultiFDMethods *multifd_ops[MULTIFD_COMPRESSION__MAX] = {
171 [MULTIFD_COMPRESSION_NONE] = &multifd_nocomp_ops,
174 void multifd_register_ops(int method, MultiFDMethods *ops)
176 assert(0 < method && method < MULTIFD_COMPRESSION__MAX);
177 multifd_ops[method] = ops;
180 static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
182 MultiFDInit_t msg = {};
183 int ret;
185 msg.magic = cpu_to_be32(MULTIFD_MAGIC);
186 msg.version = cpu_to_be32(MULTIFD_VERSION);
187 msg.id = p->id;
188 memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
190 ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), errp);
191 if (ret != 0) {
192 return -1;
194 return 0;
197 static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
199 MultiFDInit_t msg;
200 int ret;
202 ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
203 if (ret != 0) {
204 return -1;
207 msg.magic = be32_to_cpu(msg.magic);
208 msg.version = be32_to_cpu(msg.version);
210 if (msg.magic != MULTIFD_MAGIC) {
211 error_setg(errp, "multifd: received packet magic %x "
212 "expected %x", msg.magic, MULTIFD_MAGIC);
213 return -1;
216 if (msg.version != MULTIFD_VERSION) {
217 error_setg(errp, "multifd: received packet version %d "
218 "expected %d", msg.version, MULTIFD_VERSION);
219 return -1;
222 if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
223 char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
224 char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
226 error_setg(errp, "multifd: received uuid '%s' and expected "
227 "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
228 g_free(uuid);
229 g_free(msg_uuid);
230 return -1;
233 if (msg.id > migrate_multifd_channels()) {
234 error_setg(errp, "multifd: received channel version %d "
235 "expected %d", msg.version, MULTIFD_VERSION);
236 return -1;
239 return msg.id;
242 static MultiFDPages_t *multifd_pages_init(size_t size)
244 MultiFDPages_t *pages = g_new0(MultiFDPages_t, 1);
246 pages->allocated = size;
247 pages->iov = g_new0(struct iovec, size);
248 pages->offset = g_new0(ram_addr_t, size);
250 return pages;
253 static void multifd_pages_clear(MultiFDPages_t *pages)
255 pages->used = 0;
256 pages->allocated = 0;
257 pages->packet_num = 0;
258 pages->block = NULL;
259 g_free(pages->iov);
260 pages->iov = NULL;
261 g_free(pages->offset);
262 pages->offset = NULL;
263 g_free(pages);
266 static void multifd_send_fill_packet(MultiFDSendParams *p)
268 MultiFDPacket_t *packet = p->packet;
269 int i;
271 packet->flags = cpu_to_be32(p->flags);
272 packet->pages_alloc = cpu_to_be32(p->pages->allocated);
273 packet->pages_used = cpu_to_be32(p->pages->used);
274 packet->next_packet_size = cpu_to_be32(p->next_packet_size);
275 packet->packet_num = cpu_to_be64(p->packet_num);
277 if (p->pages->block) {
278 strncpy(packet->ramblock, p->pages->block->idstr, 256);
281 for (i = 0; i < p->pages->used; i++) {
282 /* there are architectures where ram_addr_t is 32 bit */
283 uint64_t temp = p->pages->offset[i];
285 packet->offset[i] = cpu_to_be64(temp);
289 static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
291 MultiFDPacket_t *packet = p->packet;
292 uint32_t pages_max = MULTIFD_PACKET_SIZE / qemu_target_page_size();
293 RAMBlock *block;
294 int i;
296 packet->magic = be32_to_cpu(packet->magic);
297 if (packet->magic != MULTIFD_MAGIC) {
298 error_setg(errp, "multifd: received packet "
299 "magic %x and expected magic %x",
300 packet->magic, MULTIFD_MAGIC);
301 return -1;
304 packet->version = be32_to_cpu(packet->version);
305 if (packet->version != MULTIFD_VERSION) {
306 error_setg(errp, "multifd: received packet "
307 "version %d and expected version %d",
308 packet->version, MULTIFD_VERSION);
309 return -1;
312 p->flags = be32_to_cpu(packet->flags);
314 packet->pages_alloc = be32_to_cpu(packet->pages_alloc);
316 * If we received a packet that is 100 times bigger than expected
317 * just stop migration. It is a magic number.
319 if (packet->pages_alloc > pages_max * 100) {
320 error_setg(errp, "multifd: received packet "
321 "with size %d and expected a maximum size of %d",
322 packet->pages_alloc, pages_max * 100) ;
323 return -1;
326 * We received a packet that is bigger than expected but inside
327 * reasonable limits (see previous comment). Just reallocate.
329 if (packet->pages_alloc > p->pages->allocated) {
330 multifd_pages_clear(p->pages);
331 p->pages = multifd_pages_init(packet->pages_alloc);
334 p->pages->used = be32_to_cpu(packet->pages_used);
335 if (p->pages->used > packet->pages_alloc) {
336 error_setg(errp, "multifd: received packet "
337 "with %d pages and expected maximum pages are %d",
338 p->pages->used, packet->pages_alloc) ;
339 return -1;
342 p->next_packet_size = be32_to_cpu(packet->next_packet_size);
343 p->packet_num = be64_to_cpu(packet->packet_num);
345 if (p->pages->used == 0) {
346 return 0;
349 /* make sure that ramblock is 0 terminated */
350 packet->ramblock[255] = 0;
351 block = qemu_ram_block_by_name(packet->ramblock);
352 if (!block) {
353 error_setg(errp, "multifd: unknown ram block %s",
354 packet->ramblock);
355 return -1;
358 for (i = 0; i < p->pages->used; i++) {
359 uint64_t offset = be64_to_cpu(packet->offset[i]);
361 if (offset > (block->used_length - qemu_target_page_size())) {
362 error_setg(errp, "multifd: offset too long %" PRIu64
363 " (max " RAM_ADDR_FMT ")",
364 offset, block->max_length);
365 return -1;
367 p->pages->iov[i].iov_base = block->host + offset;
368 p->pages->iov[i].iov_len = qemu_target_page_size();
371 return 0;
374 struct {
375 MultiFDSendParams *params;
376 /* array of pages to sent */
377 MultiFDPages_t *pages;
378 /* global number of generated multifd packets */
379 uint64_t packet_num;
380 /* send channels ready */
381 QemuSemaphore channels_ready;
383 * Have we already run terminate threads. There is a race when it
384 * happens that we got one error while we are exiting.
385 * We will use atomic operations. Only valid values are 0 and 1.
387 int exiting;
388 /* multifd ops */
389 MultiFDMethods *ops;
390 } *multifd_send_state;
393 * How we use multifd_send_state->pages and channel->pages?
395 * We create a pages for each channel, and a main one. Each time that
396 * we need to send a batch of pages we interchange the ones between
397 * multifd_send_state and the channel that is sending it. There are
398 * two reasons for that:
399 * - to not have to do so many mallocs during migration
400 * - to make easier to know what to free at the end of migration
402 * This way we always know who is the owner of each "pages" struct,
403 * and we don't need any locking. It belongs to the migration thread
404 * or to the channel thread. Switching is safe because the migration
405 * thread is using the channel mutex when changing it, and the channel
406 * have to had finish with its own, otherwise pending_job can't be
407 * false.
410 static int multifd_send_pages(QEMUFile *f)
412 int i;
413 static int next_channel;
414 MultiFDSendParams *p = NULL; /* make happy gcc */
415 MultiFDPages_t *pages = multifd_send_state->pages;
416 uint64_t transferred;
418 if (qatomic_read(&multifd_send_state->exiting)) {
419 return -1;
422 qemu_sem_wait(&multifd_send_state->channels_ready);
424 * next_channel can remain from a previous migration that was
425 * using more channels, so ensure it doesn't overflow if the
426 * limit is lower now.
428 next_channel %= migrate_multifd_channels();
429 for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
430 p = &multifd_send_state->params[i];
432 qemu_mutex_lock(&p->mutex);
433 if (p->quit) {
434 error_report("%s: channel %d has already quit!", __func__, i);
435 qemu_mutex_unlock(&p->mutex);
436 return -1;
438 if (!p->pending_job) {
439 p->pending_job++;
440 next_channel = (i + 1) % migrate_multifd_channels();
441 break;
443 qemu_mutex_unlock(&p->mutex);
445 assert(!p->pages->used);
446 assert(!p->pages->block);
448 p->packet_num = multifd_send_state->packet_num++;
449 multifd_send_state->pages = p->pages;
450 p->pages = pages;
451 transferred = ((uint64_t) pages->used) * qemu_target_page_size()
452 + p->packet_len;
453 qemu_file_update_transfer(f, transferred);
454 ram_counters.multifd_bytes += transferred;
455 ram_counters.transferred += transferred;
456 qemu_mutex_unlock(&p->mutex);
457 qemu_sem_post(&p->sem);
459 return 1;
462 int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
464 MultiFDPages_t *pages = multifd_send_state->pages;
466 if (!pages->block) {
467 pages->block = block;
470 if (pages->block == block) {
471 pages->offset[pages->used] = offset;
472 pages->iov[pages->used].iov_base = block->host + offset;
473 pages->iov[pages->used].iov_len = qemu_target_page_size();
474 pages->used++;
476 if (pages->used < pages->allocated) {
477 return 1;
481 if (multifd_send_pages(f) < 0) {
482 return -1;
485 if (pages->block != block) {
486 return multifd_queue_page(f, block, offset);
489 return 1;
492 static void multifd_send_terminate_threads(Error *err)
494 int i;
496 trace_multifd_send_terminate_threads(err != NULL);
498 if (err) {
499 MigrationState *s = migrate_get_current();
500 migrate_set_error(s, err);
501 if (s->state == MIGRATION_STATUS_SETUP ||
502 s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
503 s->state == MIGRATION_STATUS_DEVICE ||
504 s->state == MIGRATION_STATUS_ACTIVE) {
505 migrate_set_state(&s->state, s->state,
506 MIGRATION_STATUS_FAILED);
511 * We don't want to exit each threads twice. Depending on where
512 * we get the error, or if there are two independent errors in two
513 * threads at the same time, we can end calling this function
514 * twice.
516 if (qatomic_xchg(&multifd_send_state->exiting, 1)) {
517 return;
520 for (i = 0; i < migrate_multifd_channels(); i++) {
521 MultiFDSendParams *p = &multifd_send_state->params[i];
523 qemu_mutex_lock(&p->mutex);
524 p->quit = true;
525 qemu_sem_post(&p->sem);
526 qemu_mutex_unlock(&p->mutex);
530 void multifd_save_cleanup(void)
532 int i;
534 if (!migrate_use_multifd()) {
535 return;
537 multifd_send_terminate_threads(NULL);
538 for (i = 0; i < migrate_multifd_channels(); i++) {
539 MultiFDSendParams *p = &multifd_send_state->params[i];
541 if (p->running) {
542 qemu_thread_join(&p->thread);
545 for (i = 0; i < migrate_multifd_channels(); i++) {
546 MultiFDSendParams *p = &multifd_send_state->params[i];
547 Error *local_err = NULL;
549 socket_send_channel_destroy(p->c);
550 p->c = NULL;
551 qemu_mutex_destroy(&p->mutex);
552 qemu_sem_destroy(&p->sem);
553 qemu_sem_destroy(&p->sem_sync);
554 g_free(p->name);
555 p->name = NULL;
556 g_free(p->tls_hostname);
557 p->tls_hostname = NULL;
558 multifd_pages_clear(p->pages);
559 p->pages = NULL;
560 p->packet_len = 0;
561 g_free(p->packet);
562 p->packet = NULL;
563 multifd_send_state->ops->send_cleanup(p, &local_err);
564 if (local_err) {
565 migrate_set_error(migrate_get_current(), local_err);
566 error_free(local_err);
569 qemu_sem_destroy(&multifd_send_state->channels_ready);
570 g_free(multifd_send_state->params);
571 multifd_send_state->params = NULL;
572 multifd_pages_clear(multifd_send_state->pages);
573 multifd_send_state->pages = NULL;
574 g_free(multifd_send_state);
575 multifd_send_state = NULL;
578 void multifd_send_sync_main(QEMUFile *f)
580 int i;
582 if (!migrate_use_multifd()) {
583 return;
585 if (multifd_send_state->pages->used) {
586 if (multifd_send_pages(f) < 0) {
587 error_report("%s: multifd_send_pages fail", __func__);
588 return;
591 for (i = 0; i < migrate_multifd_channels(); i++) {
592 MultiFDSendParams *p = &multifd_send_state->params[i];
594 trace_multifd_send_sync_main_signal(p->id);
596 qemu_mutex_lock(&p->mutex);
598 if (p->quit) {
599 error_report("%s: channel %d has already quit", __func__, i);
600 qemu_mutex_unlock(&p->mutex);
601 return;
604 p->packet_num = multifd_send_state->packet_num++;
605 p->flags |= MULTIFD_FLAG_SYNC;
606 p->pending_job++;
607 qemu_file_update_transfer(f, p->packet_len);
608 ram_counters.multifd_bytes += p->packet_len;
609 ram_counters.transferred += p->packet_len;
610 qemu_mutex_unlock(&p->mutex);
611 qemu_sem_post(&p->sem);
613 for (i = 0; i < migrate_multifd_channels(); i++) {
614 MultiFDSendParams *p = &multifd_send_state->params[i];
616 trace_multifd_send_sync_main_wait(p->id);
617 qemu_sem_wait(&p->sem_sync);
619 trace_multifd_send_sync_main(multifd_send_state->packet_num);
622 static void *multifd_send_thread(void *opaque)
624 MultiFDSendParams *p = opaque;
625 Error *local_err = NULL;
626 int ret = 0;
627 uint32_t flags = 0;
629 trace_multifd_send_thread_start(p->id);
630 rcu_register_thread();
632 if (multifd_send_initial_packet(p, &local_err) < 0) {
633 ret = -1;
634 goto out;
636 /* initial packet */
637 p->num_packets = 1;
639 while (true) {
640 qemu_sem_wait(&p->sem);
642 if (qatomic_read(&multifd_send_state->exiting)) {
643 break;
645 qemu_mutex_lock(&p->mutex);
647 if (p->pending_job) {
648 uint32_t used = p->pages->used;
649 uint64_t packet_num = p->packet_num;
650 flags = p->flags;
652 if (used) {
653 ret = multifd_send_state->ops->send_prepare(p, used,
654 &local_err);
655 if (ret != 0) {
656 qemu_mutex_unlock(&p->mutex);
657 break;
660 multifd_send_fill_packet(p);
661 p->flags = 0;
662 p->num_packets++;
663 p->num_pages += used;
664 p->pages->used = 0;
665 p->pages->block = NULL;
666 qemu_mutex_unlock(&p->mutex);
668 trace_multifd_send(p->id, packet_num, used, flags,
669 p->next_packet_size);
671 ret = qio_channel_write_all(p->c, (void *)p->packet,
672 p->packet_len, &local_err);
673 if (ret != 0) {
674 break;
677 if (used) {
678 ret = multifd_send_state->ops->send_write(p, used, &local_err);
679 if (ret != 0) {
680 break;
684 qemu_mutex_lock(&p->mutex);
685 p->pending_job--;
686 qemu_mutex_unlock(&p->mutex);
688 if (flags & MULTIFD_FLAG_SYNC) {
689 qemu_sem_post(&p->sem_sync);
691 qemu_sem_post(&multifd_send_state->channels_ready);
692 } else if (p->quit) {
693 qemu_mutex_unlock(&p->mutex);
694 break;
695 } else {
696 qemu_mutex_unlock(&p->mutex);
697 /* sometimes there are spurious wakeups */
701 out:
702 if (local_err) {
703 trace_multifd_send_error(p->id);
704 multifd_send_terminate_threads(local_err);
705 error_free(local_err);
709 * Error happen, I will exit, but I can't just leave, tell
710 * who pay attention to me.
712 if (ret != 0) {
713 qemu_sem_post(&p->sem_sync);
714 qemu_sem_post(&multifd_send_state->channels_ready);
717 qemu_mutex_lock(&p->mutex);
718 p->running = false;
719 qemu_mutex_unlock(&p->mutex);
721 rcu_unregister_thread();
722 trace_multifd_send_thread_end(p->id, p->num_packets, p->num_pages);
724 return NULL;
727 static bool multifd_channel_connect(MultiFDSendParams *p,
728 QIOChannel *ioc,
729 Error *error);
731 static void multifd_tls_outgoing_handshake(QIOTask *task,
732 gpointer opaque)
734 MultiFDSendParams *p = opaque;
735 QIOChannel *ioc = QIO_CHANNEL(qio_task_get_source(task));
736 Error *err = NULL;
738 if (qio_task_propagate_error(task, &err)) {
739 trace_multifd_tls_outgoing_handshake_error(ioc, error_get_pretty(err));
740 } else {
741 trace_multifd_tls_outgoing_handshake_complete(ioc);
744 if (!multifd_channel_connect(p, ioc, err)) {
746 * Error happen, mark multifd_send_thread status as 'quit' although it
747 * is not created, and then tell who pay attention to me.
749 p->quit = true;
750 qemu_sem_post(&multifd_send_state->channels_ready);
751 qemu_sem_post(&p->sem_sync);
755 static void *multifd_tls_handshake_thread(void *opaque)
757 MultiFDSendParams *p = opaque;
758 QIOChannelTLS *tioc = QIO_CHANNEL_TLS(p->c);
760 qio_channel_tls_handshake(tioc,
761 multifd_tls_outgoing_handshake,
763 NULL,
764 NULL);
765 return NULL;
768 static void multifd_tls_channel_connect(MultiFDSendParams *p,
769 QIOChannel *ioc,
770 Error **errp)
772 MigrationState *s = migrate_get_current();
773 const char *hostname = p->tls_hostname;
774 QIOChannelTLS *tioc;
776 tioc = migration_tls_client_create(s, ioc, hostname, errp);
777 if (!tioc) {
778 return;
781 object_unref(OBJECT(ioc));
782 trace_multifd_tls_outgoing_handshake_start(ioc, tioc, hostname);
783 qio_channel_set_name(QIO_CHANNEL(tioc), "multifd-tls-outgoing");
784 p->c = QIO_CHANNEL(tioc);
785 qemu_thread_create(&p->thread, "multifd-tls-handshake-worker",
786 multifd_tls_handshake_thread, p,
787 QEMU_THREAD_JOINABLE);
790 static bool multifd_channel_connect(MultiFDSendParams *p,
791 QIOChannel *ioc,
792 Error *error)
794 MigrationState *s = migrate_get_current();
796 trace_multifd_set_outgoing_channel(
797 ioc, object_get_typename(OBJECT(ioc)), p->tls_hostname, error);
799 if (!error) {
800 if (s->parameters.tls_creds &&
801 *s->parameters.tls_creds &&
802 !object_dynamic_cast(OBJECT(ioc),
803 TYPE_QIO_CHANNEL_TLS)) {
804 multifd_tls_channel_connect(p, ioc, &error);
805 if (!error) {
807 * tls_channel_connect will call back to this
808 * function after the TLS handshake,
809 * so we mustn't call multifd_send_thread until then
811 return true;
812 } else {
813 return false;
815 } else {
816 /* update for tls qio channel */
817 p->c = ioc;
818 qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
819 QEMU_THREAD_JOINABLE);
821 return true;
824 return false;
827 static void multifd_new_send_channel_cleanup(MultiFDSendParams *p,
828 QIOChannel *ioc, Error *err)
830 migrate_set_error(migrate_get_current(), err);
831 /* Error happen, we need to tell who pay attention to me */
832 qemu_sem_post(&multifd_send_state->channels_ready);
833 qemu_sem_post(&p->sem_sync);
835 * Although multifd_send_thread is not created, but main migration
836 * thread neet to judge whether it is running, so we need to mark
837 * its status.
839 p->quit = true;
840 object_unref(OBJECT(ioc));
841 error_free(err);
844 static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
846 MultiFDSendParams *p = opaque;
847 QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
848 Error *local_err = NULL;
850 trace_multifd_new_send_channel_async(p->id);
851 if (qio_task_propagate_error(task, &local_err)) {
852 goto cleanup;
853 } else {
854 p->c = QIO_CHANNEL(sioc);
855 qio_channel_set_delay(p->c, false);
856 p->running = true;
857 if (!multifd_channel_connect(p, sioc, local_err)) {
858 goto cleanup;
860 return;
863 cleanup:
864 multifd_new_send_channel_cleanup(p, sioc, local_err);
867 int multifd_save_setup(Error **errp)
869 int thread_count;
870 uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
871 uint8_t i;
872 MigrationState *s;
874 if (!migrate_use_multifd()) {
875 return 0;
877 s = migrate_get_current();
878 thread_count = migrate_multifd_channels();
879 multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
880 multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
881 multifd_send_state->pages = multifd_pages_init(page_count);
882 qemu_sem_init(&multifd_send_state->channels_ready, 0);
883 qatomic_set(&multifd_send_state->exiting, 0);
884 multifd_send_state->ops = multifd_ops[migrate_multifd_compression()];
886 for (i = 0; i < thread_count; i++) {
887 MultiFDSendParams *p = &multifd_send_state->params[i];
889 qemu_mutex_init(&p->mutex);
890 qemu_sem_init(&p->sem, 0);
891 qemu_sem_init(&p->sem_sync, 0);
892 p->quit = false;
893 p->pending_job = 0;
894 p->id = i;
895 p->pages = multifd_pages_init(page_count);
896 p->packet_len = sizeof(MultiFDPacket_t)
897 + sizeof(uint64_t) * page_count;
898 p->packet = g_malloc0(p->packet_len);
899 p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
900 p->packet->version = cpu_to_be32(MULTIFD_VERSION);
901 p->name = g_strdup_printf("multifdsend_%d", i);
902 p->tls_hostname = g_strdup(s->hostname);
903 socket_send_channel_create(multifd_new_send_channel_async, p);
906 for (i = 0; i < thread_count; i++) {
907 MultiFDSendParams *p = &multifd_send_state->params[i];
908 Error *local_err = NULL;
909 int ret;
911 ret = multifd_send_state->ops->send_setup(p, &local_err);
912 if (ret) {
913 error_propagate(errp, local_err);
914 return ret;
917 return 0;
920 struct {
921 MultiFDRecvParams *params;
922 /* number of created threads */
923 int count;
924 /* syncs main thread and channels */
925 QemuSemaphore sem_sync;
926 /* global number of generated multifd packets */
927 uint64_t packet_num;
928 /* multifd ops */
929 MultiFDMethods *ops;
930 } *multifd_recv_state;
932 static void multifd_recv_terminate_threads(Error *err)
934 int i;
936 trace_multifd_recv_terminate_threads(err != NULL);
938 if (err) {
939 MigrationState *s = migrate_get_current();
940 migrate_set_error(s, err);
941 if (s->state == MIGRATION_STATUS_SETUP ||
942 s->state == MIGRATION_STATUS_ACTIVE) {
943 migrate_set_state(&s->state, s->state,
944 MIGRATION_STATUS_FAILED);
948 for (i = 0; i < migrate_multifd_channels(); i++) {
949 MultiFDRecvParams *p = &multifd_recv_state->params[i];
951 qemu_mutex_lock(&p->mutex);
952 p->quit = true;
954 * We could arrive here for two reasons:
955 * - normal quit, i.e. everything went fine, just finished
956 * - error quit: We close the channels so the channel threads
957 * finish the qio_channel_read_all_eof()
959 if (p->c) {
960 qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
962 qemu_mutex_unlock(&p->mutex);
966 int multifd_load_cleanup(Error **errp)
968 int i;
970 if (!migrate_use_multifd()) {
971 return 0;
973 multifd_recv_terminate_threads(NULL);
974 for (i = 0; i < migrate_multifd_channels(); i++) {
975 MultiFDRecvParams *p = &multifd_recv_state->params[i];
977 if (p->running) {
978 p->quit = true;
980 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
981 * however try to wakeup it without harm in cleanup phase.
983 qemu_sem_post(&p->sem_sync);
984 qemu_thread_join(&p->thread);
987 for (i = 0; i < migrate_multifd_channels(); i++) {
988 MultiFDRecvParams *p = &multifd_recv_state->params[i];
990 if (object_dynamic_cast(OBJECT(p->c), TYPE_QIO_CHANNEL_SOCKET)
991 && OBJECT(p->c)->ref == 1) {
992 yank_unregister_function(MIGRATION_YANK_INSTANCE,
993 migration_yank_iochannel,
994 QIO_CHANNEL(p->c));
997 object_unref(OBJECT(p->c));
998 p->c = NULL;
999 qemu_mutex_destroy(&p->mutex);
1000 qemu_sem_destroy(&p->sem_sync);
1001 g_free(p->name);
1002 p->name = NULL;
1003 multifd_pages_clear(p->pages);
1004 p->pages = NULL;
1005 p->packet_len = 0;
1006 g_free(p->packet);
1007 p->packet = NULL;
1008 multifd_recv_state->ops->recv_cleanup(p);
1010 qemu_sem_destroy(&multifd_recv_state->sem_sync);
1011 g_free(multifd_recv_state->params);
1012 multifd_recv_state->params = NULL;
1013 g_free(multifd_recv_state);
1014 multifd_recv_state = NULL;
1016 return 0;
1019 void multifd_recv_sync_main(void)
1021 int i;
1023 if (!migrate_use_multifd()) {
1024 return;
1026 for (i = 0; i < migrate_multifd_channels(); i++) {
1027 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1029 trace_multifd_recv_sync_main_wait(p->id);
1030 qemu_sem_wait(&multifd_recv_state->sem_sync);
1032 for (i = 0; i < migrate_multifd_channels(); i++) {
1033 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1035 WITH_QEMU_LOCK_GUARD(&p->mutex) {
1036 if (multifd_recv_state->packet_num < p->packet_num) {
1037 multifd_recv_state->packet_num = p->packet_num;
1040 trace_multifd_recv_sync_main_signal(p->id);
1041 qemu_sem_post(&p->sem_sync);
1043 trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
1046 static void *multifd_recv_thread(void *opaque)
1048 MultiFDRecvParams *p = opaque;
1049 Error *local_err = NULL;
1050 int ret;
1052 trace_multifd_recv_thread_start(p->id);
1053 rcu_register_thread();
1055 while (true) {
1056 uint32_t used;
1057 uint32_t flags;
1059 if (p->quit) {
1060 break;
1063 ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
1064 p->packet_len, &local_err);
1065 if (ret == 0) { /* EOF */
1066 break;
1068 if (ret == -1) { /* Error */
1069 break;
1072 qemu_mutex_lock(&p->mutex);
1073 ret = multifd_recv_unfill_packet(p, &local_err);
1074 if (ret) {
1075 qemu_mutex_unlock(&p->mutex);
1076 break;
1079 used = p->pages->used;
1080 flags = p->flags;
1081 /* recv methods don't know how to handle the SYNC flag */
1082 p->flags &= ~MULTIFD_FLAG_SYNC;
1083 trace_multifd_recv(p->id, p->packet_num, used, flags,
1084 p->next_packet_size);
1085 p->num_packets++;
1086 p->num_pages += used;
1087 qemu_mutex_unlock(&p->mutex);
1089 if (used) {
1090 ret = multifd_recv_state->ops->recv_pages(p, used, &local_err);
1091 if (ret != 0) {
1092 break;
1096 if (flags & MULTIFD_FLAG_SYNC) {
1097 qemu_sem_post(&multifd_recv_state->sem_sync);
1098 qemu_sem_wait(&p->sem_sync);
1102 if (local_err) {
1103 multifd_recv_terminate_threads(local_err);
1104 error_free(local_err);
1106 qemu_mutex_lock(&p->mutex);
1107 p->running = false;
1108 qemu_mutex_unlock(&p->mutex);
1110 rcu_unregister_thread();
1111 trace_multifd_recv_thread_end(p->id, p->num_packets, p->num_pages);
1113 return NULL;
1116 int multifd_load_setup(Error **errp)
1118 int thread_count;
1119 uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
1120 uint8_t i;
1122 if (!migrate_use_multifd()) {
1123 return 0;
1125 thread_count = migrate_multifd_channels();
1126 multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
1127 multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
1128 qatomic_set(&multifd_recv_state->count, 0);
1129 qemu_sem_init(&multifd_recv_state->sem_sync, 0);
1130 multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()];
1132 for (i = 0; i < thread_count; i++) {
1133 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1135 qemu_mutex_init(&p->mutex);
1136 qemu_sem_init(&p->sem_sync, 0);
1137 p->quit = false;
1138 p->id = i;
1139 p->pages = multifd_pages_init(page_count);
1140 p->packet_len = sizeof(MultiFDPacket_t)
1141 + sizeof(uint64_t) * page_count;
1142 p->packet = g_malloc0(p->packet_len);
1143 p->name = g_strdup_printf("multifdrecv_%d", i);
1146 for (i = 0; i < thread_count; i++) {
1147 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1148 Error *local_err = NULL;
1149 int ret;
1151 ret = multifd_recv_state->ops->recv_setup(p, &local_err);
1152 if (ret) {
1153 error_propagate(errp, local_err);
1154 return ret;
1157 return 0;
1160 bool multifd_recv_all_channels_created(void)
1162 int thread_count = migrate_multifd_channels();
1164 if (!migrate_use_multifd()) {
1165 return true;
1168 return thread_count == qatomic_read(&multifd_recv_state->count);
1172 * Try to receive all multifd channels to get ready for the migration.
1173 * - Return true and do not set @errp when correctly receiving all channels;
1174 * - Return false and do not set @errp when correctly receiving the current one;
1175 * - Return false and set @errp when failing to receive the current channel.
1177 bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
1179 MultiFDRecvParams *p;
1180 Error *local_err = NULL;
1181 int id;
1183 id = multifd_recv_initial_packet(ioc, &local_err);
1184 if (id < 0) {
1185 multifd_recv_terminate_threads(local_err);
1186 error_propagate_prepend(errp, local_err,
1187 "failed to receive packet"
1188 " via multifd channel %d: ",
1189 qatomic_read(&multifd_recv_state->count));
1190 return false;
1192 trace_multifd_recv_new_channel(id);
1194 p = &multifd_recv_state->params[id];
1195 if (p->c != NULL) {
1196 error_setg(&local_err, "multifd: received id '%d' already setup'",
1197 id);
1198 multifd_recv_terminate_threads(local_err);
1199 error_propagate(errp, local_err);
1200 return false;
1202 p->c = ioc;
1203 object_ref(OBJECT(ioc));
1204 /* initial packet */
1205 p->num_packets = 1;
1207 p->running = true;
1208 qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
1209 QEMU_THREAD_JOINABLE);
1210 qatomic_inc(&multifd_recv_state->count);
1211 return qatomic_read(&multifd_recv_state->count) ==
1212 migrate_multifd_channels();