xref: /openbmc/qemu/migration/rdma.c (revision 83a71719)
1 /*
2  * RDMA protocol and interfaces
3  *
4  * Copyright IBM, Corp. 2010-2013
5  * Copyright Red Hat, Inc. 2015-2016
6  *
7  * Authors:
8  *  Michael R. Hines <mrhines@us.ibm.com>
9  *  Jiuxing Liu <jl@us.ibm.com>
10  *  Daniel P. Berrange <berrange@redhat.com>
11  *
12  * This work is licensed under the terms of the GNU GPL, version 2 or
13  * later.  See the COPYING file in the top-level directory.
14  *
15  */
16 #include "qemu/osdep.h"
17 #include "qapi/error.h"
18 #include "qemu-common.h"
19 #include "qemu/cutils.h"
20 #include "rdma.h"
21 #include "migration.h"
22 #include "qemu-file.h"
23 #include "ram.h"
24 #include "qemu-file-channel.h"
25 #include "qemu/error-report.h"
26 #include "qemu/main-loop.h"
27 #include "qemu/sockets.h"
28 #include "qemu/bitmap.h"
29 #include "qemu/coroutine.h"
30 #include <sys/socket.h>
31 #include <netdb.h>
32 #include <arpa/inet.h>
33 #include <rdma/rdma_cma.h>
34 #include "trace.h"
35 
36 /*
37  * Print and error on both the Monitor and the Log file.
38  */
39 #define ERROR(errp, fmt, ...) \
40     do { \
41         fprintf(stderr, "RDMA ERROR: " fmt "\n", ## __VA_ARGS__); \
42         if (errp && (*(errp) == NULL)) { \
43             error_setg(errp, "RDMA ERROR: " fmt, ## __VA_ARGS__); \
44         } \
45     } while (0)
46 
47 #define RDMA_RESOLVE_TIMEOUT_MS 10000
48 
49 /* Do not merge data if larger than this. */
50 #define RDMA_MERGE_MAX (2 * 1024 * 1024)
51 #define RDMA_SIGNALED_SEND_MAX (RDMA_MERGE_MAX / 4096)
52 
53 #define RDMA_REG_CHUNK_SHIFT 20 /* 1 MB */
54 
55 /*
56  * This is only for non-live state being migrated.
57  * Instead of RDMA_WRITE messages, we use RDMA_SEND
58  * messages for that state, which requires a different
59  * delivery design than main memory.
60  */
61 #define RDMA_SEND_INCREMENT 32768
62 
63 /*
64  * Maximum size infiniband SEND message
65  */
66 #define RDMA_CONTROL_MAX_BUFFER (512 * 1024)
67 #define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096
68 
69 #define RDMA_CONTROL_VERSION_CURRENT 1
70 /*
71  * Capabilities for negotiation.
72  */
73 #define RDMA_CAPABILITY_PIN_ALL 0x01
74 
75 /*
76  * Add the other flags above to this list of known capabilities
77  * as they are introduced.
78  */
79 static uint32_t known_capabilities = RDMA_CAPABILITY_PIN_ALL;
80 
81 #define CHECK_ERROR_STATE() \
82     do { \
83         if (rdma->error_state) { \
84             if (!rdma->error_reported) { \
85                 error_report("RDMA is in an error state waiting migration" \
86                                 " to abort!"); \
87                 rdma->error_reported = 1; \
88             } \
89             rcu_read_unlock(); \
90             return rdma->error_state; \
91         } \
92     } while (0)
93 
94 /*
95  * A work request ID is 64-bits and we split up these bits
96  * into 3 parts:
97  *
98  * bits 0-15 : type of control message, 2^16
99  * bits 16-29: ram block index, 2^14
100  * bits 30-63: ram block chunk number, 2^34
101  *
102  * The last two bit ranges are only used for RDMA writes,
103  * in order to track their completion and potentially
104  * also track unregistration status of the message.
105  */
106 #define RDMA_WRID_TYPE_SHIFT  0UL
107 #define RDMA_WRID_BLOCK_SHIFT 16UL
108 #define RDMA_WRID_CHUNK_SHIFT 30UL
109 
110 #define RDMA_WRID_TYPE_MASK \
111     ((1UL << RDMA_WRID_BLOCK_SHIFT) - 1UL)
112 
113 #define RDMA_WRID_BLOCK_MASK \
114     (~RDMA_WRID_TYPE_MASK & ((1UL << RDMA_WRID_CHUNK_SHIFT) - 1UL))
115 
116 #define RDMA_WRID_CHUNK_MASK (~RDMA_WRID_BLOCK_MASK & ~RDMA_WRID_TYPE_MASK)
117 
118 /*
119  * RDMA migration protocol:
120  * 1. RDMA Writes (data messages, i.e. RAM)
121  * 2. IB Send/Recv (control channel messages)
122  */
123 enum {
124     RDMA_WRID_NONE = 0,
125     RDMA_WRID_RDMA_WRITE = 1,
126     RDMA_WRID_SEND_CONTROL = 2000,
127     RDMA_WRID_RECV_CONTROL = 4000,
128 };
129 
130 static const char *wrid_desc[] = {
131     [RDMA_WRID_NONE] = "NONE",
132     [RDMA_WRID_RDMA_WRITE] = "WRITE RDMA",
133     [RDMA_WRID_SEND_CONTROL] = "CONTROL SEND",
134     [RDMA_WRID_RECV_CONTROL] = "CONTROL RECV",
135 };
136 
137 /*
138  * Work request IDs for IB SEND messages only (not RDMA writes).
139  * This is used by the migration protocol to transmit
140  * control messages (such as device state and registration commands)
141  *
142  * We could use more WRs, but we have enough for now.
143  */
144 enum {
145     RDMA_WRID_READY = 0,
146     RDMA_WRID_DATA,
147     RDMA_WRID_CONTROL,
148     RDMA_WRID_MAX,
149 };
150 
151 /*
152  * SEND/RECV IB Control Messages.
153  */
154 enum {
155     RDMA_CONTROL_NONE = 0,
156     RDMA_CONTROL_ERROR,
157     RDMA_CONTROL_READY,               /* ready to receive */
158     RDMA_CONTROL_QEMU_FILE,           /* QEMUFile-transmitted bytes */
159     RDMA_CONTROL_RAM_BLOCKS_REQUEST,  /* RAMBlock synchronization */
160     RDMA_CONTROL_RAM_BLOCKS_RESULT,   /* RAMBlock synchronization */
161     RDMA_CONTROL_COMPRESS,            /* page contains repeat values */
162     RDMA_CONTROL_REGISTER_REQUEST,    /* dynamic page registration */
163     RDMA_CONTROL_REGISTER_RESULT,     /* key to use after registration */
164     RDMA_CONTROL_REGISTER_FINISHED,   /* current iteration finished */
165     RDMA_CONTROL_UNREGISTER_REQUEST,  /* dynamic UN-registration */
166     RDMA_CONTROL_UNREGISTER_FINISHED, /* unpinning finished */
167 };
168 
169 
170 /*
171  * Memory and MR structures used to represent an IB Send/Recv work request.
172  * This is *not* used for RDMA writes, only IB Send/Recv.
173  */
174 typedef struct {
175     uint8_t  control[RDMA_CONTROL_MAX_BUFFER]; /* actual buffer to register */
176     struct   ibv_mr *control_mr;               /* registration metadata */
177     size_t   control_len;                      /* length of the message */
178     uint8_t *control_curr;                     /* start of unconsumed bytes */
179 } RDMAWorkRequestData;
180 
181 /*
182  * Negotiate RDMA capabilities during connection-setup time.
183  */
184 typedef struct {
185     uint32_t version;
186     uint32_t flags;
187 } RDMACapabilities;
188 
189 static void caps_to_network(RDMACapabilities *cap)
190 {
191     cap->version = htonl(cap->version);
192     cap->flags = htonl(cap->flags);
193 }
194 
195 static void network_to_caps(RDMACapabilities *cap)
196 {
197     cap->version = ntohl(cap->version);
198     cap->flags = ntohl(cap->flags);
199 }
200 
201 /*
202  * Representation of a RAMBlock from an RDMA perspective.
203  * This is not transmitted, only local.
204  * This and subsequent structures cannot be linked lists
205  * because we're using a single IB message to transmit
206  * the information. It's small anyway, so a list is overkill.
207  */
208 typedef struct RDMALocalBlock {
209     char          *block_name;
210     uint8_t       *local_host_addr; /* local virtual address */
211     uint64_t       remote_host_addr; /* remote virtual address */
212     uint64_t       offset;
213     uint64_t       length;
214     struct         ibv_mr **pmr;    /* MRs for chunk-level registration */
215     struct         ibv_mr *mr;      /* MR for non-chunk-level registration */
216     uint32_t      *remote_keys;     /* rkeys for chunk-level registration */
217     uint32_t       remote_rkey;     /* rkeys for non-chunk-level registration */
218     int            index;           /* which block are we */
219     unsigned int   src_index;       /* (Only used on dest) */
220     bool           is_ram_block;
221     int            nb_chunks;
222     unsigned long *transit_bitmap;
223     unsigned long *unregister_bitmap;
224 } RDMALocalBlock;
225 
226 /*
227  * Also represents a RAMblock, but only on the dest.
228  * This gets transmitted by the dest during connection-time
229  * to the source VM and then is used to populate the
230  * corresponding RDMALocalBlock with
231  * the information needed to perform the actual RDMA.
232  */
233 typedef struct QEMU_PACKED RDMADestBlock {
234     uint64_t remote_host_addr;
235     uint64_t offset;
236     uint64_t length;
237     uint32_t remote_rkey;
238     uint32_t padding;
239 } RDMADestBlock;
240 
241 static const char *control_desc(unsigned int rdma_control)
242 {
243     static const char *strs[] = {
244         [RDMA_CONTROL_NONE] = "NONE",
245         [RDMA_CONTROL_ERROR] = "ERROR",
246         [RDMA_CONTROL_READY] = "READY",
247         [RDMA_CONTROL_QEMU_FILE] = "QEMU FILE",
248         [RDMA_CONTROL_RAM_BLOCKS_REQUEST] = "RAM BLOCKS REQUEST",
249         [RDMA_CONTROL_RAM_BLOCKS_RESULT] = "RAM BLOCKS RESULT",
250         [RDMA_CONTROL_COMPRESS] = "COMPRESS",
251         [RDMA_CONTROL_REGISTER_REQUEST] = "REGISTER REQUEST",
252         [RDMA_CONTROL_REGISTER_RESULT] = "REGISTER RESULT",
253         [RDMA_CONTROL_REGISTER_FINISHED] = "REGISTER FINISHED",
254         [RDMA_CONTROL_UNREGISTER_REQUEST] = "UNREGISTER REQUEST",
255         [RDMA_CONTROL_UNREGISTER_FINISHED] = "UNREGISTER FINISHED",
256     };
257 
258     if (rdma_control > RDMA_CONTROL_UNREGISTER_FINISHED) {
259         return "??BAD CONTROL VALUE??";
260     }
261 
262     return strs[rdma_control];
263 }
264 
265 static uint64_t htonll(uint64_t v)
266 {
267     union { uint32_t lv[2]; uint64_t llv; } u;
268     u.lv[0] = htonl(v >> 32);
269     u.lv[1] = htonl(v & 0xFFFFFFFFULL);
270     return u.llv;
271 }
272 
273 static uint64_t ntohll(uint64_t v) {
274     union { uint32_t lv[2]; uint64_t llv; } u;
275     u.llv = v;
276     return ((uint64_t)ntohl(u.lv[0]) << 32) | (uint64_t) ntohl(u.lv[1]);
277 }
278 
279 static void dest_block_to_network(RDMADestBlock *db)
280 {
281     db->remote_host_addr = htonll(db->remote_host_addr);
282     db->offset = htonll(db->offset);
283     db->length = htonll(db->length);
284     db->remote_rkey = htonl(db->remote_rkey);
285 }
286 
287 static void network_to_dest_block(RDMADestBlock *db)
288 {
289     db->remote_host_addr = ntohll(db->remote_host_addr);
290     db->offset = ntohll(db->offset);
291     db->length = ntohll(db->length);
292     db->remote_rkey = ntohl(db->remote_rkey);
293 }
294 
295 /*
296  * Virtual address of the above structures used for transmitting
297  * the RAMBlock descriptions at connection-time.
298  * This structure is *not* transmitted.
299  */
300 typedef struct RDMALocalBlocks {
301     int nb_blocks;
302     bool     init;             /* main memory init complete */
303     RDMALocalBlock *block;
304 } RDMALocalBlocks;
305 
306 /*
307  * Main data structure for RDMA state.
308  * While there is only one copy of this structure being allocated right now,
309  * this is the place where one would start if you wanted to consider
310  * having more than one RDMA connection open at the same time.
311  */
312 typedef struct RDMAContext {
313     char *host;
314     int port;
315 
316     RDMAWorkRequestData wr_data[RDMA_WRID_MAX];
317 
318     /*
319      * This is used by *_exchange_send() to figure out whether or not
320      * the initial "READY" message has already been received or not.
321      * This is because other functions may potentially poll() and detect
322      * the READY message before send() does, in which case we need to
323      * know if it completed.
324      */
325     int control_ready_expected;
326 
327     /* number of outstanding writes */
328     int nb_sent;
329 
330     /* store info about current buffer so that we can
331        merge it with future sends */
332     uint64_t current_addr;
333     uint64_t current_length;
334     /* index of ram block the current buffer belongs to */
335     int current_index;
336     /* index of the chunk in the current ram block */
337     int current_chunk;
338 
339     bool pin_all;
340 
341     /*
342      * infiniband-specific variables for opening the device
343      * and maintaining connection state and so forth.
344      *
345      * cm_id also has ibv_context, rdma_event_channel, and ibv_qp in
346      * cm_id->verbs, cm_id->channel, and cm_id->qp.
347      */
348     struct rdma_cm_id *cm_id;               /* connection manager ID */
349     struct rdma_cm_id *listen_id;
350     bool connected;
351 
352     struct ibv_context          *verbs;
353     struct rdma_event_channel   *channel;
354     struct ibv_qp *qp;                      /* queue pair */
355     struct ibv_comp_channel *comp_channel;  /* completion channel */
356     struct ibv_pd *pd;                      /* protection domain */
357     struct ibv_cq *cq;                      /* completion queue */
358 
359     /*
360      * If a previous write failed (perhaps because of a failed
361      * memory registration, then do not attempt any future work
362      * and remember the error state.
363      */
364     int error_state;
365     int error_reported;
366     int received_error;
367 
368     /*
369      * Description of ram blocks used throughout the code.
370      */
371     RDMALocalBlocks local_ram_blocks;
372     RDMADestBlock  *dest_blocks;
373 
374     /* Index of the next RAMBlock received during block registration */
375     unsigned int    next_src_index;
376 
377     /*
378      * Migration on *destination* started.
379      * Then use coroutine yield function.
380      * Source runs in a thread, so we don't care.
381      */
382     int migration_started_on_destination;
383 
384     int total_registrations;
385     int total_writes;
386 
387     int unregister_current, unregister_next;
388     uint64_t unregistrations[RDMA_SIGNALED_SEND_MAX];
389 
390     GHashTable *blockmap;
391 
392     /* the RDMAContext for return path */
393     struct RDMAContext *return_path;
394     bool is_return_path;
395 } RDMAContext;
396 
397 #define TYPE_QIO_CHANNEL_RDMA "qio-channel-rdma"
398 #define QIO_CHANNEL_RDMA(obj)                                     \
399     OBJECT_CHECK(QIOChannelRDMA, (obj), TYPE_QIO_CHANNEL_RDMA)
400 
401 typedef struct QIOChannelRDMA QIOChannelRDMA;
402 
403 
404 struct QIOChannelRDMA {
405     QIOChannel parent;
406     RDMAContext *rdmain;
407     RDMAContext *rdmaout;
408     QEMUFile *file;
409     bool blocking; /* XXX we don't actually honour this yet */
410 };
411 
412 /*
413  * Main structure for IB Send/Recv control messages.
414  * This gets prepended at the beginning of every Send/Recv.
415  */
416 typedef struct QEMU_PACKED {
417     uint32_t len;     /* Total length of data portion */
418     uint32_t type;    /* which control command to perform */
419     uint32_t repeat;  /* number of commands in data portion of same type */
420     uint32_t padding;
421 } RDMAControlHeader;
422 
423 static void control_to_network(RDMAControlHeader *control)
424 {
425     control->type = htonl(control->type);
426     control->len = htonl(control->len);
427     control->repeat = htonl(control->repeat);
428 }
429 
430 static void network_to_control(RDMAControlHeader *control)
431 {
432     control->type = ntohl(control->type);
433     control->len = ntohl(control->len);
434     control->repeat = ntohl(control->repeat);
435 }
436 
437 /*
438  * Register a single Chunk.
439  * Information sent by the source VM to inform the dest
440  * to register an single chunk of memory before we can perform
441  * the actual RDMA operation.
442  */
443 typedef struct QEMU_PACKED {
444     union QEMU_PACKED {
445         uint64_t current_addr;  /* offset into the ram_addr_t space */
446         uint64_t chunk;         /* chunk to lookup if unregistering */
447     } key;
448     uint32_t current_index; /* which ramblock the chunk belongs to */
449     uint32_t padding;
450     uint64_t chunks;            /* how many sequential chunks to register */
451 } RDMARegister;
452 
453 static void register_to_network(RDMAContext *rdma, RDMARegister *reg)
454 {
455     RDMALocalBlock *local_block;
456     local_block  = &rdma->local_ram_blocks.block[reg->current_index];
457 
458     if (local_block->is_ram_block) {
459         /*
460          * current_addr as passed in is an address in the local ram_addr_t
461          * space, we need to translate this for the destination
462          */
463         reg->key.current_addr -= local_block->offset;
464         reg->key.current_addr += rdma->dest_blocks[reg->current_index].offset;
465     }
466     reg->key.current_addr = htonll(reg->key.current_addr);
467     reg->current_index = htonl(reg->current_index);
468     reg->chunks = htonll(reg->chunks);
469 }
470 
471 static void network_to_register(RDMARegister *reg)
472 {
473     reg->key.current_addr = ntohll(reg->key.current_addr);
474     reg->current_index = ntohl(reg->current_index);
475     reg->chunks = ntohll(reg->chunks);
476 }
477 
478 typedef struct QEMU_PACKED {
479     uint32_t value;     /* if zero, we will madvise() */
480     uint32_t block_idx; /* which ram block index */
481     uint64_t offset;    /* Address in remote ram_addr_t space */
482     uint64_t length;    /* length of the chunk */
483 } RDMACompress;
484 
485 static void compress_to_network(RDMAContext *rdma, RDMACompress *comp)
486 {
487     comp->value = htonl(comp->value);
488     /*
489      * comp->offset as passed in is an address in the local ram_addr_t
490      * space, we need to translate this for the destination
491      */
492     comp->offset -= rdma->local_ram_blocks.block[comp->block_idx].offset;
493     comp->offset += rdma->dest_blocks[comp->block_idx].offset;
494     comp->block_idx = htonl(comp->block_idx);
495     comp->offset = htonll(comp->offset);
496     comp->length = htonll(comp->length);
497 }
498 
499 static void network_to_compress(RDMACompress *comp)
500 {
501     comp->value = ntohl(comp->value);
502     comp->block_idx = ntohl(comp->block_idx);
503     comp->offset = ntohll(comp->offset);
504     comp->length = ntohll(comp->length);
505 }
506 
507 /*
508  * The result of the dest's memory registration produces an "rkey"
509  * which the source VM must reference in order to perform
510  * the RDMA operation.
511  */
512 typedef struct QEMU_PACKED {
513     uint32_t rkey;
514     uint32_t padding;
515     uint64_t host_addr;
516 } RDMARegisterResult;
517 
518 static void result_to_network(RDMARegisterResult *result)
519 {
520     result->rkey = htonl(result->rkey);
521     result->host_addr = htonll(result->host_addr);
522 };
523 
524 static void network_to_result(RDMARegisterResult *result)
525 {
526     result->rkey = ntohl(result->rkey);
527     result->host_addr = ntohll(result->host_addr);
528 };
529 
530 const char *print_wrid(int wrid);
531 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
532                                    uint8_t *data, RDMAControlHeader *resp,
533                                    int *resp_idx,
534                                    int (*callback)(RDMAContext *rdma));
535 
536 static inline uint64_t ram_chunk_index(const uint8_t *start,
537                                        const uint8_t *host)
538 {
539     return ((uintptr_t) host - (uintptr_t) start) >> RDMA_REG_CHUNK_SHIFT;
540 }
541 
542 static inline uint8_t *ram_chunk_start(const RDMALocalBlock *rdma_ram_block,
543                                        uint64_t i)
544 {
545     return (uint8_t *)(uintptr_t)(rdma_ram_block->local_host_addr +
546                                   (i << RDMA_REG_CHUNK_SHIFT));
547 }
548 
549 static inline uint8_t *ram_chunk_end(const RDMALocalBlock *rdma_ram_block,
550                                      uint64_t i)
551 {
552     uint8_t *result = ram_chunk_start(rdma_ram_block, i) +
553                                          (1UL << RDMA_REG_CHUNK_SHIFT);
554 
555     if (result > (rdma_ram_block->local_host_addr + rdma_ram_block->length)) {
556         result = rdma_ram_block->local_host_addr + rdma_ram_block->length;
557     }
558 
559     return result;
560 }
561 
562 static int rdma_add_block(RDMAContext *rdma, const char *block_name,
563                          void *host_addr,
564                          ram_addr_t block_offset, uint64_t length)
565 {
566     RDMALocalBlocks *local = &rdma->local_ram_blocks;
567     RDMALocalBlock *block;
568     RDMALocalBlock *old = local->block;
569 
570     local->block = g_new0(RDMALocalBlock, local->nb_blocks + 1);
571 
572     if (local->nb_blocks) {
573         int x;
574 
575         if (rdma->blockmap) {
576             for (x = 0; x < local->nb_blocks; x++) {
577                 g_hash_table_remove(rdma->blockmap,
578                                     (void *)(uintptr_t)old[x].offset);
579                 g_hash_table_insert(rdma->blockmap,
580                                     (void *)(uintptr_t)old[x].offset,
581                                     &local->block[x]);
582             }
583         }
584         memcpy(local->block, old, sizeof(RDMALocalBlock) * local->nb_blocks);
585         g_free(old);
586     }
587 
588     block = &local->block[local->nb_blocks];
589 
590     block->block_name = g_strdup(block_name);
591     block->local_host_addr = host_addr;
592     block->offset = block_offset;
593     block->length = length;
594     block->index = local->nb_blocks;
595     block->src_index = ~0U; /* Filled in by the receipt of the block list */
596     block->nb_chunks = ram_chunk_index(host_addr, host_addr + length) + 1UL;
597     block->transit_bitmap = bitmap_new(block->nb_chunks);
598     bitmap_clear(block->transit_bitmap, 0, block->nb_chunks);
599     block->unregister_bitmap = bitmap_new(block->nb_chunks);
600     bitmap_clear(block->unregister_bitmap, 0, block->nb_chunks);
601     block->remote_keys = g_new0(uint32_t, block->nb_chunks);
602 
603     block->is_ram_block = local->init ? false : true;
604 
605     if (rdma->blockmap) {
606         g_hash_table_insert(rdma->blockmap, (void *)(uintptr_t)block_offset, block);
607     }
608 
609     trace_rdma_add_block(block_name, local->nb_blocks,
610                          (uintptr_t) block->local_host_addr,
611                          block->offset, block->length,
612                          (uintptr_t) (block->local_host_addr + block->length),
613                          BITS_TO_LONGS(block->nb_chunks) *
614                              sizeof(unsigned long) * 8,
615                          block->nb_chunks);
616 
617     local->nb_blocks++;
618 
619     return 0;
620 }
621 
622 /*
623  * Memory regions need to be registered with the device and queue pairs setup
624  * in advanced before the migration starts. This tells us where the RAM blocks
625  * are so that we can register them individually.
626  */
627 static int qemu_rdma_init_one_block(const char *block_name, void *host_addr,
628     ram_addr_t block_offset, ram_addr_t length, void *opaque)
629 {
630     return rdma_add_block(opaque, block_name, host_addr, block_offset, length);
631 }
632 
633 /*
634  * Identify the RAMBlocks and their quantity. They will be references to
635  * identify chunk boundaries inside each RAMBlock and also be referenced
636  * during dynamic page registration.
637  */
638 static int qemu_rdma_init_ram_blocks(RDMAContext *rdma)
639 {
640     RDMALocalBlocks *local = &rdma->local_ram_blocks;
641 
642     assert(rdma->blockmap == NULL);
643     memset(local, 0, sizeof *local);
644     qemu_ram_foreach_migratable_block(qemu_rdma_init_one_block, rdma);
645     trace_qemu_rdma_init_ram_blocks(local->nb_blocks);
646     rdma->dest_blocks = g_new0(RDMADestBlock,
647                                rdma->local_ram_blocks.nb_blocks);
648     local->init = true;
649     return 0;
650 }
651 
652 /*
653  * Note: If used outside of cleanup, the caller must ensure that the destination
654  * block structures are also updated
655  */
656 static int rdma_delete_block(RDMAContext *rdma, RDMALocalBlock *block)
657 {
658     RDMALocalBlocks *local = &rdma->local_ram_blocks;
659     RDMALocalBlock *old = local->block;
660     int x;
661 
662     if (rdma->blockmap) {
663         g_hash_table_remove(rdma->blockmap, (void *)(uintptr_t)block->offset);
664     }
665     if (block->pmr) {
666         int j;
667 
668         for (j = 0; j < block->nb_chunks; j++) {
669             if (!block->pmr[j]) {
670                 continue;
671             }
672             ibv_dereg_mr(block->pmr[j]);
673             rdma->total_registrations--;
674         }
675         g_free(block->pmr);
676         block->pmr = NULL;
677     }
678 
679     if (block->mr) {
680         ibv_dereg_mr(block->mr);
681         rdma->total_registrations--;
682         block->mr = NULL;
683     }
684 
685     g_free(block->transit_bitmap);
686     block->transit_bitmap = NULL;
687 
688     g_free(block->unregister_bitmap);
689     block->unregister_bitmap = NULL;
690 
691     g_free(block->remote_keys);
692     block->remote_keys = NULL;
693 
694     g_free(block->block_name);
695     block->block_name = NULL;
696 
697     if (rdma->blockmap) {
698         for (x = 0; x < local->nb_blocks; x++) {
699             g_hash_table_remove(rdma->blockmap,
700                                 (void *)(uintptr_t)old[x].offset);
701         }
702     }
703 
704     if (local->nb_blocks > 1) {
705 
706         local->block = g_new0(RDMALocalBlock, local->nb_blocks - 1);
707 
708         if (block->index) {
709             memcpy(local->block, old, sizeof(RDMALocalBlock) * block->index);
710         }
711 
712         if (block->index < (local->nb_blocks - 1)) {
713             memcpy(local->block + block->index, old + (block->index + 1),
714                 sizeof(RDMALocalBlock) *
715                     (local->nb_blocks - (block->index + 1)));
716             for (x = block->index; x < local->nb_blocks - 1; x++) {
717                 local->block[x].index--;
718             }
719         }
720     } else {
721         assert(block == local->block);
722         local->block = NULL;
723     }
724 
725     trace_rdma_delete_block(block, (uintptr_t)block->local_host_addr,
726                            block->offset, block->length,
727                             (uintptr_t)(block->local_host_addr + block->length),
728                            BITS_TO_LONGS(block->nb_chunks) *
729                                sizeof(unsigned long) * 8, block->nb_chunks);
730 
731     g_free(old);
732 
733     local->nb_blocks--;
734 
735     if (local->nb_blocks && rdma->blockmap) {
736         for (x = 0; x < local->nb_blocks; x++) {
737             g_hash_table_insert(rdma->blockmap,
738                                 (void *)(uintptr_t)local->block[x].offset,
739                                 &local->block[x]);
740         }
741     }
742 
743     return 0;
744 }
745 
746 /*
747  * Put in the log file which RDMA device was opened and the details
748  * associated with that device.
749  */
750 static void qemu_rdma_dump_id(const char *who, struct ibv_context *verbs)
751 {
752     struct ibv_port_attr port;
753 
754     if (ibv_query_port(verbs, 1, &port)) {
755         error_report("Failed to query port information");
756         return;
757     }
758 
759     printf("%s RDMA Device opened: kernel name %s "
760            "uverbs device name %s, "
761            "infiniband_verbs class device path %s, "
762            "infiniband class device path %s, "
763            "transport: (%d) %s\n",
764                 who,
765                 verbs->device->name,
766                 verbs->device->dev_name,
767                 verbs->device->dev_path,
768                 verbs->device->ibdev_path,
769                 port.link_layer,
770                 (port.link_layer == IBV_LINK_LAYER_INFINIBAND) ? "Infiniband" :
771                  ((port.link_layer == IBV_LINK_LAYER_ETHERNET)
772                     ? "Ethernet" : "Unknown"));
773 }
774 
775 /*
776  * Put in the log file the RDMA gid addressing information,
777  * useful for folks who have trouble understanding the
778  * RDMA device hierarchy in the kernel.
779  */
780 static void qemu_rdma_dump_gid(const char *who, struct rdma_cm_id *id)
781 {
782     char sgid[33];
783     char dgid[33];
784     inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.sgid, sgid, sizeof sgid);
785     inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.dgid, dgid, sizeof dgid);
786     trace_qemu_rdma_dump_gid(who, sgid, dgid);
787 }
788 
789 /*
790  * As of now, IPv6 over RoCE / iWARP is not supported by linux.
791  * We will try the next addrinfo struct, and fail if there are
792  * no other valid addresses to bind against.
793  *
794  * If user is listening on '[::]', then we will not have a opened a device
795  * yet and have no way of verifying if the device is RoCE or not.
796  *
797  * In this case, the source VM will throw an error for ALL types of
798  * connections (both IPv4 and IPv6) if the destination machine does not have
799  * a regular infiniband network available for use.
800  *
801  * The only way to guarantee that an error is thrown for broken kernels is
802  * for the management software to choose a *specific* interface at bind time
803  * and validate what time of hardware it is.
804  *
805  * Unfortunately, this puts the user in a fix:
806  *
807  *  If the source VM connects with an IPv4 address without knowing that the
808  *  destination has bound to '[::]' the migration will unconditionally fail
809  *  unless the management software is explicitly listening on the IPv4
810  *  address while using a RoCE-based device.
811  *
812  *  If the source VM connects with an IPv6 address, then we're OK because we can
813  *  throw an error on the source (and similarly on the destination).
814  *
815  *  But in mixed environments, this will be broken for a while until it is fixed
816  *  inside linux.
817  *
818  * We do provide a *tiny* bit of help in this function: We can list all of the
819  * devices in the system and check to see if all the devices are RoCE or
820  * Infiniband.
821  *
822  * If we detect that we have a *pure* RoCE environment, then we can safely
823  * thrown an error even if the management software has specified '[::]' as the
824  * bind address.
825  *
826  * However, if there is are multiple hetergeneous devices, then we cannot make
827  * this assumption and the user just has to be sure they know what they are
828  * doing.
829  *
830  * Patches are being reviewed on linux-rdma.
831  */
832 static int qemu_rdma_broken_ipv6_kernel(struct ibv_context *verbs, Error **errp)
833 {
834     struct ibv_port_attr port_attr;
835 
836     /* This bug only exists in linux, to our knowledge. */
837 #ifdef CONFIG_LINUX
838 
839     /*
840      * Verbs are only NULL if management has bound to '[::]'.
841      *
842      * Let's iterate through all the devices and see if there any pure IB
843      * devices (non-ethernet).
844      *
845      * If not, then we can safely proceed with the migration.
846      * Otherwise, there are no guarantees until the bug is fixed in linux.
847      */
848     if (!verbs) {
849         int num_devices, x;
850         struct ibv_device ** dev_list = ibv_get_device_list(&num_devices);
851         bool roce_found = false;
852         bool ib_found = false;
853 
854         for (x = 0; x < num_devices; x++) {
855             verbs = ibv_open_device(dev_list[x]);
856             if (!verbs) {
857                 if (errno == EPERM) {
858                     continue;
859                 } else {
860                     return -EINVAL;
861                 }
862             }
863 
864             if (ibv_query_port(verbs, 1, &port_attr)) {
865                 ibv_close_device(verbs);
866                 ERROR(errp, "Could not query initial IB port");
867                 return -EINVAL;
868             }
869 
870             if (port_attr.link_layer == IBV_LINK_LAYER_INFINIBAND) {
871                 ib_found = true;
872             } else if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
873                 roce_found = true;
874             }
875 
876             ibv_close_device(verbs);
877 
878         }
879 
880         if (roce_found) {
881             if (ib_found) {
882                 fprintf(stderr, "WARN: migrations may fail:"
883                                 " IPv6 over RoCE / iWARP in linux"
884                                 " is broken. But since you appear to have a"
885                                 " mixed RoCE / IB environment, be sure to only"
886                                 " migrate over the IB fabric until the kernel "
887                                 " fixes the bug.\n");
888             } else {
889                 ERROR(errp, "You only have RoCE / iWARP devices in your systems"
890                             " and your management software has specified '[::]'"
891                             ", but IPv6 over RoCE / iWARP is not supported in Linux.");
892                 return -ENONET;
893             }
894         }
895 
896         return 0;
897     }
898 
899     /*
900      * If we have a verbs context, that means that some other than '[::]' was
901      * used by the management software for binding. In which case we can
902      * actually warn the user about a potentially broken kernel.
903      */
904 
905     /* IB ports start with 1, not 0 */
906     if (ibv_query_port(verbs, 1, &port_attr)) {
907         ERROR(errp, "Could not query initial IB port");
908         return -EINVAL;
909     }
910 
911     if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
912         ERROR(errp, "Linux kernel's RoCE / iWARP does not support IPv6 "
913                     "(but patches on linux-rdma in progress)");
914         return -ENONET;
915     }
916 
917 #endif
918 
919     return 0;
920 }
921 
922 /*
923  * Figure out which RDMA device corresponds to the requested IP hostname
924  * Also create the initial connection manager identifiers for opening
925  * the connection.
926  */
927 static int qemu_rdma_resolve_host(RDMAContext *rdma, Error **errp)
928 {
929     int ret;
930     struct rdma_addrinfo *res;
931     char port_str[16];
932     struct rdma_cm_event *cm_event;
933     char ip[40] = "unknown";
934     struct rdma_addrinfo *e;
935 
936     if (rdma->host == NULL || !strcmp(rdma->host, "")) {
937         ERROR(errp, "RDMA hostname has not been set");
938         return -EINVAL;
939     }
940 
941     /* create CM channel */
942     rdma->channel = rdma_create_event_channel();
943     if (!rdma->channel) {
944         ERROR(errp, "could not create CM channel");
945         return -EINVAL;
946     }
947 
948     /* create CM id */
949     ret = rdma_create_id(rdma->channel, &rdma->cm_id, NULL, RDMA_PS_TCP);
950     if (ret) {
951         ERROR(errp, "could not create channel id");
952         goto err_resolve_create_id;
953     }
954 
955     snprintf(port_str, 16, "%d", rdma->port);
956     port_str[15] = '\0';
957 
958     ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
959     if (ret < 0) {
960         ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
961         goto err_resolve_get_addr;
962     }
963 
964     for (e = res; e != NULL; e = e->ai_next) {
965         inet_ntop(e->ai_family,
966             &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
967         trace_qemu_rdma_resolve_host_trying(rdma->host, ip);
968 
969         ret = rdma_resolve_addr(rdma->cm_id, NULL, e->ai_dst_addr,
970                 RDMA_RESOLVE_TIMEOUT_MS);
971         if (!ret) {
972             if (e->ai_family == AF_INET6) {
973                 ret = qemu_rdma_broken_ipv6_kernel(rdma->cm_id->verbs, errp);
974                 if (ret) {
975                     continue;
976                 }
977             }
978             goto route;
979         }
980     }
981 
982     ERROR(errp, "could not resolve address %s", rdma->host);
983     goto err_resolve_get_addr;
984 
985 route:
986     qemu_rdma_dump_gid("source_resolve_addr", rdma->cm_id);
987 
988     ret = rdma_get_cm_event(rdma->channel, &cm_event);
989     if (ret) {
990         ERROR(errp, "could not perform event_addr_resolved");
991         goto err_resolve_get_addr;
992     }
993 
994     if (cm_event->event != RDMA_CM_EVENT_ADDR_RESOLVED) {
995         ERROR(errp, "result not equal to event_addr_resolved %s",
996                 rdma_event_str(cm_event->event));
997         perror("rdma_resolve_addr");
998         rdma_ack_cm_event(cm_event);
999         ret = -EINVAL;
1000         goto err_resolve_get_addr;
1001     }
1002     rdma_ack_cm_event(cm_event);
1003 
1004     /* resolve route */
1005     ret = rdma_resolve_route(rdma->cm_id, RDMA_RESOLVE_TIMEOUT_MS);
1006     if (ret) {
1007         ERROR(errp, "could not resolve rdma route");
1008         goto err_resolve_get_addr;
1009     }
1010 
1011     ret = rdma_get_cm_event(rdma->channel, &cm_event);
1012     if (ret) {
1013         ERROR(errp, "could not perform event_route_resolved");
1014         goto err_resolve_get_addr;
1015     }
1016     if (cm_event->event != RDMA_CM_EVENT_ROUTE_RESOLVED) {
1017         ERROR(errp, "result not equal to event_route_resolved: %s",
1018                         rdma_event_str(cm_event->event));
1019         rdma_ack_cm_event(cm_event);
1020         ret = -EINVAL;
1021         goto err_resolve_get_addr;
1022     }
1023     rdma_ack_cm_event(cm_event);
1024     rdma->verbs = rdma->cm_id->verbs;
1025     qemu_rdma_dump_id("source_resolve_host", rdma->cm_id->verbs);
1026     qemu_rdma_dump_gid("source_resolve_host", rdma->cm_id);
1027     return 0;
1028 
1029 err_resolve_get_addr:
1030     rdma_destroy_id(rdma->cm_id);
1031     rdma->cm_id = NULL;
1032 err_resolve_create_id:
1033     rdma_destroy_event_channel(rdma->channel);
1034     rdma->channel = NULL;
1035     return ret;
1036 }
1037 
1038 /*
1039  * Create protection domain and completion queues
1040  */
1041 static int qemu_rdma_alloc_pd_cq(RDMAContext *rdma)
1042 {
1043     /* allocate pd */
1044     rdma->pd = ibv_alloc_pd(rdma->verbs);
1045     if (!rdma->pd) {
1046         error_report("failed to allocate protection domain");
1047         return -1;
1048     }
1049 
1050     /* create completion channel */
1051     rdma->comp_channel = ibv_create_comp_channel(rdma->verbs);
1052     if (!rdma->comp_channel) {
1053         error_report("failed to allocate completion channel");
1054         goto err_alloc_pd_cq;
1055     }
1056 
1057     /*
1058      * Completion queue can be filled by both read and write work requests,
1059      * so must reflect the sum of both possible queue sizes.
1060      */
1061     rdma->cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3),
1062             NULL, rdma->comp_channel, 0);
1063     if (!rdma->cq) {
1064         error_report("failed to allocate completion queue");
1065         goto err_alloc_pd_cq;
1066     }
1067 
1068     return 0;
1069 
1070 err_alloc_pd_cq:
1071     if (rdma->pd) {
1072         ibv_dealloc_pd(rdma->pd);
1073     }
1074     if (rdma->comp_channel) {
1075         ibv_destroy_comp_channel(rdma->comp_channel);
1076     }
1077     rdma->pd = NULL;
1078     rdma->comp_channel = NULL;
1079     return -1;
1080 
1081 }
1082 
1083 /*
1084  * Create queue pairs.
1085  */
1086 static int qemu_rdma_alloc_qp(RDMAContext *rdma)
1087 {
1088     struct ibv_qp_init_attr attr = { 0 };
1089     int ret;
1090 
1091     attr.cap.max_send_wr = RDMA_SIGNALED_SEND_MAX;
1092     attr.cap.max_recv_wr = 3;
1093     attr.cap.max_send_sge = 1;
1094     attr.cap.max_recv_sge = 1;
1095     attr.send_cq = rdma->cq;
1096     attr.recv_cq = rdma->cq;
1097     attr.qp_type = IBV_QPT_RC;
1098 
1099     ret = rdma_create_qp(rdma->cm_id, rdma->pd, &attr);
1100     if (ret) {
1101         return -1;
1102     }
1103 
1104     rdma->qp = rdma->cm_id->qp;
1105     return 0;
1106 }
1107 
1108 static int qemu_rdma_reg_whole_ram_blocks(RDMAContext *rdma)
1109 {
1110     int i;
1111     RDMALocalBlocks *local = &rdma->local_ram_blocks;
1112 
1113     for (i = 0; i < local->nb_blocks; i++) {
1114         local->block[i].mr =
1115             ibv_reg_mr(rdma->pd,
1116                     local->block[i].local_host_addr,
1117                     local->block[i].length,
1118                     IBV_ACCESS_LOCAL_WRITE |
1119                     IBV_ACCESS_REMOTE_WRITE
1120                     );
1121         if (!local->block[i].mr) {
1122             perror("Failed to register local dest ram block!\n");
1123             break;
1124         }
1125         rdma->total_registrations++;
1126     }
1127 
1128     if (i >= local->nb_blocks) {
1129         return 0;
1130     }
1131 
1132     for (i--; i >= 0; i--) {
1133         ibv_dereg_mr(local->block[i].mr);
1134         rdma->total_registrations--;
1135     }
1136 
1137     return -1;
1138 
1139 }
1140 
1141 /*
1142  * Find the ram block that corresponds to the page requested to be
1143  * transmitted by QEMU.
1144  *
1145  * Once the block is found, also identify which 'chunk' within that
1146  * block that the page belongs to.
1147  *
1148  * This search cannot fail or the migration will fail.
1149  */
1150 static int qemu_rdma_search_ram_block(RDMAContext *rdma,
1151                                       uintptr_t block_offset,
1152                                       uint64_t offset,
1153                                       uint64_t length,
1154                                       uint64_t *block_index,
1155                                       uint64_t *chunk_index)
1156 {
1157     uint64_t current_addr = block_offset + offset;
1158     RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
1159                                                 (void *) block_offset);
1160     assert(block);
1161     assert(current_addr >= block->offset);
1162     assert((current_addr + length) <= (block->offset + block->length));
1163 
1164     *block_index = block->index;
1165     *chunk_index = ram_chunk_index(block->local_host_addr,
1166                 block->local_host_addr + (current_addr - block->offset));
1167 
1168     return 0;
1169 }
1170 
1171 /*
1172  * Register a chunk with IB. If the chunk was already registered
1173  * previously, then skip.
1174  *
1175  * Also return the keys associated with the registration needed
1176  * to perform the actual RDMA operation.
1177  */
1178 static int qemu_rdma_register_and_get_keys(RDMAContext *rdma,
1179         RDMALocalBlock *block, uintptr_t host_addr,
1180         uint32_t *lkey, uint32_t *rkey, int chunk,
1181         uint8_t *chunk_start, uint8_t *chunk_end)
1182 {
1183     if (block->mr) {
1184         if (lkey) {
1185             *lkey = block->mr->lkey;
1186         }
1187         if (rkey) {
1188             *rkey = block->mr->rkey;
1189         }
1190         return 0;
1191     }
1192 
1193     /* allocate memory to store chunk MRs */
1194     if (!block->pmr) {
1195         block->pmr = g_new0(struct ibv_mr *, block->nb_chunks);
1196     }
1197 
1198     /*
1199      * If 'rkey', then we're the destination, so grant access to the source.
1200      *
1201      * If 'lkey', then we're the source VM, so grant access only to ourselves.
1202      */
1203     if (!block->pmr[chunk]) {
1204         uint64_t len = chunk_end - chunk_start;
1205 
1206         trace_qemu_rdma_register_and_get_keys(len, chunk_start);
1207 
1208         block->pmr[chunk] = ibv_reg_mr(rdma->pd,
1209                 chunk_start, len,
1210                 (rkey ? (IBV_ACCESS_LOCAL_WRITE |
1211                         IBV_ACCESS_REMOTE_WRITE) : 0));
1212 
1213         if (!block->pmr[chunk]) {
1214             perror("Failed to register chunk!");
1215             fprintf(stderr, "Chunk details: block: %d chunk index %d"
1216                             " start %" PRIuPTR " end %" PRIuPTR
1217                             " host %" PRIuPTR
1218                             " local %" PRIuPTR " registrations: %d\n",
1219                             block->index, chunk, (uintptr_t)chunk_start,
1220                             (uintptr_t)chunk_end, host_addr,
1221                             (uintptr_t)block->local_host_addr,
1222                             rdma->total_registrations);
1223             return -1;
1224         }
1225         rdma->total_registrations++;
1226     }
1227 
1228     if (lkey) {
1229         *lkey = block->pmr[chunk]->lkey;
1230     }
1231     if (rkey) {
1232         *rkey = block->pmr[chunk]->rkey;
1233     }
1234     return 0;
1235 }
1236 
1237 /*
1238  * Register (at connection time) the memory used for control
1239  * channel messages.
1240  */
1241 static int qemu_rdma_reg_control(RDMAContext *rdma, int idx)
1242 {
1243     rdma->wr_data[idx].control_mr = ibv_reg_mr(rdma->pd,
1244             rdma->wr_data[idx].control, RDMA_CONTROL_MAX_BUFFER,
1245             IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
1246     if (rdma->wr_data[idx].control_mr) {
1247         rdma->total_registrations++;
1248         return 0;
1249     }
1250     error_report("qemu_rdma_reg_control failed");
1251     return -1;
1252 }
1253 
1254 const char *print_wrid(int wrid)
1255 {
1256     if (wrid >= RDMA_WRID_RECV_CONTROL) {
1257         return wrid_desc[RDMA_WRID_RECV_CONTROL];
1258     }
1259     return wrid_desc[wrid];
1260 }
1261 
1262 /*
1263  * RDMA requires memory registration (mlock/pinning), but this is not good for
1264  * overcommitment.
1265  *
1266  * In preparation for the future where LRU information or workload-specific
1267  * writable writable working set memory access behavior is available to QEMU
1268  * it would be nice to have in place the ability to UN-register/UN-pin
1269  * particular memory regions from the RDMA hardware when it is determine that
1270  * those regions of memory will likely not be accessed again in the near future.
1271  *
1272  * While we do not yet have such information right now, the following
1273  * compile-time option allows us to perform a non-optimized version of this
1274  * behavior.
1275  *
1276  * By uncommenting this option, you will cause *all* RDMA transfers to be
1277  * unregistered immediately after the transfer completes on both sides of the
1278  * connection. This has no effect in 'rdma-pin-all' mode, only regular mode.
1279  *
1280  * This will have a terrible impact on migration performance, so until future
1281  * workload information or LRU information is available, do not attempt to use
1282  * this feature except for basic testing.
1283  */
1284 //#define RDMA_UNREGISTRATION_EXAMPLE
1285 
1286 /*
1287  * Perform a non-optimized memory unregistration after every transfer
1288  * for demonstration purposes, only if pin-all is not requested.
1289  *
1290  * Potential optimizations:
1291  * 1. Start a new thread to run this function continuously
1292         - for bit clearing
1293         - and for receipt of unregister messages
1294  * 2. Use an LRU.
1295  * 3. Use workload hints.
1296  */
1297 static int qemu_rdma_unregister_waiting(RDMAContext *rdma)
1298 {
1299     while (rdma->unregistrations[rdma->unregister_current]) {
1300         int ret;
1301         uint64_t wr_id = rdma->unregistrations[rdma->unregister_current];
1302         uint64_t chunk =
1303             (wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1304         uint64_t index =
1305             (wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1306         RDMALocalBlock *block =
1307             &(rdma->local_ram_blocks.block[index]);
1308         RDMARegister reg = { .current_index = index };
1309         RDMAControlHeader resp = { .type = RDMA_CONTROL_UNREGISTER_FINISHED,
1310                                  };
1311         RDMAControlHeader head = { .len = sizeof(RDMARegister),
1312                                    .type = RDMA_CONTROL_UNREGISTER_REQUEST,
1313                                    .repeat = 1,
1314                                  };
1315 
1316         trace_qemu_rdma_unregister_waiting_proc(chunk,
1317                                                 rdma->unregister_current);
1318 
1319         rdma->unregistrations[rdma->unregister_current] = 0;
1320         rdma->unregister_current++;
1321 
1322         if (rdma->unregister_current == RDMA_SIGNALED_SEND_MAX) {
1323             rdma->unregister_current = 0;
1324         }
1325 
1326 
1327         /*
1328          * Unregistration is speculative (because migration is single-threaded
1329          * and we cannot break the protocol's inifinband message ordering).
1330          * Thus, if the memory is currently being used for transmission,
1331          * then abort the attempt to unregister and try again
1332          * later the next time a completion is received for this memory.
1333          */
1334         clear_bit(chunk, block->unregister_bitmap);
1335 
1336         if (test_bit(chunk, block->transit_bitmap)) {
1337             trace_qemu_rdma_unregister_waiting_inflight(chunk);
1338             continue;
1339         }
1340 
1341         trace_qemu_rdma_unregister_waiting_send(chunk);
1342 
1343         ret = ibv_dereg_mr(block->pmr[chunk]);
1344         block->pmr[chunk] = NULL;
1345         block->remote_keys[chunk] = 0;
1346 
1347         if (ret != 0) {
1348             perror("unregistration chunk failed");
1349             return -ret;
1350         }
1351         rdma->total_registrations--;
1352 
1353         reg.key.chunk = chunk;
1354         register_to_network(rdma, &reg);
1355         ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) &reg,
1356                                 &resp, NULL, NULL);
1357         if (ret < 0) {
1358             return ret;
1359         }
1360 
1361         trace_qemu_rdma_unregister_waiting_complete(chunk);
1362     }
1363 
1364     return 0;
1365 }
1366 
1367 static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index,
1368                                          uint64_t chunk)
1369 {
1370     uint64_t result = wr_id & RDMA_WRID_TYPE_MASK;
1371 
1372     result |= (index << RDMA_WRID_BLOCK_SHIFT);
1373     result |= (chunk << RDMA_WRID_CHUNK_SHIFT);
1374 
1375     return result;
1376 }
1377 
1378 /*
1379  * Set bit for unregistration in the next iteration.
1380  * We cannot transmit right here, but will unpin later.
1381  */
1382 static void qemu_rdma_signal_unregister(RDMAContext *rdma, uint64_t index,
1383                                         uint64_t chunk, uint64_t wr_id)
1384 {
1385     if (rdma->unregistrations[rdma->unregister_next] != 0) {
1386         error_report("rdma migration: queue is full");
1387     } else {
1388         RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1389 
1390         if (!test_and_set_bit(chunk, block->unregister_bitmap)) {
1391             trace_qemu_rdma_signal_unregister_append(chunk,
1392                                                      rdma->unregister_next);
1393 
1394             rdma->unregistrations[rdma->unregister_next++] =
1395                     qemu_rdma_make_wrid(wr_id, index, chunk);
1396 
1397             if (rdma->unregister_next == RDMA_SIGNALED_SEND_MAX) {
1398                 rdma->unregister_next = 0;
1399             }
1400         } else {
1401             trace_qemu_rdma_signal_unregister_already(chunk);
1402         }
1403     }
1404 }
1405 
1406 /*
1407  * Consult the connection manager to see a work request
1408  * (of any kind) has completed.
1409  * Return the work request ID that completed.
1410  */
1411 static uint64_t qemu_rdma_poll(RDMAContext *rdma, uint64_t *wr_id_out,
1412                                uint32_t *byte_len)
1413 {
1414     int ret;
1415     struct ibv_wc wc;
1416     uint64_t wr_id;
1417 
1418     ret = ibv_poll_cq(rdma->cq, 1, &wc);
1419 
1420     if (!ret) {
1421         *wr_id_out = RDMA_WRID_NONE;
1422         return 0;
1423     }
1424 
1425     if (ret < 0) {
1426         error_report("ibv_poll_cq return %d", ret);
1427         return ret;
1428     }
1429 
1430     wr_id = wc.wr_id & RDMA_WRID_TYPE_MASK;
1431 
1432     if (wc.status != IBV_WC_SUCCESS) {
1433         fprintf(stderr, "ibv_poll_cq wc.status=%d %s!\n",
1434                         wc.status, ibv_wc_status_str(wc.status));
1435         fprintf(stderr, "ibv_poll_cq wrid=%s!\n", wrid_desc[wr_id]);
1436 
1437         return -1;
1438     }
1439 
1440     if (rdma->control_ready_expected &&
1441         (wr_id >= RDMA_WRID_RECV_CONTROL)) {
1442         trace_qemu_rdma_poll_recv(wrid_desc[RDMA_WRID_RECV_CONTROL],
1443                   wr_id - RDMA_WRID_RECV_CONTROL, wr_id, rdma->nb_sent);
1444         rdma->control_ready_expected = 0;
1445     }
1446 
1447     if (wr_id == RDMA_WRID_RDMA_WRITE) {
1448         uint64_t chunk =
1449             (wc.wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1450         uint64_t index =
1451             (wc.wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1452         RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1453 
1454         trace_qemu_rdma_poll_write(print_wrid(wr_id), wr_id, rdma->nb_sent,
1455                                    index, chunk, block->local_host_addr,
1456                                    (void *)(uintptr_t)block->remote_host_addr);
1457 
1458         clear_bit(chunk, block->transit_bitmap);
1459 
1460         if (rdma->nb_sent > 0) {
1461             rdma->nb_sent--;
1462         }
1463 
1464         if (!rdma->pin_all) {
1465             /*
1466              * FYI: If one wanted to signal a specific chunk to be unregistered
1467              * using LRU or workload-specific information, this is the function
1468              * you would call to do so. That chunk would then get asynchronously
1469              * unregistered later.
1470              */
1471 #ifdef RDMA_UNREGISTRATION_EXAMPLE
1472             qemu_rdma_signal_unregister(rdma, index, chunk, wc.wr_id);
1473 #endif
1474         }
1475     } else {
1476         trace_qemu_rdma_poll_other(print_wrid(wr_id), wr_id, rdma->nb_sent);
1477     }
1478 
1479     *wr_id_out = wc.wr_id;
1480     if (byte_len) {
1481         *byte_len = wc.byte_len;
1482     }
1483 
1484     return  0;
1485 }
1486 
1487 /* Wait for activity on the completion channel.
1488  * Returns 0 on success, none-0 on error.
1489  */
1490 static int qemu_rdma_wait_comp_channel(RDMAContext *rdma)
1491 {
1492     struct rdma_cm_event *cm_event;
1493     int ret = -1;
1494 
1495     /*
1496      * Coroutine doesn't start until migration_fd_process_incoming()
1497      * so don't yield unless we know we're running inside of a coroutine.
1498      */
1499     if (rdma->migration_started_on_destination &&
1500         migration_incoming_get_current()->state == MIGRATION_STATUS_ACTIVE) {
1501         yield_until_fd_readable(rdma->comp_channel->fd);
1502     } else {
1503         /* This is the source side, we're in a separate thread
1504          * or destination prior to migration_fd_process_incoming()
1505          * after postcopy, the destination also in a seprate thread.
1506          * we can't yield; so we have to poll the fd.
1507          * But we need to be able to handle 'cancel' or an error
1508          * without hanging forever.
1509          */
1510         while (!rdma->error_state  && !rdma->received_error) {
1511             GPollFD pfds[2];
1512             pfds[0].fd = rdma->comp_channel->fd;
1513             pfds[0].events = G_IO_IN | G_IO_HUP | G_IO_ERR;
1514             pfds[0].revents = 0;
1515 
1516             pfds[1].fd = rdma->channel->fd;
1517             pfds[1].events = G_IO_IN | G_IO_HUP | G_IO_ERR;
1518             pfds[1].revents = 0;
1519 
1520             /* 0.1s timeout, should be fine for a 'cancel' */
1521             switch (qemu_poll_ns(pfds, 2, 100 * 1000 * 1000)) {
1522             case 2:
1523             case 1: /* fd active */
1524                 if (pfds[0].revents) {
1525                     return 0;
1526                 }
1527 
1528                 if (pfds[1].revents) {
1529                     ret = rdma_get_cm_event(rdma->channel, &cm_event);
1530                     if (!ret) {
1531                         rdma_ack_cm_event(cm_event);
1532                     }
1533 
1534                     error_report("receive cm event while wait comp channel,"
1535                                  "cm event is %d", cm_event->event);
1536                     if (cm_event->event == RDMA_CM_EVENT_DISCONNECTED ||
1537                         cm_event->event == RDMA_CM_EVENT_DEVICE_REMOVAL) {
1538                         return -EPIPE;
1539                     }
1540                 }
1541                 break;
1542 
1543             case 0: /* Timeout, go around again */
1544                 break;
1545 
1546             default: /* Error of some type -
1547                       * I don't trust errno from qemu_poll_ns
1548                      */
1549                 error_report("%s: poll failed", __func__);
1550                 return -EPIPE;
1551             }
1552 
1553             if (migrate_get_current()->state == MIGRATION_STATUS_CANCELLING) {
1554                 /* Bail out and let the cancellation happen */
1555                 return -EPIPE;
1556             }
1557         }
1558     }
1559 
1560     if (rdma->received_error) {
1561         return -EPIPE;
1562     }
1563     return rdma->error_state;
1564 }
1565 
1566 /*
1567  * Block until the next work request has completed.
1568  *
1569  * First poll to see if a work request has already completed,
1570  * otherwise block.
1571  *
1572  * If we encounter completed work requests for IDs other than
1573  * the one we're interested in, then that's generally an error.
1574  *
1575  * The only exception is actual RDMA Write completions. These
1576  * completions only need to be recorded, but do not actually
1577  * need further processing.
1578  */
1579 static int qemu_rdma_block_for_wrid(RDMAContext *rdma, int wrid_requested,
1580                                     uint32_t *byte_len)
1581 {
1582     int num_cq_events = 0, ret = 0;
1583     struct ibv_cq *cq;
1584     void *cq_ctx;
1585     uint64_t wr_id = RDMA_WRID_NONE, wr_id_in;
1586 
1587     if (ibv_req_notify_cq(rdma->cq, 0)) {
1588         return -1;
1589     }
1590     /* poll cq first */
1591     while (wr_id != wrid_requested) {
1592         ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
1593         if (ret < 0) {
1594             return ret;
1595         }
1596 
1597         wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1598 
1599         if (wr_id == RDMA_WRID_NONE) {
1600             break;
1601         }
1602         if (wr_id != wrid_requested) {
1603             trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested),
1604                        wrid_requested, print_wrid(wr_id), wr_id);
1605         }
1606     }
1607 
1608     if (wr_id == wrid_requested) {
1609         return 0;
1610     }
1611 
1612     while (1) {
1613         ret = qemu_rdma_wait_comp_channel(rdma);
1614         if (ret) {
1615             goto err_block_for_wrid;
1616         }
1617 
1618         ret = ibv_get_cq_event(rdma->comp_channel, &cq, &cq_ctx);
1619         if (ret) {
1620             perror("ibv_get_cq_event");
1621             goto err_block_for_wrid;
1622         }
1623 
1624         num_cq_events++;
1625 
1626         ret = -ibv_req_notify_cq(cq, 0);
1627         if (ret) {
1628             goto err_block_for_wrid;
1629         }
1630 
1631         while (wr_id != wrid_requested) {
1632             ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
1633             if (ret < 0) {
1634                 goto err_block_for_wrid;
1635             }
1636 
1637             wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1638 
1639             if (wr_id == RDMA_WRID_NONE) {
1640                 break;
1641             }
1642             if (wr_id != wrid_requested) {
1643                 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested),
1644                                    wrid_requested, print_wrid(wr_id), wr_id);
1645             }
1646         }
1647 
1648         if (wr_id == wrid_requested) {
1649             goto success_block_for_wrid;
1650         }
1651     }
1652 
1653 success_block_for_wrid:
1654     if (num_cq_events) {
1655         ibv_ack_cq_events(cq, num_cq_events);
1656     }
1657     return 0;
1658 
1659 err_block_for_wrid:
1660     if (num_cq_events) {
1661         ibv_ack_cq_events(cq, num_cq_events);
1662     }
1663 
1664     rdma->error_state = ret;
1665     return ret;
1666 }
1667 
1668 /*
1669  * Post a SEND message work request for the control channel
1670  * containing some data and block until the post completes.
1671  */
1672 static int qemu_rdma_post_send_control(RDMAContext *rdma, uint8_t *buf,
1673                                        RDMAControlHeader *head)
1674 {
1675     int ret = 0;
1676     RDMAWorkRequestData *wr = &rdma->wr_data[RDMA_WRID_CONTROL];
1677     struct ibv_send_wr *bad_wr;
1678     struct ibv_sge sge = {
1679                            .addr = (uintptr_t)(wr->control),
1680                            .length = head->len + sizeof(RDMAControlHeader),
1681                            .lkey = wr->control_mr->lkey,
1682                          };
1683     struct ibv_send_wr send_wr = {
1684                                    .wr_id = RDMA_WRID_SEND_CONTROL,
1685                                    .opcode = IBV_WR_SEND,
1686                                    .send_flags = IBV_SEND_SIGNALED,
1687                                    .sg_list = &sge,
1688                                    .num_sge = 1,
1689                                 };
1690 
1691     trace_qemu_rdma_post_send_control(control_desc(head->type));
1692 
1693     /*
1694      * We don't actually need to do a memcpy() in here if we used
1695      * the "sge" properly, but since we're only sending control messages
1696      * (not RAM in a performance-critical path), then its OK for now.
1697      *
1698      * The copy makes the RDMAControlHeader simpler to manipulate
1699      * for the time being.
1700      */
1701     assert(head->len <= RDMA_CONTROL_MAX_BUFFER - sizeof(*head));
1702     memcpy(wr->control, head, sizeof(RDMAControlHeader));
1703     control_to_network((void *) wr->control);
1704 
1705     if (buf) {
1706         memcpy(wr->control + sizeof(RDMAControlHeader), buf, head->len);
1707     }
1708 
1709 
1710     ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
1711 
1712     if (ret > 0) {
1713         error_report("Failed to use post IB SEND for control");
1714         return -ret;
1715     }
1716 
1717     ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_SEND_CONTROL, NULL);
1718     if (ret < 0) {
1719         error_report("rdma migration: send polling control error");
1720     }
1721 
1722     return ret;
1723 }
1724 
1725 /*
1726  * Post a RECV work request in anticipation of some future receipt
1727  * of data on the control channel.
1728  */
1729 static int qemu_rdma_post_recv_control(RDMAContext *rdma, int idx)
1730 {
1731     struct ibv_recv_wr *bad_wr;
1732     struct ibv_sge sge = {
1733                             .addr = (uintptr_t)(rdma->wr_data[idx].control),
1734                             .length = RDMA_CONTROL_MAX_BUFFER,
1735                             .lkey = rdma->wr_data[idx].control_mr->lkey,
1736                          };
1737 
1738     struct ibv_recv_wr recv_wr = {
1739                                     .wr_id = RDMA_WRID_RECV_CONTROL + idx,
1740                                     .sg_list = &sge,
1741                                     .num_sge = 1,
1742                                  };
1743 
1744 
1745     if (ibv_post_recv(rdma->qp, &recv_wr, &bad_wr)) {
1746         return -1;
1747     }
1748 
1749     return 0;
1750 }
1751 
1752 /*
1753  * Block and wait for a RECV control channel message to arrive.
1754  */
1755 static int qemu_rdma_exchange_get_response(RDMAContext *rdma,
1756                 RDMAControlHeader *head, int expecting, int idx)
1757 {
1758     uint32_t byte_len;
1759     int ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RECV_CONTROL + idx,
1760                                        &byte_len);
1761 
1762     if (ret < 0) {
1763         error_report("rdma migration: recv polling control error!");
1764         return ret;
1765     }
1766 
1767     network_to_control((void *) rdma->wr_data[idx].control);
1768     memcpy(head, rdma->wr_data[idx].control, sizeof(RDMAControlHeader));
1769 
1770     trace_qemu_rdma_exchange_get_response_start(control_desc(expecting));
1771 
1772     if (expecting == RDMA_CONTROL_NONE) {
1773         trace_qemu_rdma_exchange_get_response_none(control_desc(head->type),
1774                                              head->type);
1775     } else if (head->type != expecting || head->type == RDMA_CONTROL_ERROR) {
1776         error_report("Was expecting a %s (%d) control message"
1777                 ", but got: %s (%d), length: %d",
1778                 control_desc(expecting), expecting,
1779                 control_desc(head->type), head->type, head->len);
1780         if (head->type == RDMA_CONTROL_ERROR) {
1781             rdma->received_error = true;
1782         }
1783         return -EIO;
1784     }
1785     if (head->len > RDMA_CONTROL_MAX_BUFFER - sizeof(*head)) {
1786         error_report("too long length: %d", head->len);
1787         return -EINVAL;
1788     }
1789     if (sizeof(*head) + head->len != byte_len) {
1790         error_report("Malformed length: %d byte_len %d", head->len, byte_len);
1791         return -EINVAL;
1792     }
1793 
1794     return 0;
1795 }
1796 
1797 /*
1798  * When a RECV work request has completed, the work request's
1799  * buffer is pointed at the header.
1800  *
1801  * This will advance the pointer to the data portion
1802  * of the control message of the work request's buffer that
1803  * was populated after the work request finished.
1804  */
1805 static void qemu_rdma_move_header(RDMAContext *rdma, int idx,
1806                                   RDMAControlHeader *head)
1807 {
1808     rdma->wr_data[idx].control_len = head->len;
1809     rdma->wr_data[idx].control_curr =
1810         rdma->wr_data[idx].control + sizeof(RDMAControlHeader);
1811 }
1812 
1813 /*
1814  * This is an 'atomic' high-level operation to deliver a single, unified
1815  * control-channel message.
1816  *
1817  * Additionally, if the user is expecting some kind of reply to this message,
1818  * they can request a 'resp' response message be filled in by posting an
1819  * additional work request on behalf of the user and waiting for an additional
1820  * completion.
1821  *
1822  * The extra (optional) response is used during registration to us from having
1823  * to perform an *additional* exchange of message just to provide a response by
1824  * instead piggy-backing on the acknowledgement.
1825  */
1826 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
1827                                    uint8_t *data, RDMAControlHeader *resp,
1828                                    int *resp_idx,
1829                                    int (*callback)(RDMAContext *rdma))
1830 {
1831     int ret = 0;
1832 
1833     /*
1834      * Wait until the dest is ready before attempting to deliver the message
1835      * by waiting for a READY message.
1836      */
1837     if (rdma->control_ready_expected) {
1838         RDMAControlHeader resp;
1839         ret = qemu_rdma_exchange_get_response(rdma,
1840                                     &resp, RDMA_CONTROL_READY, RDMA_WRID_READY);
1841         if (ret < 0) {
1842             return ret;
1843         }
1844     }
1845 
1846     /*
1847      * If the user is expecting a response, post a WR in anticipation of it.
1848      */
1849     if (resp) {
1850         ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_DATA);
1851         if (ret) {
1852             error_report("rdma migration: error posting"
1853                     " extra control recv for anticipated result!");
1854             return ret;
1855         }
1856     }
1857 
1858     /*
1859      * Post a WR to replace the one we just consumed for the READY message.
1860      */
1861     ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
1862     if (ret) {
1863         error_report("rdma migration: error posting first control recv!");
1864         return ret;
1865     }
1866 
1867     /*
1868      * Deliver the control message that was requested.
1869      */
1870     ret = qemu_rdma_post_send_control(rdma, data, head);
1871 
1872     if (ret < 0) {
1873         error_report("Failed to send control buffer!");
1874         return ret;
1875     }
1876 
1877     /*
1878      * If we're expecting a response, block and wait for it.
1879      */
1880     if (resp) {
1881         if (callback) {
1882             trace_qemu_rdma_exchange_send_issue_callback();
1883             ret = callback(rdma);
1884             if (ret < 0) {
1885                 return ret;
1886             }
1887         }
1888 
1889         trace_qemu_rdma_exchange_send_waiting(control_desc(resp->type));
1890         ret = qemu_rdma_exchange_get_response(rdma, resp,
1891                                               resp->type, RDMA_WRID_DATA);
1892 
1893         if (ret < 0) {
1894             return ret;
1895         }
1896 
1897         qemu_rdma_move_header(rdma, RDMA_WRID_DATA, resp);
1898         if (resp_idx) {
1899             *resp_idx = RDMA_WRID_DATA;
1900         }
1901         trace_qemu_rdma_exchange_send_received(control_desc(resp->type));
1902     }
1903 
1904     rdma->control_ready_expected = 1;
1905 
1906     return 0;
1907 }
1908 
1909 /*
1910  * This is an 'atomic' high-level operation to receive a single, unified
1911  * control-channel message.
1912  */
1913 static int qemu_rdma_exchange_recv(RDMAContext *rdma, RDMAControlHeader *head,
1914                                 int expecting)
1915 {
1916     RDMAControlHeader ready = {
1917                                 .len = 0,
1918                                 .type = RDMA_CONTROL_READY,
1919                                 .repeat = 1,
1920                               };
1921     int ret;
1922 
1923     /*
1924      * Inform the source that we're ready to receive a message.
1925      */
1926     ret = qemu_rdma_post_send_control(rdma, NULL, &ready);
1927 
1928     if (ret < 0) {
1929         error_report("Failed to send control buffer!");
1930         return ret;
1931     }
1932 
1933     /*
1934      * Block and wait for the message.
1935      */
1936     ret = qemu_rdma_exchange_get_response(rdma, head,
1937                                           expecting, RDMA_WRID_READY);
1938 
1939     if (ret < 0) {
1940         return ret;
1941     }
1942 
1943     qemu_rdma_move_header(rdma, RDMA_WRID_READY, head);
1944 
1945     /*
1946      * Post a new RECV work request to replace the one we just consumed.
1947      */
1948     ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
1949     if (ret) {
1950         error_report("rdma migration: error posting second control recv!");
1951         return ret;
1952     }
1953 
1954     return 0;
1955 }
1956 
1957 /*
1958  * Write an actual chunk of memory using RDMA.
1959  *
1960  * If we're using dynamic registration on the dest-side, we have to
1961  * send a registration command first.
1962  */
1963 static int qemu_rdma_write_one(QEMUFile *f, RDMAContext *rdma,
1964                                int current_index, uint64_t current_addr,
1965                                uint64_t length)
1966 {
1967     struct ibv_sge sge;
1968     struct ibv_send_wr send_wr = { 0 };
1969     struct ibv_send_wr *bad_wr;
1970     int reg_result_idx, ret, count = 0;
1971     uint64_t chunk, chunks;
1972     uint8_t *chunk_start, *chunk_end;
1973     RDMALocalBlock *block = &(rdma->local_ram_blocks.block[current_index]);
1974     RDMARegister reg;
1975     RDMARegisterResult *reg_result;
1976     RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT };
1977     RDMAControlHeader head = { .len = sizeof(RDMARegister),
1978                                .type = RDMA_CONTROL_REGISTER_REQUEST,
1979                                .repeat = 1,
1980                              };
1981 
1982 retry:
1983     sge.addr = (uintptr_t)(block->local_host_addr +
1984                             (current_addr - block->offset));
1985     sge.length = length;
1986 
1987     chunk = ram_chunk_index(block->local_host_addr,
1988                             (uint8_t *)(uintptr_t)sge.addr);
1989     chunk_start = ram_chunk_start(block, chunk);
1990 
1991     if (block->is_ram_block) {
1992         chunks = length / (1UL << RDMA_REG_CHUNK_SHIFT);
1993 
1994         if (chunks && ((length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
1995             chunks--;
1996         }
1997     } else {
1998         chunks = block->length / (1UL << RDMA_REG_CHUNK_SHIFT);
1999 
2000         if (chunks && ((block->length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
2001             chunks--;
2002         }
2003     }
2004 
2005     trace_qemu_rdma_write_one_top(chunks + 1,
2006                                   (chunks + 1) *
2007                                   (1UL << RDMA_REG_CHUNK_SHIFT) / 1024 / 1024);
2008 
2009     chunk_end = ram_chunk_end(block, chunk + chunks);
2010 
2011     if (!rdma->pin_all) {
2012 #ifdef RDMA_UNREGISTRATION_EXAMPLE
2013         qemu_rdma_unregister_waiting(rdma);
2014 #endif
2015     }
2016 
2017     while (test_bit(chunk, block->transit_bitmap)) {
2018         (void)count;
2019         trace_qemu_rdma_write_one_block(count++, current_index, chunk,
2020                 sge.addr, length, rdma->nb_sent, block->nb_chunks);
2021 
2022         ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2023 
2024         if (ret < 0) {
2025             error_report("Failed to Wait for previous write to complete "
2026                     "block %d chunk %" PRIu64
2027                     " current %" PRIu64 " len %" PRIu64 " %d",
2028                     current_index, chunk, sge.addr, length, rdma->nb_sent);
2029             return ret;
2030         }
2031     }
2032 
2033     if (!rdma->pin_all || !block->is_ram_block) {
2034         if (!block->remote_keys[chunk]) {
2035             /*
2036              * This chunk has not yet been registered, so first check to see
2037              * if the entire chunk is zero. If so, tell the other size to
2038              * memset() + madvise() the entire chunk without RDMA.
2039              */
2040 
2041             if (buffer_is_zero((void *)(uintptr_t)sge.addr, length)) {
2042                 RDMACompress comp = {
2043                                         .offset = current_addr,
2044                                         .value = 0,
2045                                         .block_idx = current_index,
2046                                         .length = length,
2047                                     };
2048 
2049                 head.len = sizeof(comp);
2050                 head.type = RDMA_CONTROL_COMPRESS;
2051 
2052                 trace_qemu_rdma_write_one_zero(chunk, sge.length,
2053                                                current_index, current_addr);
2054 
2055                 compress_to_network(rdma, &comp);
2056                 ret = qemu_rdma_exchange_send(rdma, &head,
2057                                 (uint8_t *) &comp, NULL, NULL, NULL);
2058 
2059                 if (ret < 0) {
2060                     return -EIO;
2061                 }
2062 
2063                 acct_update_position(f, sge.length, true);
2064 
2065                 return 1;
2066             }
2067 
2068             /*
2069              * Otherwise, tell other side to register.
2070              */
2071             reg.current_index = current_index;
2072             if (block->is_ram_block) {
2073                 reg.key.current_addr = current_addr;
2074             } else {
2075                 reg.key.chunk = chunk;
2076             }
2077             reg.chunks = chunks;
2078 
2079             trace_qemu_rdma_write_one_sendreg(chunk, sge.length, current_index,
2080                                               current_addr);
2081 
2082             register_to_network(rdma, &reg);
2083             ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) &reg,
2084                                     &resp, &reg_result_idx, NULL);
2085             if (ret < 0) {
2086                 return ret;
2087             }
2088 
2089             /* try to overlap this single registration with the one we sent. */
2090             if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
2091                                                 &sge.lkey, NULL, chunk,
2092                                                 chunk_start, chunk_end)) {
2093                 error_report("cannot get lkey");
2094                 return -EINVAL;
2095             }
2096 
2097             reg_result = (RDMARegisterResult *)
2098                     rdma->wr_data[reg_result_idx].control_curr;
2099 
2100             network_to_result(reg_result);
2101 
2102             trace_qemu_rdma_write_one_recvregres(block->remote_keys[chunk],
2103                                                  reg_result->rkey, chunk);
2104 
2105             block->remote_keys[chunk] = reg_result->rkey;
2106             block->remote_host_addr = reg_result->host_addr;
2107         } else {
2108             /* already registered before */
2109             if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
2110                                                 &sge.lkey, NULL, chunk,
2111                                                 chunk_start, chunk_end)) {
2112                 error_report("cannot get lkey!");
2113                 return -EINVAL;
2114             }
2115         }
2116 
2117         send_wr.wr.rdma.rkey = block->remote_keys[chunk];
2118     } else {
2119         send_wr.wr.rdma.rkey = block->remote_rkey;
2120 
2121         if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
2122                                                      &sge.lkey, NULL, chunk,
2123                                                      chunk_start, chunk_end)) {
2124             error_report("cannot get lkey!");
2125             return -EINVAL;
2126         }
2127     }
2128 
2129     /*
2130      * Encode the ram block index and chunk within this wrid.
2131      * We will use this information at the time of completion
2132      * to figure out which bitmap to check against and then which
2133      * chunk in the bitmap to look for.
2134      */
2135     send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE,
2136                                         current_index, chunk);
2137 
2138     send_wr.opcode = IBV_WR_RDMA_WRITE;
2139     send_wr.send_flags = IBV_SEND_SIGNALED;
2140     send_wr.sg_list = &sge;
2141     send_wr.num_sge = 1;
2142     send_wr.wr.rdma.remote_addr = block->remote_host_addr +
2143                                 (current_addr - block->offset);
2144 
2145     trace_qemu_rdma_write_one_post(chunk, sge.addr, send_wr.wr.rdma.remote_addr,
2146                                    sge.length);
2147 
2148     /*
2149      * ibv_post_send() does not return negative error numbers,
2150      * per the specification they are positive - no idea why.
2151      */
2152     ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
2153 
2154     if (ret == ENOMEM) {
2155         trace_qemu_rdma_write_one_queue_full();
2156         ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2157         if (ret < 0) {
2158             error_report("rdma migration: failed to make "
2159                          "room in full send queue! %d", ret);
2160             return ret;
2161         }
2162 
2163         goto retry;
2164 
2165     } else if (ret > 0) {
2166         perror("rdma migration: post rdma write failed");
2167         return -ret;
2168     }
2169 
2170     set_bit(chunk, block->transit_bitmap);
2171     acct_update_position(f, sge.length, false);
2172     rdma->total_writes++;
2173 
2174     return 0;
2175 }
2176 
2177 /*
2178  * Push out any unwritten RDMA operations.
2179  *
2180  * We support sending out multiple chunks at the same time.
2181  * Not all of them need to get signaled in the completion queue.
2182  */
2183 static int qemu_rdma_write_flush(QEMUFile *f, RDMAContext *rdma)
2184 {
2185     int ret;
2186 
2187     if (!rdma->current_length) {
2188         return 0;
2189     }
2190 
2191     ret = qemu_rdma_write_one(f, rdma,
2192             rdma->current_index, rdma->current_addr, rdma->current_length);
2193 
2194     if (ret < 0) {
2195         return ret;
2196     }
2197 
2198     if (ret == 0) {
2199         rdma->nb_sent++;
2200         trace_qemu_rdma_write_flush(rdma->nb_sent);
2201     }
2202 
2203     rdma->current_length = 0;
2204     rdma->current_addr = 0;
2205 
2206     return 0;
2207 }
2208 
2209 static inline int qemu_rdma_buffer_mergable(RDMAContext *rdma,
2210                     uint64_t offset, uint64_t len)
2211 {
2212     RDMALocalBlock *block;
2213     uint8_t *host_addr;
2214     uint8_t *chunk_end;
2215 
2216     if (rdma->current_index < 0) {
2217         return 0;
2218     }
2219 
2220     if (rdma->current_chunk < 0) {
2221         return 0;
2222     }
2223 
2224     block = &(rdma->local_ram_blocks.block[rdma->current_index]);
2225     host_addr = block->local_host_addr + (offset - block->offset);
2226     chunk_end = ram_chunk_end(block, rdma->current_chunk);
2227 
2228     if (rdma->current_length == 0) {
2229         return 0;
2230     }
2231 
2232     /*
2233      * Only merge into chunk sequentially.
2234      */
2235     if (offset != (rdma->current_addr + rdma->current_length)) {
2236         return 0;
2237     }
2238 
2239     if (offset < block->offset) {
2240         return 0;
2241     }
2242 
2243     if ((offset + len) > (block->offset + block->length)) {
2244         return 0;
2245     }
2246 
2247     if ((host_addr + len) > chunk_end) {
2248         return 0;
2249     }
2250 
2251     return 1;
2252 }
2253 
2254 /*
2255  * We're not actually writing here, but doing three things:
2256  *
2257  * 1. Identify the chunk the buffer belongs to.
2258  * 2. If the chunk is full or the buffer doesn't belong to the current
2259  *    chunk, then start a new chunk and flush() the old chunk.
2260  * 3. To keep the hardware busy, we also group chunks into batches
2261  *    and only require that a batch gets acknowledged in the completion
2262  *    qeueue instead of each individual chunk.
2263  */
2264 static int qemu_rdma_write(QEMUFile *f, RDMAContext *rdma,
2265                            uint64_t block_offset, uint64_t offset,
2266                            uint64_t len)
2267 {
2268     uint64_t current_addr = block_offset + offset;
2269     uint64_t index = rdma->current_index;
2270     uint64_t chunk = rdma->current_chunk;
2271     int ret;
2272 
2273     /* If we cannot merge it, we flush the current buffer first. */
2274     if (!qemu_rdma_buffer_mergable(rdma, current_addr, len)) {
2275         ret = qemu_rdma_write_flush(f, rdma);
2276         if (ret) {
2277             return ret;
2278         }
2279         rdma->current_length = 0;
2280         rdma->current_addr = current_addr;
2281 
2282         ret = qemu_rdma_search_ram_block(rdma, block_offset,
2283                                          offset, len, &index, &chunk);
2284         if (ret) {
2285             error_report("ram block search failed");
2286             return ret;
2287         }
2288         rdma->current_index = index;
2289         rdma->current_chunk = chunk;
2290     }
2291 
2292     /* merge it */
2293     rdma->current_length += len;
2294 
2295     /* flush it if buffer is too large */
2296     if (rdma->current_length >= RDMA_MERGE_MAX) {
2297         return qemu_rdma_write_flush(f, rdma);
2298     }
2299 
2300     return 0;
2301 }
2302 
2303 static void qemu_rdma_cleanup(RDMAContext *rdma)
2304 {
2305     int idx;
2306 
2307     if (rdma->cm_id && rdma->connected) {
2308         if ((rdma->error_state ||
2309              migrate_get_current()->state == MIGRATION_STATUS_CANCELLING) &&
2310             !rdma->received_error) {
2311             RDMAControlHeader head = { .len = 0,
2312                                        .type = RDMA_CONTROL_ERROR,
2313                                        .repeat = 1,
2314                                      };
2315             error_report("Early error. Sending error.");
2316             qemu_rdma_post_send_control(rdma, NULL, &head);
2317         }
2318 
2319         rdma_disconnect(rdma->cm_id);
2320         trace_qemu_rdma_cleanup_disconnect();
2321         rdma->connected = false;
2322     }
2323 
2324     qemu_set_fd_handler(rdma->channel->fd, NULL, NULL, NULL);
2325     g_free(rdma->dest_blocks);
2326     rdma->dest_blocks = NULL;
2327 
2328     for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2329         if (rdma->wr_data[idx].control_mr) {
2330             rdma->total_registrations--;
2331             ibv_dereg_mr(rdma->wr_data[idx].control_mr);
2332         }
2333         rdma->wr_data[idx].control_mr = NULL;
2334     }
2335 
2336     if (rdma->local_ram_blocks.block) {
2337         while (rdma->local_ram_blocks.nb_blocks) {
2338             rdma_delete_block(rdma, &rdma->local_ram_blocks.block[0]);
2339         }
2340     }
2341 
2342     if (rdma->qp) {
2343         rdma_destroy_qp(rdma->cm_id);
2344         rdma->qp = NULL;
2345     }
2346     if (rdma->cq) {
2347         ibv_destroy_cq(rdma->cq);
2348         rdma->cq = NULL;
2349     }
2350     if (rdma->comp_channel) {
2351         ibv_destroy_comp_channel(rdma->comp_channel);
2352         rdma->comp_channel = NULL;
2353     }
2354     if (rdma->pd) {
2355         ibv_dealloc_pd(rdma->pd);
2356         rdma->pd = NULL;
2357     }
2358     if (rdma->cm_id) {
2359         rdma_destroy_id(rdma->cm_id);
2360         rdma->cm_id = NULL;
2361     }
2362 
2363     /* the destination side, listen_id and channel is shared */
2364     if (rdma->listen_id) {
2365         if (!rdma->is_return_path) {
2366             rdma_destroy_id(rdma->listen_id);
2367         }
2368         rdma->listen_id = NULL;
2369 
2370         if (rdma->channel) {
2371             if (!rdma->is_return_path) {
2372                 rdma_destroy_event_channel(rdma->channel);
2373             }
2374             rdma->channel = NULL;
2375         }
2376     }
2377 
2378     if (rdma->channel) {
2379         rdma_destroy_event_channel(rdma->channel);
2380         rdma->channel = NULL;
2381     }
2382     g_free(rdma->host);
2383     rdma->host = NULL;
2384 }
2385 
2386 
2387 static int qemu_rdma_source_init(RDMAContext *rdma, bool pin_all, Error **errp)
2388 {
2389     int ret, idx;
2390     Error *local_err = NULL, **temp = &local_err;
2391 
2392     /*
2393      * Will be validated against destination's actual capabilities
2394      * after the connect() completes.
2395      */
2396     rdma->pin_all = pin_all;
2397 
2398     ret = qemu_rdma_resolve_host(rdma, temp);
2399     if (ret) {
2400         goto err_rdma_source_init;
2401     }
2402 
2403     ret = qemu_rdma_alloc_pd_cq(rdma);
2404     if (ret) {
2405         ERROR(temp, "rdma migration: error allocating pd and cq! Your mlock()"
2406                     " limits may be too low. Please check $ ulimit -a # and "
2407                     "search for 'ulimit -l' in the output");
2408         goto err_rdma_source_init;
2409     }
2410 
2411     ret = qemu_rdma_alloc_qp(rdma);
2412     if (ret) {
2413         ERROR(temp, "rdma migration: error allocating qp!");
2414         goto err_rdma_source_init;
2415     }
2416 
2417     ret = qemu_rdma_init_ram_blocks(rdma);
2418     if (ret) {
2419         ERROR(temp, "rdma migration: error initializing ram blocks!");
2420         goto err_rdma_source_init;
2421     }
2422 
2423     /* Build the hash that maps from offset to RAMBlock */
2424     rdma->blockmap = g_hash_table_new(g_direct_hash, g_direct_equal);
2425     for (idx = 0; idx < rdma->local_ram_blocks.nb_blocks; idx++) {
2426         g_hash_table_insert(rdma->blockmap,
2427                 (void *)(uintptr_t)rdma->local_ram_blocks.block[idx].offset,
2428                 &rdma->local_ram_blocks.block[idx]);
2429     }
2430 
2431     for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2432         ret = qemu_rdma_reg_control(rdma, idx);
2433         if (ret) {
2434             ERROR(temp, "rdma migration: error registering %d control!",
2435                                                             idx);
2436             goto err_rdma_source_init;
2437         }
2438     }
2439 
2440     return 0;
2441 
2442 err_rdma_source_init:
2443     error_propagate(errp, local_err);
2444     qemu_rdma_cleanup(rdma);
2445     return -1;
2446 }
2447 
2448 static int qemu_rdma_connect(RDMAContext *rdma, Error **errp)
2449 {
2450     RDMACapabilities cap = {
2451                                 .version = RDMA_CONTROL_VERSION_CURRENT,
2452                                 .flags = 0,
2453                            };
2454     struct rdma_conn_param conn_param = { .initiator_depth = 2,
2455                                           .retry_count = 5,
2456                                           .private_data = &cap,
2457                                           .private_data_len = sizeof(cap),
2458                                         };
2459     struct rdma_cm_event *cm_event;
2460     int ret;
2461 
2462     /*
2463      * Only negotiate the capability with destination if the user
2464      * on the source first requested the capability.
2465      */
2466     if (rdma->pin_all) {
2467         trace_qemu_rdma_connect_pin_all_requested();
2468         cap.flags |= RDMA_CAPABILITY_PIN_ALL;
2469     }
2470 
2471     caps_to_network(&cap);
2472 
2473     ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
2474     if (ret) {
2475         ERROR(errp, "posting second control recv");
2476         goto err_rdma_source_connect;
2477     }
2478 
2479     ret = rdma_connect(rdma->cm_id, &conn_param);
2480     if (ret) {
2481         perror("rdma_connect");
2482         ERROR(errp, "connecting to destination!");
2483         goto err_rdma_source_connect;
2484     }
2485 
2486     ret = rdma_get_cm_event(rdma->channel, &cm_event);
2487     if (ret) {
2488         perror("rdma_get_cm_event after rdma_connect");
2489         ERROR(errp, "connecting to destination!");
2490         rdma_ack_cm_event(cm_event);
2491         goto err_rdma_source_connect;
2492     }
2493 
2494     if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
2495         perror("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect");
2496         ERROR(errp, "connecting to destination!");
2497         rdma_ack_cm_event(cm_event);
2498         goto err_rdma_source_connect;
2499     }
2500     rdma->connected = true;
2501 
2502     memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
2503     network_to_caps(&cap);
2504 
2505     /*
2506      * Verify that the *requested* capabilities are supported by the destination
2507      * and disable them otherwise.
2508      */
2509     if (rdma->pin_all && !(cap.flags & RDMA_CAPABILITY_PIN_ALL)) {
2510         ERROR(errp, "Server cannot support pinning all memory. "
2511                         "Will register memory dynamically.");
2512         rdma->pin_all = false;
2513     }
2514 
2515     trace_qemu_rdma_connect_pin_all_outcome(rdma->pin_all);
2516 
2517     rdma_ack_cm_event(cm_event);
2518 
2519     rdma->control_ready_expected = 1;
2520     rdma->nb_sent = 0;
2521     return 0;
2522 
2523 err_rdma_source_connect:
2524     qemu_rdma_cleanup(rdma);
2525     return -1;
2526 }
2527 
2528 static int qemu_rdma_dest_init(RDMAContext *rdma, Error **errp)
2529 {
2530     int ret, idx;
2531     struct rdma_cm_id *listen_id;
2532     char ip[40] = "unknown";
2533     struct rdma_addrinfo *res, *e;
2534     char port_str[16];
2535 
2536     for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2537         rdma->wr_data[idx].control_len = 0;
2538         rdma->wr_data[idx].control_curr = NULL;
2539     }
2540 
2541     if (!rdma->host || !rdma->host[0]) {
2542         ERROR(errp, "RDMA host is not set!");
2543         rdma->error_state = -EINVAL;
2544         return -1;
2545     }
2546     /* create CM channel */
2547     rdma->channel = rdma_create_event_channel();
2548     if (!rdma->channel) {
2549         ERROR(errp, "could not create rdma event channel");
2550         rdma->error_state = -EINVAL;
2551         return -1;
2552     }
2553 
2554     /* create CM id */
2555     ret = rdma_create_id(rdma->channel, &listen_id, NULL, RDMA_PS_TCP);
2556     if (ret) {
2557         ERROR(errp, "could not create cm_id!");
2558         goto err_dest_init_create_listen_id;
2559     }
2560 
2561     snprintf(port_str, 16, "%d", rdma->port);
2562     port_str[15] = '\0';
2563 
2564     ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
2565     if (ret < 0) {
2566         ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
2567         goto err_dest_init_bind_addr;
2568     }
2569 
2570     for (e = res; e != NULL; e = e->ai_next) {
2571         inet_ntop(e->ai_family,
2572             &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
2573         trace_qemu_rdma_dest_init_trying(rdma->host, ip);
2574         ret = rdma_bind_addr(listen_id, e->ai_dst_addr);
2575         if (ret) {
2576             continue;
2577         }
2578         if (e->ai_family == AF_INET6) {
2579             ret = qemu_rdma_broken_ipv6_kernel(listen_id->verbs, errp);
2580             if (ret) {
2581                 continue;
2582             }
2583         }
2584         break;
2585     }
2586 
2587     if (!e) {
2588         ERROR(errp, "Error: could not rdma_bind_addr!");
2589         goto err_dest_init_bind_addr;
2590     }
2591 
2592     rdma->listen_id = listen_id;
2593     qemu_rdma_dump_gid("dest_init", listen_id);
2594     return 0;
2595 
2596 err_dest_init_bind_addr:
2597     rdma_destroy_id(listen_id);
2598 err_dest_init_create_listen_id:
2599     rdma_destroy_event_channel(rdma->channel);
2600     rdma->channel = NULL;
2601     rdma->error_state = ret;
2602     return ret;
2603 
2604 }
2605 
2606 static void qemu_rdma_return_path_dest_init(RDMAContext *rdma_return_path,
2607                                             RDMAContext *rdma)
2608 {
2609     int idx;
2610 
2611     for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2612         rdma_return_path->wr_data[idx].control_len = 0;
2613         rdma_return_path->wr_data[idx].control_curr = NULL;
2614     }
2615 
2616     /*the CM channel and CM id is shared*/
2617     rdma_return_path->channel = rdma->channel;
2618     rdma_return_path->listen_id = rdma->listen_id;
2619 
2620     rdma->return_path = rdma_return_path;
2621     rdma_return_path->return_path = rdma;
2622     rdma_return_path->is_return_path = true;
2623 }
2624 
2625 static void *qemu_rdma_data_init(const char *host_port, Error **errp)
2626 {
2627     RDMAContext *rdma = NULL;
2628     InetSocketAddress *addr;
2629 
2630     if (host_port) {
2631         rdma = g_new0(RDMAContext, 1);
2632         rdma->current_index = -1;
2633         rdma->current_chunk = -1;
2634 
2635         addr = g_new(InetSocketAddress, 1);
2636         if (!inet_parse(addr, host_port, NULL)) {
2637             rdma->port = atoi(addr->port);
2638             rdma->host = g_strdup(addr->host);
2639         } else {
2640             ERROR(errp, "bad RDMA migration address '%s'", host_port);
2641             g_free(rdma);
2642             rdma = NULL;
2643         }
2644 
2645         qapi_free_InetSocketAddress(addr);
2646     }
2647 
2648     return rdma;
2649 }
2650 
2651 /*
2652  * QEMUFile interface to the control channel.
2653  * SEND messages for control only.
2654  * VM's ram is handled with regular RDMA messages.
2655  */
2656 static ssize_t qio_channel_rdma_writev(QIOChannel *ioc,
2657                                        const struct iovec *iov,
2658                                        size_t niov,
2659                                        int *fds,
2660                                        size_t nfds,
2661                                        Error **errp)
2662 {
2663     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2664     QEMUFile *f = rioc->file;
2665     RDMAContext *rdma;
2666     int ret;
2667     ssize_t done = 0;
2668     size_t i;
2669     size_t len = 0;
2670 
2671     rcu_read_lock();
2672     rdma = atomic_rcu_read(&rioc->rdmaout);
2673 
2674     if (!rdma) {
2675         rcu_read_unlock();
2676         return -EIO;
2677     }
2678 
2679     CHECK_ERROR_STATE();
2680 
2681     /*
2682      * Push out any writes that
2683      * we're queued up for VM's ram.
2684      */
2685     ret = qemu_rdma_write_flush(f, rdma);
2686     if (ret < 0) {
2687         rdma->error_state = ret;
2688         rcu_read_unlock();
2689         return ret;
2690     }
2691 
2692     for (i = 0; i < niov; i++) {
2693         size_t remaining = iov[i].iov_len;
2694         uint8_t * data = (void *)iov[i].iov_base;
2695         while (remaining) {
2696             RDMAControlHeader head;
2697 
2698             len = MIN(remaining, RDMA_SEND_INCREMENT);
2699             remaining -= len;
2700 
2701             head.len = len;
2702             head.type = RDMA_CONTROL_QEMU_FILE;
2703 
2704             ret = qemu_rdma_exchange_send(rdma, &head, data, NULL, NULL, NULL);
2705 
2706             if (ret < 0) {
2707                 rdma->error_state = ret;
2708                 rcu_read_unlock();
2709                 return ret;
2710             }
2711 
2712             data += len;
2713             done += len;
2714         }
2715     }
2716 
2717     rcu_read_unlock();
2718     return done;
2719 }
2720 
2721 static size_t qemu_rdma_fill(RDMAContext *rdma, uint8_t *buf,
2722                              size_t size, int idx)
2723 {
2724     size_t len = 0;
2725 
2726     if (rdma->wr_data[idx].control_len) {
2727         trace_qemu_rdma_fill(rdma->wr_data[idx].control_len, size);
2728 
2729         len = MIN(size, rdma->wr_data[idx].control_len);
2730         memcpy(buf, rdma->wr_data[idx].control_curr, len);
2731         rdma->wr_data[idx].control_curr += len;
2732         rdma->wr_data[idx].control_len -= len;
2733     }
2734 
2735     return len;
2736 }
2737 
2738 /*
2739  * QEMUFile interface to the control channel.
2740  * RDMA links don't use bytestreams, so we have to
2741  * return bytes to QEMUFile opportunistically.
2742  */
2743 static ssize_t qio_channel_rdma_readv(QIOChannel *ioc,
2744                                       const struct iovec *iov,
2745                                       size_t niov,
2746                                       int **fds,
2747                                       size_t *nfds,
2748                                       Error **errp)
2749 {
2750     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2751     RDMAContext *rdma;
2752     RDMAControlHeader head;
2753     int ret = 0;
2754     ssize_t i;
2755     size_t done = 0;
2756 
2757     rcu_read_lock();
2758     rdma = atomic_rcu_read(&rioc->rdmain);
2759 
2760     if (!rdma) {
2761         rcu_read_unlock();
2762         return -EIO;
2763     }
2764 
2765     CHECK_ERROR_STATE();
2766 
2767     for (i = 0; i < niov; i++) {
2768         size_t want = iov[i].iov_len;
2769         uint8_t *data = (void *)iov[i].iov_base;
2770 
2771         /*
2772          * First, we hold on to the last SEND message we
2773          * were given and dish out the bytes until we run
2774          * out of bytes.
2775          */
2776         ret = qemu_rdma_fill(rdma, data, want, 0);
2777         done += ret;
2778         want -= ret;
2779         /* Got what we needed, so go to next iovec */
2780         if (want == 0) {
2781             continue;
2782         }
2783 
2784         /* If we got any data so far, then don't wait
2785          * for more, just return what we have */
2786         if (done > 0) {
2787             break;
2788         }
2789 
2790 
2791         /* We've got nothing at all, so lets wait for
2792          * more to arrive
2793          */
2794         ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_QEMU_FILE);
2795 
2796         if (ret < 0) {
2797             rdma->error_state = ret;
2798             rcu_read_unlock();
2799             return ret;
2800         }
2801 
2802         /*
2803          * SEND was received with new bytes, now try again.
2804          */
2805         ret = qemu_rdma_fill(rdma, data, want, 0);
2806         done += ret;
2807         want -= ret;
2808 
2809         /* Still didn't get enough, so lets just return */
2810         if (want) {
2811             if (done == 0) {
2812                 rcu_read_unlock();
2813                 return QIO_CHANNEL_ERR_BLOCK;
2814             } else {
2815                 break;
2816             }
2817         }
2818     }
2819     rcu_read_unlock();
2820     return done;
2821 }
2822 
2823 /*
2824  * Block until all the outstanding chunks have been delivered by the hardware.
2825  */
2826 static int qemu_rdma_drain_cq(QEMUFile *f, RDMAContext *rdma)
2827 {
2828     int ret;
2829 
2830     if (qemu_rdma_write_flush(f, rdma) < 0) {
2831         return -EIO;
2832     }
2833 
2834     while (rdma->nb_sent) {
2835         ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2836         if (ret < 0) {
2837             error_report("rdma migration: complete polling error!");
2838             return -EIO;
2839         }
2840     }
2841 
2842     qemu_rdma_unregister_waiting(rdma);
2843 
2844     return 0;
2845 }
2846 
2847 
2848 static int qio_channel_rdma_set_blocking(QIOChannel *ioc,
2849                                          bool blocking,
2850                                          Error **errp)
2851 {
2852     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2853     /* XXX we should make readv/writev actually honour this :-) */
2854     rioc->blocking = blocking;
2855     return 0;
2856 }
2857 
2858 
2859 typedef struct QIOChannelRDMASource QIOChannelRDMASource;
2860 struct QIOChannelRDMASource {
2861     GSource parent;
2862     QIOChannelRDMA *rioc;
2863     GIOCondition condition;
2864 };
2865 
2866 static gboolean
2867 qio_channel_rdma_source_prepare(GSource *source,
2868                                 gint *timeout)
2869 {
2870     QIOChannelRDMASource *rsource = (QIOChannelRDMASource *)source;
2871     RDMAContext *rdma;
2872     GIOCondition cond = 0;
2873     *timeout = -1;
2874 
2875     rcu_read_lock();
2876     if (rsource->condition == G_IO_IN) {
2877         rdma = atomic_rcu_read(&rsource->rioc->rdmain);
2878     } else {
2879         rdma = atomic_rcu_read(&rsource->rioc->rdmaout);
2880     }
2881 
2882     if (!rdma) {
2883         error_report("RDMAContext is NULL when prepare Gsource");
2884         rcu_read_unlock();
2885         return FALSE;
2886     }
2887 
2888     if (rdma->wr_data[0].control_len) {
2889         cond |= G_IO_IN;
2890     }
2891     cond |= G_IO_OUT;
2892 
2893     rcu_read_unlock();
2894     return cond & rsource->condition;
2895 }
2896 
2897 static gboolean
2898 qio_channel_rdma_source_check(GSource *source)
2899 {
2900     QIOChannelRDMASource *rsource = (QIOChannelRDMASource *)source;
2901     RDMAContext *rdma;
2902     GIOCondition cond = 0;
2903 
2904     rcu_read_lock();
2905     if (rsource->condition == G_IO_IN) {
2906         rdma = atomic_rcu_read(&rsource->rioc->rdmain);
2907     } else {
2908         rdma = atomic_rcu_read(&rsource->rioc->rdmaout);
2909     }
2910 
2911     if (!rdma) {
2912         error_report("RDMAContext is NULL when check Gsource");
2913         rcu_read_unlock();
2914         return FALSE;
2915     }
2916 
2917     if (rdma->wr_data[0].control_len) {
2918         cond |= G_IO_IN;
2919     }
2920     cond |= G_IO_OUT;
2921 
2922     rcu_read_unlock();
2923     return cond & rsource->condition;
2924 }
2925 
2926 static gboolean
2927 qio_channel_rdma_source_dispatch(GSource *source,
2928                                  GSourceFunc callback,
2929                                  gpointer user_data)
2930 {
2931     QIOChannelFunc func = (QIOChannelFunc)callback;
2932     QIOChannelRDMASource *rsource = (QIOChannelRDMASource *)source;
2933     RDMAContext *rdma;
2934     GIOCondition cond = 0;
2935 
2936     rcu_read_lock();
2937     if (rsource->condition == G_IO_IN) {
2938         rdma = atomic_rcu_read(&rsource->rioc->rdmain);
2939     } else {
2940         rdma = atomic_rcu_read(&rsource->rioc->rdmaout);
2941     }
2942 
2943     if (!rdma) {
2944         error_report("RDMAContext is NULL when dispatch Gsource");
2945         rcu_read_unlock();
2946         return FALSE;
2947     }
2948 
2949     if (rdma->wr_data[0].control_len) {
2950         cond |= G_IO_IN;
2951     }
2952     cond |= G_IO_OUT;
2953 
2954     rcu_read_unlock();
2955     return (*func)(QIO_CHANNEL(rsource->rioc),
2956                    (cond & rsource->condition),
2957                    user_data);
2958 }
2959 
2960 static void
2961 qio_channel_rdma_source_finalize(GSource *source)
2962 {
2963     QIOChannelRDMASource *ssource = (QIOChannelRDMASource *)source;
2964 
2965     object_unref(OBJECT(ssource->rioc));
2966 }
2967 
2968 GSourceFuncs qio_channel_rdma_source_funcs = {
2969     qio_channel_rdma_source_prepare,
2970     qio_channel_rdma_source_check,
2971     qio_channel_rdma_source_dispatch,
2972     qio_channel_rdma_source_finalize
2973 };
2974 
2975 static GSource *qio_channel_rdma_create_watch(QIOChannel *ioc,
2976                                               GIOCondition condition)
2977 {
2978     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2979     QIOChannelRDMASource *ssource;
2980     GSource *source;
2981 
2982     source = g_source_new(&qio_channel_rdma_source_funcs,
2983                           sizeof(QIOChannelRDMASource));
2984     ssource = (QIOChannelRDMASource *)source;
2985 
2986     ssource->rioc = rioc;
2987     object_ref(OBJECT(rioc));
2988 
2989     ssource->condition = condition;
2990 
2991     return source;
2992 }
2993 
2994 static void qio_channel_rdma_set_aio_fd_handler(QIOChannel *ioc,
2995                                                   AioContext *ctx,
2996                                                   IOHandler *io_read,
2997                                                   IOHandler *io_write,
2998                                                   void *opaque)
2999 {
3000     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3001     if (io_read) {
3002         aio_set_fd_handler(ctx, rioc->rdmain->comp_channel->fd,
3003                            false, io_read, io_write, NULL, opaque);
3004     } else {
3005         aio_set_fd_handler(ctx, rioc->rdmaout->comp_channel->fd,
3006                            false, io_read, io_write, NULL, opaque);
3007     }
3008 }
3009 
3010 static int qio_channel_rdma_close(QIOChannel *ioc,
3011                                   Error **errp)
3012 {
3013     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3014     RDMAContext *rdmain, *rdmaout;
3015     trace_qemu_rdma_close();
3016 
3017     rdmain = rioc->rdmain;
3018     if (rdmain) {
3019         atomic_rcu_set(&rioc->rdmain, NULL);
3020     }
3021 
3022     rdmaout = rioc->rdmaout;
3023     if (rdmaout) {
3024         atomic_rcu_set(&rioc->rdmaout, NULL);
3025     }
3026 
3027     synchronize_rcu();
3028 
3029     if (rdmain) {
3030         qemu_rdma_cleanup(rdmain);
3031     }
3032 
3033     if (rdmaout) {
3034         qemu_rdma_cleanup(rdmaout);
3035     }
3036 
3037     g_free(rdmain);
3038     g_free(rdmaout);
3039 
3040     return 0;
3041 }
3042 
3043 static int
3044 qio_channel_rdma_shutdown(QIOChannel *ioc,
3045                             QIOChannelShutdown how,
3046                             Error **errp)
3047 {
3048     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3049     RDMAContext *rdmain, *rdmaout;
3050 
3051     rcu_read_lock();
3052 
3053     rdmain = atomic_rcu_read(&rioc->rdmain);
3054     rdmaout = atomic_rcu_read(&rioc->rdmain);
3055 
3056     switch (how) {
3057     case QIO_CHANNEL_SHUTDOWN_READ:
3058         if (rdmain) {
3059             rdmain->error_state = -1;
3060         }
3061         break;
3062     case QIO_CHANNEL_SHUTDOWN_WRITE:
3063         if (rdmaout) {
3064             rdmaout->error_state = -1;
3065         }
3066         break;
3067     case QIO_CHANNEL_SHUTDOWN_BOTH:
3068     default:
3069         if (rdmain) {
3070             rdmain->error_state = -1;
3071         }
3072         if (rdmaout) {
3073             rdmaout->error_state = -1;
3074         }
3075         break;
3076     }
3077 
3078     rcu_read_unlock();
3079     return 0;
3080 }
3081 
3082 /*
3083  * Parameters:
3084  *    @offset == 0 :
3085  *        This means that 'block_offset' is a full virtual address that does not
3086  *        belong to a RAMBlock of the virtual machine and instead
3087  *        represents a private malloc'd memory area that the caller wishes to
3088  *        transfer.
3089  *
3090  *    @offset != 0 :
3091  *        Offset is an offset to be added to block_offset and used
3092  *        to also lookup the corresponding RAMBlock.
3093  *
3094  *    @size > 0 :
3095  *        Initiate an transfer this size.
3096  *
3097  *    @size == 0 :
3098  *        A 'hint' or 'advice' that means that we wish to speculatively
3099  *        and asynchronously unregister this memory. In this case, there is no
3100  *        guarantee that the unregister will actually happen, for example,
3101  *        if the memory is being actively transmitted. Additionally, the memory
3102  *        may be re-registered at any future time if a write within the same
3103  *        chunk was requested again, even if you attempted to unregister it
3104  *        here.
3105  *
3106  *    @size < 0 : TODO, not yet supported
3107  *        Unregister the memory NOW. This means that the caller does not
3108  *        expect there to be any future RDMA transfers and we just want to clean
3109  *        things up. This is used in case the upper layer owns the memory and
3110  *        cannot wait for qemu_fclose() to occur.
3111  *
3112  *    @bytes_sent : User-specificed pointer to indicate how many bytes were
3113  *                  sent. Usually, this will not be more than a few bytes of
3114  *                  the protocol because most transfers are sent asynchronously.
3115  */
3116 static size_t qemu_rdma_save_page(QEMUFile *f, void *opaque,
3117                                   ram_addr_t block_offset, ram_addr_t offset,
3118                                   size_t size, uint64_t *bytes_sent)
3119 {
3120     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(opaque);
3121     RDMAContext *rdma;
3122     int ret;
3123 
3124     rcu_read_lock();
3125     rdma = atomic_rcu_read(&rioc->rdmaout);
3126 
3127     if (!rdma) {
3128         rcu_read_unlock();
3129         return -EIO;
3130     }
3131 
3132     CHECK_ERROR_STATE();
3133 
3134     if (migrate_get_current()->state == MIGRATION_STATUS_POSTCOPY_ACTIVE) {
3135         rcu_read_unlock();
3136         return RAM_SAVE_CONTROL_NOT_SUPP;
3137     }
3138 
3139     qemu_fflush(f);
3140 
3141     if (size > 0) {
3142         /*
3143          * Add this page to the current 'chunk'. If the chunk
3144          * is full, or the page doen't belong to the current chunk,
3145          * an actual RDMA write will occur and a new chunk will be formed.
3146          */
3147         ret = qemu_rdma_write(f, rdma, block_offset, offset, size);
3148         if (ret < 0) {
3149             error_report("rdma migration: write error! %d", ret);
3150             goto err;
3151         }
3152 
3153         /*
3154          * We always return 1 bytes because the RDMA
3155          * protocol is completely asynchronous. We do not yet know
3156          * whether an  identified chunk is zero or not because we're
3157          * waiting for other pages to potentially be merged with
3158          * the current chunk. So, we have to call qemu_update_position()
3159          * later on when the actual write occurs.
3160          */
3161         if (bytes_sent) {
3162             *bytes_sent = 1;
3163         }
3164     } else {
3165         uint64_t index, chunk;
3166 
3167         /* TODO: Change QEMUFileOps prototype to be signed: size_t => long
3168         if (size < 0) {
3169             ret = qemu_rdma_drain_cq(f, rdma);
3170             if (ret < 0) {
3171                 fprintf(stderr, "rdma: failed to synchronously drain"
3172                                 " completion queue before unregistration.\n");
3173                 goto err;
3174             }
3175         }
3176         */
3177 
3178         ret = qemu_rdma_search_ram_block(rdma, block_offset,
3179                                          offset, size, &index, &chunk);
3180 
3181         if (ret) {
3182             error_report("ram block search failed");
3183             goto err;
3184         }
3185 
3186         qemu_rdma_signal_unregister(rdma, index, chunk, 0);
3187 
3188         /*
3189          * TODO: Synchronous, guaranteed unregistration (should not occur during
3190          * fast-path). Otherwise, unregisters will process on the next call to
3191          * qemu_rdma_drain_cq()
3192         if (size < 0) {
3193             qemu_rdma_unregister_waiting(rdma);
3194         }
3195         */
3196     }
3197 
3198     /*
3199      * Drain the Completion Queue if possible, but do not block,
3200      * just poll.
3201      *
3202      * If nothing to poll, the end of the iteration will do this
3203      * again to make sure we don't overflow the request queue.
3204      */
3205     while (1) {
3206         uint64_t wr_id, wr_id_in;
3207         int ret = qemu_rdma_poll(rdma, &wr_id_in, NULL);
3208         if (ret < 0) {
3209             error_report("rdma migration: polling error! %d", ret);
3210             goto err;
3211         }
3212 
3213         wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
3214 
3215         if (wr_id == RDMA_WRID_NONE) {
3216             break;
3217         }
3218     }
3219 
3220     rcu_read_unlock();
3221     return RAM_SAVE_CONTROL_DELAYED;
3222 err:
3223     rdma->error_state = ret;
3224     rcu_read_unlock();
3225     return ret;
3226 }
3227 
3228 static void rdma_accept_incoming_migration(void *opaque);
3229 
3230 static void rdma_cm_poll_handler(void *opaque)
3231 {
3232     RDMAContext *rdma = opaque;
3233     int ret;
3234     struct rdma_cm_event *cm_event;
3235     MigrationIncomingState *mis = migration_incoming_get_current();
3236 
3237     ret = rdma_get_cm_event(rdma->channel, &cm_event);
3238     if (ret) {
3239         error_report("get_cm_event failed %d", errno);
3240         return;
3241     }
3242     rdma_ack_cm_event(cm_event);
3243 
3244     if (cm_event->event == RDMA_CM_EVENT_DISCONNECTED ||
3245         cm_event->event == RDMA_CM_EVENT_DEVICE_REMOVAL) {
3246         error_report("receive cm event, cm event is %d", cm_event->event);
3247         rdma->error_state = -EPIPE;
3248         if (rdma->return_path) {
3249             rdma->return_path->error_state = -EPIPE;
3250         }
3251 
3252         if (mis->migration_incoming_co) {
3253             qemu_coroutine_enter(mis->migration_incoming_co);
3254         }
3255         return;
3256     }
3257 }
3258 
3259 static int qemu_rdma_accept(RDMAContext *rdma)
3260 {
3261     RDMACapabilities cap;
3262     struct rdma_conn_param conn_param = {
3263                                             .responder_resources = 2,
3264                                             .private_data = &cap,
3265                                             .private_data_len = sizeof(cap),
3266                                          };
3267     struct rdma_cm_event *cm_event;
3268     struct ibv_context *verbs;
3269     int ret = -EINVAL;
3270     int idx;
3271 
3272     ret = rdma_get_cm_event(rdma->channel, &cm_event);
3273     if (ret) {
3274         goto err_rdma_dest_wait;
3275     }
3276 
3277     if (cm_event->event != RDMA_CM_EVENT_CONNECT_REQUEST) {
3278         rdma_ack_cm_event(cm_event);
3279         goto err_rdma_dest_wait;
3280     }
3281 
3282     memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
3283 
3284     network_to_caps(&cap);
3285 
3286     if (cap.version < 1 || cap.version > RDMA_CONTROL_VERSION_CURRENT) {
3287             error_report("Unknown source RDMA version: %d, bailing...",
3288                             cap.version);
3289             rdma_ack_cm_event(cm_event);
3290             goto err_rdma_dest_wait;
3291     }
3292 
3293     /*
3294      * Respond with only the capabilities this version of QEMU knows about.
3295      */
3296     cap.flags &= known_capabilities;
3297 
3298     /*
3299      * Enable the ones that we do know about.
3300      * Add other checks here as new ones are introduced.
3301      */
3302     if (cap.flags & RDMA_CAPABILITY_PIN_ALL) {
3303         rdma->pin_all = true;
3304     }
3305 
3306     rdma->cm_id = cm_event->id;
3307     verbs = cm_event->id->verbs;
3308 
3309     rdma_ack_cm_event(cm_event);
3310 
3311     trace_qemu_rdma_accept_pin_state(rdma->pin_all);
3312 
3313     caps_to_network(&cap);
3314 
3315     trace_qemu_rdma_accept_pin_verbsc(verbs);
3316 
3317     if (!rdma->verbs) {
3318         rdma->verbs = verbs;
3319     } else if (rdma->verbs != verbs) {
3320             error_report("ibv context not matching %p, %p!", rdma->verbs,
3321                          verbs);
3322             goto err_rdma_dest_wait;
3323     }
3324 
3325     qemu_rdma_dump_id("dest_init", verbs);
3326 
3327     ret = qemu_rdma_alloc_pd_cq(rdma);
3328     if (ret) {
3329         error_report("rdma migration: error allocating pd and cq!");
3330         goto err_rdma_dest_wait;
3331     }
3332 
3333     ret = qemu_rdma_alloc_qp(rdma);
3334     if (ret) {
3335         error_report("rdma migration: error allocating qp!");
3336         goto err_rdma_dest_wait;
3337     }
3338 
3339     ret = qemu_rdma_init_ram_blocks(rdma);
3340     if (ret) {
3341         error_report("rdma migration: error initializing ram blocks!");
3342         goto err_rdma_dest_wait;
3343     }
3344 
3345     for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
3346         ret = qemu_rdma_reg_control(rdma, idx);
3347         if (ret) {
3348             error_report("rdma: error registering %d control", idx);
3349             goto err_rdma_dest_wait;
3350         }
3351     }
3352 
3353     /* Accept the second connection request for return path */
3354     if (migrate_postcopy() && !rdma->is_return_path) {
3355         qemu_set_fd_handler(rdma->channel->fd, rdma_accept_incoming_migration,
3356                             NULL,
3357                             (void *)(intptr_t)rdma->return_path);
3358     } else {
3359         qemu_set_fd_handler(rdma->channel->fd, rdma_cm_poll_handler,
3360                             NULL, rdma);
3361     }
3362 
3363     ret = rdma_accept(rdma->cm_id, &conn_param);
3364     if (ret) {
3365         error_report("rdma_accept returns %d", ret);
3366         goto err_rdma_dest_wait;
3367     }
3368 
3369     ret = rdma_get_cm_event(rdma->channel, &cm_event);
3370     if (ret) {
3371         error_report("rdma_accept get_cm_event failed %d", ret);
3372         goto err_rdma_dest_wait;
3373     }
3374 
3375     if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
3376         error_report("rdma_accept not event established");
3377         rdma_ack_cm_event(cm_event);
3378         goto err_rdma_dest_wait;
3379     }
3380 
3381     rdma_ack_cm_event(cm_event);
3382     rdma->connected = true;
3383 
3384     ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
3385     if (ret) {
3386         error_report("rdma migration: error posting second control recv");
3387         goto err_rdma_dest_wait;
3388     }
3389 
3390     qemu_rdma_dump_gid("dest_connect", rdma->cm_id);
3391 
3392     return 0;
3393 
3394 err_rdma_dest_wait:
3395     rdma->error_state = ret;
3396     qemu_rdma_cleanup(rdma);
3397     return ret;
3398 }
3399 
3400 static int dest_ram_sort_func(const void *a, const void *b)
3401 {
3402     unsigned int a_index = ((const RDMALocalBlock *)a)->src_index;
3403     unsigned int b_index = ((const RDMALocalBlock *)b)->src_index;
3404 
3405     return (a_index < b_index) ? -1 : (a_index != b_index);
3406 }
3407 
3408 /*
3409  * During each iteration of the migration, we listen for instructions
3410  * by the source VM to perform dynamic page registrations before they
3411  * can perform RDMA operations.
3412  *
3413  * We respond with the 'rkey'.
3414  *
3415  * Keep doing this until the source tells us to stop.
3416  */
3417 static int qemu_rdma_registration_handle(QEMUFile *f, void *opaque)
3418 {
3419     RDMAControlHeader reg_resp = { .len = sizeof(RDMARegisterResult),
3420                                .type = RDMA_CONTROL_REGISTER_RESULT,
3421                                .repeat = 0,
3422                              };
3423     RDMAControlHeader unreg_resp = { .len = 0,
3424                                .type = RDMA_CONTROL_UNREGISTER_FINISHED,
3425                                .repeat = 0,
3426                              };
3427     RDMAControlHeader blocks = { .type = RDMA_CONTROL_RAM_BLOCKS_RESULT,
3428                                  .repeat = 1 };
3429     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(opaque);
3430     RDMAContext *rdma;
3431     RDMALocalBlocks *local;
3432     RDMAControlHeader head;
3433     RDMARegister *reg, *registers;
3434     RDMACompress *comp;
3435     RDMARegisterResult *reg_result;
3436     static RDMARegisterResult results[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE];
3437     RDMALocalBlock *block;
3438     void *host_addr;
3439     int ret = 0;
3440     int idx = 0;
3441     int count = 0;
3442     int i = 0;
3443 
3444     rcu_read_lock();
3445     rdma = atomic_rcu_read(&rioc->rdmain);
3446 
3447     if (!rdma) {
3448         rcu_read_unlock();
3449         return -EIO;
3450     }
3451 
3452     CHECK_ERROR_STATE();
3453 
3454     local = &rdma->local_ram_blocks;
3455     do {
3456         trace_qemu_rdma_registration_handle_wait();
3457 
3458         ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_NONE);
3459 
3460         if (ret < 0) {
3461             break;
3462         }
3463 
3464         if (head.repeat > RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE) {
3465             error_report("rdma: Too many requests in this message (%d)."
3466                             "Bailing.", head.repeat);
3467             ret = -EIO;
3468             break;
3469         }
3470 
3471         switch (head.type) {
3472         case RDMA_CONTROL_COMPRESS:
3473             comp = (RDMACompress *) rdma->wr_data[idx].control_curr;
3474             network_to_compress(comp);
3475 
3476             trace_qemu_rdma_registration_handle_compress(comp->length,
3477                                                          comp->block_idx,
3478                                                          comp->offset);
3479             if (comp->block_idx >= rdma->local_ram_blocks.nb_blocks) {
3480                 error_report("rdma: 'compress' bad block index %u (vs %d)",
3481                              (unsigned int)comp->block_idx,
3482                              rdma->local_ram_blocks.nb_blocks);
3483                 ret = -EIO;
3484                 goto out;
3485             }
3486             block = &(rdma->local_ram_blocks.block[comp->block_idx]);
3487 
3488             host_addr = block->local_host_addr +
3489                             (comp->offset - block->offset);
3490 
3491             ram_handle_compressed(host_addr, comp->value, comp->length);
3492             break;
3493 
3494         case RDMA_CONTROL_REGISTER_FINISHED:
3495             trace_qemu_rdma_registration_handle_finished();
3496             goto out;
3497 
3498         case RDMA_CONTROL_RAM_BLOCKS_REQUEST:
3499             trace_qemu_rdma_registration_handle_ram_blocks();
3500 
3501             /* Sort our local RAM Block list so it's the same as the source,
3502              * we can do this since we've filled in a src_index in the list
3503              * as we received the RAMBlock list earlier.
3504              */
3505             qsort(rdma->local_ram_blocks.block,
3506                   rdma->local_ram_blocks.nb_blocks,
3507                   sizeof(RDMALocalBlock), dest_ram_sort_func);
3508             for (i = 0; i < local->nb_blocks; i++) {
3509                 local->block[i].index = i;
3510             }
3511 
3512             if (rdma->pin_all) {
3513                 ret = qemu_rdma_reg_whole_ram_blocks(rdma);
3514                 if (ret) {
3515                     error_report("rdma migration: error dest "
3516                                     "registering ram blocks");
3517                     goto out;
3518                 }
3519             }
3520 
3521             /*
3522              * Dest uses this to prepare to transmit the RAMBlock descriptions
3523              * to the source VM after connection setup.
3524              * Both sides use the "remote" structure to communicate and update
3525              * their "local" descriptions with what was sent.
3526              */
3527             for (i = 0; i < local->nb_blocks; i++) {
3528                 rdma->dest_blocks[i].remote_host_addr =
3529                     (uintptr_t)(local->block[i].local_host_addr);
3530 
3531                 if (rdma->pin_all) {
3532                     rdma->dest_blocks[i].remote_rkey = local->block[i].mr->rkey;
3533                 }
3534 
3535                 rdma->dest_blocks[i].offset = local->block[i].offset;
3536                 rdma->dest_blocks[i].length = local->block[i].length;
3537 
3538                 dest_block_to_network(&rdma->dest_blocks[i]);
3539                 trace_qemu_rdma_registration_handle_ram_blocks_loop(
3540                     local->block[i].block_name,
3541                     local->block[i].offset,
3542                     local->block[i].length,
3543                     local->block[i].local_host_addr,
3544                     local->block[i].src_index);
3545             }
3546 
3547             blocks.len = rdma->local_ram_blocks.nb_blocks
3548                                                 * sizeof(RDMADestBlock);
3549 
3550 
3551             ret = qemu_rdma_post_send_control(rdma,
3552                                         (uint8_t *) rdma->dest_blocks, &blocks);
3553 
3554             if (ret < 0) {
3555                 error_report("rdma migration: error sending remote info");
3556                 goto out;
3557             }
3558 
3559             break;
3560         case RDMA_CONTROL_REGISTER_REQUEST:
3561             trace_qemu_rdma_registration_handle_register(head.repeat);
3562 
3563             reg_resp.repeat = head.repeat;
3564             registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
3565 
3566             for (count = 0; count < head.repeat; count++) {
3567                 uint64_t chunk;
3568                 uint8_t *chunk_start, *chunk_end;
3569 
3570                 reg = &registers[count];
3571                 network_to_register(reg);
3572 
3573                 reg_result = &results[count];
3574 
3575                 trace_qemu_rdma_registration_handle_register_loop(count,
3576                          reg->current_index, reg->key.current_addr, reg->chunks);
3577 
3578                 if (reg->current_index >= rdma->local_ram_blocks.nb_blocks) {
3579                     error_report("rdma: 'register' bad block index %u (vs %d)",
3580                                  (unsigned int)reg->current_index,
3581                                  rdma->local_ram_blocks.nb_blocks);
3582                     ret = -ENOENT;
3583                     goto out;
3584                 }
3585                 block = &(rdma->local_ram_blocks.block[reg->current_index]);
3586                 if (block->is_ram_block) {
3587                     if (block->offset > reg->key.current_addr) {
3588                         error_report("rdma: bad register address for block %s"
3589                             " offset: %" PRIx64 " current_addr: %" PRIx64,
3590                             block->block_name, block->offset,
3591                             reg->key.current_addr);
3592                         ret = -ERANGE;
3593                         goto out;
3594                     }
3595                     host_addr = (block->local_host_addr +
3596                                 (reg->key.current_addr - block->offset));
3597                     chunk = ram_chunk_index(block->local_host_addr,
3598                                             (uint8_t *) host_addr);
3599                 } else {
3600                     chunk = reg->key.chunk;
3601                     host_addr = block->local_host_addr +
3602                         (reg->key.chunk * (1UL << RDMA_REG_CHUNK_SHIFT));
3603                     /* Check for particularly bad chunk value */
3604                     if (host_addr < (void *)block->local_host_addr) {
3605                         error_report("rdma: bad chunk for block %s"
3606                             " chunk: %" PRIx64,
3607                             block->block_name, reg->key.chunk);
3608                         ret = -ERANGE;
3609                         goto out;
3610                     }
3611                 }
3612                 chunk_start = ram_chunk_start(block, chunk);
3613                 chunk_end = ram_chunk_end(block, chunk + reg->chunks);
3614                 if (qemu_rdma_register_and_get_keys(rdma, block,
3615                             (uintptr_t)host_addr, NULL, &reg_result->rkey,
3616                             chunk, chunk_start, chunk_end)) {
3617                     error_report("cannot get rkey");
3618                     ret = -EINVAL;
3619                     goto out;
3620                 }
3621 
3622                 reg_result->host_addr = (uintptr_t)block->local_host_addr;
3623 
3624                 trace_qemu_rdma_registration_handle_register_rkey(
3625                                                            reg_result->rkey);
3626 
3627                 result_to_network(reg_result);
3628             }
3629 
3630             ret = qemu_rdma_post_send_control(rdma,
3631                             (uint8_t *) results, &reg_resp);
3632 
3633             if (ret < 0) {
3634                 error_report("Failed to send control buffer");
3635                 goto out;
3636             }
3637             break;
3638         case RDMA_CONTROL_UNREGISTER_REQUEST:
3639             trace_qemu_rdma_registration_handle_unregister(head.repeat);
3640             unreg_resp.repeat = head.repeat;
3641             registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
3642 
3643             for (count = 0; count < head.repeat; count++) {
3644                 reg = &registers[count];
3645                 network_to_register(reg);
3646 
3647                 trace_qemu_rdma_registration_handle_unregister_loop(count,
3648                            reg->current_index, reg->key.chunk);
3649 
3650                 block = &(rdma->local_ram_blocks.block[reg->current_index]);
3651 
3652                 ret = ibv_dereg_mr(block->pmr[reg->key.chunk]);
3653                 block->pmr[reg->key.chunk] = NULL;
3654 
3655                 if (ret != 0) {
3656                     perror("rdma unregistration chunk failed");
3657                     ret = -ret;
3658                     goto out;
3659                 }
3660 
3661                 rdma->total_registrations--;
3662 
3663                 trace_qemu_rdma_registration_handle_unregister_success(
3664                                                        reg->key.chunk);
3665             }
3666 
3667             ret = qemu_rdma_post_send_control(rdma, NULL, &unreg_resp);
3668 
3669             if (ret < 0) {
3670                 error_report("Failed to send control buffer");
3671                 goto out;
3672             }
3673             break;
3674         case RDMA_CONTROL_REGISTER_RESULT:
3675             error_report("Invalid RESULT message at dest.");
3676             ret = -EIO;
3677             goto out;
3678         default:
3679             error_report("Unknown control message %s", control_desc(head.type));
3680             ret = -EIO;
3681             goto out;
3682         }
3683     } while (1);
3684 out:
3685     if (ret < 0) {
3686         rdma->error_state = ret;
3687     }
3688     rcu_read_unlock();
3689     return ret;
3690 }
3691 
3692 /* Destination:
3693  * Called via a ram_control_load_hook during the initial RAM load section which
3694  * lists the RAMBlocks by name.  This lets us know the order of the RAMBlocks
3695  * on the source.
3696  * We've already built our local RAMBlock list, but not yet sent the list to
3697  * the source.
3698  */
3699 static int
3700 rdma_block_notification_handle(QIOChannelRDMA *rioc, const char *name)
3701 {
3702     RDMAContext *rdma;
3703     int curr;
3704     int found = -1;
3705 
3706     rcu_read_lock();
3707     rdma = atomic_rcu_read(&rioc->rdmain);
3708 
3709     if (!rdma) {
3710         rcu_read_unlock();
3711         return -EIO;
3712     }
3713 
3714     /* Find the matching RAMBlock in our local list */
3715     for (curr = 0; curr < rdma->local_ram_blocks.nb_blocks; curr++) {
3716         if (!strcmp(rdma->local_ram_blocks.block[curr].block_name, name)) {
3717             found = curr;
3718             break;
3719         }
3720     }
3721 
3722     if (found == -1) {
3723         error_report("RAMBlock '%s' not found on destination", name);
3724         rcu_read_unlock();
3725         return -ENOENT;
3726     }
3727 
3728     rdma->local_ram_blocks.block[curr].src_index = rdma->next_src_index;
3729     trace_rdma_block_notification_handle(name, rdma->next_src_index);
3730     rdma->next_src_index++;
3731 
3732     rcu_read_unlock();
3733     return 0;
3734 }
3735 
3736 static int rdma_load_hook(QEMUFile *f, void *opaque, uint64_t flags, void *data)
3737 {
3738     switch (flags) {
3739     case RAM_CONTROL_BLOCK_REG:
3740         return rdma_block_notification_handle(opaque, data);
3741 
3742     case RAM_CONTROL_HOOK:
3743         return qemu_rdma_registration_handle(f, opaque);
3744 
3745     default:
3746         /* Shouldn't be called with any other values */
3747         abort();
3748     }
3749 }
3750 
3751 static int qemu_rdma_registration_start(QEMUFile *f, void *opaque,
3752                                         uint64_t flags, void *data)
3753 {
3754     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(opaque);
3755     RDMAContext *rdma;
3756 
3757     rcu_read_lock();
3758     rdma = atomic_rcu_read(&rioc->rdmaout);
3759     if (!rdma) {
3760         rcu_read_unlock();
3761         return -EIO;
3762     }
3763 
3764     CHECK_ERROR_STATE();
3765 
3766     if (migrate_get_current()->state == MIGRATION_STATUS_POSTCOPY_ACTIVE) {
3767         rcu_read_unlock();
3768         return 0;
3769     }
3770 
3771     trace_qemu_rdma_registration_start(flags);
3772     qemu_put_be64(f, RAM_SAVE_FLAG_HOOK);
3773     qemu_fflush(f);
3774 
3775     rcu_read_unlock();
3776     return 0;
3777 }
3778 
3779 /*
3780  * Inform dest that dynamic registrations are done for now.
3781  * First, flush writes, if any.
3782  */
3783 static int qemu_rdma_registration_stop(QEMUFile *f, void *opaque,
3784                                        uint64_t flags, void *data)
3785 {
3786     Error *local_err = NULL, **errp = &local_err;
3787     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(opaque);
3788     RDMAContext *rdma;
3789     RDMAControlHeader head = { .len = 0, .repeat = 1 };
3790     int ret = 0;
3791 
3792     rcu_read_lock();
3793     rdma = atomic_rcu_read(&rioc->rdmaout);
3794     if (!rdma) {
3795         rcu_read_unlock();
3796         return -EIO;
3797     }
3798 
3799     CHECK_ERROR_STATE();
3800 
3801     if (migrate_get_current()->state == MIGRATION_STATUS_POSTCOPY_ACTIVE) {
3802         rcu_read_unlock();
3803         return 0;
3804     }
3805 
3806     qemu_fflush(f);
3807     ret = qemu_rdma_drain_cq(f, rdma);
3808 
3809     if (ret < 0) {
3810         goto err;
3811     }
3812 
3813     if (flags == RAM_CONTROL_SETUP) {
3814         RDMAControlHeader resp = {.type = RDMA_CONTROL_RAM_BLOCKS_RESULT };
3815         RDMALocalBlocks *local = &rdma->local_ram_blocks;
3816         int reg_result_idx, i, nb_dest_blocks;
3817 
3818         head.type = RDMA_CONTROL_RAM_BLOCKS_REQUEST;
3819         trace_qemu_rdma_registration_stop_ram();
3820 
3821         /*
3822          * Make sure that we parallelize the pinning on both sides.
3823          * For very large guests, doing this serially takes a really
3824          * long time, so we have to 'interleave' the pinning locally
3825          * with the control messages by performing the pinning on this
3826          * side before we receive the control response from the other
3827          * side that the pinning has completed.
3828          */
3829         ret = qemu_rdma_exchange_send(rdma, &head, NULL, &resp,
3830                     &reg_result_idx, rdma->pin_all ?
3831                     qemu_rdma_reg_whole_ram_blocks : NULL);
3832         if (ret < 0) {
3833             ERROR(errp, "receiving remote info!");
3834             rcu_read_unlock();
3835             return ret;
3836         }
3837 
3838         nb_dest_blocks = resp.len / sizeof(RDMADestBlock);
3839 
3840         /*
3841          * The protocol uses two different sets of rkeys (mutually exclusive):
3842          * 1. One key to represent the virtual address of the entire ram block.
3843          *    (dynamic chunk registration disabled - pin everything with one rkey.)
3844          * 2. One to represent individual chunks within a ram block.
3845          *    (dynamic chunk registration enabled - pin individual chunks.)
3846          *
3847          * Once the capability is successfully negotiated, the destination transmits
3848          * the keys to use (or sends them later) including the virtual addresses
3849          * and then propagates the remote ram block descriptions to his local copy.
3850          */
3851 
3852         if (local->nb_blocks != nb_dest_blocks) {
3853             ERROR(errp, "ram blocks mismatch (Number of blocks %d vs %d) "
3854                         "Your QEMU command line parameters are probably "
3855                         "not identical on both the source and destination.",
3856                         local->nb_blocks, nb_dest_blocks);
3857             rdma->error_state = -EINVAL;
3858             rcu_read_unlock();
3859             return -EINVAL;
3860         }
3861 
3862         qemu_rdma_move_header(rdma, reg_result_idx, &resp);
3863         memcpy(rdma->dest_blocks,
3864             rdma->wr_data[reg_result_idx].control_curr, resp.len);
3865         for (i = 0; i < nb_dest_blocks; i++) {
3866             network_to_dest_block(&rdma->dest_blocks[i]);
3867 
3868             /* We require that the blocks are in the same order */
3869             if (rdma->dest_blocks[i].length != local->block[i].length) {
3870                 ERROR(errp, "Block %s/%d has a different length %" PRIu64
3871                             "vs %" PRIu64, local->block[i].block_name, i,
3872                             local->block[i].length,
3873                             rdma->dest_blocks[i].length);
3874                 rdma->error_state = -EINVAL;
3875                 rcu_read_unlock();
3876                 return -EINVAL;
3877             }
3878             local->block[i].remote_host_addr =
3879                     rdma->dest_blocks[i].remote_host_addr;
3880             local->block[i].remote_rkey = rdma->dest_blocks[i].remote_rkey;
3881         }
3882     }
3883 
3884     trace_qemu_rdma_registration_stop(flags);
3885 
3886     head.type = RDMA_CONTROL_REGISTER_FINISHED;
3887     ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL);
3888 
3889     if (ret < 0) {
3890         goto err;
3891     }
3892 
3893     rcu_read_unlock();
3894     return 0;
3895 err:
3896     rdma->error_state = ret;
3897     rcu_read_unlock();
3898     return ret;
3899 }
3900 
3901 static const QEMUFileHooks rdma_read_hooks = {
3902     .hook_ram_load = rdma_load_hook,
3903 };
3904 
3905 static const QEMUFileHooks rdma_write_hooks = {
3906     .before_ram_iterate = qemu_rdma_registration_start,
3907     .after_ram_iterate  = qemu_rdma_registration_stop,
3908     .save_page          = qemu_rdma_save_page,
3909 };
3910 
3911 
3912 static void qio_channel_rdma_finalize(Object *obj)
3913 {
3914     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(obj);
3915     if (rioc->rdmain) {
3916         qemu_rdma_cleanup(rioc->rdmain);
3917         g_free(rioc->rdmain);
3918         rioc->rdmain = NULL;
3919     }
3920     if (rioc->rdmaout) {
3921         qemu_rdma_cleanup(rioc->rdmaout);
3922         g_free(rioc->rdmaout);
3923         rioc->rdmaout = NULL;
3924     }
3925 }
3926 
3927 static void qio_channel_rdma_class_init(ObjectClass *klass,
3928                                         void *class_data G_GNUC_UNUSED)
3929 {
3930     QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
3931 
3932     ioc_klass->io_writev = qio_channel_rdma_writev;
3933     ioc_klass->io_readv = qio_channel_rdma_readv;
3934     ioc_klass->io_set_blocking = qio_channel_rdma_set_blocking;
3935     ioc_klass->io_close = qio_channel_rdma_close;
3936     ioc_klass->io_create_watch = qio_channel_rdma_create_watch;
3937     ioc_klass->io_set_aio_fd_handler = qio_channel_rdma_set_aio_fd_handler;
3938     ioc_klass->io_shutdown = qio_channel_rdma_shutdown;
3939 }
3940 
3941 static const TypeInfo qio_channel_rdma_info = {
3942     .parent = TYPE_QIO_CHANNEL,
3943     .name = TYPE_QIO_CHANNEL_RDMA,
3944     .instance_size = sizeof(QIOChannelRDMA),
3945     .instance_finalize = qio_channel_rdma_finalize,
3946     .class_init = qio_channel_rdma_class_init,
3947 };
3948 
3949 static void qio_channel_rdma_register_types(void)
3950 {
3951     type_register_static(&qio_channel_rdma_info);
3952 }
3953 
3954 type_init(qio_channel_rdma_register_types);
3955 
3956 static QEMUFile *qemu_fopen_rdma(RDMAContext *rdma, const char *mode)
3957 {
3958     QIOChannelRDMA *rioc;
3959 
3960     if (qemu_file_mode_is_not_valid(mode)) {
3961         return NULL;
3962     }
3963 
3964     rioc = QIO_CHANNEL_RDMA(object_new(TYPE_QIO_CHANNEL_RDMA));
3965 
3966     if (mode[0] == 'w') {
3967         rioc->file = qemu_fopen_channel_output(QIO_CHANNEL(rioc));
3968         rioc->rdmaout = rdma;
3969         rioc->rdmain = rdma->return_path;
3970         qemu_file_set_hooks(rioc->file, &rdma_write_hooks);
3971     } else {
3972         rioc->file = qemu_fopen_channel_input(QIO_CHANNEL(rioc));
3973         rioc->rdmain = rdma;
3974         rioc->rdmaout = rdma->return_path;
3975         qemu_file_set_hooks(rioc->file, &rdma_read_hooks);
3976     }
3977 
3978     return rioc->file;
3979 }
3980 
3981 static void rdma_accept_incoming_migration(void *opaque)
3982 {
3983     RDMAContext *rdma = opaque;
3984     int ret;
3985     QEMUFile *f;
3986     Error *local_err = NULL, **errp = &local_err;
3987 
3988     trace_qemu_rdma_accept_incoming_migration();
3989     ret = qemu_rdma_accept(rdma);
3990 
3991     if (ret) {
3992         ERROR(errp, "RDMA Migration initialization failed!");
3993         return;
3994     }
3995 
3996     trace_qemu_rdma_accept_incoming_migration_accepted();
3997 
3998     if (rdma->is_return_path) {
3999         return;
4000     }
4001 
4002     f = qemu_fopen_rdma(rdma, "rb");
4003     if (f == NULL) {
4004         ERROR(errp, "could not qemu_fopen_rdma!");
4005         qemu_rdma_cleanup(rdma);
4006         return;
4007     }
4008 
4009     rdma->migration_started_on_destination = 1;
4010     migration_fd_process_incoming(f);
4011 }
4012 
4013 void rdma_start_incoming_migration(const char *host_port, Error **errp)
4014 {
4015     int ret;
4016     RDMAContext *rdma, *rdma_return_path = NULL;
4017     Error *local_err = NULL;
4018 
4019     trace_rdma_start_incoming_migration();
4020     rdma = qemu_rdma_data_init(host_port, &local_err);
4021 
4022     if (rdma == NULL) {
4023         goto err;
4024     }
4025 
4026     ret = qemu_rdma_dest_init(rdma, &local_err);
4027 
4028     if (ret) {
4029         goto err;
4030     }
4031 
4032     trace_rdma_start_incoming_migration_after_dest_init();
4033 
4034     ret = rdma_listen(rdma->listen_id, 5);
4035 
4036     if (ret) {
4037         ERROR(errp, "listening on socket!");
4038         goto err;
4039     }
4040 
4041     trace_rdma_start_incoming_migration_after_rdma_listen();
4042 
4043     /* initialize the RDMAContext for return path */
4044     if (migrate_postcopy()) {
4045         rdma_return_path = qemu_rdma_data_init(host_port, &local_err);
4046 
4047         if (rdma_return_path == NULL) {
4048             goto err;
4049         }
4050 
4051         qemu_rdma_return_path_dest_init(rdma_return_path, rdma);
4052     }
4053 
4054     qemu_set_fd_handler(rdma->channel->fd, rdma_accept_incoming_migration,
4055                         NULL, (void *)(intptr_t)rdma);
4056     return;
4057 err:
4058     error_propagate(errp, local_err);
4059     g_free(rdma);
4060     g_free(rdma_return_path);
4061 }
4062 
4063 void rdma_start_outgoing_migration(void *opaque,
4064                             const char *host_port, Error **errp)
4065 {
4066     MigrationState *s = opaque;
4067     RDMAContext *rdma = qemu_rdma_data_init(host_port, errp);
4068     RDMAContext *rdma_return_path = NULL;
4069     int ret = 0;
4070 
4071     if (rdma == NULL) {
4072         goto err;
4073     }
4074 
4075     ret = qemu_rdma_source_init(rdma,
4076         s->enabled_capabilities[MIGRATION_CAPABILITY_RDMA_PIN_ALL], errp);
4077 
4078     if (ret) {
4079         goto err;
4080     }
4081 
4082     trace_rdma_start_outgoing_migration_after_rdma_source_init();
4083     ret = qemu_rdma_connect(rdma, errp);
4084 
4085     if (ret) {
4086         goto err;
4087     }
4088 
4089     /* RDMA postcopy need a seprate queue pair for return path */
4090     if (migrate_postcopy()) {
4091         rdma_return_path = qemu_rdma_data_init(host_port, errp);
4092 
4093         if (rdma_return_path == NULL) {
4094             goto err;
4095         }
4096 
4097         ret = qemu_rdma_source_init(rdma_return_path,
4098             s->enabled_capabilities[MIGRATION_CAPABILITY_RDMA_PIN_ALL], errp);
4099 
4100         if (ret) {
4101             goto err;
4102         }
4103 
4104         ret = qemu_rdma_connect(rdma_return_path, errp);
4105 
4106         if (ret) {
4107             goto err;
4108         }
4109 
4110         rdma->return_path = rdma_return_path;
4111         rdma_return_path->return_path = rdma;
4112         rdma_return_path->is_return_path = true;
4113     }
4114 
4115     trace_rdma_start_outgoing_migration_after_rdma_connect();
4116 
4117     s->to_dst_file = qemu_fopen_rdma(rdma, "wb");
4118     migrate_fd_connect(s, NULL);
4119     return;
4120 err:
4121     g_free(rdma);
4122     g_free(rdma_return_path);
4123 }
4124