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