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