xref: /openbmc/qemu/migration/multifd.c (revision 22ec0c69)
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_handshake_thread(void *opaque)
743 {
744     MultiFDSendParams *p = opaque;
745     QIOChannelTLS *tioc = QIO_CHANNEL_TLS(p->c);
746 
747     qio_channel_tls_handshake(tioc,
748                               multifd_tls_outgoing_handshake,
749                               p,
750                               NULL,
751                               NULL);
752     return NULL;
753 }
754 
755 static void multifd_tls_channel_connect(MultiFDSendParams *p,
756                                         QIOChannel *ioc,
757                                         Error **errp)
758 {
759     MigrationState *s = migrate_get_current();
760     const char *hostname = p->tls_hostname;
761     QIOChannelTLS *tioc;
762 
763     tioc = migration_tls_client_create(s, ioc, hostname, errp);
764     if (!tioc) {
765         return;
766     }
767 
768     object_unref(OBJECT(ioc));
769     trace_multifd_tls_outgoing_handshake_start(ioc, tioc, hostname);
770     qio_channel_set_name(QIO_CHANNEL(tioc), "multifd-tls-outgoing");
771     p->c = QIO_CHANNEL(tioc);
772     qemu_thread_create(&p->thread, "multifd-tls-handshake-worker",
773                        multifd_tls_handshake_thread, p,
774                        QEMU_THREAD_JOINABLE);
775 }
776 
777 static bool multifd_channel_connect(MultiFDSendParams *p,
778                                     QIOChannel *ioc,
779                                     Error *error)
780 {
781     MigrationState *s = migrate_get_current();
782 
783     trace_multifd_set_outgoing_channel(
784         ioc, object_get_typename(OBJECT(ioc)), p->tls_hostname, error);
785 
786     if (!error) {
787         if (s->parameters.tls_creds &&
788             *s->parameters.tls_creds &&
789             !object_dynamic_cast(OBJECT(ioc),
790                                  TYPE_QIO_CHANNEL_TLS)) {
791             multifd_tls_channel_connect(p, ioc, &error);
792             if (!error) {
793                 /*
794                  * tls_channel_connect will call back to this
795                  * function after the TLS handshake,
796                  * so we mustn't call multifd_send_thread until then
797                  */
798                 return false;
799             } else {
800                 return true;
801             }
802         } else {
803             /* update for tls qio channel */
804             p->c = ioc;
805             qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
806                                    QEMU_THREAD_JOINABLE);
807        }
808        return false;
809     }
810 
811     return true;
812 }
813 
814 static void multifd_new_send_channel_cleanup(MultiFDSendParams *p,
815                                              QIOChannel *ioc, Error *err)
816 {
817      migrate_set_error(migrate_get_current(), err);
818      /* Error happen, we need to tell who pay attention to me */
819      qemu_sem_post(&multifd_send_state->channels_ready);
820      qemu_sem_post(&p->sem_sync);
821      /*
822       * Although multifd_send_thread is not created, but main migration
823       * thread neet to judge whether it is running, so we need to mark
824       * its status.
825       */
826      p->quit = true;
827      object_unref(OBJECT(ioc));
828      error_free(err);
829 }
830 
831 static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
832 {
833     MultiFDSendParams *p = opaque;
834     QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
835     Error *local_err = NULL;
836 
837     trace_multifd_new_send_channel_async(p->id);
838     if (qio_task_propagate_error(task, &local_err)) {
839         goto cleanup;
840     } else {
841         p->c = QIO_CHANNEL(sioc);
842         qio_channel_set_delay(p->c, false);
843         p->running = true;
844         if (multifd_channel_connect(p, sioc, local_err)) {
845             goto cleanup;
846         }
847         return;
848     }
849 
850 cleanup:
851     multifd_new_send_channel_cleanup(p, sioc, local_err);
852 }
853 
854 int multifd_save_setup(Error **errp)
855 {
856     int thread_count;
857     uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
858     uint8_t i;
859     MigrationState *s;
860 
861     if (!migrate_use_multifd()) {
862         return 0;
863     }
864     s = migrate_get_current();
865     thread_count = migrate_multifd_channels();
866     multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
867     multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
868     multifd_send_state->pages = multifd_pages_init(page_count);
869     qemu_sem_init(&multifd_send_state->channels_ready, 0);
870     qatomic_set(&multifd_send_state->exiting, 0);
871     multifd_send_state->ops = multifd_ops[migrate_multifd_compression()];
872 
873     for (i = 0; i < thread_count; i++) {
874         MultiFDSendParams *p = &multifd_send_state->params[i];
875 
876         qemu_mutex_init(&p->mutex);
877         qemu_sem_init(&p->sem, 0);
878         qemu_sem_init(&p->sem_sync, 0);
879         p->quit = false;
880         p->pending_job = 0;
881         p->id = i;
882         p->pages = multifd_pages_init(page_count);
883         p->packet_len = sizeof(MultiFDPacket_t)
884                       + sizeof(uint64_t) * page_count;
885         p->packet = g_malloc0(p->packet_len);
886         p->packet->magic = cpu_to_be32(MULTIFD_MAGIC);
887         p->packet->version = cpu_to_be32(MULTIFD_VERSION);
888         p->name = g_strdup_printf("multifdsend_%d", i);
889         p->tls_hostname = g_strdup(s->hostname);
890         socket_send_channel_create(multifd_new_send_channel_async, p);
891     }
892 
893     for (i = 0; i < thread_count; i++) {
894         MultiFDSendParams *p = &multifd_send_state->params[i];
895         Error *local_err = NULL;
896         int ret;
897 
898         ret = multifd_send_state->ops->send_setup(p, &local_err);
899         if (ret) {
900             error_propagate(errp, local_err);
901             return ret;
902         }
903     }
904     return 0;
905 }
906 
907 struct {
908     MultiFDRecvParams *params;
909     /* number of created threads */
910     int count;
911     /* syncs main thread and channels */
912     QemuSemaphore sem_sync;
913     /* global number of generated multifd packets */
914     uint64_t packet_num;
915     /* multifd ops */
916     MultiFDMethods *ops;
917 } *multifd_recv_state;
918 
919 static void multifd_recv_terminate_threads(Error *err)
920 {
921     int i;
922 
923     trace_multifd_recv_terminate_threads(err != NULL);
924 
925     if (err) {
926         MigrationState *s = migrate_get_current();
927         migrate_set_error(s, err);
928         if (s->state == MIGRATION_STATUS_SETUP ||
929             s->state == MIGRATION_STATUS_ACTIVE) {
930             migrate_set_state(&s->state, s->state,
931                               MIGRATION_STATUS_FAILED);
932         }
933     }
934 
935     for (i = 0; i < migrate_multifd_channels(); i++) {
936         MultiFDRecvParams *p = &multifd_recv_state->params[i];
937 
938         qemu_mutex_lock(&p->mutex);
939         p->quit = true;
940         /*
941          * We could arrive here for two reasons:
942          *  - normal quit, i.e. everything went fine, just finished
943          *  - error quit: We close the channels so the channel threads
944          *    finish the qio_channel_read_all_eof()
945          */
946         if (p->c) {
947             qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
948         }
949         qemu_mutex_unlock(&p->mutex);
950     }
951 }
952 
953 int multifd_load_cleanup(Error **errp)
954 {
955     int i;
956 
957     if (!migrate_use_multifd()) {
958         return 0;
959     }
960     multifd_recv_terminate_threads(NULL);
961     for (i = 0; i < migrate_multifd_channels(); i++) {
962         MultiFDRecvParams *p = &multifd_recv_state->params[i];
963 
964         if (p->running) {
965             p->quit = true;
966             /*
967              * multifd_recv_thread may hung at MULTIFD_FLAG_SYNC handle code,
968              * however try to wakeup it without harm in cleanup phase.
969              */
970             qemu_sem_post(&p->sem_sync);
971             qemu_thread_join(&p->thread);
972         }
973     }
974     for (i = 0; i < migrate_multifd_channels(); i++) {
975         MultiFDRecvParams *p = &multifd_recv_state->params[i];
976 
977         object_unref(OBJECT(p->c));
978         p->c = NULL;
979         qemu_mutex_destroy(&p->mutex);
980         qemu_sem_destroy(&p->sem_sync);
981         g_free(p->name);
982         p->name = NULL;
983         multifd_pages_clear(p->pages);
984         p->pages = NULL;
985         p->packet_len = 0;
986         g_free(p->packet);
987         p->packet = NULL;
988         multifd_recv_state->ops->recv_cleanup(p);
989     }
990     qemu_sem_destroy(&multifd_recv_state->sem_sync);
991     g_free(multifd_recv_state->params);
992     multifd_recv_state->params = NULL;
993     g_free(multifd_recv_state);
994     multifd_recv_state = NULL;
995 
996     return 0;
997 }
998 
999 void multifd_recv_sync_main(void)
1000 {
1001     int i;
1002 
1003     if (!migrate_use_multifd()) {
1004         return;
1005     }
1006     for (i = 0; i < migrate_multifd_channels(); i++) {
1007         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1008 
1009         trace_multifd_recv_sync_main_wait(p->id);
1010         qemu_sem_wait(&multifd_recv_state->sem_sync);
1011     }
1012     for (i = 0; i < migrate_multifd_channels(); i++) {
1013         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1014 
1015         WITH_QEMU_LOCK_GUARD(&p->mutex) {
1016             if (multifd_recv_state->packet_num < p->packet_num) {
1017                 multifd_recv_state->packet_num = p->packet_num;
1018             }
1019         }
1020         trace_multifd_recv_sync_main_signal(p->id);
1021         qemu_sem_post(&p->sem_sync);
1022     }
1023     trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
1024 }
1025 
1026 static void *multifd_recv_thread(void *opaque)
1027 {
1028     MultiFDRecvParams *p = opaque;
1029     Error *local_err = NULL;
1030     int ret;
1031 
1032     trace_multifd_recv_thread_start(p->id);
1033     rcu_register_thread();
1034 
1035     while (true) {
1036         uint32_t used;
1037         uint32_t flags;
1038 
1039         if (p->quit) {
1040             break;
1041         }
1042 
1043         ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
1044                                        p->packet_len, &local_err);
1045         if (ret == 0) {   /* EOF */
1046             break;
1047         }
1048         if (ret == -1) {   /* Error */
1049             break;
1050         }
1051 
1052         qemu_mutex_lock(&p->mutex);
1053         ret = multifd_recv_unfill_packet(p, &local_err);
1054         if (ret) {
1055             qemu_mutex_unlock(&p->mutex);
1056             break;
1057         }
1058 
1059         used = p->pages->used;
1060         flags = p->flags;
1061         /* recv methods don't know how to handle the SYNC flag */
1062         p->flags &= ~MULTIFD_FLAG_SYNC;
1063         trace_multifd_recv(p->id, p->packet_num, used, flags,
1064                            p->next_packet_size);
1065         p->num_packets++;
1066         p->num_pages += used;
1067         qemu_mutex_unlock(&p->mutex);
1068 
1069         if (used) {
1070             ret = multifd_recv_state->ops->recv_pages(p, used, &local_err);
1071             if (ret != 0) {
1072                 break;
1073             }
1074         }
1075 
1076         if (flags & MULTIFD_FLAG_SYNC) {
1077             qemu_sem_post(&multifd_recv_state->sem_sync);
1078             qemu_sem_wait(&p->sem_sync);
1079         }
1080     }
1081 
1082     if (local_err) {
1083         multifd_recv_terminate_threads(local_err);
1084         error_free(local_err);
1085     }
1086     qemu_mutex_lock(&p->mutex);
1087     p->running = false;
1088     qemu_mutex_unlock(&p->mutex);
1089 
1090     rcu_unregister_thread();
1091     trace_multifd_recv_thread_end(p->id, p->num_packets, p->num_pages);
1092 
1093     return NULL;
1094 }
1095 
1096 int multifd_load_setup(Error **errp)
1097 {
1098     int thread_count;
1099     uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
1100     uint8_t i;
1101 
1102     if (!migrate_use_multifd()) {
1103         return 0;
1104     }
1105     thread_count = migrate_multifd_channels();
1106     multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
1107     multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
1108     qatomic_set(&multifd_recv_state->count, 0);
1109     qemu_sem_init(&multifd_recv_state->sem_sync, 0);
1110     multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()];
1111 
1112     for (i = 0; i < thread_count; i++) {
1113         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1114 
1115         qemu_mutex_init(&p->mutex);
1116         qemu_sem_init(&p->sem_sync, 0);
1117         p->quit = false;
1118         p->id = i;
1119         p->pages = multifd_pages_init(page_count);
1120         p->packet_len = sizeof(MultiFDPacket_t)
1121                       + sizeof(uint64_t) * page_count;
1122         p->packet = g_malloc0(p->packet_len);
1123         p->name = g_strdup_printf("multifdrecv_%d", i);
1124     }
1125 
1126     for (i = 0; i < thread_count; i++) {
1127         MultiFDRecvParams *p = &multifd_recv_state->params[i];
1128         Error *local_err = NULL;
1129         int ret;
1130 
1131         ret = multifd_recv_state->ops->recv_setup(p, &local_err);
1132         if (ret) {
1133             error_propagate(errp, local_err);
1134             return ret;
1135         }
1136     }
1137     return 0;
1138 }
1139 
1140 bool multifd_recv_all_channels_created(void)
1141 {
1142     int thread_count = migrate_multifd_channels();
1143 
1144     if (!migrate_use_multifd()) {
1145         return true;
1146     }
1147 
1148     return thread_count == qatomic_read(&multifd_recv_state->count);
1149 }
1150 
1151 /*
1152  * Try to receive all multifd channels to get ready for the migration.
1153  * - Return true and do not set @errp when correctly receiving all channels;
1154  * - Return false and do not set @errp when correctly receiving the current one;
1155  * - Return false and set @errp when failing to receive the current channel.
1156  */
1157 bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
1158 {
1159     MultiFDRecvParams *p;
1160     Error *local_err = NULL;
1161     int id;
1162 
1163     id = multifd_recv_initial_packet(ioc, &local_err);
1164     if (id < 0) {
1165         multifd_recv_terminate_threads(local_err);
1166         error_propagate_prepend(errp, local_err,
1167                                 "failed to receive packet"
1168                                 " via multifd channel %d: ",
1169                                 qatomic_read(&multifd_recv_state->count));
1170         return false;
1171     }
1172     trace_multifd_recv_new_channel(id);
1173 
1174     p = &multifd_recv_state->params[id];
1175     if (p->c != NULL) {
1176         error_setg(&local_err, "multifd: received id '%d' already setup'",
1177                    id);
1178         multifd_recv_terminate_threads(local_err);
1179         error_propagate(errp, local_err);
1180         return false;
1181     }
1182     p->c = ioc;
1183     object_ref(OBJECT(ioc));
1184     /* initial packet */
1185     p->num_packets = 1;
1186 
1187     p->running = true;
1188     qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
1189                        QEMU_THREAD_JOINABLE);
1190     qatomic_inc(&multifd_recv_state->count);
1191     return qatomic_read(&multifd_recv_state->count) ==
1192            migrate_multifd_channels();
1193 }
1194