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