Merge remote-tracking branch 'remotes/mcayland/tags/qemu-sparc-20210412' into staging
[qemu.git] / migration / multifd.c
1 /*
2 * Multifd common code
3 *
4 * Copyright (c) 2019-2020 Red Hat Inc
5 *
6 * Authors:
7 * Juan Quintela <quintela@redhat.com>
8 *
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.
11 */
12
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"
27
28 #include "qemu/yank.h"
29 #include "io/channel-socket.h"
30
31 /* Multiple fd's */
32
33 #define MULTIFD_MAGIC 0x11223344U
34 #define MULTIFD_VERSION 1
35
36 typedef struct {
37 uint32_t magic;
38 uint32_t version;
39 unsigned char uuid[16]; /* QemuUUID */
40 uint8_t id;
41 uint8_t unused1[7]; /* Reserved for future use */
42 uint64_t unused2[4]; /* Reserved for future use */
43 } __attribute__((packed)) MultiFDInit_t;
44
45 /* Multifd without compression */
46
47 /**
48 * nocomp_send_setup: setup send side
49 *
50 * For no compression this function does nothing.
51 *
52 * Returns 0 for success or -1 for error
53 *
54 * @p: Params for the channel that we are using
55 * @errp: pointer to an error
56 */
57 static int nocomp_send_setup(MultiFDSendParams *p, Error **errp)
58 {
59 return 0;
60 }
61
62 /**
63 * nocomp_send_cleanup: cleanup send side
64 *
65 * For no compression this function does nothing.
66 *
67 * @p: Params for the channel that we are using
68 */
69 static void nocomp_send_cleanup(MultiFDSendParams *p, Error **errp)
70 {
71 return;
72 }
73
74 /**
75 * nocomp_send_prepare: prepare date to be able to send
76 *
77 * For no compression we just have to calculate the size of the
78 * packet.
79 *
80 * Returns 0 for success or -1 for error
81 *
82 * @p: Params for the channel that we are using
83 * @used: number of pages used
84 * @errp: pointer to an error
85 */
86 static int nocomp_send_prepare(MultiFDSendParams *p, uint32_t used,
87 Error **errp)
88 {
89 p->next_packet_size = used * qemu_target_page_size();
90 p->flags |= MULTIFD_FLAG_NOCOMP;
91 return 0;
92 }
93
94 /**
95 * nocomp_send_write: do the actual write of the data
96 *
97 * For no compression we just have to write the data.
98 *
99 * Returns 0 for success or -1 for error
100 *
101 * @p: Params for the channel that we are using
102 * @used: number of pages used
103 * @errp: pointer to an error
104 */
105 static int nocomp_send_write(MultiFDSendParams *p, uint32_t used, Error **errp)
106 {
107 return qio_channel_writev_all(p->c, p->pages->iov, used, errp);
108 }
109
110 /**
111 * nocomp_recv_setup: setup receive side
112 *
113 * For no compression this function does nothing.
114 *
115 * Returns 0 for success or -1 for error
116 *
117 * @p: Params for the channel that we are using
118 * @errp: pointer to an error
119 */
120 static int nocomp_recv_setup(MultiFDRecvParams *p, Error **errp)
121 {
122 return 0;
123 }
124
125 /**
126 * nocomp_recv_cleanup: setup receive side
127 *
128 * For no compression this function does nothing.
129 *
130 * @p: Params for the channel that we are using
131 */
132 static void nocomp_recv_cleanup(MultiFDRecvParams *p)
133 {
134 }
135
136 /**
137 * nocomp_recv_pages: read the data from the channel into actual pages
138 *
139 * For no compression we just need to read things into the correct place.
140 *
141 * Returns 0 for success or -1 for error
142 *
143 * @p: Params for the channel that we are using
144 * @used: number of pages used
145 * @errp: pointer to an error
146 */
147 static int nocomp_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
148 {
149 uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
150
151 if (flags != MULTIFD_FLAG_NOCOMP) {
152 error_setg(errp, "multifd %d: flags received %x flags expected %x",
153 p->id, flags, MULTIFD_FLAG_NOCOMP);
154 return -1;
155 }
156 return qio_channel_readv_all(p->c, p->pages->iov, used, errp);
157 }
158
159 static MultiFDMethods multifd_nocomp_ops = {
160 .send_setup = nocomp_send_setup,
161 .send_cleanup = nocomp_send_cleanup,
162 .send_prepare = nocomp_send_prepare,
163 .send_write = nocomp_send_write,
164 .recv_setup = nocomp_recv_setup,
165 .recv_cleanup = nocomp_recv_cleanup,
166 .recv_pages = nocomp_recv_pages
167 };
168
169 static MultiFDMethods *multifd_ops[MULTIFD_COMPRESSION__MAX] = {
170 [MULTIFD_COMPRESSION_NONE] = &multifd_nocomp_ops,
171 };
172
173 void multifd_register_ops(int method, MultiFDMethods *ops)
174 {
175 assert(0 < method && method < MULTIFD_COMPRESSION__MAX);
176 multifd_ops[method] = ops;
177 }
178
179 static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
180 {
181 MultiFDInit_t msg = {};
182 int ret;
183
184 msg.magic = cpu_to_be32(MULTIFD_MAGIC);
185 msg.version = cpu_to_be32(MULTIFD_VERSION);
186 msg.id = p->id;
187 memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
188
189 ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), errp);
190 if (ret != 0) {
191 return -1;
192 }
193 return 0;
194 }
195
196 static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
197 {
198 MultiFDInit_t msg;
199 int ret;
200
201 ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
202 if (ret != 0) {
203 return -1;
204 }
205
206 msg.magic = be32_to_cpu(msg.magic);
207 msg.version = be32_to_cpu(msg.version);
208
209 if (msg.magic != MULTIFD_MAGIC) {
210 error_setg(errp, "multifd: received packet magic %x "
211 "expected %x", msg.magic, MULTIFD_MAGIC);
212 return -1;
213 }
214
215 if (msg.version != MULTIFD_VERSION) {
216 error_setg(errp, "multifd: received packet version %d "
217 "expected %d", msg.version, MULTIFD_VERSION);
218 return -1;
219 }
220
221 if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
222 char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
223 char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
224
225 error_setg(errp, "multifd: received uuid '%s' and expected "
226 "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
227 g_free(uuid);
228 g_free(msg_uuid);
229 return -1;
230 }
231
232 if (msg.id > migrate_multifd_channels()) {
233 error_setg(errp, "multifd: received channel version %d "
234 "expected %d", msg.version, MULTIFD_VERSION);
235 return -1;
236 }
237
238 return msg.id;
239 }
240
241 static MultiFDPages_t *multifd_pages_init(size_t size)
242 {
243 MultiFDPages_t *pages = g_new0(MultiFDPages_t, 1);
244
245 pages->allocated = size;
246 pages->iov = g_new0(struct iovec, size);
247 pages->offset = g_new0(ram_addr_t, size);
248
249 return pages;
250 }
251
252 static void multifd_pages_clear(MultiFDPages_t *pages)
253 {
254 pages->used = 0;
255 pages->allocated = 0;
256 pages->packet_num = 0;
257 pages->block = NULL;
258 g_free(pages->iov);
259 pages->iov = NULL;
260 g_free(pages->offset);
261 pages->offset = NULL;
262 g_free(pages);
263 }
264
265 static void multifd_send_fill_packet(MultiFDSendParams *p)
266 {
267 MultiFDPacket_t *packet = p->packet;
268 int i;
269
270 packet->flags = cpu_to_be32(p->flags);
271 packet->pages_alloc = cpu_to_be32(p->pages->allocated);
272 packet->pages_used = cpu_to_be32(p->pages->used);
273 packet->next_packet_size = cpu_to_be32(p->next_packet_size);
274 packet->packet_num = cpu_to_be64(p->packet_num);
275
276 if (p->pages->block) {
277 strncpy(packet->ramblock, p->pages->block->idstr, 256);
278 }
279
280 for (i = 0; i < p->pages->used; i++) {
281 /* there are architectures where ram_addr_t is 32 bit */
282 uint64_t temp = p->pages->offset[i];
283
284 packet->offset[i] = cpu_to_be64(temp);
285 }
286 }
287
288 static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
289 {
290 MultiFDPacket_t *packet = p->packet;
291 uint32_t pages_max = MULTIFD_PACKET_SIZE / qemu_target_page_size();
292 RAMBlock *block;
293 int i;
294
295 packet->magic = be32_to_cpu(packet->magic);
296 if (packet->magic != MULTIFD_MAGIC) {
297 error_setg(errp, "multifd: received packet "
298 "magic %x and expected magic %x",
299 packet->magic, MULTIFD_MAGIC);
300 return -1;
301 }
302
303 packet->version = be32_to_cpu(packet->version);
304 if (packet->version != MULTIFD_VERSION) {
305 error_setg(errp, "multifd: received packet "
306 "version %d and expected version %d",
307 packet->version, MULTIFD_VERSION);
308 return -1;
309 }
310
311 p->flags = be32_to_cpu(packet->flags);
312
313 packet->pages_alloc = be32_to_cpu(packet->pages_alloc);
314 /*
315 * If we received a packet that is 100 times bigger than expected
316 * just stop migration. It is a magic number.
317 */
318 if (packet->pages_alloc > pages_max * 100) {
319 error_setg(errp, "multifd: received packet "
320 "with size %d and expected a maximum size of %d",
321 packet->pages_alloc, pages_max * 100) ;
322 return -1;
323 }
324 /*
325 * We received a packet that is bigger than expected but inside
326 * reasonable limits (see previous comment). Just reallocate.
327 */
328 if (packet->pages_alloc > p->pages->allocated) {
329 multifd_pages_clear(p->pages);
330 p->pages = multifd_pages_init(packet->pages_alloc);
331 }
332
333 p->pages->used = be32_to_cpu(packet->pages_used);
334 if (p->pages->used > packet->pages_alloc) {
335 error_setg(errp, "multifd: received packet "
336 "with %d pages and expected maximum pages are %d",
337 p->pages->used, packet->pages_alloc) ;
338 return -1;
339 }
340
341 p->next_packet_size = be32_to_cpu(packet->next_packet_size);
342 p->packet_num = be64_to_cpu(packet->packet_num);
343
344 if (p->pages->used == 0) {
345 return 0;
346 }
347
348 /* make sure that ramblock is 0 terminated */
349 packet->ramblock[255] = 0;
350 block = qemu_ram_block_by_name(packet->ramblock);
351 if (!block) {
352 error_setg(errp, "multifd: unknown ram block %s",
353 packet->ramblock);
354 return -1;
355 }
356
357 for (i = 0; i < p->pages->used; i++) {
358 uint64_t offset = be64_to_cpu(packet->offset[i]);
359
360 if (offset > (block->used_length - qemu_target_page_size())) {
361 error_setg(errp, "multifd: offset too long %" PRIu64
362 " (max " RAM_ADDR_FMT ")",
363 offset, block->max_length);
364 return -1;
365 }
366 p->pages->iov[i].iov_base = block->host + offset;
367 p->pages->iov[i].iov_len = qemu_target_page_size();
368 }
369
370 return 0;
371 }
372
373 struct {
374 MultiFDSendParams *params;
375 /* array of pages to sent */
376 MultiFDPages_t *pages;
377 /* global number of generated multifd packets */
378 uint64_t packet_num;
379 /* send channels ready */
380 QemuSemaphore channels_ready;
381 /*
382 * Have we already run terminate threads. There is a race when it
383 * happens that we got one error while we are exiting.
384 * We will use atomic operations. Only valid values are 0 and 1.
385 */
386 int exiting;
387 /* multifd ops */
388 MultiFDMethods *ops;
389 } *multifd_send_state;
390
391 /*
392 * How we use multifd_send_state->pages and channel->pages?
393 *
394 * We create a pages for each channel, and a main one. Each time that
395 * we need to send a batch of pages we interchange the ones between
396 * multifd_send_state and the channel that is sending it. There are
397 * two reasons for that:
398 * - to not have to do so many mallocs during migration
399 * - to make easier to know what to free at the end of migration
400 *
401 * This way we always know who is the owner of each "pages" struct,
402 * and we don't need any locking. It belongs to the migration thread
403 * or to the channel thread. Switching is safe because the migration
404 * thread is using the channel mutex when changing it, and the channel
405 * have to had finish with its own, otherwise pending_job can't be
406 * false.
407 */
408
409 static int multifd_send_pages(QEMUFile *f)
410 {
411 int i;
412 static int next_channel;
413 MultiFDSendParams *p = NULL; /* make happy gcc */
414 MultiFDPages_t *pages = multifd_send_state->pages;
415 uint64_t transferred;
416
417 if (qatomic_read(&multifd_send_state->exiting)) {
418 return -1;
419 }
420
421 qemu_sem_wait(&multifd_send_state->channels_ready);
422 /*
423 * next_channel can remain from a previous migration that was
424 * using more channels, so ensure it doesn't overflow if the
425 * limit is lower now.
426 */
427 next_channel %= migrate_multifd_channels();
428 for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
429 p = &multifd_send_state->params[i];
430
431 qemu_mutex_lock(&p->mutex);
432 if (p->quit) {
433 error_report("%s: channel %d has already quit!", __func__, i);
434 qemu_mutex_unlock(&p->mutex);
435 return -1;
436 }
437 if (!p->pending_job) {
438 p->pending_job++;
439 next_channel = (i + 1) % migrate_multifd_channels();
440 break;
441 }
442 qemu_mutex_unlock(&p->mutex);
443 }
444 assert(!p->pages->used);
445 assert(!p->pages->block);
446
447 p->packet_num = multifd_send_state->packet_num++;
448 multifd_send_state->pages = p->pages;
449 p->pages = pages;
450 transferred = ((uint64_t) pages->used) * qemu_target_page_size()
451 + p->packet_len;
452 qemu_file_update_transfer(f, transferred);
453 ram_counters.multifd_bytes += transferred;
454 ram_counters.transferred += transferred;
455 qemu_mutex_unlock(&p->mutex);
456 qemu_sem_post(&p->sem);
457
458 return 1;
459 }
460
461 int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
462 {
463 MultiFDPages_t *pages = multifd_send_state->pages;
464
465 if (!pages->block) {
466 pages->block = block;
467 }
468
469 if (pages->block == block) {
470 pages->offset[pages->used] = offset;
471 pages->iov[pages->used].iov_base = block->host + offset;
472 pages->iov[pages->used].iov_len = qemu_target_page_size();
473 pages->used++;
474
475 if (pages->used < pages->allocated) {
476 return 1;
477 }
478 }
479
480 if (multifd_send_pages(f) < 0) {
481 return -1;
482 }
483
484 if (pages->block != block) {
485 return multifd_queue_page(f, block, offset);
486 }
487
488 return 1;
489 }
490
491 static void multifd_send_terminate_threads(Error *err)
492 {
493 int i;
494
495 trace_multifd_send_terminate_threads(err != NULL);
496
497 if (err) {
498 MigrationState *s = migrate_get_current();
499 migrate_set_error(s, err);
500 if (s->state == MIGRATION_STATUS_SETUP ||
501 s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
502 s->state == MIGRATION_STATUS_DEVICE ||
503 s->state == MIGRATION_STATUS_ACTIVE) {
504 migrate_set_state(&s->state, s->state,
505 MIGRATION_STATUS_FAILED);
506 }
507 }
508
509 /*
510 * We don't want to exit each threads twice. Depending on where
511 * we get the error, or if there are two independent errors in two
512 * threads at the same time, we can end calling this function
513 * twice.
514 */
515 if (qatomic_xchg(&multifd_send_state->exiting, 1)) {
516 return;
517 }
518
519 for (i = 0; i < migrate_multifd_channels(); i++) {
520 MultiFDSendParams *p = &multifd_send_state->params[i];
521
522 qemu_mutex_lock(&p->mutex);
523 p->quit = true;
524 qemu_sem_post(&p->sem);
525 qemu_mutex_unlock(&p->mutex);
526 }
527 }
528
529 void multifd_save_cleanup(void)
530 {
531 int i;
532
533 if (!migrate_use_multifd()) {
534 return;
535 }
536 multifd_send_terminate_threads(NULL);
537 for (i = 0; i < migrate_multifd_channels(); i++) {
538 MultiFDSendParams *p = &multifd_send_state->params[i];
539
540 if (p->running) {
541 qemu_thread_join(&p->thread);
542 }
543 }
544 for (i = 0; i < migrate_multifd_channels(); i++) {
545 MultiFDSendParams *p = &multifd_send_state->params[i];
546 Error *local_err = NULL;
547
548 socket_send_channel_destroy(p->c);
549 p->c = NULL;
550 qemu_mutex_destroy(&p->mutex);
551 qemu_sem_destroy(&p->sem);
552 qemu_sem_destroy(&p->sem_sync);
553 g_free(p->name);
554 p->name = NULL;
555 g_free(p->tls_hostname);
556 p->tls_hostname = NULL;
557 multifd_pages_clear(p->pages);
558 p->pages = NULL;
559 p->packet_len = 0;
560 g_free(p->packet);
561 p->packet = NULL;
562 multifd_send_state->ops->send_cleanup(p, &local_err);
563 if (local_err) {
564 migrate_set_error(migrate_get_current(), local_err);
565 error_free(local_err);
566 }
567 }
568 qemu_sem_destroy(&multifd_send_state->channels_ready);
569 g_free(multifd_send_state->params);
570 multifd_send_state->params = NULL;
571 multifd_pages_clear(multifd_send_state->pages);
572 multifd_send_state->pages = NULL;
573 g_free(multifd_send_state);
574 multifd_send_state = NULL;
575 }
576
577 void multifd_send_sync_main(QEMUFile *f)
578 {
579 int i;
580
581 if (!migrate_use_multifd()) {
582 return;
583 }
584 if (multifd_send_state->pages->used) {
585 if (multifd_send_pages(f) < 0) {
586 error_report("%s: multifd_send_pages fail", __func__);
587 return;
588 }
589 }
590 for (i = 0; i < migrate_multifd_channels(); i++) {
591 MultiFDSendParams *p = &multifd_send_state->params[i];
592
593 trace_multifd_send_sync_main_signal(p->id);
594
595 qemu_mutex_lock(&p->mutex);
596
597 if (p->quit) {
598 error_report("%s: channel %d has already quit", __func__, i);
599 qemu_mutex_unlock(&p->mutex);
600 return;
601 }
602
603 p->packet_num = multifd_send_state->packet_num++;
604 p->flags |= MULTIFD_FLAG_SYNC;
605 p->pending_job++;
606 qemu_file_update_transfer(f, p->packet_len);
607 ram_counters.multifd_bytes += p->packet_len;
608 ram_counters.transferred += p->packet_len;
609 qemu_mutex_unlock(&p->mutex);
610 qemu_sem_post(&p->sem);
611 }
612 for (i = 0; i < migrate_multifd_channels(); i++) {
613 MultiFDSendParams *p = &multifd_send_state->params[i];
614
615 trace_multifd_send_sync_main_wait(p->id);
616 qemu_sem_wait(&p->sem_sync);
617 }
618 trace_multifd_send_sync_main(multifd_send_state->packet_num);
619 }
620
621 static void *multifd_send_thread(void *opaque)
622 {
623 MultiFDSendParams *p = opaque;
624 Error *local_err = NULL;
625 int ret = 0;
626 uint32_t flags = 0;
627
628 trace_multifd_send_thread_start(p->id);
629 rcu_register_thread();
630
631 if (multifd_send_initial_packet(p, &local_err) < 0) {
632 ret = -1;
633 goto out;
634 }
635 /* initial packet */
636 p->num_packets = 1;
637
638 while (true) {
639 qemu_sem_wait(&p->sem);
640
641 if (qatomic_read(&multifd_send_state->exiting)) {
642 break;
643 }
644 qemu_mutex_lock(&p->mutex);
645
646 if (p->pending_job) {
647 uint32_t used = p->pages->used;
648 uint64_t packet_num = p->packet_num;
649 flags = p->flags;
650
651 if (used) {
652 ret = multifd_send_state->ops->send_prepare(p, used,
653 &local_err);
654 if (ret != 0) {
655 qemu_mutex_unlock(&p->mutex);
656 break;
657 }
658 }
659 multifd_send_fill_packet(p);
660 p->flags = 0;
661 p->num_packets++;
662 p->num_pages += used;
663 p->pages->used = 0;
664 p->pages->block = NULL;
665 qemu_mutex_unlock(&p->mutex);
666
667 trace_multifd_send(p->id, packet_num, used, flags,
668 p->next_packet_size);
669
670 ret = qio_channel_write_all(p->c, (void *)p->packet,
671 p->packet_len, &local_err);
672 if (ret != 0) {
673 break;
674 }
675
676 if (used) {
677 ret = multifd_send_state->ops->send_write(p, used, &local_err);
678 if (ret != 0) {
679 break;
680 }
681 }
682
683 qemu_mutex_lock(&p->mutex);
684 p->pending_job--;
685 qemu_mutex_unlock(&p->mutex);
686
687 if (flags & MULTIFD_FLAG_SYNC) {
688 qemu_sem_post(&p->sem_sync);
689 }
690 qemu_sem_post(&multifd_send_state->channels_ready);
691 } else if (p->quit) {
692 qemu_mutex_unlock(&p->mutex);
693 break;
694 } else {
695 qemu_mutex_unlock(&p->mutex);
696 /* sometimes there are spurious wakeups */
697 }
698 }
699
700 out:
701 if (local_err) {
702 trace_multifd_send_error(p->id);
703 multifd_send_terminate_threads(local_err);
704 error_free(local_err);
705 }
706
707 /*
708 * Error happen, I will exit, but I can't just leave, tell
709 * who pay attention to me.
710 */
711 if (ret != 0) {
712 qemu_sem_post(&p->sem_sync);
713 qemu_sem_post(&multifd_send_state->channels_ready);
714 }
715
716 qemu_mutex_lock(&p->mutex);
717 p->running = false;
718 qemu_mutex_unlock(&p->mutex);
719
720 rcu_unregister_thread();
721 trace_multifd_send_thread_end(p->id, p->num_packets, p->num_pages);
722
723 return NULL;
724 }
725
726 static bool multifd_channel_connect(MultiFDSendParams *p,
727 QIOChannel *ioc,
728 Error *error);
729
730 static void multifd_tls_outgoing_handshake(QIOTask *task,
731 gpointer opaque)
732 {
733 MultiFDSendParams *p = opaque;
734 QIOChannel *ioc = QIO_CHANNEL(qio_task_get_source(task));
735 Error *err = NULL;
736
737 if (qio_task_propagate_error(task, &err)) {
738 trace_multifd_tls_outgoing_handshake_error(ioc, error_get_pretty(err));
739 } else {
740 trace_multifd_tls_outgoing_handshake_complete(ioc);
741 }
742 multifd_channel_connect(p, ioc, err);
743 }
744
745 static void *multifd_tls_handshake_thread(void *opaque)
746 {
747 MultiFDSendParams *p = opaque;
748 QIOChannelTLS *tioc = QIO_CHANNEL_TLS(p->c);
749
750 qio_channel_tls_handshake(tioc,
751 multifd_tls_outgoing_handshake,
752 p,
753 NULL,
754 NULL);
755 return NULL;
756 }
757
758 static void multifd_tls_channel_connect(MultiFDSendParams *p,
759 QIOChannel *ioc,
760 Error **errp)
761 {
762 MigrationState *s = migrate_get_current();
763 const char *hostname = p->tls_hostname;
764 QIOChannelTLS *tioc;
765
766 tioc = migration_tls_client_create(s, ioc, hostname, errp);
767 if (!tioc) {
768 return;
769 }
770
771 object_unref(OBJECT(ioc));
772 trace_multifd_tls_outgoing_handshake_start(ioc, tioc, hostname);
773 qio_channel_set_name(QIO_CHANNEL(tioc), "multifd-tls-outgoing");
774 p->c = QIO_CHANNEL(tioc);
775 qemu_thread_create(&p->thread, "multifd-tls-handshake-worker",
776 multifd_tls_handshake_thread, p,
777 QEMU_THREAD_JOINABLE);
778 }
779
780 static bool multifd_channel_connect(MultiFDSendParams *p,
781 QIOChannel *ioc,
782 Error *error)
783 {
784 MigrationState *s = migrate_get_current();
785
786 trace_multifd_set_outgoing_channel(
787 ioc, object_get_typename(OBJECT(ioc)), p->tls_hostname, error);
788
789 if (!error) {
790 if (s->parameters.tls_creds &&
791 *s->parameters.tls_creds &&
792 !object_dynamic_cast(OBJECT(ioc),
793 TYPE_QIO_CHANNEL_TLS)) {
794 multifd_tls_channel_connect(p, ioc, &error);
795 if (!error) {
796 /*
797 * tls_channel_connect will call back to this
798 * function after the TLS handshake,
799 * so we mustn't call multifd_send_thread until then
800 */
801 return false;
802 } else {
803 return true;
804 }
805 } else {
806 /* update for tls qio channel */
807 p->c = ioc;
808 qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
809 QEMU_THREAD_JOINABLE);
810 }
811 return false;
812 }
813
814 return true;
815 }
816
817 static void multifd_new_send_channel_cleanup(MultiFDSendParams *p,
818 QIOChannel *ioc, Error *err)
819 {
820 migrate_set_error(migrate_get_current(), err);
821 /* Error happen, we need to tell who pay attention to me */
822 qemu_sem_post(&multifd_send_state->channels_ready);
823 qemu_sem_post(&p->sem_sync);
824 /*
825 * Although multifd_send_thread is not created, but main migration
826 * thread neet to judge whether it is running, so we need to mark
827 * its status.
828 */
829 p->quit = true;
830 object_unref(OBJECT(ioc));
831 error_free(err);
832 }
833
834 static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
835 {
836 MultiFDSendParams *p = opaque;
837 QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
838 Error *local_err = NULL;
839
840 trace_multifd_new_send_channel_async(p->id);
841 if (qio_task_propagate_error(task, &local_err)) {
842 goto cleanup;
843 } else {
844 p->c = QIO_CHANNEL(sioc);
845 qio_channel_set_delay(p->c, false);
846 p->running = true;
847 if (multifd_channel_connect(p, sioc, local_err)) {
848 goto cleanup;
849 }
850 return;
851 }
852
853 cleanup:
854 multifd_new_send_channel_cleanup(p, sioc, local_err);
855 }
856
857 int multifd_save_setup(Error **errp)
858 {
859 int thread_count;
860 uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
861 uint8_t i;
862 MigrationState *s;
863
864 if (!migrate_use_multifd()) {
865 return 0;
866 }
867 s = migrate_get_current();
868 thread_count = migrate_multifd_channels();
869 multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
870 multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
871 multifd_send_state->pages = multifd_pages_init(page_count);
872 qemu_sem_init(&multifd_send_state->channels_ready, 0);
873 qatomic_set(&multifd_send_state->exiting, 0);
874 multifd_send_state->ops = multifd_ops[migrate_multifd_compression()];
875
876 for (i = 0; i < thread_count; i++) {
877 MultiFDSendParams *p = &multifd_send_state->params[i];
878
879 qemu_mutex_init(&p->mutex);
880 qemu_sem_init(&p->sem, 0);
881 qemu_sem_init(&p->sem_sync, 0);
882 p->quit = false;
883 p->pending_job = 0;
884 p->id = i;
885 p->pages = multifd_pages_init(page_count);
886 p->packet_len = sizeof(MultiFDPacket_t)
887 + sizeof(uint64_t) * page_count;
888 p->packet = g_malloc0(p->packet_len);
889 p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
890 p->packet->version = cpu_to_be32(MULTIFD_VERSION);
891 p->name = g_strdup_printf("multifdsend_%d", i);
892 p->tls_hostname = g_strdup(s->hostname);
893 socket_send_channel_create(multifd_new_send_channel_async, p);
894 }
895
896 for (i = 0; i < thread_count; i++) {
897 MultiFDSendParams *p = &multifd_send_state->params[i];
898 Error *local_err = NULL;
899 int ret;
900
901 ret = multifd_send_state->ops->send_setup(p, &local_err);
902 if (ret) {
903 error_propagate(errp, local_err);
904 return ret;
905 }
906 }
907 return 0;
908 }
909
910 struct {
911 MultiFDRecvParams *params;
912 /* number of created threads */
913 int count;
914 /* syncs main thread and channels */
915 QemuSemaphore sem_sync;
916 /* global number of generated multifd packets */
917 uint64_t packet_num;
918 /* multifd ops */
919 MultiFDMethods *ops;
920 } *multifd_recv_state;
921
922 static void multifd_recv_terminate_threads(Error *err)
923 {
924 int i;
925
926 trace_multifd_recv_terminate_threads(err != NULL);
927
928 if (err) {
929 MigrationState *s = migrate_get_current();
930 migrate_set_error(s, err);
931 if (s->state == MIGRATION_STATUS_SETUP ||
932 s->state == MIGRATION_STATUS_ACTIVE) {
933 migrate_set_state(&s->state, s->state,
934 MIGRATION_STATUS_FAILED);
935 }
936 }
937
938 for (i = 0; i < migrate_multifd_channels(); i++) {
939 MultiFDRecvParams *p = &multifd_recv_state->params[i];
940
941 qemu_mutex_lock(&p->mutex);
942 p->quit = true;
943 /*
944 * We could arrive here for two reasons:
945 * - normal quit, i.e. everything went fine, just finished
946 * - error quit: We close the channels so the channel threads
947 * finish the qio_channel_read_all_eof()
948 */
949 if (p->c) {
950 qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
951 }
952 qemu_mutex_unlock(&p->mutex);
953 }
954 }
955
956 int multifd_load_cleanup(Error **errp)
957 {
958 int i;
959
960 if (!migrate_use_multifd()) {
961 return 0;
962 }
963 multifd_recv_terminate_threads(NULL);
964 for (i = 0; i < migrate_multifd_channels(); i++) {
965 MultiFDRecvParams *p = &multifd_recv_state->params[i];
966
967 if (p->running) {
968 p->quit = true;
969 /*
970 * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
971 * however try to wakeup it without harm in cleanup phase.
972 */
973 qemu_sem_post(&p->sem_sync);
974 qemu_thread_join(&p->thread);
975 }
976 }
977 for (i = 0; i < migrate_multifd_channels(); i++) {
978 MultiFDRecvParams *p = &multifd_recv_state->params[i];
979
980 if (object_dynamic_cast(OBJECT(p->c), TYPE_QIO_CHANNEL_SOCKET)
981 && OBJECT(p->c)->ref == 1) {
982 yank_unregister_function(MIGRATION_YANK_INSTANCE,
983 yank_generic_iochannel,
984 QIO_CHANNEL(p->c));
985 }
986
987 object_unref(OBJECT(p->c));
988 p->c = NULL;
989 qemu_mutex_destroy(&p->mutex);
990 qemu_sem_destroy(&p->sem_sync);
991 g_free(p->name);
992 p->name = NULL;
993 multifd_pages_clear(p->pages);
994 p->pages = NULL;
995 p->packet_len = 0;
996 g_free(p->packet);
997 p->packet = NULL;
998 multifd_recv_state->ops->recv_cleanup(p);
999 }
1000 qemu_sem_destroy(&multifd_recv_state->sem_sync);
1001 g_free(multifd_recv_state->params);
1002 multifd_recv_state->params = NULL;
1003 g_free(multifd_recv_state);
1004 multifd_recv_state = NULL;
1005
1006 return 0;
1007 }
1008
1009 void multifd_recv_sync_main(void)
1010 {
1011 int i;
1012
1013 if (!migrate_use_multifd()) {
1014 return;
1015 }
1016 for (i = 0; i < migrate_multifd_channels(); i++) {
1017 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1018
1019 trace_multifd_recv_sync_main_wait(p->id);
1020 qemu_sem_wait(&multifd_recv_state->sem_sync);
1021 }
1022 for (i = 0; i < migrate_multifd_channels(); i++) {
1023 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1024
1025 WITH_QEMU_LOCK_GUARD(&p->mutex) {
1026 if (multifd_recv_state->packet_num < p->packet_num) {
1027 multifd_recv_state->packet_num = p->packet_num;
1028 }
1029 }
1030 trace_multifd_recv_sync_main_signal(p->id);
1031 qemu_sem_post(&p->sem_sync);
1032 }
1033 trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
1034 }
1035
1036 static void *multifd_recv_thread(void *opaque)
1037 {
1038 MultiFDRecvParams *p = opaque;
1039 Error *local_err = NULL;
1040 int ret;
1041
1042 trace_multifd_recv_thread_start(p->id);
1043 rcu_register_thread();
1044
1045 while (true) {
1046 uint32_t used;
1047 uint32_t flags;
1048
1049 if (p->quit) {
1050 break;
1051 }
1052
1053 ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
1054 p->packet_len, &local_err);
1055 if (ret == 0) { /* EOF */
1056 break;
1057 }
1058 if (ret == -1) { /* Error */
1059 break;
1060 }
1061
1062 qemu_mutex_lock(&p->mutex);
1063 ret = multifd_recv_unfill_packet(p, &local_err);
1064 if (ret) {
1065 qemu_mutex_unlock(&p->mutex);
1066 break;
1067 }
1068
1069 used = p->pages->used;
1070 flags = p->flags;
1071 /* recv methods don't know how to handle the SYNC flag */
1072 p->flags &= ~MULTIFD_FLAG_SYNC;
1073 trace_multifd_recv(p->id, p->packet_num, used, flags,
1074 p->next_packet_size);
1075 p->num_packets++;
1076 p->num_pages += used;
1077 qemu_mutex_unlock(&p->mutex);
1078
1079 if (used) {
1080 ret = multifd_recv_state->ops->recv_pages(p, used, &local_err);
1081 if (ret != 0) {
1082 break;
1083 }
1084 }
1085
1086 if (flags & MULTIFD_FLAG_SYNC) {
1087 qemu_sem_post(&multifd_recv_state->sem_sync);
1088 qemu_sem_wait(&p->sem_sync);
1089 }
1090 }
1091
1092 if (local_err) {
1093 multifd_recv_terminate_threads(local_err);
1094 error_free(local_err);
1095 }
1096 qemu_mutex_lock(&p->mutex);
1097 p->running = false;
1098 qemu_mutex_unlock(&p->mutex);
1099
1100 rcu_unregister_thread();
1101 trace_multifd_recv_thread_end(p->id, p->num_packets, p->num_pages);
1102
1103 return NULL;
1104 }
1105
1106 int multifd_load_setup(Error **errp)
1107 {
1108 int thread_count;
1109 uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
1110 uint8_t i;
1111
1112 if (!migrate_use_multifd()) {
1113 return 0;
1114 }
1115 thread_count = migrate_multifd_channels();
1116 multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
1117 multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
1118 qatomic_set(&multifd_recv_state->count, 0);
1119 qemu_sem_init(&multifd_recv_state->sem_sync, 0);
1120 multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()];
1121
1122 for (i = 0; i < thread_count; i++) {
1123 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1124
1125 qemu_mutex_init(&p->mutex);
1126 qemu_sem_init(&p->sem_sync, 0);
1127 p->quit = false;
1128 p->id = i;
1129 p->pages = multifd_pages_init(page_count);
1130 p->packet_len = sizeof(MultiFDPacket_t)
1131 + sizeof(uint64_t) * page_count;
1132 p->packet = g_malloc0(p->packet_len);
1133 p->name = g_strdup_printf("multifdrecv_%d", i);
1134 }
1135
1136 for (i = 0; i < thread_count; i++) {
1137 MultiFDRecvParams *p = &multifd_recv_state->params[i];
1138 Error *local_err = NULL;
1139 int ret;
1140
1141 ret = multifd_recv_state->ops->recv_setup(p, &local_err);
1142 if (ret) {
1143 error_propagate(errp, local_err);
1144 return ret;
1145 }
1146 }
1147 return 0;
1148 }
1149
1150 bool multifd_recv_all_channels_created(void)
1151 {
1152 int thread_count = migrate_multifd_channels();
1153
1154 if (!migrate_use_multifd()) {
1155 return true;
1156 }
1157
1158 return thread_count == qatomic_read(&multifd_recv_state->count);
1159 }
1160
1161 /*
1162 * Try to receive all multifd channels to get ready for the migration.
1163 * - Return true and do not set @errp when correctly receiving all channels;
1164 * - Return false and do not set @errp when correctly receiving the current one;
1165 * - Return false and set @errp when failing to receive the current channel.
1166 */
1167 bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
1168 {
1169 MultiFDRecvParams *p;
1170 Error *local_err = NULL;
1171 int id;
1172
1173 id = multifd_recv_initial_packet(ioc, &local_err);
1174 if (id < 0) {
1175 multifd_recv_terminate_threads(local_err);
1176 error_propagate_prepend(errp, local_err,
1177 "failed to receive packet"
1178 " via multifd channel %d: ",
1179 qatomic_read(&multifd_recv_state->count));
1180 return false;
1181 }
1182 trace_multifd_recv_new_channel(id);
1183
1184 p = &multifd_recv_state->params[id];
1185 if (p->c != NULL) {
1186 error_setg(&local_err, "multifd: received id '%d' already setup'",
1187 id);
1188 multifd_recv_terminate_threads(local_err);
1189 error_propagate(errp, local_err);
1190 return false;
1191 }
1192 p->c = ioc;
1193 object_ref(OBJECT(ioc));
1194 /* initial packet */
1195 p->num_packets = 1;
1196
1197 p->running = true;
1198 qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
1199 QEMU_THREAD_JOINABLE);
1200 qatomic_inc(&multifd_recv_state->count);
1201 return qatomic_read(&multifd_recv_state->count) ==
1202 migrate_multifd_channels();
1203 }