2 * RDMA protocol and interfaces
4 * Copyright IBM, Corp. 2010-2013
5 * Copyright Red Hat, Inc. 2015-2016
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.
17 #include "qemu/osdep.h"
18 #include "qapi/error.h"
19 #include "qemu/cutils.h"
20 #include "exec/target_page.h"
22 #include "migration.h"
23 #include "migration-stats.h"
24 #include "qemu-file.h"
26 #include "qemu/error-report.h"
27 #include "qemu/main-loop.h"
28 #include "qemu/module.h"
30 #include "qemu/sockets.h"
31 #include "qemu/bitmap.h"
32 #include "qemu/coroutine.h"
33 #include "exec/memory.h"
34 #include <sys/socket.h>
36 #include <arpa/inet.h>
37 #include <rdma/rdma_cma.h>
39 #include "qom/object.h"
44 * Print and error on both the Monitor and the Log file.
46 #define ERROR(errp, fmt, ...) \
48 fprintf(stderr, "RDMA ERROR: " fmt "\n", ## __VA_ARGS__); \
49 if (errp && (*(errp) == NULL)) { \
50 error_setg(errp, "RDMA ERROR: " fmt, ## __VA_ARGS__); \
54 #define RDMA_RESOLVE_TIMEOUT_MS 10000
56 /* Do not merge data if larger than this. */
57 #define RDMA_MERGE_MAX (2 * 1024 * 1024)
58 #define RDMA_SIGNALED_SEND_MAX (RDMA_MERGE_MAX / 4096)
60 #define RDMA_REG_CHUNK_SHIFT 20 /* 1 MB */
63 * This is only for non-live state being migrated.
64 * Instead of RDMA_WRITE messages, we use RDMA_SEND
65 * messages for that state, which requires a different
66 * delivery design than main memory.
68 #define RDMA_SEND_INCREMENT 32768
71 * Maximum size infiniband SEND message
73 #define RDMA_CONTROL_MAX_BUFFER (512 * 1024)
74 #define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096
76 #define RDMA_CONTROL_VERSION_CURRENT 1
78 * Capabilities for negotiation.
80 #define RDMA_CAPABILITY_PIN_ALL 0x01
83 * Add the other flags above to this list of known capabilities
84 * as they are introduced.
86 static uint32_t known_capabilities = RDMA_CAPABILITY_PIN_ALL;
88 #define CHECK_ERROR_STATE() \
90 if (rdma->error_state) { \
91 if (!rdma->error_reported) { \
92 error_report("RDMA is in an error state waiting migration" \
94 rdma->error_reported = 1; \
96 return rdma->error_state; \
101 * A work request ID is 64-bits and we split up these bits
104 * bits 0-15 : type of control message, 2^16
105 * bits 16-29: ram block index, 2^14
106 * bits 30-63: ram block chunk number, 2^34
108 * The last two bit ranges are only used for RDMA writes,
109 * in order to track their completion and potentially
110 * also track unregistration status of the message.
112 #define RDMA_WRID_TYPE_SHIFT 0UL
113 #define RDMA_WRID_BLOCK_SHIFT 16UL
114 #define RDMA_WRID_CHUNK_SHIFT 30UL
116 #define RDMA_WRID_TYPE_MASK \
117 ((1UL << RDMA_WRID_BLOCK_SHIFT) - 1UL)
119 #define RDMA_WRID_BLOCK_MASK \
120 (~RDMA_WRID_TYPE_MASK & ((1UL << RDMA_WRID_CHUNK_SHIFT) - 1UL))
122 #define RDMA_WRID_CHUNK_MASK (~RDMA_WRID_BLOCK_MASK & ~RDMA_WRID_TYPE_MASK)
125 * RDMA migration protocol:
126 * 1. RDMA Writes (data messages, i.e. RAM)
127 * 2. IB Send/Recv (control channel messages)
131 RDMA_WRID_RDMA_WRITE = 1,
132 RDMA_WRID_SEND_CONTROL = 2000,
133 RDMA_WRID_RECV_CONTROL = 4000,
136 static const char *wrid_desc[] = {
137 [RDMA_WRID_NONE] = "NONE",
138 [RDMA_WRID_RDMA_WRITE] = "WRITE RDMA",
139 [RDMA_WRID_SEND_CONTROL] = "CONTROL SEND",
140 [RDMA_WRID_RECV_CONTROL] = "CONTROL RECV",
144 * Work request IDs for IB SEND messages only (not RDMA writes).
145 * This is used by the migration protocol to transmit
146 * control messages (such as device state and registration commands)
148 * We could use more WRs, but we have enough for now.
158 * SEND/RECV IB Control Messages.
161 RDMA_CONTROL_NONE = 0,
163 RDMA_CONTROL_READY, /* ready to receive */
164 RDMA_CONTROL_QEMU_FILE, /* QEMUFile-transmitted bytes */
165 RDMA_CONTROL_RAM_BLOCKS_REQUEST, /* RAMBlock synchronization */
166 RDMA_CONTROL_RAM_BLOCKS_RESULT, /* RAMBlock synchronization */
167 RDMA_CONTROL_COMPRESS, /* page contains repeat values */
168 RDMA_CONTROL_REGISTER_REQUEST, /* dynamic page registration */
169 RDMA_CONTROL_REGISTER_RESULT, /* key to use after registration */
170 RDMA_CONTROL_REGISTER_FINISHED, /* current iteration finished */
171 RDMA_CONTROL_UNREGISTER_REQUEST, /* dynamic UN-registration */
172 RDMA_CONTROL_UNREGISTER_FINISHED, /* unpinning finished */
177 * Memory and MR structures used to represent an IB Send/Recv work request.
178 * This is *not* used for RDMA writes, only IB Send/Recv.
181 uint8_t control[RDMA_CONTROL_MAX_BUFFER]; /* actual buffer to register */
182 struct ibv_mr *control_mr; /* registration metadata */
183 size_t control_len; /* length of the message */
184 uint8_t *control_curr; /* start of unconsumed bytes */
185 } RDMAWorkRequestData;
188 * Negotiate RDMA capabilities during connection-setup time.
195 static void caps_to_network(RDMACapabilities *cap)
197 cap->version = htonl(cap->version);
198 cap->flags = htonl(cap->flags);
201 static void network_to_caps(RDMACapabilities *cap)
203 cap->version = ntohl(cap->version);
204 cap->flags = ntohl(cap->flags);
208 * Representation of a RAMBlock from an RDMA perspective.
209 * This is not transmitted, only local.
210 * This and subsequent structures cannot be linked lists
211 * because we're using a single IB message to transmit
212 * the information. It's small anyway, so a list is overkill.
214 typedef struct RDMALocalBlock {
216 uint8_t *local_host_addr; /* local virtual address */
217 uint64_t remote_host_addr; /* remote virtual address */
220 struct ibv_mr **pmr; /* MRs for chunk-level registration */
221 struct ibv_mr *mr; /* MR for non-chunk-level registration */
222 uint32_t *remote_keys; /* rkeys for chunk-level registration */
223 uint32_t remote_rkey; /* rkeys for non-chunk-level registration */
224 int index; /* which block are we */
225 unsigned int src_index; /* (Only used on dest) */
228 unsigned long *transit_bitmap;
229 unsigned long *unregister_bitmap;
233 * Also represents a RAMblock, but only on the dest.
234 * This gets transmitted by the dest during connection-time
235 * to the source VM and then is used to populate the
236 * corresponding RDMALocalBlock with
237 * the information needed to perform the actual RDMA.
239 typedef struct QEMU_PACKED RDMADestBlock {
240 uint64_t remote_host_addr;
243 uint32_t remote_rkey;
247 static const char *control_desc(unsigned int rdma_control)
249 static const char *strs[] = {
250 [RDMA_CONTROL_NONE] = "NONE",
251 [RDMA_CONTROL_ERROR] = "ERROR",
252 [RDMA_CONTROL_READY] = "READY",
253 [RDMA_CONTROL_QEMU_FILE] = "QEMU FILE",
254 [RDMA_CONTROL_RAM_BLOCKS_REQUEST] = "RAM BLOCKS REQUEST",
255 [RDMA_CONTROL_RAM_BLOCKS_RESULT] = "RAM BLOCKS RESULT",
256 [RDMA_CONTROL_COMPRESS] = "COMPRESS",
257 [RDMA_CONTROL_REGISTER_REQUEST] = "REGISTER REQUEST",
258 [RDMA_CONTROL_REGISTER_RESULT] = "REGISTER RESULT",
259 [RDMA_CONTROL_REGISTER_FINISHED] = "REGISTER FINISHED",
260 [RDMA_CONTROL_UNREGISTER_REQUEST] = "UNREGISTER REQUEST",
261 [RDMA_CONTROL_UNREGISTER_FINISHED] = "UNREGISTER FINISHED",
264 if (rdma_control > RDMA_CONTROL_UNREGISTER_FINISHED) {
265 return "??BAD CONTROL VALUE??";
268 return strs[rdma_control];
271 static uint64_t htonll(uint64_t v)
273 union { uint32_t lv[2]; uint64_t llv; } u;
274 u.lv[0] = htonl(v >> 32);
275 u.lv[1] = htonl(v & 0xFFFFFFFFULL);
279 static uint64_t ntohll(uint64_t v)
281 union { uint32_t lv[2]; uint64_t llv; } u;
283 return ((uint64_t)ntohl(u.lv[0]) << 32) | (uint64_t) ntohl(u.lv[1]);
286 static void dest_block_to_network(RDMADestBlock *db)
288 db->remote_host_addr = htonll(db->remote_host_addr);
289 db->offset = htonll(db->offset);
290 db->length = htonll(db->length);
291 db->remote_rkey = htonl(db->remote_rkey);
294 static void network_to_dest_block(RDMADestBlock *db)
296 db->remote_host_addr = ntohll(db->remote_host_addr);
297 db->offset = ntohll(db->offset);
298 db->length = ntohll(db->length);
299 db->remote_rkey = ntohl(db->remote_rkey);
303 * Virtual address of the above structures used for transmitting
304 * the RAMBlock descriptions at connection-time.
305 * This structure is *not* transmitted.
307 typedef struct RDMALocalBlocks {
309 bool init; /* main memory init complete */
310 RDMALocalBlock *block;
314 * Main data structure for RDMA state.
315 * While there is only one copy of this structure being allocated right now,
316 * this is the place where one would start if you wanted to consider
317 * having more than one RDMA connection open at the same time.
319 typedef struct RDMAContext {
324 RDMAWorkRequestData wr_data[RDMA_WRID_MAX];
327 * This is used by *_exchange_send() to figure out whether or not
328 * the initial "READY" message has already been received or not.
329 * This is because other functions may potentially poll() and detect
330 * the READY message before send() does, in which case we need to
331 * know if it completed.
333 int control_ready_expected;
335 /* number of outstanding writes */
338 /* store info about current buffer so that we can
339 merge it with future sends */
340 uint64_t current_addr;
341 uint64_t current_length;
342 /* index of ram block the current buffer belongs to */
344 /* index of the chunk in the current ram block */
350 * infiniband-specific variables for opening the device
351 * and maintaining connection state and so forth.
353 * cm_id also has ibv_context, rdma_event_channel, and ibv_qp in
354 * cm_id->verbs, cm_id->channel, and cm_id->qp.
356 struct rdma_cm_id *cm_id; /* connection manager ID */
357 struct rdma_cm_id *listen_id;
360 struct ibv_context *verbs;
361 struct rdma_event_channel *channel;
362 struct ibv_qp *qp; /* queue pair */
363 struct ibv_comp_channel *recv_comp_channel; /* recv completion channel */
364 struct ibv_comp_channel *send_comp_channel; /* send completion channel */
365 struct ibv_pd *pd; /* protection domain */
366 struct ibv_cq *recv_cq; /* recvieve completion queue */
367 struct ibv_cq *send_cq; /* send completion queue */
370 * If a previous write failed (perhaps because of a failed
371 * memory registration, then do not attempt any future work
372 * and remember the error state.
379 * Description of ram blocks used throughout the code.
381 RDMALocalBlocks local_ram_blocks;
382 RDMADestBlock *dest_blocks;
384 /* Index of the next RAMBlock received during block registration */
385 unsigned int next_src_index;
388 * Migration on *destination* started.
389 * Then use coroutine yield function.
390 * Source runs in a thread, so we don't care.
392 int migration_started_on_destination;
394 int total_registrations;
397 int unregister_current, unregister_next;
398 uint64_t unregistrations[RDMA_SIGNALED_SEND_MAX];
400 GHashTable *blockmap;
402 /* the RDMAContext for return path */
403 struct RDMAContext *return_path;
407 #define TYPE_QIO_CHANNEL_RDMA "qio-channel-rdma"
408 OBJECT_DECLARE_SIMPLE_TYPE(QIOChannelRDMA, QIO_CHANNEL_RDMA)
412 struct QIOChannelRDMA {
415 RDMAContext *rdmaout;
417 bool blocking; /* XXX we don't actually honour this yet */
421 * Main structure for IB Send/Recv control messages.
422 * This gets prepended at the beginning of every Send/Recv.
424 typedef struct QEMU_PACKED {
425 uint32_t len; /* Total length of data portion */
426 uint32_t type; /* which control command to perform */
427 uint32_t repeat; /* number of commands in data portion of same type */
431 static void control_to_network(RDMAControlHeader *control)
433 control->type = htonl(control->type);
434 control->len = htonl(control->len);
435 control->repeat = htonl(control->repeat);
438 static void network_to_control(RDMAControlHeader *control)
440 control->type = ntohl(control->type);
441 control->len = ntohl(control->len);
442 control->repeat = ntohl(control->repeat);
446 * Register a single Chunk.
447 * Information sent by the source VM to inform the dest
448 * to register an single chunk of memory before we can perform
449 * the actual RDMA operation.
451 typedef struct QEMU_PACKED {
453 uint64_t current_addr; /* offset into the ram_addr_t space */
454 uint64_t chunk; /* chunk to lookup if unregistering */
456 uint32_t current_index; /* which ramblock the chunk belongs to */
458 uint64_t chunks; /* how many sequential chunks to register */
461 static void register_to_network(RDMAContext *rdma, RDMARegister *reg)
463 RDMALocalBlock *local_block;
464 local_block = &rdma->local_ram_blocks.block[reg->current_index];
466 if (local_block->is_ram_block) {
468 * current_addr as passed in is an address in the local ram_addr_t
469 * space, we need to translate this for the destination
471 reg->key.current_addr -= local_block->offset;
472 reg->key.current_addr += rdma->dest_blocks[reg->current_index].offset;
474 reg->key.current_addr = htonll(reg->key.current_addr);
475 reg->current_index = htonl(reg->current_index);
476 reg->chunks = htonll(reg->chunks);
479 static void network_to_register(RDMARegister *reg)
481 reg->key.current_addr = ntohll(reg->key.current_addr);
482 reg->current_index = ntohl(reg->current_index);
483 reg->chunks = ntohll(reg->chunks);
486 typedef struct QEMU_PACKED {
487 uint32_t value; /* if zero, we will madvise() */
488 uint32_t block_idx; /* which ram block index */
489 uint64_t offset; /* Address in remote ram_addr_t space */
490 uint64_t length; /* length of the chunk */
493 static void compress_to_network(RDMAContext *rdma, RDMACompress *comp)
495 comp->value = htonl(comp->value);
497 * comp->offset as passed in is an address in the local ram_addr_t
498 * space, we need to translate this for the destination
500 comp->offset -= rdma->local_ram_blocks.block[comp->block_idx].offset;
501 comp->offset += rdma->dest_blocks[comp->block_idx].offset;
502 comp->block_idx = htonl(comp->block_idx);
503 comp->offset = htonll(comp->offset);
504 comp->length = htonll(comp->length);
507 static void network_to_compress(RDMACompress *comp)
509 comp->value = ntohl(comp->value);
510 comp->block_idx = ntohl(comp->block_idx);
511 comp->offset = ntohll(comp->offset);
512 comp->length = ntohll(comp->length);
516 * The result of the dest's memory registration produces an "rkey"
517 * which the source VM must reference in order to perform
518 * the RDMA operation.
520 typedef struct QEMU_PACKED {
524 } RDMARegisterResult;
526 static void result_to_network(RDMARegisterResult *result)
528 result->rkey = htonl(result->rkey);
529 result->host_addr = htonll(result->host_addr);
532 static void network_to_result(RDMARegisterResult *result)
534 result->rkey = ntohl(result->rkey);
535 result->host_addr = ntohll(result->host_addr);
538 const char *print_wrid(int wrid);
539 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
540 uint8_t *data, RDMAControlHeader *resp,
542 int (*callback)(RDMAContext *rdma));
544 static inline uint64_t ram_chunk_index(const uint8_t *start,
547 return ((uintptr_t) host - (uintptr_t) start) >> RDMA_REG_CHUNK_SHIFT;
550 static inline uint8_t *ram_chunk_start(const RDMALocalBlock *rdma_ram_block,
553 return (uint8_t *)(uintptr_t)(rdma_ram_block->local_host_addr +
554 (i << RDMA_REG_CHUNK_SHIFT));
557 static inline uint8_t *ram_chunk_end(const RDMALocalBlock *rdma_ram_block,
560 uint8_t *result = ram_chunk_start(rdma_ram_block, i) +
561 (1UL << RDMA_REG_CHUNK_SHIFT);
563 if (result > (rdma_ram_block->local_host_addr + rdma_ram_block->length)) {
564 result = rdma_ram_block->local_host_addr + rdma_ram_block->length;
570 static int rdma_add_block(RDMAContext *rdma, const char *block_name,
572 ram_addr_t block_offset, uint64_t length)
574 RDMALocalBlocks *local = &rdma->local_ram_blocks;
575 RDMALocalBlock *block;
576 RDMALocalBlock *old = local->block;
578 local->block = g_new0(RDMALocalBlock, local->nb_blocks + 1);
580 if (local->nb_blocks) {
583 if (rdma->blockmap) {
584 for (x = 0; x < local->nb_blocks; x++) {
585 g_hash_table_remove(rdma->blockmap,
586 (void *)(uintptr_t)old[x].offset);
587 g_hash_table_insert(rdma->blockmap,
588 (void *)(uintptr_t)old[x].offset,
592 memcpy(local->block, old, sizeof(RDMALocalBlock) * local->nb_blocks);
596 block = &local->block[local->nb_blocks];
598 block->block_name = g_strdup(block_name);
599 block->local_host_addr = host_addr;
600 block->offset = block_offset;
601 block->length = length;
602 block->index = local->nb_blocks;
603 block->src_index = ~0U; /* Filled in by the receipt of the block list */
604 block->nb_chunks = ram_chunk_index(host_addr, host_addr + length) + 1UL;
605 block->transit_bitmap = bitmap_new(block->nb_chunks);
606 bitmap_clear(block->transit_bitmap, 0, block->nb_chunks);
607 block->unregister_bitmap = bitmap_new(block->nb_chunks);
608 bitmap_clear(block->unregister_bitmap, 0, block->nb_chunks);
609 block->remote_keys = g_new0(uint32_t, block->nb_chunks);
611 block->is_ram_block = local->init ? false : true;
613 if (rdma->blockmap) {
614 g_hash_table_insert(rdma->blockmap, (void *)(uintptr_t)block_offset, block);
617 trace_rdma_add_block(block_name, local->nb_blocks,
618 (uintptr_t) block->local_host_addr,
619 block->offset, block->length,
620 (uintptr_t) (block->local_host_addr + block->length),
621 BITS_TO_LONGS(block->nb_chunks) *
622 sizeof(unsigned long) * 8,
631 * Memory regions need to be registered with the device and queue pairs setup
632 * in advanced before the migration starts. This tells us where the RAM blocks
633 * are so that we can register them individually.
635 static int qemu_rdma_init_one_block(RAMBlock *rb, void *opaque)
637 const char *block_name = qemu_ram_get_idstr(rb);
638 void *host_addr = qemu_ram_get_host_addr(rb);
639 ram_addr_t block_offset = qemu_ram_get_offset(rb);
640 ram_addr_t length = qemu_ram_get_used_length(rb);
641 return rdma_add_block(opaque, block_name, host_addr, block_offset, length);
645 * Identify the RAMBlocks and their quantity. They will be references to
646 * identify chunk boundaries inside each RAMBlock and also be referenced
647 * during dynamic page registration.
649 static int qemu_rdma_init_ram_blocks(RDMAContext *rdma)
651 RDMALocalBlocks *local = &rdma->local_ram_blocks;
654 assert(rdma->blockmap == NULL);
655 memset(local, 0, sizeof *local);
656 ret = foreach_not_ignored_block(qemu_rdma_init_one_block, rdma);
660 trace_qemu_rdma_init_ram_blocks(local->nb_blocks);
661 rdma->dest_blocks = g_new0(RDMADestBlock,
662 rdma->local_ram_blocks.nb_blocks);
668 * Note: If used outside of cleanup, the caller must ensure that the destination
669 * block structures are also updated
671 static int rdma_delete_block(RDMAContext *rdma, RDMALocalBlock *block)
673 RDMALocalBlocks *local = &rdma->local_ram_blocks;
674 RDMALocalBlock *old = local->block;
677 if (rdma->blockmap) {
678 g_hash_table_remove(rdma->blockmap, (void *)(uintptr_t)block->offset);
683 for (j = 0; j < block->nb_chunks; j++) {
684 if (!block->pmr[j]) {
687 ibv_dereg_mr(block->pmr[j]);
688 rdma->total_registrations--;
695 ibv_dereg_mr(block->mr);
696 rdma->total_registrations--;
700 g_free(block->transit_bitmap);
701 block->transit_bitmap = NULL;
703 g_free(block->unregister_bitmap);
704 block->unregister_bitmap = NULL;
706 g_free(block->remote_keys);
707 block->remote_keys = NULL;
709 g_free(block->block_name);
710 block->block_name = NULL;
712 if (rdma->blockmap) {
713 for (x = 0; x < local->nb_blocks; x++) {
714 g_hash_table_remove(rdma->blockmap,
715 (void *)(uintptr_t)old[x].offset);
719 if (local->nb_blocks > 1) {
721 local->block = g_new0(RDMALocalBlock, local->nb_blocks - 1);
724 memcpy(local->block, old, sizeof(RDMALocalBlock) * block->index);
727 if (block->index < (local->nb_blocks - 1)) {
728 memcpy(local->block + block->index, old + (block->index + 1),
729 sizeof(RDMALocalBlock) *
730 (local->nb_blocks - (block->index + 1)));
731 for (x = block->index; x < local->nb_blocks - 1; x++) {
732 local->block[x].index--;
736 assert(block == local->block);
740 trace_rdma_delete_block(block, (uintptr_t)block->local_host_addr,
741 block->offset, block->length,
742 (uintptr_t)(block->local_host_addr + block->length),
743 BITS_TO_LONGS(block->nb_chunks) *
744 sizeof(unsigned long) * 8, block->nb_chunks);
750 if (local->nb_blocks && rdma->blockmap) {
751 for (x = 0; x < local->nb_blocks; x++) {
752 g_hash_table_insert(rdma->blockmap,
753 (void *)(uintptr_t)local->block[x].offset,
762 * Put in the log file which RDMA device was opened and the details
763 * associated with that device.
765 static void qemu_rdma_dump_id(const char *who, struct ibv_context *verbs)
767 struct ibv_port_attr port;
769 if (ibv_query_port(verbs, 1, &port)) {
770 error_report("Failed to query port information");
774 printf("%s RDMA Device opened: kernel name %s "
775 "uverbs device name %s, "
776 "infiniband_verbs class device path %s, "
777 "infiniband class device path %s, "
778 "transport: (%d) %s\n",
781 verbs->device->dev_name,
782 verbs->device->dev_path,
783 verbs->device->ibdev_path,
785 (port.link_layer == IBV_LINK_LAYER_INFINIBAND) ? "Infiniband" :
786 ((port.link_layer == IBV_LINK_LAYER_ETHERNET)
787 ? "Ethernet" : "Unknown"));
791 * Put in the log file the RDMA gid addressing information,
792 * useful for folks who have trouble understanding the
793 * RDMA device hierarchy in the kernel.
795 static void qemu_rdma_dump_gid(const char *who, struct rdma_cm_id *id)
799 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.sgid, sgid, sizeof sgid);
800 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.dgid, dgid, sizeof dgid);
801 trace_qemu_rdma_dump_gid(who, sgid, dgid);
805 * As of now, IPv6 over RoCE / iWARP is not supported by linux.
806 * We will try the next addrinfo struct, and fail if there are
807 * no other valid addresses to bind against.
809 * If user is listening on '[::]', then we will not have a opened a device
810 * yet and have no way of verifying if the device is RoCE or not.
812 * In this case, the source VM will throw an error for ALL types of
813 * connections (both IPv4 and IPv6) if the destination machine does not have
814 * a regular infiniband network available for use.
816 * The only way to guarantee that an error is thrown for broken kernels is
817 * for the management software to choose a *specific* interface at bind time
818 * and validate what time of hardware it is.
820 * Unfortunately, this puts the user in a fix:
822 * If the source VM connects with an IPv4 address without knowing that the
823 * destination has bound to '[::]' the migration will unconditionally fail
824 * unless the management software is explicitly listening on the IPv4
825 * address while using a RoCE-based device.
827 * If the source VM connects with an IPv6 address, then we're OK because we can
828 * throw an error on the source (and similarly on the destination).
830 * But in mixed environments, this will be broken for a while until it is fixed
833 * We do provide a *tiny* bit of help in this function: We can list all of the
834 * devices in the system and check to see if all the devices are RoCE or
837 * If we detect that we have a *pure* RoCE environment, then we can safely
838 * thrown an error even if the management software has specified '[::]' as the
841 * However, if there is are multiple hetergeneous devices, then we cannot make
842 * this assumption and the user just has to be sure they know what they are
845 * Patches are being reviewed on linux-rdma.
847 static int qemu_rdma_broken_ipv6_kernel(struct ibv_context *verbs, Error **errp)
849 /* This bug only exists in linux, to our knowledge. */
851 struct ibv_port_attr port_attr;
854 * Verbs are only NULL if management has bound to '[::]'.
856 * Let's iterate through all the devices and see if there any pure IB
857 * devices (non-ethernet).
859 * If not, then we can safely proceed with the migration.
860 * Otherwise, there are no guarantees until the bug is fixed in linux.
864 struct ibv_device **dev_list = ibv_get_device_list(&num_devices);
865 bool roce_found = false;
866 bool ib_found = false;
868 for (x = 0; x < num_devices; x++) {
869 verbs = ibv_open_device(dev_list[x]);
871 if (errno == EPERM) {
878 if (ibv_query_port(verbs, 1, &port_attr)) {
879 ibv_close_device(verbs);
880 ERROR(errp, "Could not query initial IB port");
884 if (port_attr.link_layer == IBV_LINK_LAYER_INFINIBAND) {
886 } else if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
890 ibv_close_device(verbs);
896 fprintf(stderr, "WARN: migrations may fail:"
897 " IPv6 over RoCE / iWARP in linux"
898 " is broken. But since you appear to have a"
899 " mixed RoCE / IB environment, be sure to only"
900 " migrate over the IB fabric until the kernel "
901 " fixes the bug.\n");
903 ERROR(errp, "You only have RoCE / iWARP devices in your systems"
904 " and your management software has specified '[::]'"
905 ", but IPv6 over RoCE / iWARP is not supported in Linux.");
914 * If we have a verbs context, that means that some other than '[::]' was
915 * used by the management software for binding. In which case we can
916 * actually warn the user about a potentially broken kernel.
919 /* IB ports start with 1, not 0 */
920 if (ibv_query_port(verbs, 1, &port_attr)) {
921 ERROR(errp, "Could not query initial IB port");
925 if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
926 ERROR(errp, "Linux kernel's RoCE / iWARP does not support IPv6 "
927 "(but patches on linux-rdma in progress)");
937 * Figure out which RDMA device corresponds to the requested IP hostname
938 * Also create the initial connection manager identifiers for opening
941 static int qemu_rdma_resolve_host(RDMAContext *rdma, Error **errp)
944 struct rdma_addrinfo *res;
946 struct rdma_cm_event *cm_event;
947 char ip[40] = "unknown";
948 struct rdma_addrinfo *e;
950 if (rdma->host == NULL || !strcmp(rdma->host, "")) {
951 ERROR(errp, "RDMA hostname has not been set");
955 /* create CM channel */
956 rdma->channel = rdma_create_event_channel();
957 if (!rdma->channel) {
958 ERROR(errp, "could not create CM channel");
963 ret = rdma_create_id(rdma->channel, &rdma->cm_id, NULL, RDMA_PS_TCP);
965 ERROR(errp, "could not create channel id");
966 goto err_resolve_create_id;
969 snprintf(port_str, 16, "%d", rdma->port);
972 ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
974 ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
975 goto err_resolve_get_addr;
978 for (e = res; e != NULL; e = e->ai_next) {
979 inet_ntop(e->ai_family,
980 &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
981 trace_qemu_rdma_resolve_host_trying(rdma->host, ip);
983 ret = rdma_resolve_addr(rdma->cm_id, NULL, e->ai_dst_addr,
984 RDMA_RESOLVE_TIMEOUT_MS);
986 if (e->ai_family == AF_INET6) {
987 ret = qemu_rdma_broken_ipv6_kernel(rdma->cm_id->verbs, errp);
996 rdma_freeaddrinfo(res);
997 ERROR(errp, "could not resolve address %s", rdma->host);
998 goto err_resolve_get_addr;
1001 rdma_freeaddrinfo(res);
1002 qemu_rdma_dump_gid("source_resolve_addr", rdma->cm_id);
1004 ret = rdma_get_cm_event(rdma->channel, &cm_event);
1006 ERROR(errp, "could not perform event_addr_resolved");
1007 goto err_resolve_get_addr;
1010 if (cm_event->event != RDMA_CM_EVENT_ADDR_RESOLVED) {
1011 ERROR(errp, "result not equal to event_addr_resolved %s",
1012 rdma_event_str(cm_event->event));
1013 error_report("rdma_resolve_addr");
1014 rdma_ack_cm_event(cm_event);
1016 goto err_resolve_get_addr;
1018 rdma_ack_cm_event(cm_event);
1021 ret = rdma_resolve_route(rdma->cm_id, RDMA_RESOLVE_TIMEOUT_MS);
1023 ERROR(errp, "could not resolve rdma route");
1024 goto err_resolve_get_addr;
1027 ret = rdma_get_cm_event(rdma->channel, &cm_event);
1029 ERROR(errp, "could not perform event_route_resolved");
1030 goto err_resolve_get_addr;
1032 if (cm_event->event != RDMA_CM_EVENT_ROUTE_RESOLVED) {
1033 ERROR(errp, "result not equal to event_route_resolved: %s",
1034 rdma_event_str(cm_event->event));
1035 rdma_ack_cm_event(cm_event);
1037 goto err_resolve_get_addr;
1039 rdma_ack_cm_event(cm_event);
1040 rdma->verbs = rdma->cm_id->verbs;
1041 qemu_rdma_dump_id("source_resolve_host", rdma->cm_id->verbs);
1042 qemu_rdma_dump_gid("source_resolve_host", rdma->cm_id);
1045 err_resolve_get_addr:
1046 rdma_destroy_id(rdma->cm_id);
1048 err_resolve_create_id:
1049 rdma_destroy_event_channel(rdma->channel);
1050 rdma->channel = NULL;
1055 * Create protection domain and completion queues
1057 static int qemu_rdma_alloc_pd_cq(RDMAContext *rdma)
1060 rdma->pd = ibv_alloc_pd(rdma->verbs);
1062 error_report("failed to allocate protection domain");
1066 /* create receive completion channel */
1067 rdma->recv_comp_channel = ibv_create_comp_channel(rdma->verbs);
1068 if (!rdma->recv_comp_channel) {
1069 error_report("failed to allocate receive completion channel");
1070 goto err_alloc_pd_cq;
1074 * Completion queue can be filled by read work requests.
1076 rdma->recv_cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3),
1077 NULL, rdma->recv_comp_channel, 0);
1078 if (!rdma->recv_cq) {
1079 error_report("failed to allocate receive completion queue");
1080 goto err_alloc_pd_cq;
1083 /* create send completion channel */
1084 rdma->send_comp_channel = ibv_create_comp_channel(rdma->verbs);
1085 if (!rdma->send_comp_channel) {
1086 error_report("failed to allocate send completion channel");
1087 goto err_alloc_pd_cq;
1090 rdma->send_cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3),
1091 NULL, rdma->send_comp_channel, 0);
1092 if (!rdma->send_cq) {
1093 error_report("failed to allocate send completion queue");
1094 goto err_alloc_pd_cq;
1101 ibv_dealloc_pd(rdma->pd);
1103 if (rdma->recv_comp_channel) {
1104 ibv_destroy_comp_channel(rdma->recv_comp_channel);
1106 if (rdma->send_comp_channel) {
1107 ibv_destroy_comp_channel(rdma->send_comp_channel);
1109 if (rdma->recv_cq) {
1110 ibv_destroy_cq(rdma->recv_cq);
1111 rdma->recv_cq = NULL;
1114 rdma->recv_comp_channel = NULL;
1115 rdma->send_comp_channel = NULL;
1121 * Create queue pairs.
1123 static int qemu_rdma_alloc_qp(RDMAContext *rdma)
1125 struct ibv_qp_init_attr attr = { 0 };
1128 attr.cap.max_send_wr = RDMA_SIGNALED_SEND_MAX;
1129 attr.cap.max_recv_wr = 3;
1130 attr.cap.max_send_sge = 1;
1131 attr.cap.max_recv_sge = 1;
1132 attr.send_cq = rdma->send_cq;
1133 attr.recv_cq = rdma->recv_cq;
1134 attr.qp_type = IBV_QPT_RC;
1136 ret = rdma_create_qp(rdma->cm_id, rdma->pd, &attr);
1141 rdma->qp = rdma->cm_id->qp;
1145 /* Check whether On-Demand Paging is supported by RDAM device */
1146 static bool rdma_support_odp(struct ibv_context *dev)
1148 struct ibv_device_attr_ex attr = {0};
1149 int ret = ibv_query_device_ex(dev, NULL, &attr);
1154 if (attr.odp_caps.general_caps & IBV_ODP_SUPPORT) {
1162 * ibv_advise_mr to avoid RNR NAK error as far as possible.
1163 * The responder mr registering with ODP will sent RNR NAK back to
1164 * the requester in the face of the page fault.
1166 static void qemu_rdma_advise_prefetch_mr(struct ibv_pd *pd, uint64_t addr,
1167 uint32_t len, uint32_t lkey,
1168 const char *name, bool wr)
1170 #ifdef HAVE_IBV_ADVISE_MR
1172 int advice = wr ? IBV_ADVISE_MR_ADVICE_PREFETCH_WRITE :
1173 IBV_ADVISE_MR_ADVICE_PREFETCH;
1174 struct ibv_sge sg_list = {.lkey = lkey, .addr = addr, .length = len};
1176 ret = ibv_advise_mr(pd, advice,
1177 IBV_ADVISE_MR_FLAG_FLUSH, &sg_list, 1);
1178 /* ignore the error */
1180 trace_qemu_rdma_advise_mr(name, len, addr, strerror(errno));
1182 trace_qemu_rdma_advise_mr(name, len, addr, "successed");
1187 static int qemu_rdma_reg_whole_ram_blocks(RDMAContext *rdma)
1190 RDMALocalBlocks *local = &rdma->local_ram_blocks;
1192 for (i = 0; i < local->nb_blocks; i++) {
1193 int access = IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE;
1195 local->block[i].mr =
1196 ibv_reg_mr(rdma->pd,
1197 local->block[i].local_host_addr,
1198 local->block[i].length, access
1201 if (!local->block[i].mr &&
1202 errno == ENOTSUP && rdma_support_odp(rdma->verbs)) {
1203 access |= IBV_ACCESS_ON_DEMAND;
1204 /* register ODP mr */
1205 local->block[i].mr =
1206 ibv_reg_mr(rdma->pd,
1207 local->block[i].local_host_addr,
1208 local->block[i].length, access);
1209 trace_qemu_rdma_register_odp_mr(local->block[i].block_name);
1211 if (local->block[i].mr) {
1212 qemu_rdma_advise_prefetch_mr(rdma->pd,
1213 (uintptr_t)local->block[i].local_host_addr,
1214 local->block[i].length,
1215 local->block[i].mr->lkey,
1216 local->block[i].block_name,
1221 if (!local->block[i].mr) {
1222 perror("Failed to register local dest ram block!");
1225 rdma->total_registrations++;
1228 if (i >= local->nb_blocks) {
1232 for (i--; i >= 0; i--) {
1233 ibv_dereg_mr(local->block[i].mr);
1234 local->block[i].mr = NULL;
1235 rdma->total_registrations--;
1243 * Find the ram block that corresponds to the page requested to be
1244 * transmitted by QEMU.
1246 * Once the block is found, also identify which 'chunk' within that
1247 * block that the page belongs to.
1249 * This search cannot fail or the migration will fail.
1251 static int qemu_rdma_search_ram_block(RDMAContext *rdma,
1252 uintptr_t block_offset,
1255 uint64_t *block_index,
1256 uint64_t *chunk_index)
1258 uint64_t current_addr = block_offset + offset;
1259 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
1260 (void *) block_offset);
1262 assert(current_addr >= block->offset);
1263 assert((current_addr + length) <= (block->offset + block->length));
1265 *block_index = block->index;
1266 *chunk_index = ram_chunk_index(block->local_host_addr,
1267 block->local_host_addr + (current_addr - block->offset));
1273 * Register a chunk with IB. If the chunk was already registered
1274 * previously, then skip.
1276 * Also return the keys associated with the registration needed
1277 * to perform the actual RDMA operation.
1279 static int qemu_rdma_register_and_get_keys(RDMAContext *rdma,
1280 RDMALocalBlock *block, uintptr_t host_addr,
1281 uint32_t *lkey, uint32_t *rkey, int chunk,
1282 uint8_t *chunk_start, uint8_t *chunk_end)
1286 *lkey = block->mr->lkey;
1289 *rkey = block->mr->rkey;
1294 /* allocate memory to store chunk MRs */
1296 block->pmr = g_new0(struct ibv_mr *, block->nb_chunks);
1300 * If 'rkey', then we're the destination, so grant access to the source.
1302 * If 'lkey', then we're the source VM, so grant access only to ourselves.
1304 if (!block->pmr[chunk]) {
1305 uint64_t len = chunk_end - chunk_start;
1306 int access = rkey ? IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE :
1309 trace_qemu_rdma_register_and_get_keys(len, chunk_start);
1311 block->pmr[chunk] = ibv_reg_mr(rdma->pd, chunk_start, len, access);
1312 if (!block->pmr[chunk] &&
1313 errno == ENOTSUP && rdma_support_odp(rdma->verbs)) {
1314 access |= IBV_ACCESS_ON_DEMAND;
1315 /* register ODP mr */
1316 block->pmr[chunk] = ibv_reg_mr(rdma->pd, chunk_start, len, access);
1317 trace_qemu_rdma_register_odp_mr(block->block_name);
1319 if (block->pmr[chunk]) {
1320 qemu_rdma_advise_prefetch_mr(rdma->pd, (uintptr_t)chunk_start,
1321 len, block->pmr[chunk]->lkey,
1322 block->block_name, rkey);
1327 if (!block->pmr[chunk]) {
1328 perror("Failed to register chunk!");
1329 fprintf(stderr, "Chunk details: block: %d chunk index %d"
1330 " start %" PRIuPTR " end %" PRIuPTR
1332 " local %" PRIuPTR " registrations: %d\n",
1333 block->index, chunk, (uintptr_t)chunk_start,
1334 (uintptr_t)chunk_end, host_addr,
1335 (uintptr_t)block->local_host_addr,
1336 rdma->total_registrations);
1339 rdma->total_registrations++;
1342 *lkey = block->pmr[chunk]->lkey;
1345 *rkey = block->pmr[chunk]->rkey;
1351 * Register (at connection time) the memory used for control
1354 static int qemu_rdma_reg_control(RDMAContext *rdma, int idx)
1356 rdma->wr_data[idx].control_mr = ibv_reg_mr(rdma->pd,
1357 rdma->wr_data[idx].control, RDMA_CONTROL_MAX_BUFFER,
1358 IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
1359 if (rdma->wr_data[idx].control_mr) {
1360 rdma->total_registrations++;
1363 error_report("qemu_rdma_reg_control failed");
1367 const char *print_wrid(int wrid)
1369 if (wrid >= RDMA_WRID_RECV_CONTROL) {
1370 return wrid_desc[RDMA_WRID_RECV_CONTROL];
1372 return wrid_desc[wrid];
1376 * Perform a non-optimized memory unregistration after every transfer
1377 * for demonstration purposes, only if pin-all is not requested.
1379 * Potential optimizations:
1380 * 1. Start a new thread to run this function continuously
1382 - and for receipt of unregister messages
1384 * 3. Use workload hints.
1386 static int qemu_rdma_unregister_waiting(RDMAContext *rdma)
1388 while (rdma->unregistrations[rdma->unregister_current]) {
1390 uint64_t wr_id = rdma->unregistrations[rdma->unregister_current];
1392 (wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1394 (wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1395 RDMALocalBlock *block =
1396 &(rdma->local_ram_blocks.block[index]);
1397 RDMARegister reg = { .current_index = index };
1398 RDMAControlHeader resp = { .type = RDMA_CONTROL_UNREGISTER_FINISHED,
1400 RDMAControlHeader head = { .len = sizeof(RDMARegister),
1401 .type = RDMA_CONTROL_UNREGISTER_REQUEST,
1405 trace_qemu_rdma_unregister_waiting_proc(chunk,
1406 rdma->unregister_current);
1408 rdma->unregistrations[rdma->unregister_current] = 0;
1409 rdma->unregister_current++;
1411 if (rdma->unregister_current == RDMA_SIGNALED_SEND_MAX) {
1412 rdma->unregister_current = 0;
1417 * Unregistration is speculative (because migration is single-threaded
1418 * and we cannot break the protocol's inifinband message ordering).
1419 * Thus, if the memory is currently being used for transmission,
1420 * then abort the attempt to unregister and try again
1421 * later the next time a completion is received for this memory.
1423 clear_bit(chunk, block->unregister_bitmap);
1425 if (test_bit(chunk, block->transit_bitmap)) {
1426 trace_qemu_rdma_unregister_waiting_inflight(chunk);
1430 trace_qemu_rdma_unregister_waiting_send(chunk);
1432 ret = ibv_dereg_mr(block->pmr[chunk]);
1433 block->pmr[chunk] = NULL;
1434 block->remote_keys[chunk] = 0;
1437 perror("unregistration chunk failed");
1440 rdma->total_registrations--;
1442 reg.key.chunk = chunk;
1443 register_to_network(rdma, ®);
1444 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) ®,
1450 trace_qemu_rdma_unregister_waiting_complete(chunk);
1456 static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index,
1459 uint64_t result = wr_id & RDMA_WRID_TYPE_MASK;
1461 result |= (index << RDMA_WRID_BLOCK_SHIFT);
1462 result |= (chunk << RDMA_WRID_CHUNK_SHIFT);
1468 * Consult the connection manager to see a work request
1469 * (of any kind) has completed.
1470 * Return the work request ID that completed.
1472 static uint64_t qemu_rdma_poll(RDMAContext *rdma, struct ibv_cq *cq,
1473 uint64_t *wr_id_out, uint32_t *byte_len)
1479 ret = ibv_poll_cq(cq, 1, &wc);
1482 *wr_id_out = RDMA_WRID_NONE;
1487 error_report("ibv_poll_cq return %d", ret);
1491 wr_id = wc.wr_id & RDMA_WRID_TYPE_MASK;
1493 if (wc.status != IBV_WC_SUCCESS) {
1494 fprintf(stderr, "ibv_poll_cq wc.status=%d %s!\n",
1495 wc.status, ibv_wc_status_str(wc.status));
1496 fprintf(stderr, "ibv_poll_cq wrid=%s!\n", wrid_desc[wr_id]);
1501 if (rdma->control_ready_expected &&
1502 (wr_id >= RDMA_WRID_RECV_CONTROL)) {
1503 trace_qemu_rdma_poll_recv(wrid_desc[RDMA_WRID_RECV_CONTROL],
1504 wr_id - RDMA_WRID_RECV_CONTROL, wr_id, rdma->nb_sent);
1505 rdma->control_ready_expected = 0;
1508 if (wr_id == RDMA_WRID_RDMA_WRITE) {
1510 (wc.wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1512 (wc.wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1513 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1515 trace_qemu_rdma_poll_write(print_wrid(wr_id), wr_id, rdma->nb_sent,
1516 index, chunk, block->local_host_addr,
1517 (void *)(uintptr_t)block->remote_host_addr);
1519 clear_bit(chunk, block->transit_bitmap);
1521 if (rdma->nb_sent > 0) {
1525 trace_qemu_rdma_poll_other(print_wrid(wr_id), wr_id, rdma->nb_sent);
1528 *wr_id_out = wc.wr_id;
1530 *byte_len = wc.byte_len;
1536 /* Wait for activity on the completion channel.
1537 * Returns 0 on success, none-0 on error.
1539 static int qemu_rdma_wait_comp_channel(RDMAContext *rdma,
1540 struct ibv_comp_channel *comp_channel)
1542 struct rdma_cm_event *cm_event;
1546 * Coroutine doesn't start until migration_fd_process_incoming()
1547 * so don't yield unless we know we're running inside of a coroutine.
1549 if (rdma->migration_started_on_destination &&
1550 migration_incoming_get_current()->state == MIGRATION_STATUS_ACTIVE) {
1551 yield_until_fd_readable(comp_channel->fd);
1553 /* This is the source side, we're in a separate thread
1554 * or destination prior to migration_fd_process_incoming()
1555 * after postcopy, the destination also in a separate thread.
1556 * we can't yield; so we have to poll the fd.
1557 * But we need to be able to handle 'cancel' or an error
1558 * without hanging forever.
1560 while (!rdma->error_state && !rdma->received_error) {
1562 pfds[0].fd = comp_channel->fd;
1563 pfds[0].events = G_IO_IN | G_IO_HUP | G_IO_ERR;
1564 pfds[0].revents = 0;
1566 pfds[1].fd = rdma->channel->fd;
1567 pfds[1].events = G_IO_IN | G_IO_HUP | G_IO_ERR;
1568 pfds[1].revents = 0;
1570 /* 0.1s timeout, should be fine for a 'cancel' */
1571 switch (qemu_poll_ns(pfds, 2, 100 * 1000 * 1000)) {
1573 case 1: /* fd active */
1574 if (pfds[0].revents) {
1578 if (pfds[1].revents) {
1579 ret = rdma_get_cm_event(rdma->channel, &cm_event);
1581 error_report("failed to get cm event while wait "
1582 "completion channel");
1586 error_report("receive cm event while wait comp channel,"
1587 "cm event is %d", cm_event->event);
1588 if (cm_event->event == RDMA_CM_EVENT_DISCONNECTED ||
1589 cm_event->event == RDMA_CM_EVENT_DEVICE_REMOVAL) {
1590 rdma_ack_cm_event(cm_event);
1593 rdma_ack_cm_event(cm_event);
1597 case 0: /* Timeout, go around again */
1600 default: /* Error of some type -
1601 * I don't trust errno from qemu_poll_ns
1603 error_report("%s: poll failed", __func__);
1607 if (migrate_get_current()->state == MIGRATION_STATUS_CANCELLING) {
1608 /* Bail out and let the cancellation happen */
1614 if (rdma->received_error) {
1617 return rdma->error_state;
1620 static struct ibv_comp_channel *to_channel(RDMAContext *rdma, int wrid)
1622 return wrid < RDMA_WRID_RECV_CONTROL ? rdma->send_comp_channel :
1623 rdma->recv_comp_channel;
1626 static struct ibv_cq *to_cq(RDMAContext *rdma, int wrid)
1628 return wrid < RDMA_WRID_RECV_CONTROL ? rdma->send_cq : rdma->recv_cq;
1632 * Block until the next work request has completed.
1634 * First poll to see if a work request has already completed,
1637 * If we encounter completed work requests for IDs other than
1638 * the one we're interested in, then that's generally an error.
1640 * The only exception is actual RDMA Write completions. These
1641 * completions only need to be recorded, but do not actually
1642 * need further processing.
1644 static int qemu_rdma_block_for_wrid(RDMAContext *rdma, int wrid_requested,
1647 int num_cq_events = 0, ret = 0;
1650 uint64_t wr_id = RDMA_WRID_NONE, wr_id_in;
1651 struct ibv_comp_channel *ch = to_channel(rdma, wrid_requested);
1652 struct ibv_cq *poll_cq = to_cq(rdma, wrid_requested);
1654 if (ibv_req_notify_cq(poll_cq, 0)) {
1658 while (wr_id != wrid_requested) {
1659 ret = qemu_rdma_poll(rdma, poll_cq, &wr_id_in, byte_len);
1664 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1666 if (wr_id == RDMA_WRID_NONE) {
1669 if (wr_id != wrid_requested) {
1670 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested),
1671 wrid_requested, print_wrid(wr_id), wr_id);
1675 if (wr_id == wrid_requested) {
1680 ret = qemu_rdma_wait_comp_channel(rdma, ch);
1682 goto err_block_for_wrid;
1685 ret = ibv_get_cq_event(ch, &cq, &cq_ctx);
1687 perror("ibv_get_cq_event");
1688 goto err_block_for_wrid;
1693 ret = -ibv_req_notify_cq(cq, 0);
1695 goto err_block_for_wrid;
1698 while (wr_id != wrid_requested) {
1699 ret = qemu_rdma_poll(rdma, poll_cq, &wr_id_in, byte_len);
1701 goto err_block_for_wrid;
1704 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1706 if (wr_id == RDMA_WRID_NONE) {
1709 if (wr_id != wrid_requested) {
1710 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested),
1711 wrid_requested, print_wrid(wr_id), wr_id);
1715 if (wr_id == wrid_requested) {
1716 goto success_block_for_wrid;
1720 success_block_for_wrid:
1721 if (num_cq_events) {
1722 ibv_ack_cq_events(cq, num_cq_events);
1727 if (num_cq_events) {
1728 ibv_ack_cq_events(cq, num_cq_events);
1731 rdma->error_state = ret;
1736 * Post a SEND message work request for the control channel
1737 * containing some data and block until the post completes.
1739 static int qemu_rdma_post_send_control(RDMAContext *rdma, uint8_t *buf,
1740 RDMAControlHeader *head)
1743 RDMAWorkRequestData *wr = &rdma->wr_data[RDMA_WRID_CONTROL];
1744 struct ibv_send_wr *bad_wr;
1745 struct ibv_sge sge = {
1746 .addr = (uintptr_t)(wr->control),
1747 .length = head->len + sizeof(RDMAControlHeader),
1748 .lkey = wr->control_mr->lkey,
1750 struct ibv_send_wr send_wr = {
1751 .wr_id = RDMA_WRID_SEND_CONTROL,
1752 .opcode = IBV_WR_SEND,
1753 .send_flags = IBV_SEND_SIGNALED,
1758 trace_qemu_rdma_post_send_control(control_desc(head->type));
1761 * We don't actually need to do a memcpy() in here if we used
1762 * the "sge" properly, but since we're only sending control messages
1763 * (not RAM in a performance-critical path), then its OK for now.
1765 * The copy makes the RDMAControlHeader simpler to manipulate
1766 * for the time being.
1768 assert(head->len <= RDMA_CONTROL_MAX_BUFFER - sizeof(*head));
1769 memcpy(wr->control, head, sizeof(RDMAControlHeader));
1770 control_to_network((void *) wr->control);
1773 memcpy(wr->control + sizeof(RDMAControlHeader), buf, head->len);
1777 ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
1780 error_report("Failed to use post IB SEND for control");
1784 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_SEND_CONTROL, NULL);
1786 error_report("rdma migration: send polling control error");
1793 * Post a RECV work request in anticipation of some future receipt
1794 * of data on the control channel.
1796 static int qemu_rdma_post_recv_control(RDMAContext *rdma, int idx)
1798 struct ibv_recv_wr *bad_wr;
1799 struct ibv_sge sge = {
1800 .addr = (uintptr_t)(rdma->wr_data[idx].control),
1801 .length = RDMA_CONTROL_MAX_BUFFER,
1802 .lkey = rdma->wr_data[idx].control_mr->lkey,
1805 struct ibv_recv_wr recv_wr = {
1806 .wr_id = RDMA_WRID_RECV_CONTROL + idx,
1812 if (ibv_post_recv(rdma->qp, &recv_wr, &bad_wr)) {
1820 * Block and wait for a RECV control channel message to arrive.
1822 static int qemu_rdma_exchange_get_response(RDMAContext *rdma,
1823 RDMAControlHeader *head, int expecting, int idx)
1826 int ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RECV_CONTROL + idx,
1830 error_report("rdma migration: recv polling control error!");
1834 network_to_control((void *) rdma->wr_data[idx].control);
1835 memcpy(head, rdma->wr_data[idx].control, sizeof(RDMAControlHeader));
1837 trace_qemu_rdma_exchange_get_response_start(control_desc(expecting));
1839 if (expecting == RDMA_CONTROL_NONE) {
1840 trace_qemu_rdma_exchange_get_response_none(control_desc(head->type),
1842 } else if (head->type != expecting || head->type == RDMA_CONTROL_ERROR) {
1843 error_report("Was expecting a %s (%d) control message"
1844 ", but got: %s (%d), length: %d",
1845 control_desc(expecting), expecting,
1846 control_desc(head->type), head->type, head->len);
1847 if (head->type == RDMA_CONTROL_ERROR) {
1848 rdma->received_error = true;
1852 if (head->len > RDMA_CONTROL_MAX_BUFFER - sizeof(*head)) {
1853 error_report("too long length: %d", head->len);
1856 if (sizeof(*head) + head->len != byte_len) {
1857 error_report("Malformed length: %d byte_len %d", head->len, byte_len);
1865 * When a RECV work request has completed, the work request's
1866 * buffer is pointed at the header.
1868 * This will advance the pointer to the data portion
1869 * of the control message of the work request's buffer that
1870 * was populated after the work request finished.
1872 static void qemu_rdma_move_header(RDMAContext *rdma, int idx,
1873 RDMAControlHeader *head)
1875 rdma->wr_data[idx].control_len = head->len;
1876 rdma->wr_data[idx].control_curr =
1877 rdma->wr_data[idx].control + sizeof(RDMAControlHeader);
1881 * This is an 'atomic' high-level operation to deliver a single, unified
1882 * control-channel message.
1884 * Additionally, if the user is expecting some kind of reply to this message,
1885 * they can request a 'resp' response message be filled in by posting an
1886 * additional work request on behalf of the user and waiting for an additional
1889 * The extra (optional) response is used during registration to us from having
1890 * to perform an *additional* exchange of message just to provide a response by
1891 * instead piggy-backing on the acknowledgement.
1893 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
1894 uint8_t *data, RDMAControlHeader *resp,
1896 int (*callback)(RDMAContext *rdma))
1901 * Wait until the dest is ready before attempting to deliver the message
1902 * by waiting for a READY message.
1904 if (rdma->control_ready_expected) {
1905 RDMAControlHeader resp;
1906 ret = qemu_rdma_exchange_get_response(rdma,
1907 &resp, RDMA_CONTROL_READY, RDMA_WRID_READY);
1914 * If the user is expecting a response, post a WR in anticipation of it.
1917 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_DATA);
1919 error_report("rdma migration: error posting"
1920 " extra control recv for anticipated result!");
1926 * Post a WR to replace the one we just consumed for the READY message.
1928 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
1930 error_report("rdma migration: error posting first control recv!");
1935 * Deliver the control message that was requested.
1937 ret = qemu_rdma_post_send_control(rdma, data, head);
1940 error_report("Failed to send control buffer!");
1945 * If we're expecting a response, block and wait for it.
1949 trace_qemu_rdma_exchange_send_issue_callback();
1950 ret = callback(rdma);
1956 trace_qemu_rdma_exchange_send_waiting(control_desc(resp->type));
1957 ret = qemu_rdma_exchange_get_response(rdma, resp,
1958 resp->type, RDMA_WRID_DATA);
1964 qemu_rdma_move_header(rdma, RDMA_WRID_DATA, resp);
1966 *resp_idx = RDMA_WRID_DATA;
1968 trace_qemu_rdma_exchange_send_received(control_desc(resp->type));
1971 rdma->control_ready_expected = 1;
1977 * This is an 'atomic' high-level operation to receive a single, unified
1978 * control-channel message.
1980 static int qemu_rdma_exchange_recv(RDMAContext *rdma, RDMAControlHeader *head,
1983 RDMAControlHeader ready = {
1985 .type = RDMA_CONTROL_READY,
1991 * Inform the source that we're ready to receive a message.
1993 ret = qemu_rdma_post_send_control(rdma, NULL, &ready);
1996 error_report("Failed to send control buffer!");
2001 * Block and wait for the message.
2003 ret = qemu_rdma_exchange_get_response(rdma, head,
2004 expecting, RDMA_WRID_READY);
2010 qemu_rdma_move_header(rdma, RDMA_WRID_READY, head);
2013 * Post a new RECV work request to replace the one we just consumed.
2015 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
2017 error_report("rdma migration: error posting second control recv!");
2025 * Write an actual chunk of memory using RDMA.
2027 * If we're using dynamic registration on the dest-side, we have to
2028 * send a registration command first.
2030 static int qemu_rdma_write_one(QEMUFile *f, RDMAContext *rdma,
2031 int current_index, uint64_t current_addr,
2035 struct ibv_send_wr send_wr = { 0 };
2036 struct ibv_send_wr *bad_wr;
2037 int reg_result_idx, ret, count = 0;
2038 uint64_t chunk, chunks;
2039 uint8_t *chunk_start, *chunk_end;
2040 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[current_index]);
2042 RDMARegisterResult *reg_result;
2043 RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT };
2044 RDMAControlHeader head = { .len = sizeof(RDMARegister),
2045 .type = RDMA_CONTROL_REGISTER_REQUEST,
2050 sge.addr = (uintptr_t)(block->local_host_addr +
2051 (current_addr - block->offset));
2052 sge.length = length;
2054 chunk = ram_chunk_index(block->local_host_addr,
2055 (uint8_t *)(uintptr_t)sge.addr);
2056 chunk_start = ram_chunk_start(block, chunk);
2058 if (block->is_ram_block) {
2059 chunks = length / (1UL << RDMA_REG_CHUNK_SHIFT);
2061 if (chunks && ((length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
2065 chunks = block->length / (1UL << RDMA_REG_CHUNK_SHIFT);
2067 if (chunks && ((block->length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
2072 trace_qemu_rdma_write_one_top(chunks + 1,
2074 (1UL << RDMA_REG_CHUNK_SHIFT) / 1024 / 1024);
2076 chunk_end = ram_chunk_end(block, chunk + chunks);
2079 while (test_bit(chunk, block->transit_bitmap)) {
2081 trace_qemu_rdma_write_one_block(count++, current_index, chunk,
2082 sge.addr, length, rdma->nb_sent, block->nb_chunks);
2084 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2087 error_report("Failed to Wait for previous write to complete "
2088 "block %d chunk %" PRIu64
2089 " current %" PRIu64 " len %" PRIu64 " %d",
2090 current_index, chunk, sge.addr, length, rdma->nb_sent);
2095 if (!rdma->pin_all || !block->is_ram_block) {
2096 if (!block->remote_keys[chunk]) {
2098 * This chunk has not yet been registered, so first check to see
2099 * if the entire chunk is zero. If so, tell the other size to
2100 * memset() + madvise() the entire chunk without RDMA.
2103 if (buffer_is_zero((void *)(uintptr_t)sge.addr, length)) {
2104 RDMACompress comp = {
2105 .offset = current_addr,
2107 .block_idx = current_index,
2111 head.len = sizeof(comp);
2112 head.type = RDMA_CONTROL_COMPRESS;
2114 trace_qemu_rdma_write_one_zero(chunk, sge.length,
2115 current_index, current_addr);
2117 compress_to_network(rdma, &comp);
2118 ret = qemu_rdma_exchange_send(rdma, &head,
2119 (uint8_t *) &comp, NULL, NULL, NULL);
2125 stat64_add(&mig_stats.zero_pages,
2126 sge.length / qemu_target_page_size());
2132 * Otherwise, tell other side to register.
2134 reg.current_index = current_index;
2135 if (block->is_ram_block) {
2136 reg.key.current_addr = current_addr;
2138 reg.key.chunk = chunk;
2140 reg.chunks = chunks;
2142 trace_qemu_rdma_write_one_sendreg(chunk, sge.length, current_index,
2145 register_to_network(rdma, ®);
2146 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) ®,
2147 &resp, ®_result_idx, NULL);
2152 /* try to overlap this single registration with the one we sent. */
2153 if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
2154 &sge.lkey, NULL, chunk,
2155 chunk_start, chunk_end)) {
2156 error_report("cannot get lkey");
2160 reg_result = (RDMARegisterResult *)
2161 rdma->wr_data[reg_result_idx].control_curr;
2163 network_to_result(reg_result);
2165 trace_qemu_rdma_write_one_recvregres(block->remote_keys[chunk],
2166 reg_result->rkey, chunk);
2168 block->remote_keys[chunk] = reg_result->rkey;
2169 block->remote_host_addr = reg_result->host_addr;
2171 /* already registered before */
2172 if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
2173 &sge.lkey, NULL, chunk,
2174 chunk_start, chunk_end)) {
2175 error_report("cannot get lkey!");
2180 send_wr.wr.rdma.rkey = block->remote_keys[chunk];
2182 send_wr.wr.rdma.rkey = block->remote_rkey;
2184 if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
2185 &sge.lkey, NULL, chunk,
2186 chunk_start, chunk_end)) {
2187 error_report("cannot get lkey!");
2193 * Encode the ram block index and chunk within this wrid.
2194 * We will use this information at the time of completion
2195 * to figure out which bitmap to check against and then which
2196 * chunk in the bitmap to look for.
2198 send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE,
2199 current_index, chunk);
2201 send_wr.opcode = IBV_WR_RDMA_WRITE;
2202 send_wr.send_flags = IBV_SEND_SIGNALED;
2203 send_wr.sg_list = &sge;
2204 send_wr.num_sge = 1;
2205 send_wr.wr.rdma.remote_addr = block->remote_host_addr +
2206 (current_addr - block->offset);
2208 trace_qemu_rdma_write_one_post(chunk, sge.addr, send_wr.wr.rdma.remote_addr,
2212 * ibv_post_send() does not return negative error numbers,
2213 * per the specification they are positive - no idea why.
2215 ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
2217 if (ret == ENOMEM) {
2218 trace_qemu_rdma_write_one_queue_full();
2219 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2221 error_report("rdma migration: failed to make "
2222 "room in full send queue! %d", ret);
2228 } else if (ret > 0) {
2229 perror("rdma migration: post rdma write failed");
2233 set_bit(chunk, block->transit_bitmap);
2234 stat64_add(&mig_stats.normal_pages, sge.length / qemu_target_page_size());
2235 ram_transferred_add(sge.length);
2236 qemu_file_credit_transfer(f, sge.length);
2237 rdma->total_writes++;
2243 * Push out any unwritten RDMA operations.
2245 * We support sending out multiple chunks at the same time.
2246 * Not all of them need to get signaled in the completion queue.
2248 static int qemu_rdma_write_flush(QEMUFile *f, RDMAContext *rdma)
2252 if (!rdma->current_length) {
2256 ret = qemu_rdma_write_one(f, rdma,
2257 rdma->current_index, rdma->current_addr, rdma->current_length);
2265 trace_qemu_rdma_write_flush(rdma->nb_sent);
2268 rdma->current_length = 0;
2269 rdma->current_addr = 0;
2274 static inline int qemu_rdma_buffer_mergable(RDMAContext *rdma,
2275 uint64_t offset, uint64_t len)
2277 RDMALocalBlock *block;
2281 if (rdma->current_index < 0) {
2285 if (rdma->current_chunk < 0) {
2289 block = &(rdma->local_ram_blocks.block[rdma->current_index]);
2290 host_addr = block->local_host_addr + (offset - block->offset);
2291 chunk_end = ram_chunk_end(block, rdma->current_chunk);
2293 if (rdma->current_length == 0) {
2298 * Only merge into chunk sequentially.
2300 if (offset != (rdma->current_addr + rdma->current_length)) {
2304 if (offset < block->offset) {
2308 if ((offset + len) > (block->offset + block->length)) {
2312 if ((host_addr + len) > chunk_end) {
2320 * We're not actually writing here, but doing three things:
2322 * 1. Identify the chunk the buffer belongs to.
2323 * 2. If the chunk is full or the buffer doesn't belong to the current
2324 * chunk, then start a new chunk and flush() the old chunk.
2325 * 3. To keep the hardware busy, we also group chunks into batches
2326 * and only require that a batch gets acknowledged in the completion
2327 * queue instead of each individual chunk.
2329 static int qemu_rdma_write(QEMUFile *f, RDMAContext *rdma,
2330 uint64_t block_offset, uint64_t offset,
2333 uint64_t current_addr = block_offset + offset;
2334 uint64_t index = rdma->current_index;
2335 uint64_t chunk = rdma->current_chunk;
2338 /* If we cannot merge it, we flush the current buffer first. */
2339 if (!qemu_rdma_buffer_mergable(rdma, current_addr, len)) {
2340 ret = qemu_rdma_write_flush(f, rdma);
2344 rdma->current_length = 0;
2345 rdma->current_addr = current_addr;
2347 ret = qemu_rdma_search_ram_block(rdma, block_offset,
2348 offset, len, &index, &chunk);
2350 error_report("ram block search failed");
2353 rdma->current_index = index;
2354 rdma->current_chunk = chunk;
2358 rdma->current_length += len;
2360 /* flush it if buffer is too large */
2361 if (rdma->current_length >= RDMA_MERGE_MAX) {
2362 return qemu_rdma_write_flush(f, rdma);
2368 static void qemu_rdma_cleanup(RDMAContext *rdma)
2372 if (rdma->cm_id && rdma->connected) {
2373 if ((rdma->error_state ||
2374 migrate_get_current()->state == MIGRATION_STATUS_CANCELLING) &&
2375 !rdma->received_error) {
2376 RDMAControlHeader head = { .len = 0,
2377 .type = RDMA_CONTROL_ERROR,
2380 error_report("Early error. Sending error.");
2381 qemu_rdma_post_send_control(rdma, NULL, &head);
2384 rdma_disconnect(rdma->cm_id);
2385 trace_qemu_rdma_cleanup_disconnect();
2386 rdma->connected = false;
2389 if (rdma->channel) {
2390 qemu_set_fd_handler(rdma->channel->fd, NULL, NULL, NULL);
2392 g_free(rdma->dest_blocks);
2393 rdma->dest_blocks = NULL;
2395 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2396 if (rdma->wr_data[idx].control_mr) {
2397 rdma->total_registrations--;
2398 ibv_dereg_mr(rdma->wr_data[idx].control_mr);
2400 rdma->wr_data[idx].control_mr = NULL;
2403 if (rdma->local_ram_blocks.block) {
2404 while (rdma->local_ram_blocks.nb_blocks) {
2405 rdma_delete_block(rdma, &rdma->local_ram_blocks.block[0]);
2410 rdma_destroy_qp(rdma->cm_id);
2413 if (rdma->recv_cq) {
2414 ibv_destroy_cq(rdma->recv_cq);
2415 rdma->recv_cq = NULL;
2417 if (rdma->send_cq) {
2418 ibv_destroy_cq(rdma->send_cq);
2419 rdma->send_cq = NULL;
2421 if (rdma->recv_comp_channel) {
2422 ibv_destroy_comp_channel(rdma->recv_comp_channel);
2423 rdma->recv_comp_channel = NULL;
2425 if (rdma->send_comp_channel) {
2426 ibv_destroy_comp_channel(rdma->send_comp_channel);
2427 rdma->send_comp_channel = NULL;
2430 ibv_dealloc_pd(rdma->pd);
2434 rdma_destroy_id(rdma->cm_id);
2438 /* the destination side, listen_id and channel is shared */
2439 if (rdma->listen_id) {
2440 if (!rdma->is_return_path) {
2441 rdma_destroy_id(rdma->listen_id);
2443 rdma->listen_id = NULL;
2445 if (rdma->channel) {
2446 if (!rdma->is_return_path) {
2447 rdma_destroy_event_channel(rdma->channel);
2449 rdma->channel = NULL;
2453 if (rdma->channel) {
2454 rdma_destroy_event_channel(rdma->channel);
2455 rdma->channel = NULL;
2458 g_free(rdma->host_port);
2460 rdma->host_port = NULL;
2464 static int qemu_rdma_source_init(RDMAContext *rdma, bool pin_all, Error **errp)
2467 Error *local_err = NULL, **temp = &local_err;
2470 * Will be validated against destination's actual capabilities
2471 * after the connect() completes.
2473 rdma->pin_all = pin_all;
2475 ret = qemu_rdma_resolve_host(rdma, temp);
2477 goto err_rdma_source_init;
2480 ret = qemu_rdma_alloc_pd_cq(rdma);
2482 ERROR(temp, "rdma migration: error allocating pd and cq! Your mlock()"
2483 " limits may be too low. Please check $ ulimit -a # and "
2484 "search for 'ulimit -l' in the output");
2485 goto err_rdma_source_init;
2488 ret = qemu_rdma_alloc_qp(rdma);
2490 ERROR(temp, "rdma migration: error allocating qp!");
2491 goto err_rdma_source_init;
2494 ret = qemu_rdma_init_ram_blocks(rdma);
2496 ERROR(temp, "rdma migration: error initializing ram blocks!");
2497 goto err_rdma_source_init;
2500 /* Build the hash that maps from offset to RAMBlock */
2501 rdma->blockmap = g_hash_table_new(g_direct_hash, g_direct_equal);
2502 for (idx = 0; idx < rdma->local_ram_blocks.nb_blocks; idx++) {
2503 g_hash_table_insert(rdma->blockmap,
2504 (void *)(uintptr_t)rdma->local_ram_blocks.block[idx].offset,
2505 &rdma->local_ram_blocks.block[idx]);
2508 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2509 ret = qemu_rdma_reg_control(rdma, idx);
2511 ERROR(temp, "rdma migration: error registering %d control!",
2513 goto err_rdma_source_init;
2519 err_rdma_source_init:
2520 error_propagate(errp, local_err);
2521 qemu_rdma_cleanup(rdma);
2525 static int qemu_get_cm_event_timeout(RDMAContext *rdma,
2526 struct rdma_cm_event **cm_event,
2527 long msec, Error **errp)
2530 struct pollfd poll_fd = {
2531 .fd = rdma->channel->fd,
2537 ret = poll(&poll_fd, 1, msec);
2538 } while (ret < 0 && errno == EINTR);
2541 ERROR(errp, "poll cm event timeout");
2543 } else if (ret < 0) {
2544 ERROR(errp, "failed to poll cm event, errno=%i", errno);
2546 } else if (poll_fd.revents & POLLIN) {
2547 return rdma_get_cm_event(rdma->channel, cm_event);
2549 ERROR(errp, "no POLLIN event, revent=%x", poll_fd.revents);
2554 static int qemu_rdma_connect(RDMAContext *rdma, Error **errp, bool return_path)
2556 RDMACapabilities cap = {
2557 .version = RDMA_CONTROL_VERSION_CURRENT,
2560 struct rdma_conn_param conn_param = { .initiator_depth = 2,
2562 .private_data = &cap,
2563 .private_data_len = sizeof(cap),
2565 struct rdma_cm_event *cm_event;
2569 * Only negotiate the capability with destination if the user
2570 * on the source first requested the capability.
2572 if (rdma->pin_all) {
2573 trace_qemu_rdma_connect_pin_all_requested();
2574 cap.flags |= RDMA_CAPABILITY_PIN_ALL;
2577 caps_to_network(&cap);
2579 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
2581 ERROR(errp, "posting second control recv");
2582 goto err_rdma_source_connect;
2585 ret = rdma_connect(rdma->cm_id, &conn_param);
2587 perror("rdma_connect");
2588 ERROR(errp, "connecting to destination!");
2589 goto err_rdma_source_connect;
2593 ret = qemu_get_cm_event_timeout(rdma, &cm_event, 5000, errp);
2595 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2598 perror("rdma_get_cm_event after rdma_connect");
2599 ERROR(errp, "connecting to destination!");
2600 goto err_rdma_source_connect;
2603 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
2604 error_report("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect");
2605 ERROR(errp, "connecting to destination!");
2606 rdma_ack_cm_event(cm_event);
2607 goto err_rdma_source_connect;
2609 rdma->connected = true;
2611 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
2612 network_to_caps(&cap);
2615 * Verify that the *requested* capabilities are supported by the destination
2616 * and disable them otherwise.
2618 if (rdma->pin_all && !(cap.flags & RDMA_CAPABILITY_PIN_ALL)) {
2619 ERROR(errp, "Server cannot support pinning all memory. "
2620 "Will register memory dynamically.");
2621 rdma->pin_all = false;
2624 trace_qemu_rdma_connect_pin_all_outcome(rdma->pin_all);
2626 rdma_ack_cm_event(cm_event);
2628 rdma->control_ready_expected = 1;
2632 err_rdma_source_connect:
2633 qemu_rdma_cleanup(rdma);
2637 static int qemu_rdma_dest_init(RDMAContext *rdma, Error **errp)
2640 struct rdma_cm_id *listen_id;
2641 char ip[40] = "unknown";
2642 struct rdma_addrinfo *res, *e;
2646 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2647 rdma->wr_data[idx].control_len = 0;
2648 rdma->wr_data[idx].control_curr = NULL;
2651 if (!rdma->host || !rdma->host[0]) {
2652 ERROR(errp, "RDMA host is not set!");
2653 rdma->error_state = -EINVAL;
2656 /* create CM channel */
2657 rdma->channel = rdma_create_event_channel();
2658 if (!rdma->channel) {
2659 ERROR(errp, "could not create rdma event channel");
2660 rdma->error_state = -EINVAL;
2665 ret = rdma_create_id(rdma->channel, &listen_id, NULL, RDMA_PS_TCP);
2667 ERROR(errp, "could not create cm_id!");
2668 goto err_dest_init_create_listen_id;
2671 snprintf(port_str, 16, "%d", rdma->port);
2672 port_str[15] = '\0';
2674 ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
2676 ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
2677 goto err_dest_init_bind_addr;
2680 ret = rdma_set_option(listen_id, RDMA_OPTION_ID, RDMA_OPTION_ID_REUSEADDR,
2681 &reuse, sizeof reuse);
2683 ERROR(errp, "Error: could not set REUSEADDR option");
2684 goto err_dest_init_bind_addr;
2686 for (e = res; e != NULL; e = e->ai_next) {
2687 inet_ntop(e->ai_family,
2688 &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
2689 trace_qemu_rdma_dest_init_trying(rdma->host, ip);
2690 ret = rdma_bind_addr(listen_id, e->ai_dst_addr);
2694 if (e->ai_family == AF_INET6) {
2695 ret = qemu_rdma_broken_ipv6_kernel(listen_id->verbs, errp);
2703 rdma_freeaddrinfo(res);
2705 ERROR(errp, "Error: could not rdma_bind_addr!");
2706 goto err_dest_init_bind_addr;
2709 rdma->listen_id = listen_id;
2710 qemu_rdma_dump_gid("dest_init", listen_id);
2713 err_dest_init_bind_addr:
2714 rdma_destroy_id(listen_id);
2715 err_dest_init_create_listen_id:
2716 rdma_destroy_event_channel(rdma->channel);
2717 rdma->channel = NULL;
2718 rdma->error_state = ret;
2723 static void qemu_rdma_return_path_dest_init(RDMAContext *rdma_return_path,
2728 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2729 rdma_return_path->wr_data[idx].control_len = 0;
2730 rdma_return_path->wr_data[idx].control_curr = NULL;
2733 /*the CM channel and CM id is shared*/
2734 rdma_return_path->channel = rdma->channel;
2735 rdma_return_path->listen_id = rdma->listen_id;
2737 rdma->return_path = rdma_return_path;
2738 rdma_return_path->return_path = rdma;
2739 rdma_return_path->is_return_path = true;
2742 static void *qemu_rdma_data_init(const char *host_port, Error **errp)
2744 RDMAContext *rdma = NULL;
2745 InetSocketAddress *addr;
2748 rdma = g_new0(RDMAContext, 1);
2749 rdma->current_index = -1;
2750 rdma->current_chunk = -1;
2752 addr = g_new(InetSocketAddress, 1);
2753 if (!inet_parse(addr, host_port, NULL)) {
2754 rdma->port = atoi(addr->port);
2755 rdma->host = g_strdup(addr->host);
2756 rdma->host_port = g_strdup(host_port);
2758 ERROR(errp, "bad RDMA migration address '%s'", host_port);
2763 qapi_free_InetSocketAddress(addr);
2770 * QEMUFile interface to the control channel.
2771 * SEND messages for control only.
2772 * VM's ram is handled with regular RDMA messages.
2774 static ssize_t qio_channel_rdma_writev(QIOChannel *ioc,
2775 const struct iovec *iov,
2782 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2783 QEMUFile *f = rioc->file;
2790 RCU_READ_LOCK_GUARD();
2791 rdma = qatomic_rcu_read(&rioc->rdmaout);
2794 error_setg(errp, "RDMA control channel output is not set");
2798 CHECK_ERROR_STATE();
2801 * Push out any writes that
2802 * we're queued up for VM's ram.
2804 ret = qemu_rdma_write_flush(f, rdma);
2806 rdma->error_state = ret;
2807 error_setg(errp, "qemu_rdma_write_flush returned %d", ret);
2811 for (i = 0; i < niov; i++) {
2812 size_t remaining = iov[i].iov_len;
2813 uint8_t * data = (void *)iov[i].iov_base;
2815 RDMAControlHeader head;
2817 len = MIN(remaining, RDMA_SEND_INCREMENT);
2821 head.type = RDMA_CONTROL_QEMU_FILE;
2823 ret = qemu_rdma_exchange_send(rdma, &head, data, NULL, NULL, NULL);
2826 rdma->error_state = ret;
2827 error_setg(errp, "qemu_rdma_exchange_send returned %d", ret);
2839 static size_t qemu_rdma_fill(RDMAContext *rdma, uint8_t *buf,
2840 size_t size, int idx)
2844 if (rdma->wr_data[idx].control_len) {
2845 trace_qemu_rdma_fill(rdma->wr_data[idx].control_len, size);
2847 len = MIN(size, rdma->wr_data[idx].control_len);
2848 memcpy(buf, rdma->wr_data[idx].control_curr, len);
2849 rdma->wr_data[idx].control_curr += len;
2850 rdma->wr_data[idx].control_len -= len;
2857 * QEMUFile interface to the control channel.
2858 * RDMA links don't use bytestreams, so we have to
2859 * return bytes to QEMUFile opportunistically.
2861 static ssize_t qio_channel_rdma_readv(QIOChannel *ioc,
2862 const struct iovec *iov,
2869 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2871 RDMAControlHeader head;
2876 RCU_READ_LOCK_GUARD();
2877 rdma = qatomic_rcu_read(&rioc->rdmain);
2880 error_setg(errp, "RDMA control channel input is not set");
2884 CHECK_ERROR_STATE();
2886 for (i = 0; i < niov; i++) {
2887 size_t want = iov[i].iov_len;
2888 uint8_t *data = (void *)iov[i].iov_base;
2891 * First, we hold on to the last SEND message we
2892 * were given and dish out the bytes until we run
2895 ret = qemu_rdma_fill(rdma, data, want, 0);
2898 /* Got what we needed, so go to next iovec */
2903 /* If we got any data so far, then don't wait
2904 * for more, just return what we have */
2910 /* We've got nothing at all, so lets wait for
2913 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_QEMU_FILE);
2916 rdma->error_state = ret;
2917 error_setg(errp, "qemu_rdma_exchange_recv returned %d", ret);
2922 * SEND was received with new bytes, now try again.
2924 ret = qemu_rdma_fill(rdma, data, want, 0);
2928 /* Still didn't get enough, so lets just return */
2931 return QIO_CHANNEL_ERR_BLOCK;
2941 * Block until all the outstanding chunks have been delivered by the hardware.
2943 static int qemu_rdma_drain_cq(QEMUFile *f, RDMAContext *rdma)
2947 if (qemu_rdma_write_flush(f, rdma) < 0) {
2951 while (rdma->nb_sent) {
2952 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2954 error_report("rdma migration: complete polling error!");
2959 qemu_rdma_unregister_waiting(rdma);
2965 static int qio_channel_rdma_set_blocking(QIOChannel *ioc,
2969 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2970 /* XXX we should make readv/writev actually honour this :-) */
2971 rioc->blocking = blocking;
2976 typedef struct QIOChannelRDMASource QIOChannelRDMASource;
2977 struct QIOChannelRDMASource {
2979 QIOChannelRDMA *rioc;
2980 GIOCondition condition;
2984 qio_channel_rdma_source_prepare(GSource *source,
2987 QIOChannelRDMASource *rsource = (QIOChannelRDMASource *)source;
2989 GIOCondition cond = 0;
2992 RCU_READ_LOCK_GUARD();
2993 if (rsource->condition == G_IO_IN) {
2994 rdma = qatomic_rcu_read(&rsource->rioc->rdmain);
2996 rdma = qatomic_rcu_read(&rsource->rioc->rdmaout);
3000 error_report("RDMAContext is NULL when prepare Gsource");
3004 if (rdma->wr_data[0].control_len) {
3009 return cond & rsource->condition;
3013 qio_channel_rdma_source_check(GSource *source)
3015 QIOChannelRDMASource *rsource = (QIOChannelRDMASource *)source;
3017 GIOCondition cond = 0;
3019 RCU_READ_LOCK_GUARD();
3020 if (rsource->condition == G_IO_IN) {
3021 rdma = qatomic_rcu_read(&rsource->rioc->rdmain);
3023 rdma = qatomic_rcu_read(&rsource->rioc->rdmaout);
3027 error_report("RDMAContext is NULL when check Gsource");
3031 if (rdma->wr_data[0].control_len) {
3036 return cond & rsource->condition;
3040 qio_channel_rdma_source_dispatch(GSource *source,
3041 GSourceFunc callback,
3044 QIOChannelFunc func = (QIOChannelFunc)callback;
3045 QIOChannelRDMASource *rsource = (QIOChannelRDMASource *)source;
3047 GIOCondition cond = 0;
3049 RCU_READ_LOCK_GUARD();
3050 if (rsource->condition == G_IO_IN) {
3051 rdma = qatomic_rcu_read(&rsource->rioc->rdmain);
3053 rdma = qatomic_rcu_read(&rsource->rioc->rdmaout);
3057 error_report("RDMAContext is NULL when dispatch Gsource");
3061 if (rdma->wr_data[0].control_len) {
3066 return (*func)(QIO_CHANNEL(rsource->rioc),
3067 (cond & rsource->condition),
3072 qio_channel_rdma_source_finalize(GSource *source)
3074 QIOChannelRDMASource *ssource = (QIOChannelRDMASource *)source;
3076 object_unref(OBJECT(ssource->rioc));
3079 GSourceFuncs qio_channel_rdma_source_funcs = {
3080 qio_channel_rdma_source_prepare,
3081 qio_channel_rdma_source_check,
3082 qio_channel_rdma_source_dispatch,
3083 qio_channel_rdma_source_finalize
3086 static GSource *qio_channel_rdma_create_watch(QIOChannel *ioc,
3087 GIOCondition condition)
3089 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3090 QIOChannelRDMASource *ssource;
3093 source = g_source_new(&qio_channel_rdma_source_funcs,
3094 sizeof(QIOChannelRDMASource));
3095 ssource = (QIOChannelRDMASource *)source;
3097 ssource->rioc = rioc;
3098 object_ref(OBJECT(rioc));
3100 ssource->condition = condition;
3105 static void qio_channel_rdma_set_aio_fd_handler(QIOChannel *ioc,
3108 IOHandler *io_write,
3111 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3113 aio_set_fd_handler(ctx, rioc->rdmain->recv_comp_channel->fd, io_read,
3114 io_write, NULL, NULL, opaque);
3115 aio_set_fd_handler(ctx, rioc->rdmain->send_comp_channel->fd, io_read,
3116 io_write, NULL, NULL, opaque);
3118 aio_set_fd_handler(ctx, rioc->rdmaout->recv_comp_channel->fd, io_read,
3119 io_write, NULL, NULL, opaque);
3120 aio_set_fd_handler(ctx, rioc->rdmaout->send_comp_channel->fd, io_read,
3121 io_write, NULL, NULL, opaque);
3125 struct rdma_close_rcu {
3126 struct rcu_head rcu;
3127 RDMAContext *rdmain;
3128 RDMAContext *rdmaout;
3131 /* callback from qio_channel_rdma_close via call_rcu */
3132 static void qio_channel_rdma_close_rcu(struct rdma_close_rcu *rcu)
3135 qemu_rdma_cleanup(rcu->rdmain);
3139 qemu_rdma_cleanup(rcu->rdmaout);
3142 g_free(rcu->rdmain);
3143 g_free(rcu->rdmaout);
3147 static int qio_channel_rdma_close(QIOChannel *ioc,
3150 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3151 RDMAContext *rdmain, *rdmaout;
3152 struct rdma_close_rcu *rcu = g_new(struct rdma_close_rcu, 1);
3154 trace_qemu_rdma_close();
3156 rdmain = rioc->rdmain;
3158 qatomic_rcu_set(&rioc->rdmain, NULL);
3161 rdmaout = rioc->rdmaout;
3163 qatomic_rcu_set(&rioc->rdmaout, NULL);
3166 rcu->rdmain = rdmain;
3167 rcu->rdmaout = rdmaout;
3168 call_rcu(rcu, qio_channel_rdma_close_rcu, rcu);
3174 qio_channel_rdma_shutdown(QIOChannel *ioc,
3175 QIOChannelShutdown how,
3178 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
3179 RDMAContext *rdmain, *rdmaout;
3181 RCU_READ_LOCK_GUARD();
3183 rdmain = qatomic_rcu_read(&rioc->rdmain);
3184 rdmaout = qatomic_rcu_read(&rioc->rdmain);
3187 case QIO_CHANNEL_SHUTDOWN_READ:
3189 rdmain->error_state = -1;
3192 case QIO_CHANNEL_SHUTDOWN_WRITE:
3194 rdmaout->error_state = -1;
3197 case QIO_CHANNEL_SHUTDOWN_BOTH:
3200 rdmain->error_state = -1;
3203 rdmaout->error_state = -1;
3214 * This means that 'block_offset' is a full virtual address that does not
3215 * belong to a RAMBlock of the virtual machine and instead
3216 * represents a private malloc'd memory area that the caller wishes to
3220 * Offset is an offset to be added to block_offset and used
3221 * to also lookup the corresponding RAMBlock.
3223 * @size : Number of bytes to transfer
3225 * @bytes_sent : User-specificed pointer to indicate how many bytes were
3226 * sent. Usually, this will not be more than a few bytes of
3227 * the protocol because most transfers are sent asynchronously.
3229 static size_t qemu_rdma_save_page(QEMUFile *f,
3230 ram_addr_t block_offset, ram_addr_t offset,
3231 size_t size, uint64_t *bytes_sent)
3233 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
3237 if (migration_in_postcopy()) {
3238 return RAM_SAVE_CONTROL_NOT_SUPP;
3241 RCU_READ_LOCK_GUARD();
3242 rdma = qatomic_rcu_read(&rioc->rdmaout);
3248 CHECK_ERROR_STATE();
3253 * Add this page to the current 'chunk'. If the chunk
3254 * is full, or the page doesn't belong to the current chunk,
3255 * an actual RDMA write will occur and a new chunk will be formed.
3257 ret = qemu_rdma_write(f, rdma, block_offset, offset, size);
3259 error_report("rdma migration: write error! %d", ret);
3264 * We always return 1 bytes because the RDMA
3265 * protocol is completely asynchronous. We do not yet know
3266 * whether an identified chunk is zero or not because we're
3267 * waiting for other pages to potentially be merged with
3268 * the current chunk. So, we have to call qemu_update_position()
3269 * later on when the actual write occurs.
3276 * Drain the Completion Queue if possible, but do not block,
3279 * If nothing to poll, the end of the iteration will do this
3280 * again to make sure we don't overflow the request queue.
3283 uint64_t wr_id, wr_id_in;
3284 int ret = qemu_rdma_poll(rdma, rdma->recv_cq, &wr_id_in, NULL);
3286 error_report("rdma migration: polling error! %d", ret);
3290 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
3292 if (wr_id == RDMA_WRID_NONE) {
3298 uint64_t wr_id, wr_id_in;
3299 int ret = qemu_rdma_poll(rdma, rdma->send_cq, &wr_id_in, NULL);
3301 error_report("rdma migration: polling error! %d", ret);
3305 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
3307 if (wr_id == RDMA_WRID_NONE) {
3312 return RAM_SAVE_CONTROL_DELAYED;
3314 rdma->error_state = ret;
3318 static void rdma_accept_incoming_migration(void *opaque);
3320 static void rdma_cm_poll_handler(void *opaque)
3322 RDMAContext *rdma = opaque;
3324 struct rdma_cm_event *cm_event;
3325 MigrationIncomingState *mis = migration_incoming_get_current();
3327 ret = rdma_get_cm_event(rdma->channel, &cm_event);
3329 error_report("get_cm_event failed %d", errno);
3333 if (cm_event->event == RDMA_CM_EVENT_DISCONNECTED ||
3334 cm_event->event == RDMA_CM_EVENT_DEVICE_REMOVAL) {
3335 if (!rdma->error_state &&
3336 migration_incoming_get_current()->state !=
3337 MIGRATION_STATUS_COMPLETED) {
3338 error_report("receive cm event, cm event is %d", cm_event->event);
3339 rdma->error_state = -EPIPE;
3340 if (rdma->return_path) {
3341 rdma->return_path->error_state = -EPIPE;
3344 rdma_ack_cm_event(cm_event);
3345 if (mis->loadvm_co) {
3346 qemu_coroutine_enter(mis->loadvm_co);
3350 rdma_ack_cm_event(cm_event);
3353 static int qemu_rdma_accept(RDMAContext *rdma)
3355 RDMACapabilities cap;
3356 struct rdma_conn_param conn_param = {
3357 .responder_resources = 2,
3358 .private_data = &cap,
3359 .private_data_len = sizeof(cap),
3361 RDMAContext *rdma_return_path = NULL;
3362 struct rdma_cm_event *cm_event;
3363 struct ibv_context *verbs;
3367 ret = rdma_get_cm_event(rdma->channel, &cm_event);
3369 goto err_rdma_dest_wait;
3372 if (cm_event->event != RDMA_CM_EVENT_CONNECT_REQUEST) {
3373 rdma_ack_cm_event(cm_event);
3374 goto err_rdma_dest_wait;
3378 * initialize the RDMAContext for return path for postcopy after first
3379 * connection request reached.
3381 if ((migrate_postcopy() || migrate_return_path())
3382 && !rdma->is_return_path) {
3383 rdma_return_path = qemu_rdma_data_init(rdma->host_port, NULL);
3384 if (rdma_return_path == NULL) {
3385 rdma_ack_cm_event(cm_event);
3386 goto err_rdma_dest_wait;
3389 qemu_rdma_return_path_dest_init(rdma_return_path, rdma);
3392 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
3394 network_to_caps(&cap);
3396 if (cap.version < 1 || cap.version > RDMA_CONTROL_VERSION_CURRENT) {
3397 error_report("Unknown source RDMA version: %d, bailing...",
3399 rdma_ack_cm_event(cm_event);
3400 goto err_rdma_dest_wait;
3404 * Respond with only the capabilities this version of QEMU knows about.
3406 cap.flags &= known_capabilities;
3409 * Enable the ones that we do know about.
3410 * Add other checks here as new ones are introduced.
3412 if (cap.flags & RDMA_CAPABILITY_PIN_ALL) {
3413 rdma->pin_all = true;
3416 rdma->cm_id = cm_event->id;
3417 verbs = cm_event->id->verbs;
3419 rdma_ack_cm_event(cm_event);
3421 trace_qemu_rdma_accept_pin_state(rdma->pin_all);
3423 caps_to_network(&cap);
3425 trace_qemu_rdma_accept_pin_verbsc(verbs);
3428 rdma->verbs = verbs;
3429 } else if (rdma->verbs != verbs) {
3430 error_report("ibv context not matching %p, %p!", rdma->verbs,
3432 goto err_rdma_dest_wait;
3435 qemu_rdma_dump_id("dest_init", verbs);
3437 ret = qemu_rdma_alloc_pd_cq(rdma);
3439 error_report("rdma migration: error allocating pd and cq!");
3440 goto err_rdma_dest_wait;
3443 ret = qemu_rdma_alloc_qp(rdma);
3445 error_report("rdma migration: error allocating qp!");
3446 goto err_rdma_dest_wait;
3449 ret = qemu_rdma_init_ram_blocks(rdma);
3451 error_report("rdma migration: error initializing ram blocks!");
3452 goto err_rdma_dest_wait;
3455 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
3456 ret = qemu_rdma_reg_control(rdma, idx);
3458 error_report("rdma: error registering %d control", idx);
3459 goto err_rdma_dest_wait;
3463 /* Accept the second connection request for return path */
3464 if ((migrate_postcopy() || migrate_return_path())
3465 && !rdma->is_return_path) {
3466 qemu_set_fd_handler(rdma->channel->fd, rdma_accept_incoming_migration,
3468 (void *)(intptr_t)rdma->return_path);
3470 qemu_set_fd_handler(rdma->channel->fd, rdma_cm_poll_handler,
3474 ret = rdma_accept(rdma->cm_id, &conn_param);
3476 error_report("rdma_accept returns %d", ret);
3477 goto err_rdma_dest_wait;
3480 ret = rdma_get_cm_event(rdma->channel, &cm_event);
3482 error_report("rdma_accept get_cm_event failed %d", ret);
3483 goto err_rdma_dest_wait;
3486 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
3487 error_report("rdma_accept not event established");
3488 rdma_ack_cm_event(cm_event);
3489 goto err_rdma_dest_wait;
3492 rdma_ack_cm_event(cm_event);
3493 rdma->connected = true;
3495 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
3497 error_report("rdma migration: error posting second control recv");
3498 goto err_rdma_dest_wait;
3501 qemu_rdma_dump_gid("dest_connect", rdma->cm_id);
3506 rdma->error_state = ret;
3507 qemu_rdma_cleanup(rdma);
3508 g_free(rdma_return_path);
3512 static int dest_ram_sort_func(const void *a, const void *b)
3514 unsigned int a_index = ((const RDMALocalBlock *)a)->src_index;
3515 unsigned int b_index = ((const RDMALocalBlock *)b)->src_index;
3517 return (a_index < b_index) ? -1 : (a_index != b_index);
3521 * During each iteration of the migration, we listen for instructions
3522 * by the source VM to perform dynamic page registrations before they
3523 * can perform RDMA operations.
3525 * We respond with the 'rkey'.
3527 * Keep doing this until the source tells us to stop.
3529 static int qemu_rdma_registration_handle(QEMUFile *f)
3531 RDMAControlHeader reg_resp = { .len = sizeof(RDMARegisterResult),
3532 .type = RDMA_CONTROL_REGISTER_RESULT,
3535 RDMAControlHeader unreg_resp = { .len = 0,
3536 .type = RDMA_CONTROL_UNREGISTER_FINISHED,
3539 RDMAControlHeader blocks = { .type = RDMA_CONTROL_RAM_BLOCKS_RESULT,
3541 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
3543 RDMALocalBlocks *local;
3544 RDMAControlHeader head;
3545 RDMARegister *reg, *registers;
3547 RDMARegisterResult *reg_result;
3548 static RDMARegisterResult results[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE];
3549 RDMALocalBlock *block;
3556 RCU_READ_LOCK_GUARD();
3557 rdma = qatomic_rcu_read(&rioc->rdmain);
3563 CHECK_ERROR_STATE();
3565 local = &rdma->local_ram_blocks;
3567 trace_qemu_rdma_registration_handle_wait();
3569 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_NONE);
3575 if (head.repeat > RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE) {
3576 error_report("rdma: Too many requests in this message (%d)."
3577 "Bailing.", head.repeat);
3582 switch (head.type) {
3583 case RDMA_CONTROL_COMPRESS:
3584 comp = (RDMACompress *) rdma->wr_data[idx].control_curr;
3585 network_to_compress(comp);
3587 trace_qemu_rdma_registration_handle_compress(comp->length,
3590 if (comp->block_idx >= rdma->local_ram_blocks.nb_blocks) {
3591 error_report("rdma: 'compress' bad block index %u (vs %d)",
3592 (unsigned int)comp->block_idx,
3593 rdma->local_ram_blocks.nb_blocks);
3597 block = &(rdma->local_ram_blocks.block[comp->block_idx]);
3599 host_addr = block->local_host_addr +
3600 (comp->offset - block->offset);
3602 ram_handle_compressed(host_addr, comp->value, comp->length);
3605 case RDMA_CONTROL_REGISTER_FINISHED:
3606 trace_qemu_rdma_registration_handle_finished();
3609 case RDMA_CONTROL_RAM_BLOCKS_REQUEST:
3610 trace_qemu_rdma_registration_handle_ram_blocks();
3612 /* Sort our local RAM Block list so it's the same as the source,
3613 * we can do this since we've filled in a src_index in the list
3614 * as we received the RAMBlock list earlier.
3616 qsort(rdma->local_ram_blocks.block,
3617 rdma->local_ram_blocks.nb_blocks,
3618 sizeof(RDMALocalBlock), dest_ram_sort_func);
3619 for (i = 0; i < local->nb_blocks; i++) {
3620 local->block[i].index = i;
3623 if (rdma->pin_all) {
3624 ret = qemu_rdma_reg_whole_ram_blocks(rdma);
3626 error_report("rdma migration: error dest "
3627 "registering ram blocks");
3633 * Dest uses this to prepare to transmit the RAMBlock descriptions
3634 * to the source VM after connection setup.
3635 * Both sides use the "remote" structure to communicate and update
3636 * their "local" descriptions with what was sent.
3638 for (i = 0; i < local->nb_blocks; i++) {
3639 rdma->dest_blocks[i].remote_host_addr =
3640 (uintptr_t)(local->block[i].local_host_addr);
3642 if (rdma->pin_all) {
3643 rdma->dest_blocks[i].remote_rkey = local->block[i].mr->rkey;
3646 rdma->dest_blocks[i].offset = local->block[i].offset;
3647 rdma->dest_blocks[i].length = local->block[i].length;
3649 dest_block_to_network(&rdma->dest_blocks[i]);
3650 trace_qemu_rdma_registration_handle_ram_blocks_loop(
3651 local->block[i].block_name,
3652 local->block[i].offset,
3653 local->block[i].length,
3654 local->block[i].local_host_addr,
3655 local->block[i].src_index);
3658 blocks.len = rdma->local_ram_blocks.nb_blocks
3659 * sizeof(RDMADestBlock);
3662 ret = qemu_rdma_post_send_control(rdma,
3663 (uint8_t *) rdma->dest_blocks, &blocks);
3666 error_report("rdma migration: error sending remote info");
3671 case RDMA_CONTROL_REGISTER_REQUEST:
3672 trace_qemu_rdma_registration_handle_register(head.repeat);
3674 reg_resp.repeat = head.repeat;
3675 registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
3677 for (count = 0; count < head.repeat; count++) {
3679 uint8_t *chunk_start, *chunk_end;
3681 reg = ®isters[count];
3682 network_to_register(reg);
3684 reg_result = &results[count];
3686 trace_qemu_rdma_registration_handle_register_loop(count,
3687 reg->current_index, reg->key.current_addr, reg->chunks);
3689 if (reg->current_index >= rdma->local_ram_blocks.nb_blocks) {
3690 error_report("rdma: 'register' bad block index %u (vs %d)",
3691 (unsigned int)reg->current_index,
3692 rdma->local_ram_blocks.nb_blocks);
3696 block = &(rdma->local_ram_blocks.block[reg->current_index]);
3697 if (block->is_ram_block) {
3698 if (block->offset > reg->key.current_addr) {
3699 error_report("rdma: bad register address for block %s"
3700 " offset: %" PRIx64 " current_addr: %" PRIx64,
3701 block->block_name, block->offset,
3702 reg->key.current_addr);
3706 host_addr = (block->local_host_addr +
3707 (reg->key.current_addr - block->offset));
3708 chunk = ram_chunk_index(block->local_host_addr,
3709 (uint8_t *) host_addr);
3711 chunk = reg->key.chunk;
3712 host_addr = block->local_host_addr +
3713 (reg->key.chunk * (1UL << RDMA_REG_CHUNK_SHIFT));
3714 /* Check for particularly bad chunk value */
3715 if (host_addr < (void *)block->local_host_addr) {
3716 error_report("rdma: bad chunk for block %s"
3718 block->block_name, reg->key.chunk);
3723 chunk_start = ram_chunk_start(block, chunk);
3724 chunk_end = ram_chunk_end(block, chunk + reg->chunks);
3725 /* avoid "-Waddress-of-packed-member" warning */
3726 uint32_t tmp_rkey = 0;
3727 if (qemu_rdma_register_and_get_keys(rdma, block,
3728 (uintptr_t)host_addr, NULL, &tmp_rkey,
3729 chunk, chunk_start, chunk_end)) {
3730 error_report("cannot get rkey");
3734 reg_result->rkey = tmp_rkey;
3736 reg_result->host_addr = (uintptr_t)block->local_host_addr;
3738 trace_qemu_rdma_registration_handle_register_rkey(
3741 result_to_network(reg_result);
3744 ret = qemu_rdma_post_send_control(rdma,
3745 (uint8_t *) results, ®_resp);
3748 error_report("Failed to send control buffer");
3752 case RDMA_CONTROL_UNREGISTER_REQUEST:
3753 trace_qemu_rdma_registration_handle_unregister(head.repeat);
3754 unreg_resp.repeat = head.repeat;
3755 registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
3757 for (count = 0; count < head.repeat; count++) {
3758 reg = ®isters[count];
3759 network_to_register(reg);
3761 trace_qemu_rdma_registration_handle_unregister_loop(count,
3762 reg->current_index, reg->key.chunk);
3764 block = &(rdma->local_ram_blocks.block[reg->current_index]);
3766 ret = ibv_dereg_mr(block->pmr[reg->key.chunk]);
3767 block->pmr[reg->key.chunk] = NULL;
3770 perror("rdma unregistration chunk failed");
3775 rdma->total_registrations--;
3777 trace_qemu_rdma_registration_handle_unregister_success(
3781 ret = qemu_rdma_post_send_control(rdma, NULL, &unreg_resp);
3784 error_report("Failed to send control buffer");
3788 case RDMA_CONTROL_REGISTER_RESULT:
3789 error_report("Invalid RESULT message at dest.");
3793 error_report("Unknown control message %s", control_desc(head.type));
3800 rdma->error_state = ret;
3806 * Called via a ram_control_load_hook during the initial RAM load section which
3807 * lists the RAMBlocks by name. This lets us know the order of the RAMBlocks
3809 * We've already built our local RAMBlock list, but not yet sent the list to
3813 rdma_block_notification_handle(QEMUFile *f, const char *name)
3816 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
3820 RCU_READ_LOCK_GUARD();
3821 rdma = qatomic_rcu_read(&rioc->rdmain);
3827 /* Find the matching RAMBlock in our local list */
3828 for (curr = 0; curr < rdma->local_ram_blocks.nb_blocks; curr++) {
3829 if (!strcmp(rdma->local_ram_blocks.block[curr].block_name, name)) {
3836 error_report("RAMBlock '%s' not found on destination", name);
3840 rdma->local_ram_blocks.block[curr].src_index = rdma->next_src_index;
3841 trace_rdma_block_notification_handle(name, rdma->next_src_index);
3842 rdma->next_src_index++;
3847 static int rdma_load_hook(QEMUFile *f, uint64_t flags, void *data)
3850 case RAM_CONTROL_BLOCK_REG:
3851 return rdma_block_notification_handle(f, data);
3853 case RAM_CONTROL_HOOK:
3854 return qemu_rdma_registration_handle(f);
3857 /* Shouldn't be called with any other values */
3862 static int qemu_rdma_registration_start(QEMUFile *f,
3863 uint64_t flags, void *data)
3865 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
3868 if (migration_in_postcopy()) {
3872 RCU_READ_LOCK_GUARD();
3873 rdma = qatomic_rcu_read(&rioc->rdmaout);
3878 CHECK_ERROR_STATE();
3880 trace_qemu_rdma_registration_start(flags);
3881 qemu_put_be64(f, RAM_SAVE_FLAG_HOOK);
3888 * Inform dest that dynamic registrations are done for now.
3889 * First, flush writes, if any.
3891 static int qemu_rdma_registration_stop(QEMUFile *f,
3892 uint64_t flags, void *data)
3894 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
3896 RDMAControlHeader head = { .len = 0, .repeat = 1 };
3899 if (migration_in_postcopy()) {
3903 RCU_READ_LOCK_GUARD();
3904 rdma = qatomic_rcu_read(&rioc->rdmaout);
3909 CHECK_ERROR_STATE();
3912 ret = qemu_rdma_drain_cq(f, rdma);
3918 if (flags == RAM_CONTROL_SETUP) {
3919 RDMAControlHeader resp = {.type = RDMA_CONTROL_RAM_BLOCKS_RESULT };
3920 RDMALocalBlocks *local = &rdma->local_ram_blocks;
3921 int reg_result_idx, i, nb_dest_blocks;
3923 head.type = RDMA_CONTROL_RAM_BLOCKS_REQUEST;
3924 trace_qemu_rdma_registration_stop_ram();
3927 * Make sure that we parallelize the pinning on both sides.
3928 * For very large guests, doing this serially takes a really
3929 * long time, so we have to 'interleave' the pinning locally
3930 * with the control messages by performing the pinning on this
3931 * side before we receive the control response from the other
3932 * side that the pinning has completed.
3934 ret = qemu_rdma_exchange_send(rdma, &head, NULL, &resp,
3935 ®_result_idx, rdma->pin_all ?
3936 qemu_rdma_reg_whole_ram_blocks : NULL);
3938 fprintf(stderr, "receiving remote info!");
3942 nb_dest_blocks = resp.len / sizeof(RDMADestBlock);
3945 * The protocol uses two different sets of rkeys (mutually exclusive):
3946 * 1. One key to represent the virtual address of the entire ram block.
3947 * (dynamic chunk registration disabled - pin everything with one rkey.)
3948 * 2. One to represent individual chunks within a ram block.
3949 * (dynamic chunk registration enabled - pin individual chunks.)
3951 * Once the capability is successfully negotiated, the destination transmits
3952 * the keys to use (or sends them later) including the virtual addresses
3953 * and then propagates the remote ram block descriptions to his local copy.
3956 if (local->nb_blocks != nb_dest_blocks) {
3957 fprintf(stderr, "ram blocks mismatch (Number of blocks %d vs %d) "
3958 "Your QEMU command line parameters are probably "
3959 "not identical on both the source and destination.",
3960 local->nb_blocks, nb_dest_blocks);
3961 rdma->error_state = -EINVAL;
3965 qemu_rdma_move_header(rdma, reg_result_idx, &resp);
3966 memcpy(rdma->dest_blocks,
3967 rdma->wr_data[reg_result_idx].control_curr, resp.len);
3968 for (i = 0; i < nb_dest_blocks; i++) {
3969 network_to_dest_block(&rdma->dest_blocks[i]);
3971 /* We require that the blocks are in the same order */
3972 if (rdma->dest_blocks[i].length != local->block[i].length) {
3973 fprintf(stderr, "Block %s/%d has a different length %" PRIu64
3974 "vs %" PRIu64, local->block[i].block_name, i,
3975 local->block[i].length,
3976 rdma->dest_blocks[i].length);
3977 rdma->error_state = -EINVAL;
3980 local->block[i].remote_host_addr =
3981 rdma->dest_blocks[i].remote_host_addr;
3982 local->block[i].remote_rkey = rdma->dest_blocks[i].remote_rkey;
3986 trace_qemu_rdma_registration_stop(flags);
3988 head.type = RDMA_CONTROL_REGISTER_FINISHED;
3989 ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL);
3997 rdma->error_state = ret;
4001 static const QEMUFileHooks rdma_read_hooks = {
4002 .hook_ram_load = rdma_load_hook,
4005 static const QEMUFileHooks rdma_write_hooks = {
4006 .before_ram_iterate = qemu_rdma_registration_start,
4007 .after_ram_iterate = qemu_rdma_registration_stop,
4008 .save_page = qemu_rdma_save_page,
4012 static void qio_channel_rdma_finalize(Object *obj)
4014 QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(obj);
4016 qemu_rdma_cleanup(rioc->rdmain);
4017 g_free(rioc->rdmain);
4018 rioc->rdmain = NULL;
4020 if (rioc->rdmaout) {
4021 qemu_rdma_cleanup(rioc->rdmaout);
4022 g_free(rioc->rdmaout);
4023 rioc->rdmaout = NULL;
4027 static void qio_channel_rdma_class_init(ObjectClass *klass,
4028 void *class_data G_GNUC_UNUSED)
4030 QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
4032 ioc_klass->io_writev = qio_channel_rdma_writev;
4033 ioc_klass->io_readv = qio_channel_rdma_readv;
4034 ioc_klass->io_set_blocking = qio_channel_rdma_set_blocking;
4035 ioc_klass->io_close = qio_channel_rdma_close;
4036 ioc_klass->io_create_watch = qio_channel_rdma_create_watch;
4037 ioc_klass->io_set_aio_fd_handler = qio_channel_rdma_set_aio_fd_handler;
4038 ioc_klass->io_shutdown = qio_channel_rdma_shutdown;
4041 static const TypeInfo qio_channel_rdma_info = {
4042 .parent = TYPE_QIO_CHANNEL,
4043 .name = TYPE_QIO_CHANNEL_RDMA,
4044 .instance_size = sizeof(QIOChannelRDMA),
4045 .instance_finalize = qio_channel_rdma_finalize,
4046 .class_init = qio_channel_rdma_class_init,
4049 static void qio_channel_rdma_register_types(void)
4051 type_register_static(&qio_channel_rdma_info);
4054 type_init(qio_channel_rdma_register_types);
4056 static QEMUFile *qemu_fopen_rdma(RDMAContext *rdma, const char *mode)
4058 QIOChannelRDMA *rioc;
4060 if (qemu_file_mode_is_not_valid(mode)) {
4064 rioc = QIO_CHANNEL_RDMA(object_new(TYPE_QIO_CHANNEL_RDMA));
4066 if (mode[0] == 'w') {
4067 rioc->file = qemu_file_new_output(QIO_CHANNEL(rioc));
4068 rioc->rdmaout = rdma;
4069 rioc->rdmain = rdma->return_path;
4070 qemu_file_set_hooks(rioc->file, &rdma_write_hooks);
4072 rioc->file = qemu_file_new_input(QIO_CHANNEL(rioc));
4073 rioc->rdmain = rdma;
4074 rioc->rdmaout = rdma->return_path;
4075 qemu_file_set_hooks(rioc->file, &rdma_read_hooks);
4081 static void rdma_accept_incoming_migration(void *opaque)
4083 RDMAContext *rdma = opaque;
4086 Error *local_err = NULL;
4088 trace_qemu_rdma_accept_incoming_migration();
4089 ret = qemu_rdma_accept(rdma);
4092 fprintf(stderr, "RDMA ERROR: Migration initialization failed\n");
4096 trace_qemu_rdma_accept_incoming_migration_accepted();
4098 if (rdma->is_return_path) {
4102 f = qemu_fopen_rdma(rdma, "rb");
4104 fprintf(stderr, "RDMA ERROR: could not qemu_fopen_rdma\n");
4105 qemu_rdma_cleanup(rdma);
4109 rdma->migration_started_on_destination = 1;
4110 migration_fd_process_incoming(f, &local_err);
4112 error_reportf_err(local_err, "RDMA ERROR:");
4116 void rdma_start_incoming_migration(const char *host_port, Error **errp)
4120 Error *local_err = NULL;
4122 trace_rdma_start_incoming_migration();
4124 /* Avoid ram_block_discard_disable(), cannot change during migration. */
4125 if (ram_block_discard_is_required()) {
4126 error_setg(errp, "RDMA: cannot disable RAM discard");
4130 rdma = qemu_rdma_data_init(host_port, &local_err);
4135 ret = qemu_rdma_dest_init(rdma, &local_err);
4141 trace_rdma_start_incoming_migration_after_dest_init();
4143 ret = rdma_listen(rdma->listen_id, 5);
4146 ERROR(errp, "listening on socket!");
4150 trace_rdma_start_incoming_migration_after_rdma_listen();
4152 qemu_set_fd_handler(rdma->channel->fd, rdma_accept_incoming_migration,
4153 NULL, (void *)(intptr_t)rdma);
4157 qemu_rdma_cleanup(rdma);
4159 error_propagate(errp, local_err);
4162 g_free(rdma->host_port);
4167 void rdma_start_outgoing_migration(void *opaque,
4168 const char *host_port, Error **errp)
4170 MigrationState *s = opaque;
4171 RDMAContext *rdma_return_path = NULL;
4175 /* Avoid ram_block_discard_disable(), cannot change during migration. */
4176 if (ram_block_discard_is_required()) {
4177 error_setg(errp, "RDMA: cannot disable RAM discard");
4181 rdma = qemu_rdma_data_init(host_port, errp);
4186 ret = qemu_rdma_source_init(rdma, migrate_rdma_pin_all(), errp);
4192 trace_rdma_start_outgoing_migration_after_rdma_source_init();
4193 ret = qemu_rdma_connect(rdma, errp, false);
4199 /* RDMA postcopy need a separate queue pair for return path */
4200 if (migrate_postcopy() || migrate_return_path()) {
4201 rdma_return_path = qemu_rdma_data_init(host_port, errp);
4203 if (rdma_return_path == NULL) {
4204 goto return_path_err;
4207 ret = qemu_rdma_source_init(rdma_return_path,
4208 migrate_rdma_pin_all(), errp);
4211 goto return_path_err;
4214 ret = qemu_rdma_connect(rdma_return_path, errp, true);
4217 goto return_path_err;
4220 rdma->return_path = rdma_return_path;
4221 rdma_return_path->return_path = rdma;
4222 rdma_return_path->is_return_path = true;
4225 trace_rdma_start_outgoing_migration_after_rdma_connect();
4227 s->to_dst_file = qemu_fopen_rdma(rdma, "wb");
4228 migrate_fd_connect(s, NULL);
4231 qemu_rdma_cleanup(rdma);
4234 g_free(rdma_return_path);