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