hw/arm/musca: Add MPCs
[qemu/ar7.git] / migration / rdma.c
blob54a3c11540d947a741b97afdcef8cf68bc70da4a
1 /*
2 * RDMA protocol and interfaces
4 * Copyright IBM, Corp. 2010-2013
5 * Copyright Red Hat, Inc. 2015-2016
7 * Authors:
8 * Michael R. Hines <mrhines@us.ibm.com>
9 * Jiuxing Liu <jl@us.ibm.com>
10 * Daniel P. Berrange <berrange@redhat.com>
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.
16 #include "qemu/osdep.h"
17 #include "qapi/error.h"
18 #include "qemu-common.h"
19 #include "qemu/cutils.h"
20 #include "rdma.h"
21 #include "migration.h"
22 #include "qemu-file.h"
23 #include "ram.h"
24 #include "qemu-file-channel.h"
25 #include "qemu/error-report.h"
26 #include "qemu/main-loop.h"
27 #include "qemu/sockets.h"
28 #include "qemu/bitmap.h"
29 #include "qemu/coroutine.h"
30 #include <sys/socket.h>
31 #include <netdb.h>
32 #include <arpa/inet.h>
33 #include <rdma/rdma_cma.h>
34 #include "trace.h"
37 * Print and error on both the Monitor and the Log file.
39 #define ERROR(errp, fmt, ...) \
40 do { \
41 fprintf(stderr, "RDMA ERROR: " fmt "\n", ## __VA_ARGS__); \
42 if (errp && (*(errp) == NULL)) { \
43 error_setg(errp, "RDMA ERROR: " fmt, ## __VA_ARGS__); \
44 } \
45 } while (0)
47 #define RDMA_RESOLVE_TIMEOUT_MS 10000
49 /* Do not merge data if larger than this. */
50 #define RDMA_MERGE_MAX (2 * 1024 * 1024)
51 #define RDMA_SIGNALED_SEND_MAX (RDMA_MERGE_MAX / 4096)
53 #define RDMA_REG_CHUNK_SHIFT 20 /* 1 MB */
56 * This is only for non-live state being migrated.
57 * Instead of RDMA_WRITE messages, we use RDMA_SEND
58 * messages for that state, which requires a different
59 * delivery design than main memory.
61 #define RDMA_SEND_INCREMENT 32768
64 * Maximum size infiniband SEND message
66 #define RDMA_CONTROL_MAX_BUFFER (512 * 1024)
67 #define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096
69 #define RDMA_CONTROL_VERSION_CURRENT 1
71 * Capabilities for negotiation.
73 #define RDMA_CAPABILITY_PIN_ALL 0x01
76 * Add the other flags above to this list of known capabilities
77 * as they are introduced.
79 static uint32_t known_capabilities = RDMA_CAPABILITY_PIN_ALL;
81 #define CHECK_ERROR_STATE() \
82 do { \
83 if (rdma->error_state) { \
84 if (!rdma->error_reported) { \
85 error_report("RDMA is in an error state waiting migration" \
86 " to abort!"); \
87 rdma->error_reported = 1; \
88 } \
89 rcu_read_unlock(); \
90 return rdma->error_state; \
91 } \
92 } while (0)
95 * A work request ID is 64-bits and we split up these bits
96 * into 3 parts:
98 * bits 0-15 : type of control message, 2^16
99 * bits 16-29: ram block index, 2^14
100 * bits 30-63: ram block chunk number, 2^34
102 * The last two bit ranges are only used for RDMA writes,
103 * in order to track their completion and potentially
104 * also track unregistration status of the message.
106 #define RDMA_WRID_TYPE_SHIFT 0UL
107 #define RDMA_WRID_BLOCK_SHIFT 16UL
108 #define RDMA_WRID_CHUNK_SHIFT 30UL
110 #define RDMA_WRID_TYPE_MASK \
111 ((1UL << RDMA_WRID_BLOCK_SHIFT) - 1UL)
113 #define RDMA_WRID_BLOCK_MASK \
114 (~RDMA_WRID_TYPE_MASK & ((1UL << RDMA_WRID_CHUNK_SHIFT) - 1UL))
116 #define RDMA_WRID_CHUNK_MASK (~RDMA_WRID_BLOCK_MASK & ~RDMA_WRID_TYPE_MASK)
119 * RDMA migration protocol:
120 * 1. RDMA Writes (data messages, i.e. RAM)
121 * 2. IB Send/Recv (control channel messages)
123 enum {
124 RDMA_WRID_NONE = 0,
125 RDMA_WRID_RDMA_WRITE = 1,
126 RDMA_WRID_SEND_CONTROL = 2000,
127 RDMA_WRID_RECV_CONTROL = 4000,
130 static const char *wrid_desc[] = {
131 [RDMA_WRID_NONE] = "NONE",
132 [RDMA_WRID_RDMA_WRITE] = "WRITE RDMA",
133 [RDMA_WRID_SEND_CONTROL] = "CONTROL SEND",
134 [RDMA_WRID_RECV_CONTROL] = "CONTROL RECV",
138 * Work request IDs for IB SEND messages only (not RDMA writes).
139 * This is used by the migration protocol to transmit
140 * control messages (such as device state and registration commands)
142 * We could use more WRs, but we have enough for now.
144 enum {
145 RDMA_WRID_READY = 0,
146 RDMA_WRID_DATA,
147 RDMA_WRID_CONTROL,
148 RDMA_WRID_MAX,
152 * SEND/RECV IB Control Messages.
154 enum {
155 RDMA_CONTROL_NONE = 0,
156 RDMA_CONTROL_ERROR,
157 RDMA_CONTROL_READY, /* ready to receive */
158 RDMA_CONTROL_QEMU_FILE, /* QEMUFile-transmitted bytes */
159 RDMA_CONTROL_RAM_BLOCKS_REQUEST, /* RAMBlock synchronization */
160 RDMA_CONTROL_RAM_BLOCKS_RESULT, /* RAMBlock synchronization */
161 RDMA_CONTROL_COMPRESS, /* page contains repeat values */
162 RDMA_CONTROL_REGISTER_REQUEST, /* dynamic page registration */
163 RDMA_CONTROL_REGISTER_RESULT, /* key to use after registration */
164 RDMA_CONTROL_REGISTER_FINISHED, /* current iteration finished */
165 RDMA_CONTROL_UNREGISTER_REQUEST, /* dynamic UN-registration */
166 RDMA_CONTROL_UNREGISTER_FINISHED, /* unpinning finished */
171 * Memory and MR structures used to represent an IB Send/Recv work request.
172 * This is *not* used for RDMA writes, only IB Send/Recv.
174 typedef struct {
175 uint8_t control[RDMA_CONTROL_MAX_BUFFER]; /* actual buffer to register */
176 struct ibv_mr *control_mr; /* registration metadata */
177 size_t control_len; /* length of the message */
178 uint8_t *control_curr; /* start of unconsumed bytes */
179 } RDMAWorkRequestData;
182 * Negotiate RDMA capabilities during connection-setup time.
184 typedef struct {
185 uint32_t version;
186 uint32_t flags;
187 } RDMACapabilities;
189 static void caps_to_network(RDMACapabilities *cap)
191 cap->version = htonl(cap->version);
192 cap->flags = htonl(cap->flags);
195 static void network_to_caps(RDMACapabilities *cap)
197 cap->version = ntohl(cap->version);
198 cap->flags = ntohl(cap->flags);
202 * Representation of a RAMBlock from an RDMA perspective.
203 * This is not transmitted, only local.
204 * This and subsequent structures cannot be linked lists
205 * because we're using a single IB message to transmit
206 * the information. It's small anyway, so a list is overkill.
208 typedef struct RDMALocalBlock {
209 char *block_name;
210 uint8_t *local_host_addr; /* local virtual address */
211 uint64_t remote_host_addr; /* remote virtual address */
212 uint64_t offset;
213 uint64_t length;
214 struct ibv_mr **pmr; /* MRs for chunk-level registration */
215 struct ibv_mr *mr; /* MR for non-chunk-level registration */
216 uint32_t *remote_keys; /* rkeys for chunk-level registration */
217 uint32_t remote_rkey; /* rkeys for non-chunk-level registration */
218 int index; /* which block are we */
219 unsigned int src_index; /* (Only used on dest) */
220 bool is_ram_block;
221 int nb_chunks;
222 unsigned long *transit_bitmap;
223 unsigned long *unregister_bitmap;
224 } RDMALocalBlock;
227 * Also represents a RAMblock, but only on the dest.
228 * This gets transmitted by the dest during connection-time
229 * to the source VM and then is used to populate the
230 * corresponding RDMALocalBlock with
231 * the information needed to perform the actual RDMA.
233 typedef struct QEMU_PACKED RDMADestBlock {
234 uint64_t remote_host_addr;
235 uint64_t offset;
236 uint64_t length;
237 uint32_t remote_rkey;
238 uint32_t padding;
239 } RDMADestBlock;
241 static const char *control_desc(unsigned int rdma_control)
243 static const char *strs[] = {
244 [RDMA_CONTROL_NONE] = "NONE",
245 [RDMA_CONTROL_ERROR] = "ERROR",
246 [RDMA_CONTROL_READY] = "READY",
247 [RDMA_CONTROL_QEMU_FILE] = "QEMU FILE",
248 [RDMA_CONTROL_RAM_BLOCKS_REQUEST] = "RAM BLOCKS REQUEST",
249 [RDMA_CONTROL_RAM_BLOCKS_RESULT] = "RAM BLOCKS RESULT",
250 [RDMA_CONTROL_COMPRESS] = "COMPRESS",
251 [RDMA_CONTROL_REGISTER_REQUEST] = "REGISTER REQUEST",
252 [RDMA_CONTROL_REGISTER_RESULT] = "REGISTER RESULT",
253 [RDMA_CONTROL_REGISTER_FINISHED] = "REGISTER FINISHED",
254 [RDMA_CONTROL_UNREGISTER_REQUEST] = "UNREGISTER REQUEST",
255 [RDMA_CONTROL_UNREGISTER_FINISHED] = "UNREGISTER FINISHED",
258 if (rdma_control > RDMA_CONTROL_UNREGISTER_FINISHED) {
259 return "??BAD CONTROL VALUE??";
262 return strs[rdma_control];
265 static uint64_t htonll(uint64_t v)
267 union { uint32_t lv[2]; uint64_t llv; } u;
268 u.lv[0] = htonl(v >> 32);
269 u.lv[1] = htonl(v & 0xFFFFFFFFULL);
270 return u.llv;
273 static uint64_t ntohll(uint64_t v) {
274 union { uint32_t lv[2]; uint64_t llv; } u;
275 u.llv = v;
276 return ((uint64_t)ntohl(u.lv[0]) << 32) | (uint64_t) ntohl(u.lv[1]);
279 static void dest_block_to_network(RDMADestBlock *db)
281 db->remote_host_addr = htonll(db->remote_host_addr);
282 db->offset = htonll(db->offset);
283 db->length = htonll(db->length);
284 db->remote_rkey = htonl(db->remote_rkey);
287 static void network_to_dest_block(RDMADestBlock *db)
289 db->remote_host_addr = ntohll(db->remote_host_addr);
290 db->offset = ntohll(db->offset);
291 db->length = ntohll(db->length);
292 db->remote_rkey = ntohl(db->remote_rkey);
296 * Virtual address of the above structures used for transmitting
297 * the RAMBlock descriptions at connection-time.
298 * This structure is *not* transmitted.
300 typedef struct RDMALocalBlocks {
301 int nb_blocks;
302 bool init; /* main memory init complete */
303 RDMALocalBlock *block;
304 } RDMALocalBlocks;
307 * Main data structure for RDMA state.
308 * While there is only one copy of this structure being allocated right now,
309 * this is the place where one would start if you wanted to consider
310 * having more than one RDMA connection open at the same time.
312 typedef struct RDMAContext {
313 char *host;
314 int port;
316 RDMAWorkRequestData wr_data[RDMA_WRID_MAX];
319 * This is used by *_exchange_send() to figure out whether or not
320 * the initial "READY" message has already been received or not.
321 * This is because other functions may potentially poll() and detect
322 * the READY message before send() does, in which case we need to
323 * know if it completed.
325 int control_ready_expected;
327 /* number of outstanding writes */
328 int nb_sent;
330 /* store info about current buffer so that we can
331 merge it with future sends */
332 uint64_t current_addr;
333 uint64_t current_length;
334 /* index of ram block the current buffer belongs to */
335 int current_index;
336 /* index of the chunk in the current ram block */
337 int current_chunk;
339 bool pin_all;
342 * infiniband-specific variables for opening the device
343 * and maintaining connection state and so forth.
345 * cm_id also has ibv_context, rdma_event_channel, and ibv_qp in
346 * cm_id->verbs, cm_id->channel, and cm_id->qp.
348 struct rdma_cm_id *cm_id; /* connection manager ID */
349 struct rdma_cm_id *listen_id;
350 bool connected;
352 struct ibv_context *verbs;
353 struct rdma_event_channel *channel;
354 struct ibv_qp *qp; /* queue pair */
355 struct ibv_comp_channel *comp_channel; /* completion channel */
356 struct ibv_pd *pd; /* protection domain */
357 struct ibv_cq *cq; /* completion queue */
360 * If a previous write failed (perhaps because of a failed
361 * memory registration, then do not attempt any future work
362 * and remember the error state.
364 int error_state;
365 int error_reported;
366 int received_error;
369 * Description of ram blocks used throughout the code.
371 RDMALocalBlocks local_ram_blocks;
372 RDMADestBlock *dest_blocks;
374 /* Index of the next RAMBlock received during block registration */
375 unsigned int next_src_index;
378 * Migration on *destination* started.
379 * Then use coroutine yield function.
380 * Source runs in a thread, so we don't care.
382 int migration_started_on_destination;
384 int total_registrations;
385 int total_writes;
387 int unregister_current, unregister_next;
388 uint64_t unregistrations[RDMA_SIGNALED_SEND_MAX];
390 GHashTable *blockmap;
392 /* the RDMAContext for return path */
393 struct RDMAContext *return_path;
394 bool is_return_path;
395 } RDMAContext;
397 #define TYPE_QIO_CHANNEL_RDMA "qio-channel-rdma"
398 #define QIO_CHANNEL_RDMA(obj) \
399 OBJECT_CHECK(QIOChannelRDMA, (obj), TYPE_QIO_CHANNEL_RDMA)
401 typedef struct QIOChannelRDMA QIOChannelRDMA;
404 struct QIOChannelRDMA {
405 QIOChannel parent;
406 RDMAContext *rdmain;
407 RDMAContext *rdmaout;
408 QEMUFile *file;
409 bool blocking; /* XXX we don't actually honour this yet */
413 * Main structure for IB Send/Recv control messages.
414 * This gets prepended at the beginning of every Send/Recv.
416 typedef struct QEMU_PACKED {
417 uint32_t len; /* Total length of data portion */
418 uint32_t type; /* which control command to perform */
419 uint32_t repeat; /* number of commands in data portion of same type */
420 uint32_t padding;
421 } RDMAControlHeader;
423 static void control_to_network(RDMAControlHeader *control)
425 control->type = htonl(control->type);
426 control->len = htonl(control->len);
427 control->repeat = htonl(control->repeat);
430 static void network_to_control(RDMAControlHeader *control)
432 control->type = ntohl(control->type);
433 control->len = ntohl(control->len);
434 control->repeat = ntohl(control->repeat);
438 * Register a single Chunk.
439 * Information sent by the source VM to inform the dest
440 * to register an single chunk of memory before we can perform
441 * the actual RDMA operation.
443 typedef struct QEMU_PACKED {
444 union QEMU_PACKED {
445 uint64_t current_addr; /* offset into the ram_addr_t space */
446 uint64_t chunk; /* chunk to lookup if unregistering */
447 } key;
448 uint32_t current_index; /* which ramblock the chunk belongs to */
449 uint32_t padding;
450 uint64_t chunks; /* how many sequential chunks to register */
451 } RDMARegister;
453 static void register_to_network(RDMAContext *rdma, RDMARegister *reg)
455 RDMALocalBlock *local_block;
456 local_block = &rdma->local_ram_blocks.block[reg->current_index];
458 if (local_block->is_ram_block) {
460 * current_addr as passed in is an address in the local ram_addr_t
461 * space, we need to translate this for the destination
463 reg->key.current_addr -= local_block->offset;
464 reg->key.current_addr += rdma->dest_blocks[reg->current_index].offset;
466 reg->key.current_addr = htonll(reg->key.current_addr);
467 reg->current_index = htonl(reg->current_index);
468 reg->chunks = htonll(reg->chunks);
471 static void network_to_register(RDMARegister *reg)
473 reg->key.current_addr = ntohll(reg->key.current_addr);
474 reg->current_index = ntohl(reg->current_index);
475 reg->chunks = ntohll(reg->chunks);
478 typedef struct QEMU_PACKED {
479 uint32_t value; /* if zero, we will madvise() */
480 uint32_t block_idx; /* which ram block index */
481 uint64_t offset; /* Address in remote ram_addr_t space */
482 uint64_t length; /* length of the chunk */
483 } RDMACompress;
485 static void compress_to_network(RDMAContext *rdma, RDMACompress *comp)
487 comp->value = htonl(comp->value);
489 * comp->offset as passed in is an address in the local ram_addr_t
490 * space, we need to translate this for the destination
492 comp->offset -= rdma->local_ram_blocks.block[comp->block_idx].offset;
493 comp->offset += rdma->dest_blocks[comp->block_idx].offset;
494 comp->block_idx = htonl(comp->block_idx);
495 comp->offset = htonll(comp->offset);
496 comp->length = htonll(comp->length);
499 static void network_to_compress(RDMACompress *comp)
501 comp->value = ntohl(comp->value);
502 comp->block_idx = ntohl(comp->block_idx);
503 comp->offset = ntohll(comp->offset);
504 comp->length = ntohll(comp->length);
508 * The result of the dest's memory registration produces an "rkey"
509 * which the source VM must reference in order to perform
510 * the RDMA operation.
512 typedef struct QEMU_PACKED {
513 uint32_t rkey;
514 uint32_t padding;
515 uint64_t host_addr;
516 } RDMARegisterResult;
518 static void result_to_network(RDMARegisterResult *result)
520 result->rkey = htonl(result->rkey);
521 result->host_addr = htonll(result->host_addr);
524 static void network_to_result(RDMARegisterResult *result)
526 result->rkey = ntohl(result->rkey);
527 result->host_addr = ntohll(result->host_addr);
530 const char *print_wrid(int wrid);
531 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
532 uint8_t *data, RDMAControlHeader *resp,
533 int *resp_idx,
534 int (*callback)(RDMAContext *rdma));
536 static inline uint64_t ram_chunk_index(const uint8_t *start,
537 const uint8_t *host)
539 return ((uintptr_t) host - (uintptr_t) start) >> RDMA_REG_CHUNK_SHIFT;
542 static inline uint8_t *ram_chunk_start(const RDMALocalBlock *rdma_ram_block,
543 uint64_t i)
545 return (uint8_t *)(uintptr_t)(rdma_ram_block->local_host_addr +
546 (i << RDMA_REG_CHUNK_SHIFT));
549 static inline uint8_t *ram_chunk_end(const RDMALocalBlock *rdma_ram_block,
550 uint64_t i)
552 uint8_t *result = ram_chunk_start(rdma_ram_block, i) +
553 (1UL << RDMA_REG_CHUNK_SHIFT);
555 if (result > (rdma_ram_block->local_host_addr + rdma_ram_block->length)) {
556 result = rdma_ram_block->local_host_addr + rdma_ram_block->length;
559 return result;
562 static int rdma_add_block(RDMAContext *rdma, const char *block_name,
563 void *host_addr,
564 ram_addr_t block_offset, uint64_t length)
566 RDMALocalBlocks *local = &rdma->local_ram_blocks;
567 RDMALocalBlock *block;
568 RDMALocalBlock *old = local->block;
570 local->block = g_new0(RDMALocalBlock, local->nb_blocks + 1);
572 if (local->nb_blocks) {
573 int x;
575 if (rdma->blockmap) {
576 for (x = 0; x < local->nb_blocks; x++) {
577 g_hash_table_remove(rdma->blockmap,
578 (void *)(uintptr_t)old[x].offset);
579 g_hash_table_insert(rdma->blockmap,
580 (void *)(uintptr_t)old[x].offset,
581 &local->block[x]);
584 memcpy(local->block, old, sizeof(RDMALocalBlock) * local->nb_blocks);
585 g_free(old);
588 block = &local->block[local->nb_blocks];
590 block->block_name = g_strdup(block_name);
591 block->local_host_addr = host_addr;
592 block->offset = block_offset;
593 block->length = length;
594 block->index = local->nb_blocks;
595 block->src_index = ~0U; /* Filled in by the receipt of the block list */
596 block->nb_chunks = ram_chunk_index(host_addr, host_addr + length) + 1UL;
597 block->transit_bitmap = bitmap_new(block->nb_chunks);
598 bitmap_clear(block->transit_bitmap, 0, block->nb_chunks);
599 block->unregister_bitmap = bitmap_new(block->nb_chunks);
600 bitmap_clear(block->unregister_bitmap, 0, block->nb_chunks);
601 block->remote_keys = g_new0(uint32_t, block->nb_chunks);
603 block->is_ram_block = local->init ? false : true;
605 if (rdma->blockmap) {
606 g_hash_table_insert(rdma->blockmap, (void *)(uintptr_t)block_offset, block);
609 trace_rdma_add_block(block_name, local->nb_blocks,
610 (uintptr_t) block->local_host_addr,
611 block->offset, block->length,
612 (uintptr_t) (block->local_host_addr + block->length),
613 BITS_TO_LONGS(block->nb_chunks) *
614 sizeof(unsigned long) * 8,
615 block->nb_chunks);
617 local->nb_blocks++;
619 return 0;
623 * Memory regions need to be registered with the device and queue pairs setup
624 * in advanced before the migration starts. This tells us where the RAM blocks
625 * are so that we can register them individually.
627 static int qemu_rdma_init_one_block(const char *block_name, void *host_addr,
628 ram_addr_t block_offset, ram_addr_t length, void *opaque)
630 return rdma_add_block(opaque, block_name, host_addr, block_offset, length);
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.
638 static int qemu_rdma_init_ram_blocks(RDMAContext *rdma)
640 RDMALocalBlocks *local = &rdma->local_ram_blocks;
642 assert(rdma->blockmap == NULL);
643 memset(local, 0, sizeof *local);
644 qemu_ram_foreach_migratable_block(qemu_rdma_init_one_block, rdma);
645 trace_qemu_rdma_init_ram_blocks(local->nb_blocks);
646 rdma->dest_blocks = g_new0(RDMADestBlock,
647 rdma->local_ram_blocks.nb_blocks);
648 local->init = true;
649 return 0;
653 * Note: If used outside of cleanup, the caller must ensure that the destination
654 * block structures are also updated
656 static int rdma_delete_block(RDMAContext *rdma, RDMALocalBlock *block)
658 RDMALocalBlocks *local = &rdma->local_ram_blocks;
659 RDMALocalBlock *old = local->block;
660 int x;
662 if (rdma->blockmap) {
663 g_hash_table_remove(rdma->blockmap, (void *)(uintptr_t)block->offset);
665 if (block->pmr) {
666 int j;
668 for (j = 0; j < block->nb_chunks; j++) {
669 if (!block->pmr[j]) {
670 continue;
672 ibv_dereg_mr(block->pmr[j]);
673 rdma->total_registrations--;
675 g_free(block->pmr);
676 block->pmr = NULL;
679 if (block->mr) {
680 ibv_dereg_mr(block->mr);
681 rdma->total_registrations--;
682 block->mr = NULL;
685 g_free(block->transit_bitmap);
686 block->transit_bitmap = NULL;
688 g_free(block->unregister_bitmap);
689 block->unregister_bitmap = NULL;
691 g_free(block->remote_keys);
692 block->remote_keys = NULL;
694 g_free(block->block_name);
695 block->block_name = NULL;
697 if (rdma->blockmap) {
698 for (x = 0; x < local->nb_blocks; x++) {
699 g_hash_table_remove(rdma->blockmap,
700 (void *)(uintptr_t)old[x].offset);
704 if (local->nb_blocks > 1) {
706 local->block = g_new0(RDMALocalBlock, local->nb_blocks - 1);
708 if (block->index) {
709 memcpy(local->block, old, sizeof(RDMALocalBlock) * block->index);
712 if (block->index < (local->nb_blocks - 1)) {
713 memcpy(local->block + block->index, old + (block->index + 1),
714 sizeof(RDMALocalBlock) *
715 (local->nb_blocks - (block->index + 1)));
716 for (x = block->index; x < local->nb_blocks - 1; x++) {
717 local->block[x].index--;
720 } else {
721 assert(block == local->block);
722 local->block = NULL;
725 trace_rdma_delete_block(block, (uintptr_t)block->local_host_addr,
726 block->offset, block->length,
727 (uintptr_t)(block->local_host_addr + block->length),
728 BITS_TO_LONGS(block->nb_chunks) *
729 sizeof(unsigned long) * 8, block->nb_chunks);
731 g_free(old);
733 local->nb_blocks--;
735 if (local->nb_blocks && rdma->blockmap) {
736 for (x = 0; x < local->nb_blocks; x++) {
737 g_hash_table_insert(rdma->blockmap,
738 (void *)(uintptr_t)local->block[x].offset,
739 &local->block[x]);
743 return 0;
747 * Put in the log file which RDMA device was opened and the details
748 * associated with that device.
750 static void qemu_rdma_dump_id(const char *who, struct ibv_context *verbs)
752 struct ibv_port_attr port;
754 if (ibv_query_port(verbs, 1, &port)) {
755 error_report("Failed to query port information");
756 return;
759 printf("%s RDMA Device opened: kernel name %s "
760 "uverbs device name %s, "
761 "infiniband_verbs class device path %s, "
762 "infiniband class device path %s, "
763 "transport: (%d) %s\n",
764 who,
765 verbs->device->name,
766 verbs->device->dev_name,
767 verbs->device->dev_path,
768 verbs->device->ibdev_path,
769 port.link_layer,
770 (port.link_layer == IBV_LINK_LAYER_INFINIBAND) ? "Infiniband" :
771 ((port.link_layer == IBV_LINK_LAYER_ETHERNET)
772 ? "Ethernet" : "Unknown"));
776 * Put in the log file the RDMA gid addressing information,
777 * useful for folks who have trouble understanding the
778 * RDMA device hierarchy in the kernel.
780 static void qemu_rdma_dump_gid(const char *who, struct rdma_cm_id *id)
782 char sgid[33];
783 char dgid[33];
784 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.sgid, sgid, sizeof sgid);
785 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.dgid, dgid, sizeof dgid);
786 trace_qemu_rdma_dump_gid(who, sgid, dgid);
790 * As of now, IPv6 over RoCE / iWARP is not supported by linux.
791 * We will try the next addrinfo struct, and fail if there are
792 * no other valid addresses to bind against.
794 * If user is listening on '[::]', then we will not have a opened a device
795 * yet and have no way of verifying if the device is RoCE or not.
797 * In this case, the source VM will throw an error for ALL types of
798 * connections (both IPv4 and IPv6) if the destination machine does not have
799 * a regular infiniband network available for use.
801 * The only way to guarantee that an error is thrown for broken kernels is
802 * for the management software to choose a *specific* interface at bind time
803 * and validate what time of hardware it is.
805 * Unfortunately, this puts the user in a fix:
807 * If the source VM connects with an IPv4 address without knowing that the
808 * destination has bound to '[::]' the migration will unconditionally fail
809 * unless the management software is explicitly listening on the IPv4
810 * address while using a RoCE-based device.
812 * If the source VM connects with an IPv6 address, then we're OK because we can
813 * throw an error on the source (and similarly on the destination).
815 * But in mixed environments, this will be broken for a while until it is fixed
816 * inside linux.
818 * We do provide a *tiny* bit of help in this function: We can list all of the
819 * devices in the system and check to see if all the devices are RoCE or
820 * Infiniband.
822 * If we detect that we have a *pure* RoCE environment, then we can safely
823 * thrown an error even if the management software has specified '[::]' as the
824 * bind address.
826 * However, if there is are multiple hetergeneous devices, then we cannot make
827 * this assumption and the user just has to be sure they know what they are
828 * doing.
830 * Patches are being reviewed on linux-rdma.
832 static int qemu_rdma_broken_ipv6_kernel(struct ibv_context *verbs, Error **errp)
834 struct ibv_port_attr port_attr;
836 /* This bug only exists in linux, to our knowledge. */
837 #ifdef CONFIG_LINUX
840 * Verbs are only NULL if management has bound to '[::]'.
842 * Let's iterate through all the devices and see if there any pure IB
843 * devices (non-ethernet).
845 * If not, then we can safely proceed with the migration.
846 * Otherwise, there are no guarantees until the bug is fixed in linux.
848 if (!verbs) {
849 int num_devices, x;
850 struct ibv_device ** dev_list = ibv_get_device_list(&num_devices);
851 bool roce_found = false;
852 bool ib_found = false;
854 for (x = 0; x < num_devices; x++) {
855 verbs = ibv_open_device(dev_list[x]);
856 if (!verbs) {
857 if (errno == EPERM) {
858 continue;
859 } else {
860 return -EINVAL;
864 if (ibv_query_port(verbs, 1, &port_attr)) {
865 ibv_close_device(verbs);
866 ERROR(errp, "Could not query initial IB port");
867 return -EINVAL;
870 if (port_attr.link_layer == IBV_LINK_LAYER_INFINIBAND) {
871 ib_found = true;
872 } else if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
873 roce_found = true;
876 ibv_close_device(verbs);
880 if (roce_found) {
881 if (ib_found) {
882 fprintf(stderr, "WARN: migrations may fail:"
883 " IPv6 over RoCE / iWARP in linux"
884 " is broken. But since you appear to have a"
885 " mixed RoCE / IB environment, be sure to only"
886 " migrate over the IB fabric until the kernel "
887 " fixes the bug.\n");
888 } else {
889 ERROR(errp, "You only have RoCE / iWARP devices in your systems"
890 " and your management software has specified '[::]'"
891 ", but IPv6 over RoCE / iWARP is not supported in Linux.");
892 return -ENONET;
896 return 0;
900 * If we have a verbs context, that means that some other than '[::]' was
901 * used by the management software for binding. In which case we can
902 * actually warn the user about a potentially broken kernel.
905 /* IB ports start with 1, not 0 */
906 if (ibv_query_port(verbs, 1, &port_attr)) {
907 ERROR(errp, "Could not query initial IB port");
908 return -EINVAL;
911 if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
912 ERROR(errp, "Linux kernel's RoCE / iWARP does not support IPv6 "
913 "(but patches on linux-rdma in progress)");
914 return -ENONET;
917 #endif
919 return 0;
923 * Figure out which RDMA device corresponds to the requested IP hostname
924 * Also create the initial connection manager identifiers for opening
925 * the connection.
927 static int qemu_rdma_resolve_host(RDMAContext *rdma, Error **errp)
929 int ret;
930 struct rdma_addrinfo *res;
931 char port_str[16];
932 struct rdma_cm_event *cm_event;
933 char ip[40] = "unknown";
934 struct rdma_addrinfo *e;
936 if (rdma->host == NULL || !strcmp(rdma->host, "")) {
937 ERROR(errp, "RDMA hostname has not been set");
938 return -EINVAL;
941 /* create CM channel */
942 rdma->channel = rdma_create_event_channel();
943 if (!rdma->channel) {
944 ERROR(errp, "could not create CM channel");
945 return -EINVAL;
948 /* create CM id */
949 ret = rdma_create_id(rdma->channel, &rdma->cm_id, NULL, RDMA_PS_TCP);
950 if (ret) {
951 ERROR(errp, "could not create channel id");
952 goto err_resolve_create_id;
955 snprintf(port_str, 16, "%d", rdma->port);
956 port_str[15] = '\0';
958 ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
959 if (ret < 0) {
960 ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
961 goto err_resolve_get_addr;
964 for (e = res; e != NULL; e = e->ai_next) {
965 inet_ntop(e->ai_family,
966 &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
967 trace_qemu_rdma_resolve_host_trying(rdma->host, ip);
969 ret = rdma_resolve_addr(rdma->cm_id, NULL, e->ai_dst_addr,
970 RDMA_RESOLVE_TIMEOUT_MS);
971 if (!ret) {
972 if (e->ai_family == AF_INET6) {
973 ret = qemu_rdma_broken_ipv6_kernel(rdma->cm_id->verbs, errp);
974 if (ret) {
975 continue;
978 goto route;
982 ERROR(errp, "could not resolve address %s", rdma->host);
983 goto err_resolve_get_addr;
985 route:
986 qemu_rdma_dump_gid("source_resolve_addr", rdma->cm_id);
988 ret = rdma_get_cm_event(rdma->channel, &cm_event);
989 if (ret) {
990 ERROR(errp, "could not perform event_addr_resolved");
991 goto err_resolve_get_addr;
994 if (cm_event->event != RDMA_CM_EVENT_ADDR_RESOLVED) {
995 ERROR(errp, "result not equal to event_addr_resolved %s",
996 rdma_event_str(cm_event->event));
997 perror("rdma_resolve_addr");
998 rdma_ack_cm_event(cm_event);
999 ret = -EINVAL;
1000 goto err_resolve_get_addr;
1002 rdma_ack_cm_event(cm_event);
1004 /* resolve route */
1005 ret = rdma_resolve_route(rdma->cm_id, RDMA_RESOLVE_TIMEOUT_MS);
1006 if (ret) {
1007 ERROR(errp, "could not resolve rdma route");
1008 goto err_resolve_get_addr;
1011 ret = rdma_get_cm_event(rdma->channel, &cm_event);
1012 if (ret) {
1013 ERROR(errp, "could not perform event_route_resolved");
1014 goto err_resolve_get_addr;
1016 if (cm_event->event != RDMA_CM_EVENT_ROUTE_RESOLVED) {
1017 ERROR(errp, "result not equal to event_route_resolved: %s",
1018 rdma_event_str(cm_event->event));
1019 rdma_ack_cm_event(cm_event);
1020 ret = -EINVAL;
1021 goto err_resolve_get_addr;
1023 rdma_ack_cm_event(cm_event);
1024 rdma->verbs = rdma->cm_id->verbs;
1025 qemu_rdma_dump_id("source_resolve_host", rdma->cm_id->verbs);
1026 qemu_rdma_dump_gid("source_resolve_host", rdma->cm_id);
1027 return 0;
1029 err_resolve_get_addr:
1030 rdma_destroy_id(rdma->cm_id);
1031 rdma->cm_id = NULL;
1032 err_resolve_create_id:
1033 rdma_destroy_event_channel(rdma->channel);
1034 rdma->channel = NULL;
1035 return ret;
1039 * Create protection domain and completion queues
1041 static int qemu_rdma_alloc_pd_cq(RDMAContext *rdma)
1043 /* allocate pd */
1044 rdma->pd = ibv_alloc_pd(rdma->verbs);
1045 if (!rdma->pd) {
1046 error_report("failed to allocate protection domain");
1047 return -1;
1050 /* create completion channel */
1051 rdma->comp_channel = ibv_create_comp_channel(rdma->verbs);
1052 if (!rdma->comp_channel) {
1053 error_report("failed to allocate completion channel");
1054 goto err_alloc_pd_cq;
1058 * Completion queue can be filled by both read and write work requests,
1059 * so must reflect the sum of both possible queue sizes.
1061 rdma->cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3),
1062 NULL, rdma->comp_channel, 0);
1063 if (!rdma->cq) {
1064 error_report("failed to allocate completion queue");
1065 goto err_alloc_pd_cq;
1068 return 0;
1070 err_alloc_pd_cq:
1071 if (rdma->pd) {
1072 ibv_dealloc_pd(rdma->pd);
1074 if (rdma->comp_channel) {
1075 ibv_destroy_comp_channel(rdma->comp_channel);
1077 rdma->pd = NULL;
1078 rdma->comp_channel = NULL;
1079 return -1;
1084 * Create queue pairs.
1086 static int qemu_rdma_alloc_qp(RDMAContext *rdma)
1088 struct ibv_qp_init_attr attr = { 0 };
1089 int ret;
1091 attr.cap.max_send_wr = RDMA_SIGNALED_SEND_MAX;
1092 attr.cap.max_recv_wr = 3;
1093 attr.cap.max_send_sge = 1;
1094 attr.cap.max_recv_sge = 1;
1095 attr.send_cq = rdma->cq;
1096 attr.recv_cq = rdma->cq;
1097 attr.qp_type = IBV_QPT_RC;
1099 ret = rdma_create_qp(rdma->cm_id, rdma->pd, &attr);
1100 if (ret) {
1101 return -1;
1104 rdma->qp = rdma->cm_id->qp;
1105 return 0;
1108 static int qemu_rdma_reg_whole_ram_blocks(RDMAContext *rdma)
1110 int i;
1111 RDMALocalBlocks *local = &rdma->local_ram_blocks;
1113 for (i = 0; i < local->nb_blocks; i++) {
1114 local->block[i].mr =
1115 ibv_reg_mr(rdma->pd,
1116 local->block[i].local_host_addr,
1117 local->block[i].length,
1118 IBV_ACCESS_LOCAL_WRITE |
1119 IBV_ACCESS_REMOTE_WRITE
1121 if (!local->block[i].mr) {
1122 perror("Failed to register local dest ram block!\n");
1123 break;
1125 rdma->total_registrations++;
1128 if (i >= local->nb_blocks) {
1129 return 0;
1132 for (i--; i >= 0; i--) {
1133 ibv_dereg_mr(local->block[i].mr);
1134 rdma->total_registrations--;
1137 return -1;
1142 * Find the ram block that corresponds to the page requested to be
1143 * transmitted by QEMU.
1145 * Once the block is found, also identify which 'chunk' within that
1146 * block that the page belongs to.
1148 * This search cannot fail or the migration will fail.
1150 static int qemu_rdma_search_ram_block(RDMAContext *rdma,
1151 uintptr_t block_offset,
1152 uint64_t offset,
1153 uint64_t length,
1154 uint64_t *block_index,
1155 uint64_t *chunk_index)
1157 uint64_t current_addr = block_offset + offset;
1158 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
1159 (void *) block_offset);
1160 assert(block);
1161 assert(current_addr >= block->offset);
1162 assert((current_addr + length) <= (block->offset + block->length));
1164 *block_index = block->index;
1165 *chunk_index = ram_chunk_index(block->local_host_addr,
1166 block->local_host_addr + (current_addr - block->offset));
1168 return 0;
1172 * Register a chunk with IB. If the chunk was already registered
1173 * previously, then skip.
1175 * Also return the keys associated with the registration needed
1176 * to perform the actual RDMA operation.
1178 static int qemu_rdma_register_and_get_keys(RDMAContext *rdma,
1179 RDMALocalBlock *block, uintptr_t host_addr,
1180 uint32_t *lkey, uint32_t *rkey, int chunk,
1181 uint8_t *chunk_start, uint8_t *chunk_end)
1183 if (block->mr) {
1184 if (lkey) {
1185 *lkey = block->mr->lkey;
1187 if (rkey) {
1188 *rkey = block->mr->rkey;
1190 return 0;
1193 /* allocate memory to store chunk MRs */
1194 if (!block->pmr) {
1195 block->pmr = g_new0(struct ibv_mr *, block->nb_chunks);
1199 * If 'rkey', then we're the destination, so grant access to the source.
1201 * If 'lkey', then we're the source VM, so grant access only to ourselves.
1203 if (!block->pmr[chunk]) {
1204 uint64_t len = chunk_end - chunk_start;
1206 trace_qemu_rdma_register_and_get_keys(len, chunk_start);
1208 block->pmr[chunk] = ibv_reg_mr(rdma->pd,
1209 chunk_start, len,
1210 (rkey ? (IBV_ACCESS_LOCAL_WRITE |
1211 IBV_ACCESS_REMOTE_WRITE) : 0));
1213 if (!block->pmr[chunk]) {
1214 perror("Failed to register chunk!");
1215 fprintf(stderr, "Chunk details: block: %d chunk index %d"
1216 " start %" PRIuPTR " end %" PRIuPTR
1217 " host %" PRIuPTR
1218 " local %" PRIuPTR " registrations: %d\n",
1219 block->index, chunk, (uintptr_t)chunk_start,
1220 (uintptr_t)chunk_end, host_addr,
1221 (uintptr_t)block->local_host_addr,
1222 rdma->total_registrations);
1223 return -1;
1225 rdma->total_registrations++;
1228 if (lkey) {
1229 *lkey = block->pmr[chunk]->lkey;
1231 if (rkey) {
1232 *rkey = block->pmr[chunk]->rkey;
1234 return 0;
1238 * Register (at connection time) the memory used for control
1239 * channel messages.
1241 static int qemu_rdma_reg_control(RDMAContext *rdma, int idx)
1243 rdma->wr_data[idx].control_mr = ibv_reg_mr(rdma->pd,
1244 rdma->wr_data[idx].control, RDMA_CONTROL_MAX_BUFFER,
1245 IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
1246 if (rdma->wr_data[idx].control_mr) {
1247 rdma->total_registrations++;
1248 return 0;
1250 error_report("qemu_rdma_reg_control failed");
1251 return -1;
1254 const char *print_wrid(int wrid)
1256 if (wrid >= RDMA_WRID_RECV_CONTROL) {
1257 return wrid_desc[RDMA_WRID_RECV_CONTROL];
1259 return wrid_desc[wrid];
1263 * RDMA requires memory registration (mlock/pinning), but this is not good for
1264 * overcommitment.
1266 * In preparation for the future where LRU information or workload-specific
1267 * writable writable working set memory access behavior is available to QEMU
1268 * it would be nice to have in place the ability to UN-register/UN-pin
1269 * particular memory regions from the RDMA hardware when it is determine that
1270 * those regions of memory will likely not be accessed again in the near future.
1272 * While we do not yet have such information right now, the following
1273 * compile-time option allows us to perform a non-optimized version of this
1274 * behavior.
1276 * By uncommenting this option, you will cause *all* RDMA transfers to be
1277 * unregistered immediately after the transfer completes on both sides of the
1278 * connection. This has no effect in 'rdma-pin-all' mode, only regular mode.
1280 * This will have a terrible impact on migration performance, so until future
1281 * workload information or LRU information is available, do not attempt to use
1282 * this feature except for basic testing.
1284 //#define RDMA_UNREGISTRATION_EXAMPLE
1287 * Perform a non-optimized memory unregistration after every transfer
1288 * for demonstration purposes, only if pin-all is not requested.
1290 * Potential optimizations:
1291 * 1. Start a new thread to run this function continuously
1292 - for bit clearing
1293 - and for receipt of unregister messages
1294 * 2. Use an LRU.
1295 * 3. Use workload hints.
1297 static int qemu_rdma_unregister_waiting(RDMAContext *rdma)
1299 while (rdma->unregistrations[rdma->unregister_current]) {
1300 int ret;
1301 uint64_t wr_id = rdma->unregistrations[rdma->unregister_current];
1302 uint64_t chunk =
1303 (wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1304 uint64_t index =
1305 (wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1306 RDMALocalBlock *block =
1307 &(rdma->local_ram_blocks.block[index]);
1308 RDMARegister reg = { .current_index = index };
1309 RDMAControlHeader resp = { .type = RDMA_CONTROL_UNREGISTER_FINISHED,
1311 RDMAControlHeader head = { .len = sizeof(RDMARegister),
1312 .type = RDMA_CONTROL_UNREGISTER_REQUEST,
1313 .repeat = 1,
1316 trace_qemu_rdma_unregister_waiting_proc(chunk,
1317 rdma->unregister_current);
1319 rdma->unregistrations[rdma->unregister_current] = 0;
1320 rdma->unregister_current++;
1322 if (rdma->unregister_current == RDMA_SIGNALED_SEND_MAX) {
1323 rdma->unregister_current = 0;
1328 * Unregistration is speculative (because migration is single-threaded
1329 * and we cannot break the protocol's inifinband message ordering).
1330 * Thus, if the memory is currently being used for transmission,
1331 * then abort the attempt to unregister and try again
1332 * later the next time a completion is received for this memory.
1334 clear_bit(chunk, block->unregister_bitmap);
1336 if (test_bit(chunk, block->transit_bitmap)) {
1337 trace_qemu_rdma_unregister_waiting_inflight(chunk);
1338 continue;
1341 trace_qemu_rdma_unregister_waiting_send(chunk);
1343 ret = ibv_dereg_mr(block->pmr[chunk]);
1344 block->pmr[chunk] = NULL;
1345 block->remote_keys[chunk] = 0;
1347 if (ret != 0) {
1348 perror("unregistration chunk failed");
1349 return -ret;
1351 rdma->total_registrations--;
1353 reg.key.chunk = chunk;
1354 register_to_network(rdma, &reg);
1355 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) &reg,
1356 &resp, NULL, NULL);
1357 if (ret < 0) {
1358 return ret;
1361 trace_qemu_rdma_unregister_waiting_complete(chunk);
1364 return 0;
1367 static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index,
1368 uint64_t chunk)
1370 uint64_t result = wr_id & RDMA_WRID_TYPE_MASK;
1372 result |= (index << RDMA_WRID_BLOCK_SHIFT);
1373 result |= (chunk << RDMA_WRID_CHUNK_SHIFT);
1375 return result;
1379 * Set bit for unregistration in the next iteration.
1380 * We cannot transmit right here, but will unpin later.
1382 static void qemu_rdma_signal_unregister(RDMAContext *rdma, uint64_t index,
1383 uint64_t chunk, uint64_t wr_id)
1385 if (rdma->unregistrations[rdma->unregister_next] != 0) {
1386 error_report("rdma migration: queue is full");
1387 } else {
1388 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1390 if (!test_and_set_bit(chunk, block->unregister_bitmap)) {
1391 trace_qemu_rdma_signal_unregister_append(chunk,
1392 rdma->unregister_next);
1394 rdma->unregistrations[rdma->unregister_next++] =
1395 qemu_rdma_make_wrid(wr_id, index, chunk);
1397 if (rdma->unregister_next == RDMA_SIGNALED_SEND_MAX) {
1398 rdma->unregister_next = 0;
1400 } else {
1401 trace_qemu_rdma_signal_unregister_already(chunk);
1407 * Consult the connection manager to see a work request
1408 * (of any kind) has completed.
1409 * Return the work request ID that completed.
1411 static uint64_t qemu_rdma_poll(RDMAContext *rdma, uint64_t *wr_id_out,
1412 uint32_t *byte_len)
1414 int ret;
1415 struct ibv_wc wc;
1416 uint64_t wr_id;
1418 ret = ibv_poll_cq(rdma->cq, 1, &wc);
1420 if (!ret) {
1421 *wr_id_out = RDMA_WRID_NONE;
1422 return 0;
1425 if (ret < 0) {
1426 error_report("ibv_poll_cq return %d", ret);
1427 return ret;
1430 wr_id = wc.wr_id & RDMA_WRID_TYPE_MASK;
1432 if (wc.status != IBV_WC_SUCCESS) {
1433 fprintf(stderr, "ibv_poll_cq wc.status=%d %s!\n",
1434 wc.status, ibv_wc_status_str(wc.status));
1435 fprintf(stderr, "ibv_poll_cq wrid=%s!\n", wrid_desc[wr_id]);
1437 return -1;
1440 if (rdma->control_ready_expected &&
1441 (wr_id >= RDMA_WRID_RECV_CONTROL)) {
1442 trace_qemu_rdma_poll_recv(wrid_desc[RDMA_WRID_RECV_CONTROL],
1443 wr_id - RDMA_WRID_RECV_CONTROL, wr_id, rdma->nb_sent);
1444 rdma->control_ready_expected = 0;
1447 if (wr_id == RDMA_WRID_RDMA_WRITE) {
1448 uint64_t chunk =
1449 (wc.wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1450 uint64_t index =
1451 (wc.wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1452 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1454 trace_qemu_rdma_poll_write(print_wrid(wr_id), wr_id, rdma->nb_sent,
1455 index, chunk, block->local_host_addr,
1456 (void *)(uintptr_t)block->remote_host_addr);
1458 clear_bit(chunk, block->transit_bitmap);
1460 if (rdma->nb_sent > 0) {
1461 rdma->nb_sent--;
1464 if (!rdma->pin_all) {
1466 * FYI: If one wanted to signal a specific chunk to be unregistered
1467 * using LRU or workload-specific information, this is the function
1468 * you would call to do so. That chunk would then get asynchronously
1469 * unregistered later.
1471 #ifdef RDMA_UNREGISTRATION_EXAMPLE
1472 qemu_rdma_signal_unregister(rdma, index, chunk, wc.wr_id);
1473 #endif
1475 } else {
1476 trace_qemu_rdma_poll_other(print_wrid(wr_id), wr_id, rdma->nb_sent);
1479 *wr_id_out = wc.wr_id;
1480 if (byte_len) {
1481 *byte_len = wc.byte_len;
1484 return 0;
1487 /* Wait for activity on the completion channel.
1488 * Returns 0 on success, none-0 on error.
1490 static int qemu_rdma_wait_comp_channel(RDMAContext *rdma)
1492 struct rdma_cm_event *cm_event;
1493 int ret = -1;
1496 * Coroutine doesn't start until migration_fd_process_incoming()
1497 * so don't yield unless we know we're running inside of a coroutine.
1499 if (rdma->migration_started_on_destination &&
1500 migration_incoming_get_current()->state == MIGRATION_STATUS_ACTIVE) {
1501 yield_until_fd_readable(rdma->comp_channel->fd);
1502 } else {
1503 /* This is the source side, we're in a separate thread
1504 * or destination prior to migration_fd_process_incoming()
1505 * after postcopy, the destination also in a seprate thread.
1506 * we can't yield; so we have to poll the fd.
1507 * But we need to be able to handle 'cancel' or an error
1508 * without hanging forever.
1510 while (!rdma->error_state && !rdma->received_error) {
1511 GPollFD pfds[2];
1512 pfds[0].fd = rdma->comp_channel->fd;
1513 pfds[0].events = G_IO_IN | G_IO_HUP | G_IO_ERR;
1514 pfds[0].revents = 0;
1516 pfds[1].fd = rdma->channel->fd;
1517 pfds[1].events = G_IO_IN | G_IO_HUP | G_IO_ERR;
1518 pfds[1].revents = 0;
1520 /* 0.1s timeout, should be fine for a 'cancel' */
1521 switch (qemu_poll_ns(pfds, 2, 100 * 1000 * 1000)) {
1522 case 2:
1523 case 1: /* fd active */
1524 if (pfds[0].revents) {
1525 return 0;
1528 if (pfds[1].revents) {
1529 ret = rdma_get_cm_event(rdma->channel, &cm_event);
1530 if (!ret) {
1531 rdma_ack_cm_event(cm_event);
1534 error_report("receive cm event while wait comp channel,"
1535 "cm event is %d", cm_event->event);
1536 if (cm_event->event == RDMA_CM_EVENT_DISCONNECTED ||
1537 cm_event->event == RDMA_CM_EVENT_DEVICE_REMOVAL) {
1538 return -EPIPE;
1541 break;
1543 case 0: /* Timeout, go around again */
1544 break;
1546 default: /* Error of some type -
1547 * I don't trust errno from qemu_poll_ns
1549 error_report("%s: poll failed", __func__);
1550 return -EPIPE;
1553 if (migrate_get_current()->state == MIGRATION_STATUS_CANCELLING) {
1554 /* Bail out and let the cancellation happen */
1555 return -EPIPE;
1560 if (rdma->received_error) {
1561 return -EPIPE;
1563 return rdma->error_state;
1567 * Block until the next work request has completed.
1569 * First poll to see if a work request has already completed,
1570 * otherwise block.
1572 * If we encounter completed work requests for IDs other than
1573 * the one we're interested in, then that's generally an error.
1575 * The only exception is actual RDMA Write completions. These
1576 * completions only need to be recorded, but do not actually
1577 * need further processing.
1579 static int qemu_rdma_block_for_wrid(RDMAContext *rdma, int wrid_requested,
1580 uint32_t *byte_len)
1582 int num_cq_events = 0, ret = 0;
1583 struct ibv_cq *cq;
1584 void *cq_ctx;
1585 uint64_t wr_id = RDMA_WRID_NONE, wr_id_in;
1587 if (ibv_req_notify_cq(rdma->cq, 0)) {
1588 return -1;
1590 /* poll cq first */
1591 while (wr_id != wrid_requested) {
1592 ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
1593 if (ret < 0) {
1594 return ret;
1597 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1599 if (wr_id == RDMA_WRID_NONE) {
1600 break;
1602 if (wr_id != wrid_requested) {
1603 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested),
1604 wrid_requested, print_wrid(wr_id), wr_id);
1608 if (wr_id == wrid_requested) {
1609 return 0;
1612 while (1) {
1613 ret = qemu_rdma_wait_comp_channel(rdma);
1614 if (ret) {
1615 goto err_block_for_wrid;
1618 ret = ibv_get_cq_event(rdma->comp_channel, &cq, &cq_ctx);
1619 if (ret) {
1620 perror("ibv_get_cq_event");
1621 goto err_block_for_wrid;
1624 num_cq_events++;
1626 ret = -ibv_req_notify_cq(cq, 0);
1627 if (ret) {
1628 goto err_block_for_wrid;
1631 while (wr_id != wrid_requested) {
1632 ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
1633 if (ret < 0) {
1634 goto err_block_for_wrid;
1637 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1639 if (wr_id == RDMA_WRID_NONE) {
1640 break;
1642 if (wr_id != wrid_requested) {
1643 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested),
1644 wrid_requested, print_wrid(wr_id), wr_id);
1648 if (wr_id == wrid_requested) {
1649 goto success_block_for_wrid;
1653 success_block_for_wrid:
1654 if (num_cq_events) {
1655 ibv_ack_cq_events(cq, num_cq_events);
1657 return 0;
1659 err_block_for_wrid:
1660 if (num_cq_events) {
1661 ibv_ack_cq_events(cq, num_cq_events);
1664 rdma->error_state = ret;
1665 return ret;
1669 * Post a SEND message work request for the control channel
1670 * containing some data and block until the post completes.
1672 static int qemu_rdma_post_send_control(RDMAContext *rdma, uint8_t *buf,
1673 RDMAControlHeader *head)
1675 int ret = 0;
1676 RDMAWorkRequestData *wr = &rdma->wr_data[RDMA_WRID_CONTROL];
1677 struct ibv_send_wr *bad_wr;
1678 struct ibv_sge sge = {
1679 .addr = (uintptr_t)(wr->control),
1680 .length = head->len + sizeof(RDMAControlHeader),
1681 .lkey = wr->control_mr->lkey,
1683 struct ibv_send_wr send_wr = {
1684 .wr_id = RDMA_WRID_SEND_CONTROL,
1685 .opcode = IBV_WR_SEND,
1686 .send_flags = IBV_SEND_SIGNALED,
1687 .sg_list = &sge,
1688 .num_sge = 1,
1691 trace_qemu_rdma_post_send_control(control_desc(head->type));
1694 * We don't actually need to do a memcpy() in here if we used
1695 * the "sge" properly, but since we're only sending control messages
1696 * (not RAM in a performance-critical path), then its OK for now.
1698 * The copy makes the RDMAControlHeader simpler to manipulate
1699 * for the time being.
1701 assert(head->len <= RDMA_CONTROL_MAX_BUFFER - sizeof(*head));
1702 memcpy(wr->control, head, sizeof(RDMAControlHeader));
1703 control_to_network((void *) wr->control);
1705 if (buf) {
1706 memcpy(wr->control + sizeof(RDMAControlHeader), buf, head->len);
1710 ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
1712 if (ret > 0) {
1713 error_report("Failed to use post IB SEND for control");
1714 return -ret;
1717 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_SEND_CONTROL, NULL);
1718 if (ret < 0) {
1719 error_report("rdma migration: send polling control error");
1722 return ret;
1726 * Post a RECV work request in anticipation of some future receipt
1727 * of data on the control channel.
1729 static int qemu_rdma_post_recv_control(RDMAContext *rdma, int idx)
1731 struct ibv_recv_wr *bad_wr;
1732 struct ibv_sge sge = {
1733 .addr = (uintptr_t)(rdma->wr_data[idx].control),
1734 .length = RDMA_CONTROL_MAX_BUFFER,
1735 .lkey = rdma->wr_data[idx].control_mr->lkey,
1738 struct ibv_recv_wr recv_wr = {
1739 .wr_id = RDMA_WRID_RECV_CONTROL + idx,
1740 .sg_list = &sge,
1741 .num_sge = 1,
1745 if (ibv_post_recv(rdma->qp, &recv_wr, &bad_wr)) {
1746 return -1;
1749 return 0;
1753 * Block and wait for a RECV control channel message to arrive.
1755 static int qemu_rdma_exchange_get_response(RDMAContext *rdma,
1756 RDMAControlHeader *head, int expecting, int idx)
1758 uint32_t byte_len;
1759 int ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RECV_CONTROL + idx,
1760 &byte_len);
1762 if (ret < 0) {
1763 error_report("rdma migration: recv polling control error!");
1764 return ret;
1767 network_to_control((void *) rdma->wr_data[idx].control);
1768 memcpy(head, rdma->wr_data[idx].control, sizeof(RDMAControlHeader));
1770 trace_qemu_rdma_exchange_get_response_start(control_desc(expecting));
1772 if (expecting == RDMA_CONTROL_NONE) {
1773 trace_qemu_rdma_exchange_get_response_none(control_desc(head->type),
1774 head->type);
1775 } else if (head->type != expecting || head->type == RDMA_CONTROL_ERROR) {
1776 error_report("Was expecting a %s (%d) control message"
1777 ", but got: %s (%d), length: %d",
1778 control_desc(expecting), expecting,
1779 control_desc(head->type), head->type, head->len);
1780 if (head->type == RDMA_CONTROL_ERROR) {
1781 rdma->received_error = true;
1783 return -EIO;
1785 if (head->len > RDMA_CONTROL_MAX_BUFFER - sizeof(*head)) {
1786 error_report("too long length: %d", head->len);
1787 return -EINVAL;
1789 if (sizeof(*head) + head->len != byte_len) {
1790 error_report("Malformed length: %d byte_len %d", head->len, byte_len);
1791 return -EINVAL;
1794 return 0;
1798 * When a RECV work request has completed, the work request's
1799 * buffer is pointed at the header.
1801 * This will advance the pointer to the data portion
1802 * of the control message of the work request's buffer that
1803 * was populated after the work request finished.
1805 static void qemu_rdma_move_header(RDMAContext *rdma, int idx,
1806 RDMAControlHeader *head)
1808 rdma->wr_data[idx].control_len = head->len;
1809 rdma->wr_data[idx].control_curr =
1810 rdma->wr_data[idx].control + sizeof(RDMAControlHeader);
1814 * This is an 'atomic' high-level operation to deliver a single, unified
1815 * control-channel message.
1817 * Additionally, if the user is expecting some kind of reply to this message,
1818 * they can request a 'resp' response message be filled in by posting an
1819 * additional work request on behalf of the user and waiting for an additional
1820 * completion.
1822 * The extra (optional) response is used during registration to us from having
1823 * to perform an *additional* exchange of message just to provide a response by
1824 * instead piggy-backing on the acknowledgement.
1826 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
1827 uint8_t *data, RDMAControlHeader *resp,
1828 int *resp_idx,
1829 int (*callback)(RDMAContext *rdma))
1831 int ret = 0;
1834 * Wait until the dest is ready before attempting to deliver the message
1835 * by waiting for a READY message.
1837 if (rdma->control_ready_expected) {
1838 RDMAControlHeader resp;
1839 ret = qemu_rdma_exchange_get_response(rdma,
1840 &resp, RDMA_CONTROL_READY, RDMA_WRID_READY);
1841 if (ret < 0) {
1842 return ret;
1847 * If the user is expecting a response, post a WR in anticipation of it.
1849 if (resp) {
1850 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_DATA);
1851 if (ret) {
1852 error_report("rdma migration: error posting"
1853 " extra control recv for anticipated result!");
1854 return ret;
1859 * Post a WR to replace the one we just consumed for the READY message.
1861 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
1862 if (ret) {
1863 error_report("rdma migration: error posting first control recv!");
1864 return ret;
1868 * Deliver the control message that was requested.
1870 ret = qemu_rdma_post_send_control(rdma, data, head);
1872 if (ret < 0) {
1873 error_report("Failed to send control buffer!");
1874 return ret;
1878 * If we're expecting a response, block and wait for it.
1880 if (resp) {
1881 if (callback) {
1882 trace_qemu_rdma_exchange_send_issue_callback();
1883 ret = callback(rdma);
1884 if (ret < 0) {
1885 return ret;
1889 trace_qemu_rdma_exchange_send_waiting(control_desc(resp->type));
1890 ret = qemu_rdma_exchange_get_response(rdma, resp,
1891 resp->type, RDMA_WRID_DATA);
1893 if (ret < 0) {
1894 return ret;
1897 qemu_rdma_move_header(rdma, RDMA_WRID_DATA, resp);
1898 if (resp_idx) {
1899 *resp_idx = RDMA_WRID_DATA;
1901 trace_qemu_rdma_exchange_send_received(control_desc(resp->type));
1904 rdma->control_ready_expected = 1;
1906 return 0;
1910 * This is an 'atomic' high-level operation to receive a single, unified
1911 * control-channel message.
1913 static int qemu_rdma_exchange_recv(RDMAContext *rdma, RDMAControlHeader *head,
1914 int expecting)
1916 RDMAControlHeader ready = {
1917 .len = 0,
1918 .type = RDMA_CONTROL_READY,
1919 .repeat = 1,
1921 int ret;
1924 * Inform the source that we're ready to receive a message.
1926 ret = qemu_rdma_post_send_control(rdma, NULL, &ready);
1928 if (ret < 0) {
1929 error_report("Failed to send control buffer!");
1930 return ret;
1934 * Block and wait for the message.
1936 ret = qemu_rdma_exchange_get_response(rdma, head,
1937 expecting, RDMA_WRID_READY);
1939 if (ret < 0) {
1940 return ret;
1943 qemu_rdma_move_header(rdma, RDMA_WRID_READY, head);
1946 * Post a new RECV work request to replace the one we just consumed.
1948 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
1949 if (ret) {
1950 error_report("rdma migration: error posting second control recv!");
1951 return ret;
1954 return 0;
1958 * Write an actual chunk of memory using RDMA.
1960 * If we're using dynamic registration on the dest-side, we have to
1961 * send a registration command first.
1963 static int qemu_rdma_write_one(QEMUFile *f, RDMAContext *rdma,
1964 int current_index, uint64_t current_addr,
1965 uint64_t length)
1967 struct ibv_sge sge;
1968 struct ibv_send_wr send_wr = { 0 };
1969 struct ibv_send_wr *bad_wr;
1970 int reg_result_idx, ret, count = 0;
1971 uint64_t chunk, chunks;
1972 uint8_t *chunk_start, *chunk_end;
1973 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[current_index]);
1974 RDMARegister reg;
1975 RDMARegisterResult *reg_result;
1976 RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT };
1977 RDMAControlHeader head = { .len = sizeof(RDMARegister),
1978 .type = RDMA_CONTROL_REGISTER_REQUEST,
1979 .repeat = 1,
1982 retry:
1983 sge.addr = (uintptr_t)(block->local_host_addr +
1984 (current_addr - block->offset));
1985 sge.length = length;
1987 chunk = ram_chunk_index(block->local_host_addr,
1988 (uint8_t *)(uintptr_t)sge.addr);
1989 chunk_start = ram_chunk_start(block, chunk);
1991 if (block->is_ram_block) {
1992 chunks = length / (1UL << RDMA_REG_CHUNK_SHIFT);
1994 if (chunks && ((length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
1995 chunks--;
1997 } else {
1998 chunks = block->length / (1UL << RDMA_REG_CHUNK_SHIFT);
2000 if (chunks && ((block->length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
2001 chunks--;
2005 trace_qemu_rdma_write_one_top(chunks + 1,
2006 (chunks + 1) *
2007 (1UL << RDMA_REG_CHUNK_SHIFT) / 1024 / 1024);
2009 chunk_end = ram_chunk_end(block, chunk + chunks);
2011 if (!rdma->pin_all) {
2012 #ifdef RDMA_UNREGISTRATION_EXAMPLE
2013 qemu_rdma_unregister_waiting(rdma);
2014 #endif
2017 while (test_bit(chunk, block->transit_bitmap)) {
2018 (void)count;
2019 trace_qemu_rdma_write_one_block(count++, current_index, chunk,
2020 sge.addr, length, rdma->nb_sent, block->nb_chunks);
2022 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2024 if (ret < 0) {
2025 error_report("Failed to Wait for previous write to complete "
2026 "block %d chunk %" PRIu64
2027 " current %" PRIu64 " len %" PRIu64 " %d",
2028 current_index, chunk, sge.addr, length, rdma->nb_sent);
2029 return ret;
2033 if (!rdma->pin_all || !block->is_ram_block) {
2034 if (!block->remote_keys[chunk]) {
2036 * This chunk has not yet been registered, so first check to see
2037 * if the entire chunk is zero. If so, tell the other size to
2038 * memset() + madvise() the entire chunk without RDMA.
2041 if (buffer_is_zero((void *)(uintptr_t)sge.addr, length)) {
2042 RDMACompress comp = {
2043 .offset = current_addr,
2044 .value = 0,
2045 .block_idx = current_index,
2046 .length = length,
2049 head.len = sizeof(comp);
2050 head.type = RDMA_CONTROL_COMPRESS;
2052 trace_qemu_rdma_write_one_zero(chunk, sge.length,
2053 current_index, current_addr);
2055 compress_to_network(rdma, &comp);
2056 ret = qemu_rdma_exchange_send(rdma, &head,
2057 (uint8_t *) &comp, NULL, NULL, NULL);
2059 if (ret < 0) {
2060 return -EIO;
2063 acct_update_position(f, sge.length, true);
2065 return 1;
2069 * Otherwise, tell other side to register.
2071 reg.current_index = current_index;
2072 if (block->is_ram_block) {
2073 reg.key.current_addr = current_addr;
2074 } else {
2075 reg.key.chunk = chunk;
2077 reg.chunks = chunks;
2079 trace_qemu_rdma_write_one_sendreg(chunk, sge.length, current_index,
2080 current_addr);
2082 register_to_network(rdma, &reg);
2083 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) &reg,
2084 &resp, &reg_result_idx, NULL);
2085 if (ret < 0) {
2086 return ret;
2089 /* try to overlap this single registration with the one we sent. */
2090 if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
2091 &sge.lkey, NULL, chunk,
2092 chunk_start, chunk_end)) {
2093 error_report("cannot get lkey");
2094 return -EINVAL;
2097 reg_result = (RDMARegisterResult *)
2098 rdma->wr_data[reg_result_idx].control_curr;
2100 network_to_result(reg_result);
2102 trace_qemu_rdma_write_one_recvregres(block->remote_keys[chunk],
2103 reg_result->rkey, chunk);
2105 block->remote_keys[chunk] = reg_result->rkey;
2106 block->remote_host_addr = reg_result->host_addr;
2107 } else {
2108 /* already registered before */
2109 if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
2110 &sge.lkey, NULL, chunk,
2111 chunk_start, chunk_end)) {
2112 error_report("cannot get lkey!");
2113 return -EINVAL;
2117 send_wr.wr.rdma.rkey = block->remote_keys[chunk];
2118 } else {
2119 send_wr.wr.rdma.rkey = block->remote_rkey;
2121 if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
2122 &sge.lkey, NULL, chunk,
2123 chunk_start, chunk_end)) {
2124 error_report("cannot get lkey!");
2125 return -EINVAL;
2130 * Encode the ram block index and chunk within this wrid.
2131 * We will use this information at the time of completion
2132 * to figure out which bitmap to check against and then which
2133 * chunk in the bitmap to look for.
2135 send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE,
2136 current_index, chunk);
2138 send_wr.opcode = IBV_WR_RDMA_WRITE;
2139 send_wr.send_flags = IBV_SEND_SIGNALED;
2140 send_wr.sg_list = &sge;
2141 send_wr.num_sge = 1;
2142 send_wr.wr.rdma.remote_addr = block->remote_host_addr +
2143 (current_addr - block->offset);
2145 trace_qemu_rdma_write_one_post(chunk, sge.addr, send_wr.wr.rdma.remote_addr,
2146 sge.length);
2149 * ibv_post_send() does not return negative error numbers,
2150 * per the specification they are positive - no idea why.
2152 ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
2154 if (ret == ENOMEM) {
2155 trace_qemu_rdma_write_one_queue_full();
2156 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2157 if (ret < 0) {
2158 error_report("rdma migration: failed to make "
2159 "room in full send queue! %d", ret);
2160 return ret;
2163 goto retry;
2165 } else if (ret > 0) {
2166 perror("rdma migration: post rdma write failed");
2167 return -ret;
2170 set_bit(chunk, block->transit_bitmap);
2171 acct_update_position(f, sge.length, false);
2172 rdma->total_writes++;
2174 return 0;
2178 * Push out any unwritten RDMA operations.
2180 * We support sending out multiple chunks at the same time.
2181 * Not all of them need to get signaled in the completion queue.
2183 static int qemu_rdma_write_flush(QEMUFile *f, RDMAContext *rdma)
2185 int ret;
2187 if (!rdma->current_length) {
2188 return 0;
2191 ret = qemu_rdma_write_one(f, rdma,
2192 rdma->current_index, rdma->current_addr, rdma->current_length);
2194 if (ret < 0) {
2195 return ret;
2198 if (ret == 0) {
2199 rdma->nb_sent++;
2200 trace_qemu_rdma_write_flush(rdma->nb_sent);
2203 rdma->current_length = 0;
2204 rdma->current_addr = 0;
2206 return 0;
2209 static inline int qemu_rdma_buffer_mergable(RDMAContext *rdma,
2210 uint64_t offset, uint64_t len)
2212 RDMALocalBlock *block;
2213 uint8_t *host_addr;
2214 uint8_t *chunk_end;
2216 if (rdma->current_index < 0) {
2217 return 0;
2220 if (rdma->current_chunk < 0) {
2221 return 0;
2224 block = &(rdma->local_ram_blocks.block[rdma->current_index]);
2225 host_addr = block->local_host_addr + (offset - block->offset);
2226 chunk_end = ram_chunk_end(block, rdma->current_chunk);
2228 if (rdma->current_length == 0) {
2229 return 0;
2233 * Only merge into chunk sequentially.
2235 if (offset != (rdma->current_addr + rdma->current_length)) {
2236 return 0;
2239 if (offset < block->offset) {
2240 return 0;
2243 if ((offset + len) > (block->offset + block->length)) {
2244 return 0;
2247 if ((host_addr + len) > chunk_end) {
2248 return 0;
2251 return 1;
2255 * We're not actually writing here, but doing three things:
2257 * 1. Identify the chunk the buffer belongs to.
2258 * 2. If the chunk is full or the buffer doesn't belong to the current
2259 * chunk, then start a new chunk and flush() the old chunk.
2260 * 3. To keep the hardware busy, we also group chunks into batches
2261 * and only require that a batch gets acknowledged in the completion
2262 * qeueue instead of each individual chunk.
2264 static int qemu_rdma_write(QEMUFile *f, RDMAContext *rdma,
2265 uint64_t block_offset, uint64_t offset,
2266 uint64_t len)
2268 uint64_t current_addr = block_offset + offset;
2269 uint64_t index = rdma->current_index;
2270 uint64_t chunk = rdma->current_chunk;
2271 int ret;
2273 /* If we cannot merge it, we flush the current buffer first. */
2274 if (!qemu_rdma_buffer_mergable(rdma, current_addr, len)) {
2275 ret = qemu_rdma_write_flush(f, rdma);
2276 if (ret) {
2277 return ret;
2279 rdma->current_length = 0;
2280 rdma->current_addr = current_addr;
2282 ret = qemu_rdma_search_ram_block(rdma, block_offset,
2283 offset, len, &index, &chunk);
2284 if (ret) {
2285 error_report("ram block search failed");
2286 return ret;
2288 rdma->current_index = index;
2289 rdma->current_chunk = chunk;
2292 /* merge it */
2293 rdma->current_length += len;
2295 /* flush it if buffer is too large */
2296 if (rdma->current_length >= RDMA_MERGE_MAX) {
2297 return qemu_rdma_write_flush(f, rdma);
2300 return 0;
2303 static void qemu_rdma_cleanup(RDMAContext *rdma)
2305 int idx;
2307 if (rdma->cm_id && rdma->connected) {
2308 if ((rdma->error_state ||
2309 migrate_get_current()->state == MIGRATION_STATUS_CANCELLING) &&
2310 !rdma->received_error) {
2311 RDMAControlHeader head = { .len = 0,
2312 .type = RDMA_CONTROL_ERROR,
2313 .repeat = 1,
2315 error_report("Early error. Sending error.");
2316 qemu_rdma_post_send_control(rdma, NULL, &head);
2319 rdma_disconnect(rdma->cm_id);
2320 trace_qemu_rdma_cleanup_disconnect();
2321 rdma->connected = false;
2324 qemu_set_fd_handler(rdma->channel->fd, NULL, NULL, NULL);
2325 g_free(rdma->dest_blocks);
2326 rdma->dest_blocks = NULL;
2328 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2329 if (rdma->wr_data[idx].control_mr) {
2330 rdma->total_registrations--;
2331 ibv_dereg_mr(rdma->wr_data[idx].control_mr);
2333 rdma->wr_data[idx].control_mr = NULL;
2336 if (rdma->local_ram_blocks.block) {
2337 while (rdma->local_ram_blocks.nb_blocks) {
2338 rdma_delete_block(rdma, &rdma->local_ram_blocks.block[0]);
2342 if (rdma->qp) {
2343 rdma_destroy_qp(rdma->cm_id);
2344 rdma->qp = NULL;
2346 if (rdma->cq) {
2347 ibv_destroy_cq(rdma->cq);
2348 rdma->cq = NULL;
2350 if (rdma->comp_channel) {
2351 ibv_destroy_comp_channel(rdma->comp_channel);
2352 rdma->comp_channel = NULL;
2354 if (rdma->pd) {
2355 ibv_dealloc_pd(rdma->pd);
2356 rdma->pd = NULL;
2358 if (rdma->cm_id) {
2359 rdma_destroy_id(rdma->cm_id);
2360 rdma->cm_id = NULL;
2363 /* the destination side, listen_id and channel is shared */
2364 if (rdma->listen_id) {
2365 if (!rdma->is_return_path) {
2366 rdma_destroy_id(rdma->listen_id);
2368 rdma->listen_id = NULL;
2370 if (rdma->channel) {
2371 if (!rdma->is_return_path) {
2372 rdma_destroy_event_channel(rdma->channel);
2374 rdma->channel = NULL;
2378 if (rdma->channel) {
2379 rdma_destroy_event_channel(rdma->channel);
2380 rdma->channel = NULL;
2382 g_free(rdma->host);
2383 rdma->host = NULL;
2387 static int qemu_rdma_source_init(RDMAContext *rdma, bool pin_all, Error **errp)
2389 int ret, idx;
2390 Error *local_err = NULL, **temp = &local_err;
2393 * Will be validated against destination's actual capabilities
2394 * after the connect() completes.
2396 rdma->pin_all = pin_all;
2398 ret = qemu_rdma_resolve_host(rdma, temp);
2399 if (ret) {
2400 goto err_rdma_source_init;
2403 ret = qemu_rdma_alloc_pd_cq(rdma);
2404 if (ret) {
2405 ERROR(temp, "rdma migration: error allocating pd and cq! Your mlock()"
2406 " limits may be too low. Please check $ ulimit -a # and "
2407 "search for 'ulimit -l' in the output");
2408 goto err_rdma_source_init;
2411 ret = qemu_rdma_alloc_qp(rdma);
2412 if (ret) {
2413 ERROR(temp, "rdma migration: error allocating qp!");
2414 goto err_rdma_source_init;
2417 ret = qemu_rdma_init_ram_blocks(rdma);
2418 if (ret) {
2419 ERROR(temp, "rdma migration: error initializing ram blocks!");
2420 goto err_rdma_source_init;
2423 /* Build the hash that maps from offset to RAMBlock */
2424 rdma->blockmap = g_hash_table_new(g_direct_hash, g_direct_equal);
2425 for (idx = 0; idx < rdma->local_ram_blocks.nb_blocks; idx++) {
2426 g_hash_table_insert(rdma->blockmap,
2427 (void *)(uintptr_t)rdma->local_ram_blocks.block[idx].offset,
2428 &rdma->local_ram_blocks.block[idx]);
2431 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2432 ret = qemu_rdma_reg_control(rdma, idx);
2433 if (ret) {
2434 ERROR(temp, "rdma migration: error registering %d control!",
2435 idx);
2436 goto err_rdma_source_init;
2440 return 0;
2442 err_rdma_source_init:
2443 error_propagate(errp, local_err);
2444 qemu_rdma_cleanup(rdma);
2445 return -1;
2448 static int qemu_rdma_connect(RDMAContext *rdma, Error **errp)
2450 RDMACapabilities cap = {
2451 .version = RDMA_CONTROL_VERSION_CURRENT,
2452 .flags = 0,
2454 struct rdma_conn_param conn_param = { .initiator_depth = 2,
2455 .retry_count = 5,
2456 .private_data = &cap,
2457 .private_data_len = sizeof(cap),
2459 struct rdma_cm_event *cm_event;
2460 int ret;
2463 * Only negotiate the capability with destination if the user
2464 * on the source first requested the capability.
2466 if (rdma->pin_all) {
2467 trace_qemu_rdma_connect_pin_all_requested();
2468 cap.flags |= RDMA_CAPABILITY_PIN_ALL;
2471 caps_to_network(&cap);
2473 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
2474 if (ret) {
2475 ERROR(errp, "posting second control recv");
2476 goto err_rdma_source_connect;
2479 ret = rdma_connect(rdma->cm_id, &conn_param);
2480 if (ret) {
2481 perror("rdma_connect");
2482 ERROR(errp, "connecting to destination!");
2483 goto err_rdma_source_connect;
2486 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2487 if (ret) {
2488 perror("rdma_get_cm_event after rdma_connect");
2489 ERROR(errp, "connecting to destination!");
2490 rdma_ack_cm_event(cm_event);
2491 goto err_rdma_source_connect;
2494 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
2495 perror("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect");
2496 ERROR(errp, "connecting to destination!");
2497 rdma_ack_cm_event(cm_event);
2498 goto err_rdma_source_connect;
2500 rdma->connected = true;
2502 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
2503 network_to_caps(&cap);
2506 * Verify that the *requested* capabilities are supported by the destination
2507 * and disable them otherwise.
2509 if (rdma->pin_all && !(cap.flags & RDMA_CAPABILITY_PIN_ALL)) {
2510 ERROR(errp, "Server cannot support pinning all memory. "
2511 "Will register memory dynamically.");
2512 rdma->pin_all = false;
2515 trace_qemu_rdma_connect_pin_all_outcome(rdma->pin_all);
2517 rdma_ack_cm_event(cm_event);
2519 rdma->control_ready_expected = 1;
2520 rdma->nb_sent = 0;
2521 return 0;
2523 err_rdma_source_connect:
2524 qemu_rdma_cleanup(rdma);
2525 return -1;
2528 static int qemu_rdma_dest_init(RDMAContext *rdma, Error **errp)
2530 int ret, idx;
2531 struct rdma_cm_id *listen_id;
2532 char ip[40] = "unknown";
2533 struct rdma_addrinfo *res, *e;
2534 char port_str[16];
2536 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2537 rdma->wr_data[idx].control_len = 0;
2538 rdma->wr_data[idx].control_curr = NULL;
2541 if (!rdma->host || !rdma->host[0]) {
2542 ERROR(errp, "RDMA host is not set!");
2543 rdma->error_state = -EINVAL;
2544 return -1;
2546 /* create CM channel */
2547 rdma->channel = rdma_create_event_channel();
2548 if (!rdma->channel) {
2549 ERROR(errp, "could not create rdma event channel");
2550 rdma->error_state = -EINVAL;
2551 return -1;
2554 /* create CM id */
2555 ret = rdma_create_id(rdma->channel, &listen_id, NULL, RDMA_PS_TCP);
2556 if (ret) {
2557 ERROR(errp, "could not create cm_id!");
2558 goto err_dest_init_create_listen_id;
2561 snprintf(port_str, 16, "%d", rdma->port);
2562 port_str[15] = '\0';
2564 ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
2565 if (ret < 0) {
2566 ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
2567 goto err_dest_init_bind_addr;
2570 for (e = res; e != NULL; e = e->ai_next) {
2571 inet_ntop(e->ai_family,
2572 &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
2573 trace_qemu_rdma_dest_init_trying(rdma->host, ip);
2574 ret = rdma_bind_addr(listen_id, e->ai_dst_addr);
2575 if (ret) {
2576 continue;
2578 if (e->ai_family == AF_INET6) {
2579 ret = qemu_rdma_broken_ipv6_kernel(listen_id->verbs, errp);
2580 if (ret) {
2581 continue;
2584 break;
2587 if (!e) {
2588 ERROR(errp, "Error: could not rdma_bind_addr!");
2589 goto err_dest_init_bind_addr;
2592 rdma->listen_id = listen_id;
2593 qemu_rdma_dump_gid("dest_init", listen_id);
2594 return 0;
2596 err_dest_init_bind_addr:
2597 rdma_destroy_id(listen_id);
2598 err_dest_init_create_listen_id:
2599 rdma_destroy_event_channel(rdma->channel);
2600 rdma->channel = NULL;
2601 rdma->error_state = ret;
2602 return ret;
2606 static void qemu_rdma_return_path_dest_init(RDMAContext *rdma_return_path,
2607 RDMAContext *rdma)
2609 int idx;
2611 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2612 rdma_return_path->wr_data[idx].control_len = 0;
2613 rdma_return_path->wr_data[idx].control_curr = NULL;
2616 /*the CM channel and CM id is shared*/
2617 rdma_return_path->channel = rdma->channel;
2618 rdma_return_path->listen_id = rdma->listen_id;
2620 rdma->return_path = rdma_return_path;
2621 rdma_return_path->return_path = rdma;
2622 rdma_return_path->is_return_path = true;
2625 static void *qemu_rdma_data_init(const char *host_port, Error **errp)
2627 RDMAContext *rdma = NULL;
2628 InetSocketAddress *addr;
2630 if (host_port) {
2631 rdma = g_new0(RDMAContext, 1);
2632 rdma->current_index = -1;
2633 rdma->current_chunk = -1;
2635 addr = g_new(InetSocketAddress, 1);
2636 if (!inet_parse(addr, host_port, NULL)) {
2637 rdma->port = atoi(addr->port);
2638 rdma->host = g_strdup(addr->host);
2639 } else {
2640 ERROR(errp, "bad RDMA migration address '%s'", host_port);
2641 g_free(rdma);
2642 rdma = NULL;
2645 qapi_free_InetSocketAddress(addr);
2648 return rdma;
2652 * QEMUFile interface to the control channel.
2653 * SEND messages for control only.
2654 * VM's ram is handled with regular RDMA messages.
2656 static ssize_t qio_channel_rdma_writev(QIOChannel *ioc,
2657 const struct iovec *iov,
2658 size_t niov,
2659 int *fds,
2660 size_t nfds,
2661 Error **errp)
2663 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2664 QEMUFile *f = rioc->file;
2665 RDMAContext *rdma;
2666 int ret;
2667 ssize_t done = 0;
2668 size_t i;
2669 size_t len = 0;
2671 rcu_read_lock();
2672 rdma = atomic_rcu_read(&rioc->rdmaout);
2674 if (!rdma) {
2675 rcu_read_unlock();
2676 return -EIO;
2679 CHECK_ERROR_STATE();
2682 * Push out any writes that
2683 * we're queued up for VM's ram.
2685 ret = qemu_rdma_write_flush(f, rdma);
2686 if (ret < 0) {
2687 rdma->error_state = ret;
2688 rcu_read_unlock();
2689 return ret;
2692 for (i = 0; i < niov; i++) {
2693 size_t remaining = iov[i].iov_len;
2694 uint8_t * data = (void *)iov[i].iov_base;
2695 while (remaining) {
2696 RDMAControlHeader head;
2698 len = MIN(remaining, RDMA_SEND_INCREMENT);
2699 remaining -= len;
2701 head.len = len;
2702 head.type = RDMA_CONTROL_QEMU_FILE;
2704 ret = qemu_rdma_exchange_send(rdma, &head, data, NULL, NULL, NULL);
2706 if (ret < 0) {
2707 rdma->error_state = ret;
2708 rcu_read_unlock();
2709 return ret;
2712 data += len;
2713 done += len;
2717 rcu_read_unlock();
2718 return done;
2721 static size_t qemu_rdma_fill(RDMAContext *rdma, uint8_t *buf,
2722 size_t size, int idx)
2724 size_t len = 0;
2726 if (rdma->wr_data[idx].control_len) {
2727 trace_qemu_rdma_fill(rdma->wr_data[idx].control_len, size);
2729 len = MIN(size, rdma->wr_data[idx].control_len);
2730 memcpy(buf, rdma->wr_data[idx].control_curr, len);
2731 rdma->wr_data[idx].control_curr += len;
2732 rdma->wr_data[idx].control_len -= len;
2735 return len;
2739 * QEMUFile interface to the control channel.
2740 * RDMA links don't use bytestreams, so we have to
2741 * return bytes to QEMUFile opportunistically.
2743 static ssize_t qio_channel_rdma_readv(QIOChannel *ioc,
2744 const struct iovec *iov,
2745 size_t niov,
2746 int **fds,
2747 size_t *nfds,
2748 Error **errp)
2750 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2751 RDMAContext *rdma;
2752 RDMAControlHeader head;
2753 int ret = 0;
2754 ssize_t i;
2755 size_t done = 0;
2757 rcu_read_lock();
2758 rdma = atomic_rcu_read(&rioc->rdmain);
2760 if (!rdma) {
2761 rcu_read_unlock();
2762 return -EIO;
2765 CHECK_ERROR_STATE();
2767 for (i = 0; i < niov; i++) {
2768 size_t want = iov[i].iov_len;
2769 uint8_t *data = (void *)iov[i].iov_base;
2772 * First, we hold on to the last SEND message we
2773 * were given and dish out the bytes until we run
2774 * out of bytes.
2776 ret = qemu_rdma_fill(rdma, data, want, 0);
2777 done += ret;
2778 want -= ret;
2779 /* Got what we needed, so go to next iovec */
2780 if (want == 0) {
2781 continue;
2784 /* If we got any data so far, then don't wait
2785 * for more, just return what we have */
2786 if (done > 0) {
2787 break;
2791 /* We've got nothing at all, so lets wait for
2792 * more to arrive
2794 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_QEMU_FILE);
2796 if (ret < 0) {
2797 rdma->error_state = ret;
2798 rcu_read_unlock();
2799 return ret;
2803 * SEND was received with new bytes, now try again.
2805 ret = qemu_rdma_fill(rdma, data, want, 0);
2806 done += ret;
2807 want -= ret;
2809 /* Still didn't get enough, so lets just return */
2810 if (want) {
2811 if (done == 0) {
2812 rcu_read_unlock();
2813 return QIO_CHANNEL_ERR_BLOCK;
2814 } else {
2815 break;
2819 rcu_read_unlock();
2820 return done;
2824 * Block until all the outstanding chunks have been delivered by the hardware.
2826 static int qemu_rdma_drain_cq(QEMUFile *f, RDMAContext *rdma)
2828 int ret;
2830 if (qemu_rdma_write_flush(f, rdma) < 0) {
2831 return -EIO;
2834 while (rdma->nb_sent) {
2835 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2836 if (ret < 0) {
2837 error_report("rdma migration: complete polling error!");
2838 return -EIO;
2842 qemu_rdma_unregister_waiting(rdma);
2844 return 0;
2848 static int qio_channel_rdma_set_blocking(QIOChannel *ioc,
2849 bool blocking,
2850 Error **errp)
2852 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2853 /* XXX we should make readv/writev actually honour this :-) */
2854 rioc->blocking = blocking;
2855 return 0;
2859 typedef struct QIOChannelRDMASource QIOChannelRDMASource;
2860 struct QIOChannelRDMASource {
2861 GSource parent;
2862 QIOChannelRDMA *rioc;
2863 GIOCondition condition;
2866 static gboolean
2867 qio_channel_rdma_source_prepare(GSource *source,
2868 gint *timeout)
2870 QIOChannelRDMASource *rsource = (QIOChannelRDMASource *)source;
2871 RDMAContext *rdma;
2872 GIOCondition cond = 0;
2873 *timeout = -1;
2875 rcu_read_lock();
2876 if (rsource->condition == G_IO_IN) {
2877 rdma = atomic_rcu_read(&rsource->rioc->rdmain);
2878 } else {
2879 rdma = atomic_rcu_read(&rsource->rioc->rdmaout);
2882 if (!rdma) {
2883 error_report("RDMAContext is NULL when prepare Gsource");
2884 rcu_read_unlock();
2885 return FALSE;
2888 if (rdma->wr_data[0].control_len) {
2889 cond |= G_IO_IN;
2891 cond |= G_IO_OUT;
2893 rcu_read_unlock();
2894 return cond & rsource->condition;
2897 static gboolean
2898 qio_channel_rdma_source_check(GSource *source)
2900 QIOChannelRDMASource *rsource = (QIOChannelRDMASource *)source;
2901 RDMAContext *rdma;
2902 GIOCondition cond = 0;
2904 rcu_read_lock();
2905 if (rsource->condition == G_IO_IN) {
2906 rdma = atomic_rcu_read(&rsource->rioc->rdmain);
2907 } else {
2908 rdma = atomic_rcu_read(&rsource->rioc->rdmaout);
2911 if (!rdma) {
2912 error_report("RDMAContext is NULL when check Gsource");
2913 rcu_read_unlock();
2914 return FALSE;
2917 if (rdma->wr_data[0].control_len) {
2918 cond |= G_IO_IN;
2920 cond |= G_IO_OUT;
2922 rcu_read_unlock();
2923 return cond & rsource->condition;
2926 static gboolean
2927 qio_channel_rdma_source_dispatch(GSource *source,
2928 GSourceFunc callback,
2929 gpointer user_data)
2931 QIOChannelFunc func = (QIOChannelFunc)callback;
2932 QIOChannelRDMASource *rsource = (QIOChannelRDMASource *)source;
2933 RDMAContext *rdma;
2934 GIOCondition cond = 0;
2936 rcu_read_lock();
2937 if (rsource->condition == G_IO_IN) {
2938 rdma = atomic_rcu_read(&rsource->rioc->rdmain);
2939 } else {
2940 rdma = atomic_rcu_read(&rsource->rioc->rdmaout);
2943 if (!rdma) {
2944 error_report("RDMAContext is NULL when dispatch Gsource");
2945 rcu_read_unlock();
2946 return FALSE;
2949 if (rdma->wr_data[0].control_len) {
2950 cond |= G_IO_IN;
2952 cond |= G_IO_OUT;
2954 rcu_read_unlock();
2955 return (*func)(QIO_CHANNEL(rsource->rioc),
2956 (cond & rsource->condition),
2957 user_data);
2960 static void
2961 qio_channel_rdma_source_finalize(GSource *source)
2963 QIOChannelRDMASource *ssource = (QIOChannelRDMASource *)source;
2965 object_unref(OBJECT(ssource->rioc));
2968 GSourceFuncs qio_channel_rdma_source_funcs = {
2969 qio_channel_rdma_source_prepare,
2970 qio_channel_rdma_source_check,
2971 qio_channel_rdma_source_dispatch,
2972 qio_channel_rdma_source_finalize
2975 static GSource *qio_channel_rdma_create_watch(QIOChannel *ioc,
2976 GIOCondition condition)
2978 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2979 QIOChannelRDMASource *ssource;
2980 GSource *source;
2982 source = g_source_new(&qio_channel_rdma_source_funcs,
2983 sizeof(QIOChannelRDMASource));
2984 ssource = (QIOChannelRDMASource *)source;
2986 ssource->rioc = rioc;
2987 object_ref(OBJECT(rioc));
2989 ssource->condition = condition;
2991 return source;
2994 static void qio_channel_rdma_set_aio_fd_handler(QIOChannel *ioc,
2995 AioContext *ctx,
2996 IOHandler *io_read,
2997 IOHandler *io_write,
2998 void *opaque)
3000 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3001 if (io_read) {
3002 aio_set_fd_handler(ctx, rioc->rdmain->comp_channel->fd,
3003 false, io_read, io_write, NULL, opaque);
3004 } else {
3005 aio_set_fd_handler(ctx, rioc->rdmaout->comp_channel->fd,
3006 false, io_read, io_write, NULL, opaque);
3010 static int qio_channel_rdma_close(QIOChannel *ioc,
3011 Error **errp)
3013 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3014 RDMAContext *rdmain, *rdmaout;
3015 trace_qemu_rdma_close();
3017 rdmain = rioc->rdmain;
3018 if (rdmain) {
3019 atomic_rcu_set(&rioc->rdmain, NULL);
3022 rdmaout = rioc->rdmaout;
3023 if (rdmaout) {
3024 atomic_rcu_set(&rioc->rdmaout, NULL);
3027 synchronize_rcu();
3029 if (rdmain) {
3030 qemu_rdma_cleanup(rdmain);
3033 if (rdmaout) {
3034 qemu_rdma_cleanup(rdmaout);
3037 g_free(rdmain);
3038 g_free(rdmaout);
3040 return 0;
3043 static int
3044 qio_channel_rdma_shutdown(QIOChannel *ioc,
3045 QIOChannelShutdown how,
3046 Error **errp)
3048 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3049 RDMAContext *rdmain, *rdmaout;
3051 rcu_read_lock();
3053 rdmain = atomic_rcu_read(&rioc->rdmain);
3054 rdmaout = atomic_rcu_read(&rioc->rdmain);
3056 switch (how) {
3057 case QIO_CHANNEL_SHUTDOWN_READ:
3058 if (rdmain) {
3059 rdmain->error_state = -1;
3061 break;
3062 case QIO_CHANNEL_SHUTDOWN_WRITE:
3063 if (rdmaout) {
3064 rdmaout->error_state = -1;
3066 break;
3067 case QIO_CHANNEL_SHUTDOWN_BOTH:
3068 default:
3069 if (rdmain) {
3070 rdmain->error_state = -1;
3072 if (rdmaout) {
3073 rdmaout->error_state = -1;
3075 break;
3078 rcu_read_unlock();
3079 return 0;
3083 * Parameters:
3084 * @offset == 0 :
3085 * This means that 'block_offset' is a full virtual address that does not
3086 * belong to a RAMBlock of the virtual machine and instead
3087 * represents a private malloc'd memory area that the caller wishes to
3088 * transfer.
3090 * @offset != 0 :
3091 * Offset is an offset to be added to block_offset and used
3092 * to also lookup the corresponding RAMBlock.
3094 * @size > 0 :
3095 * Initiate an transfer this size.
3097 * @size == 0 :
3098 * A 'hint' or 'advice' that means that we wish to speculatively
3099 * and asynchronously unregister this memory. In this case, there is no
3100 * guarantee that the unregister will actually happen, for example,
3101 * if the memory is being actively transmitted. Additionally, the memory
3102 * may be re-registered at any future time if a write within the same
3103 * chunk was requested again, even if you attempted to unregister it
3104 * here.
3106 * @size < 0 : TODO, not yet supported
3107 * Unregister the memory NOW. This means that the caller does not
3108 * expect there to be any future RDMA transfers and we just want to clean
3109 * things up. This is used in case the upper layer owns the memory and
3110 * cannot wait for qemu_fclose() to occur.
3112 * @bytes_sent : User-specificed pointer to indicate how many bytes were
3113 * sent. Usually, this will not be more than a few bytes of
3114 * the protocol because most transfers are sent asynchronously.
3116 static size_t qemu_rdma_save_page(QEMUFile *f, void *opaque,
3117 ram_addr_t block_offset, ram_addr_t offset,
3118 size_t size, uint64_t *bytes_sent)
3120 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(opaque);
3121 RDMAContext *rdma;
3122 int ret;
3124 rcu_read_lock();
3125 rdma = atomic_rcu_read(&rioc->rdmaout);
3127 if (!rdma) {
3128 rcu_read_unlock();
3129 return -EIO;
3132 CHECK_ERROR_STATE();
3134 if (migrate_get_current()->state == MIGRATION_STATUS_POSTCOPY_ACTIVE) {
3135 rcu_read_unlock();
3136 return RAM_SAVE_CONTROL_NOT_SUPP;
3139 qemu_fflush(f);
3141 if (size > 0) {
3143 * Add this page to the current 'chunk'. If the chunk
3144 * is full, or the page doen't belong to the current chunk,
3145 * an actual RDMA write will occur and a new chunk will be formed.
3147 ret = qemu_rdma_write(f, rdma, block_offset, offset, size);
3148 if (ret < 0) {
3149 error_report("rdma migration: write error! %d", ret);
3150 goto err;
3154 * We always return 1 bytes because the RDMA
3155 * protocol is completely asynchronous. We do not yet know
3156 * whether an identified chunk is zero or not because we're
3157 * waiting for other pages to potentially be merged with
3158 * the current chunk. So, we have to call qemu_update_position()
3159 * later on when the actual write occurs.
3161 if (bytes_sent) {
3162 *bytes_sent = 1;
3164 } else {
3165 uint64_t index, chunk;
3167 /* TODO: Change QEMUFileOps prototype to be signed: size_t => long
3168 if (size < 0) {
3169 ret = qemu_rdma_drain_cq(f, rdma);
3170 if (ret < 0) {
3171 fprintf(stderr, "rdma: failed to synchronously drain"
3172 " completion queue before unregistration.\n");
3173 goto err;
3178 ret = qemu_rdma_search_ram_block(rdma, block_offset,
3179 offset, size, &index, &chunk);
3181 if (ret) {
3182 error_report("ram block search failed");
3183 goto err;
3186 qemu_rdma_signal_unregister(rdma, index, chunk, 0);
3189 * TODO: Synchronous, guaranteed unregistration (should not occur during
3190 * fast-path). Otherwise, unregisters will process on the next call to
3191 * qemu_rdma_drain_cq()
3192 if (size < 0) {
3193 qemu_rdma_unregister_waiting(rdma);
3199 * Drain the Completion Queue if possible, but do not block,
3200 * just poll.
3202 * If nothing to poll, the end of the iteration will do this
3203 * again to make sure we don't overflow the request queue.
3205 while (1) {
3206 uint64_t wr_id, wr_id_in;
3207 int ret = qemu_rdma_poll(rdma, &wr_id_in, NULL);
3208 if (ret < 0) {
3209 error_report("rdma migration: polling error! %d", ret);
3210 goto err;
3213 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
3215 if (wr_id == RDMA_WRID_NONE) {
3216 break;
3220 rcu_read_unlock();
3221 return RAM_SAVE_CONTROL_DELAYED;
3222 err:
3223 rdma->error_state = ret;
3224 rcu_read_unlock();
3225 return ret;
3228 static void rdma_accept_incoming_migration(void *opaque);
3230 static void rdma_cm_poll_handler(void *opaque)
3232 RDMAContext *rdma = opaque;
3233 int ret;
3234 struct rdma_cm_event *cm_event;
3235 MigrationIncomingState *mis = migration_incoming_get_current();
3237 ret = rdma_get_cm_event(rdma->channel, &cm_event);
3238 if (ret) {
3239 error_report("get_cm_event failed %d", errno);
3240 return;
3242 rdma_ack_cm_event(cm_event);
3244 if (cm_event->event == RDMA_CM_EVENT_DISCONNECTED ||
3245 cm_event->event == RDMA_CM_EVENT_DEVICE_REMOVAL) {
3246 error_report("receive cm event, cm event is %d", cm_event->event);
3247 rdma->error_state = -EPIPE;
3248 if (rdma->return_path) {
3249 rdma->return_path->error_state = -EPIPE;
3252 if (mis->migration_incoming_co) {
3253 qemu_coroutine_enter(mis->migration_incoming_co);
3255 return;
3259 static int qemu_rdma_accept(RDMAContext *rdma)
3261 RDMACapabilities cap;
3262 struct rdma_conn_param conn_param = {
3263 .responder_resources = 2,
3264 .private_data = &cap,
3265 .private_data_len = sizeof(cap),
3267 struct rdma_cm_event *cm_event;
3268 struct ibv_context *verbs;
3269 int ret = -EINVAL;
3270 int idx;
3272 ret = rdma_get_cm_event(rdma->channel, &cm_event);
3273 if (ret) {
3274 goto err_rdma_dest_wait;
3277 if (cm_event->event != RDMA_CM_EVENT_CONNECT_REQUEST) {
3278 rdma_ack_cm_event(cm_event);
3279 goto err_rdma_dest_wait;
3282 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
3284 network_to_caps(&cap);
3286 if (cap.version < 1 || cap.version > RDMA_CONTROL_VERSION_CURRENT) {
3287 error_report("Unknown source RDMA version: %d, bailing...",
3288 cap.version);
3289 rdma_ack_cm_event(cm_event);
3290 goto err_rdma_dest_wait;
3294 * Respond with only the capabilities this version of QEMU knows about.
3296 cap.flags &= known_capabilities;
3299 * Enable the ones that we do know about.
3300 * Add other checks here as new ones are introduced.
3302 if (cap.flags & RDMA_CAPABILITY_PIN_ALL) {
3303 rdma->pin_all = true;
3306 rdma->cm_id = cm_event->id;
3307 verbs = cm_event->id->verbs;
3309 rdma_ack_cm_event(cm_event);
3311 trace_qemu_rdma_accept_pin_state(rdma->pin_all);
3313 caps_to_network(&cap);
3315 trace_qemu_rdma_accept_pin_verbsc(verbs);
3317 if (!rdma->verbs) {
3318 rdma->verbs = verbs;
3319 } else if (rdma->verbs != verbs) {
3320 error_report("ibv context not matching %p, %p!", rdma->verbs,
3321 verbs);
3322 goto err_rdma_dest_wait;
3325 qemu_rdma_dump_id("dest_init", verbs);
3327 ret = qemu_rdma_alloc_pd_cq(rdma);
3328 if (ret) {
3329 error_report("rdma migration: error allocating pd and cq!");
3330 goto err_rdma_dest_wait;
3333 ret = qemu_rdma_alloc_qp(rdma);
3334 if (ret) {
3335 error_report("rdma migration: error allocating qp!");
3336 goto err_rdma_dest_wait;
3339 ret = qemu_rdma_init_ram_blocks(rdma);
3340 if (ret) {
3341 error_report("rdma migration: error initializing ram blocks!");
3342 goto err_rdma_dest_wait;
3345 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
3346 ret = qemu_rdma_reg_control(rdma, idx);
3347 if (ret) {
3348 error_report("rdma: error registering %d control", idx);
3349 goto err_rdma_dest_wait;
3353 /* Accept the second connection request for return path */
3354 if (migrate_postcopy() && !rdma->is_return_path) {
3355 qemu_set_fd_handler(rdma->channel->fd, rdma_accept_incoming_migration,
3356 NULL,
3357 (void *)(intptr_t)rdma->return_path);
3358 } else {
3359 qemu_set_fd_handler(rdma->channel->fd, rdma_cm_poll_handler,
3360 NULL, rdma);
3363 ret = rdma_accept(rdma->cm_id, &conn_param);
3364 if (ret) {
3365 error_report("rdma_accept returns %d", ret);
3366 goto err_rdma_dest_wait;
3369 ret = rdma_get_cm_event(rdma->channel, &cm_event);
3370 if (ret) {
3371 error_report("rdma_accept get_cm_event failed %d", ret);
3372 goto err_rdma_dest_wait;
3375 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
3376 error_report("rdma_accept not event established");
3377 rdma_ack_cm_event(cm_event);
3378 goto err_rdma_dest_wait;
3381 rdma_ack_cm_event(cm_event);
3382 rdma->connected = true;
3384 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
3385 if (ret) {
3386 error_report("rdma migration: error posting second control recv");
3387 goto err_rdma_dest_wait;
3390 qemu_rdma_dump_gid("dest_connect", rdma->cm_id);
3392 return 0;
3394 err_rdma_dest_wait:
3395 rdma->error_state = ret;
3396 qemu_rdma_cleanup(rdma);
3397 return ret;
3400 static int dest_ram_sort_func(const void *a, const void *b)
3402 unsigned int a_index = ((const RDMALocalBlock *)a)->src_index;
3403 unsigned int b_index = ((const RDMALocalBlock *)b)->src_index;
3405 return (a_index < b_index) ? -1 : (a_index != b_index);
3409 * During each iteration of the migration, we listen for instructions
3410 * by the source VM to perform dynamic page registrations before they
3411 * can perform RDMA operations.
3413 * We respond with the 'rkey'.
3415 * Keep doing this until the source tells us to stop.
3417 static int qemu_rdma_registration_handle(QEMUFile *f, void *opaque)
3419 RDMAControlHeader reg_resp = { .len = sizeof(RDMARegisterResult),
3420 .type = RDMA_CONTROL_REGISTER_RESULT,
3421 .repeat = 0,
3423 RDMAControlHeader unreg_resp = { .len = 0,
3424 .type = RDMA_CONTROL_UNREGISTER_FINISHED,
3425 .repeat = 0,
3427 RDMAControlHeader blocks = { .type = RDMA_CONTROL_RAM_BLOCKS_RESULT,
3428 .repeat = 1 };
3429 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(opaque);
3430 RDMAContext *rdma;
3431 RDMALocalBlocks *local;
3432 RDMAControlHeader head;
3433 RDMARegister *reg, *registers;
3434 RDMACompress *comp;
3435 RDMARegisterResult *reg_result;
3436 static RDMARegisterResult results[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE];
3437 RDMALocalBlock *block;
3438 void *host_addr;
3439 int ret = 0;
3440 int idx = 0;
3441 int count = 0;
3442 int i = 0;
3444 rcu_read_lock();
3445 rdma = atomic_rcu_read(&rioc->rdmain);
3447 if (!rdma) {
3448 rcu_read_unlock();
3449 return -EIO;
3452 CHECK_ERROR_STATE();
3454 local = &rdma->local_ram_blocks;
3455 do {
3456 trace_qemu_rdma_registration_handle_wait();
3458 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_NONE);
3460 if (ret < 0) {
3461 break;
3464 if (head.repeat > RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE) {
3465 error_report("rdma: Too many requests in this message (%d)."
3466 "Bailing.", head.repeat);
3467 ret = -EIO;
3468 break;
3471 switch (head.type) {
3472 case RDMA_CONTROL_COMPRESS:
3473 comp = (RDMACompress *) rdma->wr_data[idx].control_curr;
3474 network_to_compress(comp);
3476 trace_qemu_rdma_registration_handle_compress(comp->length,
3477 comp->block_idx,
3478 comp->offset);
3479 if (comp->block_idx >= rdma->local_ram_blocks.nb_blocks) {
3480 error_report("rdma: 'compress' bad block index %u (vs %d)",
3481 (unsigned int)comp->block_idx,
3482 rdma->local_ram_blocks.nb_blocks);
3483 ret = -EIO;
3484 goto out;
3486 block = &(rdma->local_ram_blocks.block[comp->block_idx]);
3488 host_addr = block->local_host_addr +
3489 (comp->offset - block->offset);
3491 ram_handle_compressed(host_addr, comp->value, comp->length);
3492 break;
3494 case RDMA_CONTROL_REGISTER_FINISHED:
3495 trace_qemu_rdma_registration_handle_finished();
3496 goto out;
3498 case RDMA_CONTROL_RAM_BLOCKS_REQUEST:
3499 trace_qemu_rdma_registration_handle_ram_blocks();
3501 /* Sort our local RAM Block list so it's the same as the source,
3502 * we can do this since we've filled in a src_index in the list
3503 * as we received the RAMBlock list earlier.
3505 qsort(rdma->local_ram_blocks.block,
3506 rdma->local_ram_blocks.nb_blocks,
3507 sizeof(RDMALocalBlock), dest_ram_sort_func);
3508 for (i = 0; i < local->nb_blocks; i++) {
3509 local->block[i].index = i;
3512 if (rdma->pin_all) {
3513 ret = qemu_rdma_reg_whole_ram_blocks(rdma);
3514 if (ret) {
3515 error_report("rdma migration: error dest "
3516 "registering ram blocks");
3517 goto out;
3522 * Dest uses this to prepare to transmit the RAMBlock descriptions
3523 * to the source VM after connection setup.
3524 * Both sides use the "remote" structure to communicate and update
3525 * their "local" descriptions with what was sent.
3527 for (i = 0; i < local->nb_blocks; i++) {
3528 rdma->dest_blocks[i].remote_host_addr =
3529 (uintptr_t)(local->block[i].local_host_addr);
3531 if (rdma->pin_all) {
3532 rdma->dest_blocks[i].remote_rkey = local->block[i].mr->rkey;
3535 rdma->dest_blocks[i].offset = local->block[i].offset;
3536 rdma->dest_blocks[i].length = local->block[i].length;
3538 dest_block_to_network(&rdma->dest_blocks[i]);
3539 trace_qemu_rdma_registration_handle_ram_blocks_loop(
3540 local->block[i].block_name,
3541 local->block[i].offset,
3542 local->block[i].length,
3543 local->block[i].local_host_addr,
3544 local->block[i].src_index);
3547 blocks.len = rdma->local_ram_blocks.nb_blocks
3548 * sizeof(RDMADestBlock);
3551 ret = qemu_rdma_post_send_control(rdma,
3552 (uint8_t *) rdma->dest_blocks, &blocks);
3554 if (ret < 0) {
3555 error_report("rdma migration: error sending remote info");
3556 goto out;
3559 break;
3560 case RDMA_CONTROL_REGISTER_REQUEST:
3561 trace_qemu_rdma_registration_handle_register(head.repeat);
3563 reg_resp.repeat = head.repeat;
3564 registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
3566 for (count = 0; count < head.repeat; count++) {
3567 uint64_t chunk;
3568 uint8_t *chunk_start, *chunk_end;
3570 reg = &registers[count];
3571 network_to_register(reg);
3573 reg_result = &results[count];
3575 trace_qemu_rdma_registration_handle_register_loop(count,
3576 reg->current_index, reg->key.current_addr, reg->chunks);
3578 if (reg->current_index >= rdma->local_ram_blocks.nb_blocks) {
3579 error_report("rdma: 'register' bad block index %u (vs %d)",
3580 (unsigned int)reg->current_index,
3581 rdma->local_ram_blocks.nb_blocks);
3582 ret = -ENOENT;
3583 goto out;
3585 block = &(rdma->local_ram_blocks.block[reg->current_index]);
3586 if (block->is_ram_block) {
3587 if (block->offset > reg->key.current_addr) {
3588 error_report("rdma: bad register address for block %s"
3589 " offset: %" PRIx64 " current_addr: %" PRIx64,
3590 block->block_name, block->offset,
3591 reg->key.current_addr);
3592 ret = -ERANGE;
3593 goto out;
3595 host_addr = (block->local_host_addr +
3596 (reg->key.current_addr - block->offset));
3597 chunk = ram_chunk_index(block->local_host_addr,
3598 (uint8_t *) host_addr);
3599 } else {
3600 chunk = reg->key.chunk;
3601 host_addr = block->local_host_addr +
3602 (reg->key.chunk * (1UL << RDMA_REG_CHUNK_SHIFT));
3603 /* Check for particularly bad chunk value */
3604 if (host_addr < (void *)block->local_host_addr) {
3605 error_report("rdma: bad chunk for block %s"
3606 " chunk: %" PRIx64,
3607 block->block_name, reg->key.chunk);
3608 ret = -ERANGE;
3609 goto out;
3612 chunk_start = ram_chunk_start(block, chunk);
3613 chunk_end = ram_chunk_end(block, chunk + reg->chunks);
3614 if (qemu_rdma_register_and_get_keys(rdma, block,
3615 (uintptr_t)host_addr, NULL, &reg_result->rkey,
3616 chunk, chunk_start, chunk_end)) {
3617 error_report("cannot get rkey");
3618 ret = -EINVAL;
3619 goto out;
3622 reg_result->host_addr = (uintptr_t)block->local_host_addr;
3624 trace_qemu_rdma_registration_handle_register_rkey(
3625 reg_result->rkey);
3627 result_to_network(reg_result);
3630 ret = qemu_rdma_post_send_control(rdma,
3631 (uint8_t *) results, &reg_resp);
3633 if (ret < 0) {
3634 error_report("Failed to send control buffer");
3635 goto out;
3637 break;
3638 case RDMA_CONTROL_UNREGISTER_REQUEST:
3639 trace_qemu_rdma_registration_handle_unregister(head.repeat);
3640 unreg_resp.repeat = head.repeat;
3641 registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
3643 for (count = 0; count < head.repeat; count++) {
3644 reg = &registers[count];
3645 network_to_register(reg);
3647 trace_qemu_rdma_registration_handle_unregister_loop(count,
3648 reg->current_index, reg->key.chunk);
3650 block = &(rdma->local_ram_blocks.block[reg->current_index]);
3652 ret = ibv_dereg_mr(block->pmr[reg->key.chunk]);
3653 block->pmr[reg->key.chunk] = NULL;
3655 if (ret != 0) {
3656 perror("rdma unregistration chunk failed");
3657 ret = -ret;
3658 goto out;
3661 rdma->total_registrations--;
3663 trace_qemu_rdma_registration_handle_unregister_success(
3664 reg->key.chunk);
3667 ret = qemu_rdma_post_send_control(rdma, NULL, &unreg_resp);
3669 if (ret < 0) {
3670 error_report("Failed to send control buffer");
3671 goto out;
3673 break;
3674 case RDMA_CONTROL_REGISTER_RESULT:
3675 error_report("Invalid RESULT message at dest.");
3676 ret = -EIO;
3677 goto out;
3678 default:
3679 error_report("Unknown control message %s", control_desc(head.type));
3680 ret = -EIO;
3681 goto out;
3683 } while (1);
3684 out:
3685 if (ret < 0) {
3686 rdma->error_state = ret;
3688 rcu_read_unlock();
3689 return ret;
3692 /* Destination:
3693 * Called via a ram_control_load_hook during the initial RAM load section which
3694 * lists the RAMBlocks by name. This lets us know the order of the RAMBlocks
3695 * on the source.
3696 * We've already built our local RAMBlock list, but not yet sent the list to
3697 * the source.
3699 static int
3700 rdma_block_notification_handle(QIOChannelRDMA *rioc, const char *name)
3702 RDMAContext *rdma;
3703 int curr;
3704 int found = -1;
3706 rcu_read_lock();
3707 rdma = atomic_rcu_read(&rioc->rdmain);
3709 if (!rdma) {
3710 rcu_read_unlock();
3711 return -EIO;
3714 /* Find the matching RAMBlock in our local list */
3715 for (curr = 0; curr < rdma->local_ram_blocks.nb_blocks; curr++) {
3716 if (!strcmp(rdma->local_ram_blocks.block[curr].block_name, name)) {
3717 found = curr;
3718 break;
3722 if (found == -1) {
3723 error_report("RAMBlock '%s' not found on destination", name);
3724 rcu_read_unlock();
3725 return -ENOENT;
3728 rdma->local_ram_blocks.block[curr].src_index = rdma->next_src_index;
3729 trace_rdma_block_notification_handle(name, rdma->next_src_index);
3730 rdma->next_src_index++;
3732 rcu_read_unlock();
3733 return 0;
3736 static int rdma_load_hook(QEMUFile *f, void *opaque, uint64_t flags, void *data)
3738 switch (flags) {
3739 case RAM_CONTROL_BLOCK_REG:
3740 return rdma_block_notification_handle(opaque, data);
3742 case RAM_CONTROL_HOOK:
3743 return qemu_rdma_registration_handle(f, opaque);
3745 default:
3746 /* Shouldn't be called with any other values */
3747 abort();
3751 static int qemu_rdma_registration_start(QEMUFile *f, void *opaque,
3752 uint64_t flags, void *data)
3754 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(opaque);
3755 RDMAContext *rdma;
3757 rcu_read_lock();
3758 rdma = atomic_rcu_read(&rioc->rdmaout);
3759 if (!rdma) {
3760 rcu_read_unlock();
3761 return -EIO;
3764 CHECK_ERROR_STATE();
3766 if (migrate_get_current()->state == MIGRATION_STATUS_POSTCOPY_ACTIVE) {
3767 rcu_read_unlock();
3768 return 0;
3771 trace_qemu_rdma_registration_start(flags);
3772 qemu_put_be64(f, RAM_SAVE_FLAG_HOOK);
3773 qemu_fflush(f);
3775 rcu_read_unlock();
3776 return 0;
3780 * Inform dest that dynamic registrations are done for now.
3781 * First, flush writes, if any.
3783 static int qemu_rdma_registration_stop(QEMUFile *f, void *opaque,
3784 uint64_t flags, void *data)
3786 Error *local_err = NULL, **errp = &local_err;
3787 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(opaque);
3788 RDMAContext *rdma;
3789 RDMAControlHeader head = { .len = 0, .repeat = 1 };
3790 int ret = 0;
3792 rcu_read_lock();
3793 rdma = atomic_rcu_read(&rioc->rdmaout);
3794 if (!rdma) {
3795 rcu_read_unlock();
3796 return -EIO;
3799 CHECK_ERROR_STATE();
3801 if (migrate_get_current()->state == MIGRATION_STATUS_POSTCOPY_ACTIVE) {
3802 rcu_read_unlock();
3803 return 0;
3806 qemu_fflush(f);
3807 ret = qemu_rdma_drain_cq(f, rdma);
3809 if (ret < 0) {
3810 goto err;
3813 if (flags == RAM_CONTROL_SETUP) {
3814 RDMAControlHeader resp = {.type = RDMA_CONTROL_RAM_BLOCKS_RESULT };
3815 RDMALocalBlocks *local = &rdma->local_ram_blocks;
3816 int reg_result_idx, i, nb_dest_blocks;
3818 head.type = RDMA_CONTROL_RAM_BLOCKS_REQUEST;
3819 trace_qemu_rdma_registration_stop_ram();
3822 * Make sure that we parallelize the pinning on both sides.
3823 * For very large guests, doing this serially takes a really
3824 * long time, so we have to 'interleave' the pinning locally
3825 * with the control messages by performing the pinning on this
3826 * side before we receive the control response from the other
3827 * side that the pinning has completed.
3829 ret = qemu_rdma_exchange_send(rdma, &head, NULL, &resp,
3830 &reg_result_idx, rdma->pin_all ?
3831 qemu_rdma_reg_whole_ram_blocks : NULL);
3832 if (ret < 0) {
3833 ERROR(errp, "receiving remote info!");
3834 rcu_read_unlock();
3835 return ret;
3838 nb_dest_blocks = resp.len / sizeof(RDMADestBlock);
3841 * The protocol uses two different sets of rkeys (mutually exclusive):
3842 * 1. One key to represent the virtual address of the entire ram block.
3843 * (dynamic chunk registration disabled - pin everything with one rkey.)
3844 * 2. One to represent individual chunks within a ram block.
3845 * (dynamic chunk registration enabled - pin individual chunks.)
3847 * Once the capability is successfully negotiated, the destination transmits
3848 * the keys to use (or sends them later) including the virtual addresses
3849 * and then propagates the remote ram block descriptions to his local copy.
3852 if (local->nb_blocks != nb_dest_blocks) {
3853 ERROR(errp, "ram blocks mismatch (Number of blocks %d vs %d) "
3854 "Your QEMU command line parameters are probably "
3855 "not identical on both the source and destination.",
3856 local->nb_blocks, nb_dest_blocks);
3857 rdma->error_state = -EINVAL;
3858 rcu_read_unlock();
3859 return -EINVAL;
3862 qemu_rdma_move_header(rdma, reg_result_idx, &resp);
3863 memcpy(rdma->dest_blocks,
3864 rdma->wr_data[reg_result_idx].control_curr, resp.len);
3865 for (i = 0; i < nb_dest_blocks; i++) {
3866 network_to_dest_block(&rdma->dest_blocks[i]);
3868 /* We require that the blocks are in the same order */
3869 if (rdma->dest_blocks[i].length != local->block[i].length) {
3870 ERROR(errp, "Block %s/%d has a different length %" PRIu64
3871 "vs %" PRIu64, local->block[i].block_name, i,
3872 local->block[i].length,
3873 rdma->dest_blocks[i].length);
3874 rdma->error_state = -EINVAL;
3875 rcu_read_unlock();
3876 return -EINVAL;
3878 local->block[i].remote_host_addr =
3879 rdma->dest_blocks[i].remote_host_addr;
3880 local->block[i].remote_rkey = rdma->dest_blocks[i].remote_rkey;
3884 trace_qemu_rdma_registration_stop(flags);
3886 head.type = RDMA_CONTROL_REGISTER_FINISHED;
3887 ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL);
3889 if (ret < 0) {
3890 goto err;
3893 rcu_read_unlock();
3894 return 0;
3895 err:
3896 rdma->error_state = ret;
3897 rcu_read_unlock();
3898 return ret;
3901 static const QEMUFileHooks rdma_read_hooks = {
3902 .hook_ram_load = rdma_load_hook,
3905 static const QEMUFileHooks rdma_write_hooks = {
3906 .before_ram_iterate = qemu_rdma_registration_start,
3907 .after_ram_iterate = qemu_rdma_registration_stop,
3908 .save_page = qemu_rdma_save_page,
3912 static void qio_channel_rdma_finalize(Object *obj)
3914 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(obj);
3915 if (rioc->rdmain) {
3916 qemu_rdma_cleanup(rioc->rdmain);
3917 g_free(rioc->rdmain);
3918 rioc->rdmain = NULL;
3920 if (rioc->rdmaout) {
3921 qemu_rdma_cleanup(rioc->rdmaout);
3922 g_free(rioc->rdmaout);
3923 rioc->rdmaout = NULL;
3927 static void qio_channel_rdma_class_init(ObjectClass *klass,
3928 void *class_data G_GNUC_UNUSED)
3930 QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
3932 ioc_klass->io_writev = qio_channel_rdma_writev;
3933 ioc_klass->io_readv = qio_channel_rdma_readv;
3934 ioc_klass->io_set_blocking = qio_channel_rdma_set_blocking;
3935 ioc_klass->io_close = qio_channel_rdma_close;
3936 ioc_klass->io_create_watch = qio_channel_rdma_create_watch;
3937 ioc_klass->io_set_aio_fd_handler = qio_channel_rdma_set_aio_fd_handler;
3938 ioc_klass->io_shutdown = qio_channel_rdma_shutdown;
3941 static const TypeInfo qio_channel_rdma_info = {
3942 .parent = TYPE_QIO_CHANNEL,
3943 .name = TYPE_QIO_CHANNEL_RDMA,
3944 .instance_size = sizeof(QIOChannelRDMA),
3945 .instance_finalize = qio_channel_rdma_finalize,
3946 .class_init = qio_channel_rdma_class_init,
3949 static void qio_channel_rdma_register_types(void)
3951 type_register_static(&qio_channel_rdma_info);
3954 type_init(qio_channel_rdma_register_types);
3956 static QEMUFile *qemu_fopen_rdma(RDMAContext *rdma, const char *mode)
3958 QIOChannelRDMA *rioc;
3960 if (qemu_file_mode_is_not_valid(mode)) {
3961 return NULL;
3964 rioc = QIO_CHANNEL_RDMA(object_new(TYPE_QIO_CHANNEL_RDMA));
3966 if (mode[0] == 'w') {
3967 rioc->file = qemu_fopen_channel_output(QIO_CHANNEL(rioc));
3968 rioc->rdmaout = rdma;
3969 rioc->rdmain = rdma->return_path;
3970 qemu_file_set_hooks(rioc->file, &rdma_write_hooks);
3971 } else {
3972 rioc->file = qemu_fopen_channel_input(QIO_CHANNEL(rioc));
3973 rioc->rdmain = rdma;
3974 rioc->rdmaout = rdma->return_path;
3975 qemu_file_set_hooks(rioc->file, &rdma_read_hooks);
3978 return rioc->file;
3981 static void rdma_accept_incoming_migration(void *opaque)
3983 RDMAContext *rdma = opaque;
3984 int ret;
3985 QEMUFile *f;
3986 Error *local_err = NULL, **errp = &local_err;
3988 trace_qemu_rdma_accept_incoming_migration();
3989 ret = qemu_rdma_accept(rdma);
3991 if (ret) {
3992 ERROR(errp, "RDMA Migration initialization failed!");
3993 return;
3996 trace_qemu_rdma_accept_incoming_migration_accepted();
3998 if (rdma->is_return_path) {
3999 return;
4002 f = qemu_fopen_rdma(rdma, "rb");
4003 if (f == NULL) {
4004 ERROR(errp, "could not qemu_fopen_rdma!");
4005 qemu_rdma_cleanup(rdma);
4006 return;
4009 rdma->migration_started_on_destination = 1;
4010 migration_fd_process_incoming(f);
4013 void rdma_start_incoming_migration(const char *host_port, Error **errp)
4015 int ret;
4016 RDMAContext *rdma, *rdma_return_path = NULL;
4017 Error *local_err = NULL;
4019 trace_rdma_start_incoming_migration();
4020 rdma = qemu_rdma_data_init(host_port, &local_err);
4022 if (rdma == NULL) {
4023 goto err;
4026 ret = qemu_rdma_dest_init(rdma, &local_err);
4028 if (ret) {
4029 goto err;
4032 trace_rdma_start_incoming_migration_after_dest_init();
4034 ret = rdma_listen(rdma->listen_id, 5);
4036 if (ret) {
4037 ERROR(errp, "listening on socket!");
4038 goto err;
4041 trace_rdma_start_incoming_migration_after_rdma_listen();
4043 /* initialize the RDMAContext for return path */
4044 if (migrate_postcopy()) {
4045 rdma_return_path = qemu_rdma_data_init(host_port, &local_err);
4047 if (rdma_return_path == NULL) {
4048 goto err;
4051 qemu_rdma_return_path_dest_init(rdma_return_path, rdma);
4054 qemu_set_fd_handler(rdma->channel->fd, rdma_accept_incoming_migration,
4055 NULL, (void *)(intptr_t)rdma);
4056 return;
4057 err:
4058 error_propagate(errp, local_err);
4059 g_free(rdma);
4060 g_free(rdma_return_path);
4063 void rdma_start_outgoing_migration(void *opaque,
4064 const char *host_port, Error **errp)
4066 MigrationState *s = opaque;
4067 RDMAContext *rdma = qemu_rdma_data_init(host_port, errp);
4068 RDMAContext *rdma_return_path = NULL;
4069 int ret = 0;
4071 if (rdma == NULL) {
4072 goto err;
4075 ret = qemu_rdma_source_init(rdma,
4076 s->enabled_capabilities[MIGRATION_CAPABILITY_RDMA_PIN_ALL], errp);
4078 if (ret) {
4079 goto err;
4082 trace_rdma_start_outgoing_migration_after_rdma_source_init();
4083 ret = qemu_rdma_connect(rdma, errp);
4085 if (ret) {
4086 goto err;
4089 /* RDMA postcopy need a seprate queue pair for return path */
4090 if (migrate_postcopy()) {
4091 rdma_return_path = qemu_rdma_data_init(host_port, errp);
4093 if (rdma_return_path == NULL) {
4094 goto err;
4097 ret = qemu_rdma_source_init(rdma_return_path,
4098 s->enabled_capabilities[MIGRATION_CAPABILITY_RDMA_PIN_ALL], errp);
4100 if (ret) {
4101 goto err;
4104 ret = qemu_rdma_connect(rdma_return_path, errp);
4106 if (ret) {
4107 goto err;
4110 rdma->return_path = rdma_return_path;
4111 rdma_return_path->return_path = rdma;
4112 rdma_return_path->is_return_path = true;
4115 trace_rdma_start_outgoing_migration_after_rdma_connect();
4117 s->to_dst_file = qemu_fopen_rdma(rdma, "wb");
4118 migrate_fd_connect(s, NULL);
4119 return;
4120 err:
4121 g_free(rdma);
4122 g_free(rdma_return_path);