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