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