2 * RDMA protocol and interfaces
4 * Copyright IBM, Corp. 2010-2013
7 * Michael R. Hines <mrhines@us.ibm.com>
8 * Jiuxing Liu <jl@us.ibm.com>
10 * This work is licensed under the terms of the GNU GPL, version 2 or
11 * later. See the COPYING file in the top-level directory.
14 #include "qemu-common.h"
15 #include "migration/migration.h"
16 #include "migration/qemu-file.h"
17 #include "exec/cpu-common.h"
18 #include "qemu/main-loop.h"
19 #include "qemu/sockets.h"
20 #include "qemu/bitmap.h"
21 #include "block/coroutine.h"
23 #include <sys/types.h>
24 #include <sys/socket.h>
26 #include <arpa/inet.h>
28 #include <rdma/rdma_cma.h>
31 //#define DEBUG_RDMA_VERBOSE
32 //#define DEBUG_RDMA_REALLY_VERBOSE
35 #define DPRINTF(fmt, ...) \
36 do { printf("rdma: " fmt, ## __VA_ARGS__); } while (0)
38 #define DPRINTF(fmt, ...) \
42 #ifdef DEBUG_RDMA_VERBOSE
43 #define DDPRINTF(fmt, ...) \
44 do { printf("rdma: " fmt, ## __VA_ARGS__); } while (0)
46 #define DDPRINTF(fmt, ...) \
50 #ifdef DEBUG_RDMA_REALLY_VERBOSE
51 #define DDDPRINTF(fmt, ...) \
52 do { printf("rdma: " fmt, ## __VA_ARGS__); } while (0)
54 #define DDDPRINTF(fmt, ...) \
59 * Print and error on both the Monitor and the Log file.
61 #define ERROR(errp, fmt, ...) \
63 fprintf(stderr, "RDMA ERROR: " fmt "\n", ## __VA_ARGS__); \
64 if (errp && (*(errp) == NULL)) { \
65 error_setg(errp, "RDMA ERROR: " fmt, ## __VA_ARGS__); \
69 #define RDMA_RESOLVE_TIMEOUT_MS 10000
71 /* Do not merge data if larger than this. */
72 #define RDMA_MERGE_MAX (2 * 1024 * 1024)
73 #define RDMA_SIGNALED_SEND_MAX (RDMA_MERGE_MAX / 4096)
75 #define RDMA_REG_CHUNK_SHIFT 20 /* 1 MB */
78 * This is only for non-live state being migrated.
79 * Instead of RDMA_WRITE messages, we use RDMA_SEND
80 * messages for that state, which requires a different
81 * delivery design than main memory.
83 #define RDMA_SEND_INCREMENT 32768
86 * Maximum size infiniband SEND message
88 #define RDMA_CONTROL_MAX_BUFFER (512 * 1024)
89 #define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096
91 #define RDMA_CONTROL_VERSION_CURRENT 1
93 * Capabilities for negotiation.
95 #define RDMA_CAPABILITY_PIN_ALL 0x01
98 * Add the other flags above to this list of known capabilities
99 * as they are introduced.
101 static uint32_t known_capabilities = RDMA_CAPABILITY_PIN_ALL;
103 #define CHECK_ERROR_STATE() \
105 if (rdma->error_state) { \
106 if (!rdma->error_reported) { \
107 fprintf(stderr, "RDMA is in an error state waiting migration" \
109 rdma->error_reported = 1; \
111 return rdma->error_state; \
116 * A work request ID is 64-bits and we split up these bits
119 * bits 0-15 : type of control message, 2^16
120 * bits 16-29: ram block index, 2^14
121 * bits 30-63: ram block chunk number, 2^34
123 * The last two bit ranges are only used for RDMA writes,
124 * in order to track their completion and potentially
125 * also track unregistration status of the message.
127 #define RDMA_WRID_TYPE_SHIFT 0UL
128 #define RDMA_WRID_BLOCK_SHIFT 16UL
129 #define RDMA_WRID_CHUNK_SHIFT 30UL
131 #define RDMA_WRID_TYPE_MASK \
132 ((1UL << RDMA_WRID_BLOCK_SHIFT) - 1UL)
134 #define RDMA_WRID_BLOCK_MASK \
135 (~RDMA_WRID_TYPE_MASK & ((1UL << RDMA_WRID_CHUNK_SHIFT) - 1UL))
137 #define RDMA_WRID_CHUNK_MASK (~RDMA_WRID_BLOCK_MASK & ~RDMA_WRID_TYPE_MASK)
140 * RDMA migration protocol:
141 * 1. RDMA Writes (data messages, i.e. RAM)
142 * 2. IB Send/Recv (control channel messages)
146 RDMA_WRID_RDMA_WRITE = 1,
147 RDMA_WRID_SEND_CONTROL = 2000,
148 RDMA_WRID_RECV_CONTROL = 4000,
151 const char *wrid_desc[] = {
152 [RDMA_WRID_NONE] = "NONE",
153 [RDMA_WRID_RDMA_WRITE] = "WRITE RDMA",
154 [RDMA_WRID_SEND_CONTROL] = "CONTROL SEND",
155 [RDMA_WRID_RECV_CONTROL] = "CONTROL RECV",
159 * Work request IDs for IB SEND messages only (not RDMA writes).
160 * This is used by the migration protocol to transmit
161 * control messages (such as device state and registration commands)
163 * We could use more WRs, but we have enough for now.
173 * SEND/RECV IB Control Messages.
176 RDMA_CONTROL_NONE = 0,
178 RDMA_CONTROL_READY, /* ready to receive */
179 RDMA_CONTROL_QEMU_FILE, /* QEMUFile-transmitted bytes */
180 RDMA_CONTROL_RAM_BLOCKS_REQUEST, /* RAMBlock synchronization */
181 RDMA_CONTROL_RAM_BLOCKS_RESULT, /* RAMBlock synchronization */
182 RDMA_CONTROL_COMPRESS, /* page contains repeat values */
183 RDMA_CONTROL_REGISTER_REQUEST, /* dynamic page registration */
184 RDMA_CONTROL_REGISTER_RESULT, /* key to use after registration */
185 RDMA_CONTROL_REGISTER_FINISHED, /* current iteration finished */
186 RDMA_CONTROL_UNREGISTER_REQUEST, /* dynamic UN-registration */
187 RDMA_CONTROL_UNREGISTER_FINISHED, /* unpinning finished */
190 const char *control_desc[] = {
191 [RDMA_CONTROL_NONE] = "NONE",
192 [RDMA_CONTROL_ERROR] = "ERROR",
193 [RDMA_CONTROL_READY] = "READY",
194 [RDMA_CONTROL_QEMU_FILE] = "QEMU FILE",
195 [RDMA_CONTROL_RAM_BLOCKS_REQUEST] = "RAM BLOCKS REQUEST",
196 [RDMA_CONTROL_RAM_BLOCKS_RESULT] = "RAM BLOCKS RESULT",
197 [RDMA_CONTROL_COMPRESS] = "COMPRESS",
198 [RDMA_CONTROL_REGISTER_REQUEST] = "REGISTER REQUEST",
199 [RDMA_CONTROL_REGISTER_RESULT] = "REGISTER RESULT",
200 [RDMA_CONTROL_REGISTER_FINISHED] = "REGISTER FINISHED",
201 [RDMA_CONTROL_UNREGISTER_REQUEST] = "UNREGISTER REQUEST",
202 [RDMA_CONTROL_UNREGISTER_FINISHED] = "UNREGISTER FINISHED",
206 * Memory and MR structures used to represent an IB Send/Recv work request.
207 * This is *not* used for RDMA writes, only IB Send/Recv.
210 uint8_t control[RDMA_CONTROL_MAX_BUFFER]; /* actual buffer to register */
211 struct ibv_mr *control_mr; /* registration metadata */
212 size_t control_len; /* length of the message */
213 uint8_t *control_curr; /* start of unconsumed bytes */
214 } RDMAWorkRequestData;
217 * Negotiate RDMA capabilities during connection-setup time.
224 static void caps_to_network(RDMACapabilities *cap)
226 cap->version = htonl(cap->version);
227 cap->flags = htonl(cap->flags);
230 static void network_to_caps(RDMACapabilities *cap)
232 cap->version = ntohl(cap->version);
233 cap->flags = ntohl(cap->flags);
237 * Representation of a RAMBlock from an RDMA perspective.
238 * This is not transmitted, only local.
239 * This and subsequent structures cannot be linked lists
240 * because we're using a single IB message to transmit
241 * the information. It's small anyway, so a list is overkill.
243 typedef struct RDMALocalBlock {
244 uint8_t *local_host_addr; /* local virtual address */
245 uint64_t remote_host_addr; /* remote virtual address */
248 struct ibv_mr **pmr; /* MRs for chunk-level registration */
249 struct ibv_mr *mr; /* MR for non-chunk-level registration */
250 uint32_t *remote_keys; /* rkeys for chunk-level registration */
251 uint32_t remote_rkey; /* rkeys for non-chunk-level registration */
252 int index; /* which block are we */
255 unsigned long *transit_bitmap;
256 unsigned long *unregister_bitmap;
260 * Also represents a RAMblock, but only on the dest.
261 * This gets transmitted by the dest during connection-time
262 * to the source VM and then is used to populate the
263 * corresponding RDMALocalBlock with
264 * the information needed to perform the actual RDMA.
266 typedef struct QEMU_PACKED RDMARemoteBlock {
267 uint64_t remote_host_addr;
270 uint32_t remote_rkey;
274 static uint64_t htonll(uint64_t v)
276 union { uint32_t lv[2]; uint64_t llv; } u;
277 u.lv[0] = htonl(v >> 32);
278 u.lv[1] = htonl(v & 0xFFFFFFFFULL);
282 static uint64_t ntohll(uint64_t v) {
283 union { uint32_t lv[2]; uint64_t llv; } u;
285 return ((uint64_t)ntohl(u.lv[0]) << 32) | (uint64_t) ntohl(u.lv[1]);
288 static void remote_block_to_network(RDMARemoteBlock *rb)
290 rb->remote_host_addr = htonll(rb->remote_host_addr);
291 rb->offset = htonll(rb->offset);
292 rb->length = htonll(rb->length);
293 rb->remote_rkey = htonl(rb->remote_rkey);
296 static void network_to_remote_block(RDMARemoteBlock *rb)
298 rb->remote_host_addr = ntohll(rb->remote_host_addr);
299 rb->offset = ntohll(rb->offset);
300 rb->length = ntohll(rb->length);
301 rb->remote_rkey = ntohl(rb->remote_rkey);
305 * Virtual address of the above structures used for transmitting
306 * the RAMBlock descriptions at connection-time.
307 * This structure is *not* transmitted.
309 typedef struct RDMALocalBlocks {
311 bool init; /* main memory init complete */
312 RDMALocalBlock *block;
316 * Main data structure for RDMA state.
317 * While there is only one copy of this structure being allocated right now,
318 * this is the place where one would start if you wanted to consider
319 * having more than one RDMA connection open at the same time.
321 typedef struct RDMAContext {
325 RDMAWorkRequestData wr_data[RDMA_WRID_MAX];
328 * This is used by *_exchange_send() to figure out whether or not
329 * the initial "READY" message has already been received or not.
330 * This is because other functions may potentially poll() and detect
331 * the READY message before send() does, in which case we need to
332 * know if it completed.
334 int control_ready_expected;
336 /* number of outstanding writes */
339 /* store info about current buffer so that we can
340 merge it with future sends */
341 uint64_t current_addr;
342 uint64_t current_length;
343 /* index of ram block the current buffer belongs to */
345 /* index of the chunk in the current ram block */
351 * infiniband-specific variables for opening the device
352 * and maintaining connection state and so forth.
354 * cm_id also has ibv_context, rdma_event_channel, and ibv_qp in
355 * cm_id->verbs, cm_id->channel, and cm_id->qp.
357 struct rdma_cm_id *cm_id; /* connection manager ID */
358 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 *comp_channel; /* completion channel */
364 struct ibv_pd *pd; /* protection domain */
365 struct ibv_cq *cq; /* completion queue */
368 * If a previous write failed (perhaps because of a failed
369 * memory registration, then do not attempt any future work
370 * and remember the error state.
376 * Description of ram blocks used throughout the code.
378 RDMALocalBlocks local_ram_blocks;
379 RDMARemoteBlock *block;
382 * Migration on *destination* started.
383 * Then use coroutine yield function.
384 * Source runs in a thread, so we don't care.
386 int migration_started_on_destination;
388 int total_registrations;
391 int unregister_current, unregister_next;
392 uint64_t unregistrations[RDMA_SIGNALED_SEND_MAX];
394 GHashTable *blockmap;
398 * Interface to the rest of the migration call stack.
400 typedef struct QEMUFileRDMA {
407 * Main structure for IB Send/Recv control messages.
408 * This gets prepended at the beginning of every Send/Recv.
410 typedef struct QEMU_PACKED {
411 uint32_t len; /* Total length of data portion */
412 uint32_t type; /* which control command to perform */
413 uint32_t repeat; /* number of commands in data portion of same type */
417 static void control_to_network(RDMAControlHeader *control)
419 control->type = htonl(control->type);
420 control->len = htonl(control->len);
421 control->repeat = htonl(control->repeat);
424 static void network_to_control(RDMAControlHeader *control)
426 control->type = ntohl(control->type);
427 control->len = ntohl(control->len);
428 control->repeat = ntohl(control->repeat);
432 * Register a single Chunk.
433 * Information sent by the source VM to inform the dest
434 * to register an single chunk of memory before we can perform
435 * the actual RDMA operation.
437 typedef struct QEMU_PACKED {
439 uint64_t current_addr; /* offset into the ramblock of the chunk */
440 uint64_t chunk; /* chunk to lookup if unregistering */
442 uint32_t current_index; /* which ramblock the chunk belongs to */
444 uint64_t chunks; /* how many sequential chunks to register */
447 static void register_to_network(RDMARegister *reg)
449 reg->key.current_addr = htonll(reg->key.current_addr);
450 reg->current_index = htonl(reg->current_index);
451 reg->chunks = htonll(reg->chunks);
454 static void network_to_register(RDMARegister *reg)
456 reg->key.current_addr = ntohll(reg->key.current_addr);
457 reg->current_index = ntohl(reg->current_index);
458 reg->chunks = ntohll(reg->chunks);
461 typedef struct QEMU_PACKED {
462 uint32_t value; /* if zero, we will madvise() */
463 uint32_t block_idx; /* which ram block index */
464 uint64_t offset; /* where in the remote ramblock this chunk */
465 uint64_t length; /* length of the chunk */
468 static void compress_to_network(RDMACompress *comp)
470 comp->value = htonl(comp->value);
471 comp->block_idx = htonl(comp->block_idx);
472 comp->offset = htonll(comp->offset);
473 comp->length = htonll(comp->length);
476 static void network_to_compress(RDMACompress *comp)
478 comp->value = ntohl(comp->value);
479 comp->block_idx = ntohl(comp->block_idx);
480 comp->offset = ntohll(comp->offset);
481 comp->length = ntohll(comp->length);
485 * The result of the dest's memory registration produces an "rkey"
486 * which the source VM must reference in order to perform
487 * the RDMA operation.
489 typedef struct QEMU_PACKED {
493 } RDMARegisterResult;
495 static void result_to_network(RDMARegisterResult *result)
497 result->rkey = htonl(result->rkey);
498 result->host_addr = htonll(result->host_addr);
501 static void network_to_result(RDMARegisterResult *result)
503 result->rkey = ntohl(result->rkey);
504 result->host_addr = ntohll(result->host_addr);
507 const char *print_wrid(int wrid);
508 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
509 uint8_t *data, RDMAControlHeader *resp,
511 int (*callback)(RDMAContext *rdma));
513 static inline uint64_t ram_chunk_index(uint8_t *start, uint8_t *host)
515 return ((uintptr_t) host - (uintptr_t) start) >> RDMA_REG_CHUNK_SHIFT;
518 static inline uint8_t *ram_chunk_start(RDMALocalBlock *rdma_ram_block,
521 return (uint8_t *) (((uintptr_t) rdma_ram_block->local_host_addr)
522 + (i << RDMA_REG_CHUNK_SHIFT));
525 static inline uint8_t *ram_chunk_end(RDMALocalBlock *rdma_ram_block, uint64_t i)
527 uint8_t *result = ram_chunk_start(rdma_ram_block, i) +
528 (1UL << RDMA_REG_CHUNK_SHIFT);
530 if (result > (rdma_ram_block->local_host_addr + rdma_ram_block->length)) {
531 result = rdma_ram_block->local_host_addr + rdma_ram_block->length;
537 static int __qemu_rdma_add_block(RDMAContext *rdma, void *host_addr,
538 ram_addr_t block_offset, uint64_t length)
540 RDMALocalBlocks *local = &rdma->local_ram_blocks;
541 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
542 (void *) block_offset);
543 RDMALocalBlock *old = local->block;
545 assert(block == NULL);
547 local->block = g_malloc0(sizeof(RDMALocalBlock) * (local->nb_blocks + 1));
549 if (local->nb_blocks) {
552 for (x = 0; x < local->nb_blocks; x++) {
553 g_hash_table_remove(rdma->blockmap, (void *)old[x].offset);
554 g_hash_table_insert(rdma->blockmap, (void *)old[x].offset,
557 memcpy(local->block, old, sizeof(RDMALocalBlock) * local->nb_blocks);
561 block = &local->block[local->nb_blocks];
563 block->local_host_addr = host_addr;
564 block->offset = block_offset;
565 block->length = length;
566 block->index = local->nb_blocks;
567 block->nb_chunks = ram_chunk_index(host_addr, host_addr + length) + 1UL;
568 block->transit_bitmap = bitmap_new(block->nb_chunks);
569 bitmap_clear(block->transit_bitmap, 0, block->nb_chunks);
570 block->unregister_bitmap = bitmap_new(block->nb_chunks);
571 bitmap_clear(block->unregister_bitmap, 0, block->nb_chunks);
572 block->remote_keys = g_malloc0(block->nb_chunks * sizeof(uint32_t));
574 block->is_ram_block = local->init ? false : true;
576 g_hash_table_insert(rdma->blockmap, (void *) block_offset, block);
578 DDPRINTF("Added Block: %d, addr: %" PRIu64 ", offset: %" PRIu64
579 " length: %" PRIu64 " end: %" PRIu64 " bits %" PRIu64 " chunks %d\n",
580 local->nb_blocks, (uint64_t) block->local_host_addr, block->offset,
581 block->length, (uint64_t) (block->local_host_addr + block->length),
582 BITS_TO_LONGS(block->nb_chunks) *
583 sizeof(unsigned long) * 8, block->nb_chunks);
591 * Memory regions need to be registered with the device and queue pairs setup
592 * in advanced before the migration starts. This tells us where the RAM blocks
593 * are so that we can register them individually.
595 static void qemu_rdma_init_one_block(void *host_addr,
596 ram_addr_t block_offset, ram_addr_t length, void *opaque)
598 __qemu_rdma_add_block(opaque, host_addr, block_offset, length);
602 * Identify the RAMBlocks and their quantity. They will be references to
603 * identify chunk boundaries inside each RAMBlock and also be referenced
604 * during dynamic page registration.
606 static int qemu_rdma_init_ram_blocks(RDMAContext *rdma)
608 RDMALocalBlocks *local = &rdma->local_ram_blocks;
610 assert(rdma->blockmap == NULL);
611 rdma->blockmap = g_hash_table_new(g_direct_hash, g_direct_equal);
612 memset(local, 0, sizeof *local);
613 qemu_ram_foreach_block(qemu_rdma_init_one_block, rdma);
614 DPRINTF("Allocated %d local ram block structures\n", local->nb_blocks);
615 rdma->block = (RDMARemoteBlock *) g_malloc0(sizeof(RDMARemoteBlock) *
616 rdma->local_ram_blocks.nb_blocks);
621 static int __qemu_rdma_delete_block(RDMAContext *rdma, ram_addr_t block_offset)
623 RDMALocalBlocks *local = &rdma->local_ram_blocks;
624 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
625 (void *) block_offset);
626 RDMALocalBlock *old = local->block;
634 for (j = 0; j < block->nb_chunks; j++) {
635 if (!block->pmr[j]) {
638 ibv_dereg_mr(block->pmr[j]);
639 rdma->total_registrations--;
646 ibv_dereg_mr(block->mr);
647 rdma->total_registrations--;
651 g_free(block->transit_bitmap);
652 block->transit_bitmap = NULL;
654 g_free(block->unregister_bitmap);
655 block->unregister_bitmap = NULL;
657 g_free(block->remote_keys);
658 block->remote_keys = NULL;
660 for (x = 0; x < local->nb_blocks; x++) {
661 g_hash_table_remove(rdma->blockmap, (void *)old[x].offset);
664 if (local->nb_blocks > 1) {
666 local->block = g_malloc0(sizeof(RDMALocalBlock) *
667 (local->nb_blocks - 1));
670 memcpy(local->block, old, sizeof(RDMALocalBlock) * block->index);
673 if (block->index < (local->nb_blocks - 1)) {
674 memcpy(local->block + block->index, old + (block->index + 1),
675 sizeof(RDMALocalBlock) *
676 (local->nb_blocks - (block->index + 1)));
679 assert(block == local->block);
683 DDPRINTF("Deleted Block: %d, addr: %" PRIu64 ", offset: %" PRIu64
684 " length: %" PRIu64 " end: %" PRIu64 " bits %" PRIu64 " chunks %d\n",
685 local->nb_blocks, (uint64_t) block->local_host_addr, block->offset,
686 block->length, (uint64_t) (block->local_host_addr + block->length),
687 BITS_TO_LONGS(block->nb_chunks) *
688 sizeof(unsigned long) * 8, block->nb_chunks);
694 if (local->nb_blocks) {
695 for (x = 0; x < local->nb_blocks; x++) {
696 g_hash_table_insert(rdma->blockmap, (void *)local->block[x].offset,
705 * Put in the log file which RDMA device was opened and the details
706 * associated with that device.
708 static void qemu_rdma_dump_id(const char *who, struct ibv_context *verbs)
710 printf("%s RDMA Device opened: kernel name %s "
711 "uverbs device name %s, "
712 "infiniband_verbs class device path %s,"
713 " infiniband class device path %s\n",
716 verbs->device->dev_name,
717 verbs->device->dev_path,
718 verbs->device->ibdev_path);
722 * Put in the log file the RDMA gid addressing information,
723 * useful for folks who have trouble understanding the
724 * RDMA device hierarchy in the kernel.
726 static void qemu_rdma_dump_gid(const char *who, struct rdma_cm_id *id)
730 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.sgid, sgid, sizeof sgid);
731 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.dgid, dgid, sizeof dgid);
732 DPRINTF("%s Source GID: %s, Dest GID: %s\n", who, sgid, dgid);
736 * Figure out which RDMA device corresponds to the requested IP hostname
737 * Also create the initial connection manager identifiers for opening
740 static int qemu_rdma_resolve_host(RDMAContext *rdma, Error **errp)
743 struct addrinfo *res;
745 struct rdma_cm_event *cm_event;
746 char ip[40] = "unknown";
749 if (rdma->host == NULL || !strcmp(rdma->host, "")) {
750 ERROR(errp, "RDMA hostname has not been set");
754 /* create CM channel */
755 rdma->channel = rdma_create_event_channel();
756 if (!rdma->channel) {
757 ERROR(errp, "could not create CM channel");
762 ret = rdma_create_id(rdma->channel, &rdma->cm_id, NULL, RDMA_PS_TCP);
764 ERROR(errp, "could not create channel id");
765 goto err_resolve_create_id;
768 snprintf(port_str, 16, "%d", rdma->port);
771 ret = getaddrinfo(rdma->host, port_str, NULL, &res);
773 ERROR(errp, "could not getaddrinfo address %s", rdma->host);
774 goto err_resolve_get_addr;
777 for (e = res; e != NULL; e = e->ai_next) {
778 inet_ntop(e->ai_family,
779 &((struct sockaddr_in *) e->ai_addr)->sin_addr, ip, sizeof ip);
780 DPRINTF("Trying %s => %s\n", rdma->host, ip);
782 /* resolve the first address */
783 ret = rdma_resolve_addr(rdma->cm_id, NULL, e->ai_addr,
784 RDMA_RESOLVE_TIMEOUT_MS);
790 ERROR(errp, "could not resolve address %s", rdma->host);
791 goto err_resolve_get_addr;
794 qemu_rdma_dump_gid("source_resolve_addr", rdma->cm_id);
796 ret = rdma_get_cm_event(rdma->channel, &cm_event);
798 ERROR(errp, "could not perform event_addr_resolved");
799 goto err_resolve_get_addr;
802 if (cm_event->event != RDMA_CM_EVENT_ADDR_RESOLVED) {
803 ERROR(errp, "result not equal to event_addr_resolved %s",
804 rdma_event_str(cm_event->event));
805 perror("rdma_resolve_addr");
806 goto err_resolve_get_addr;
808 rdma_ack_cm_event(cm_event);
811 ret = rdma_resolve_route(rdma->cm_id, RDMA_RESOLVE_TIMEOUT_MS);
813 ERROR(errp, "could not resolve rdma route");
814 goto err_resolve_get_addr;
817 ret = rdma_get_cm_event(rdma->channel, &cm_event);
819 ERROR(errp, "could not perform event_route_resolved");
820 goto err_resolve_get_addr;
822 if (cm_event->event != RDMA_CM_EVENT_ROUTE_RESOLVED) {
823 ERROR(errp, "result not equal to event_route_resolved: %s",
824 rdma_event_str(cm_event->event));
825 rdma_ack_cm_event(cm_event);
826 goto err_resolve_get_addr;
828 rdma_ack_cm_event(cm_event);
829 rdma->verbs = rdma->cm_id->verbs;
830 qemu_rdma_dump_id("source_resolve_host", rdma->cm_id->verbs);
831 qemu_rdma_dump_gid("source_resolve_host", rdma->cm_id);
834 err_resolve_get_addr:
835 rdma_destroy_id(rdma->cm_id);
837 err_resolve_create_id:
838 rdma_destroy_event_channel(rdma->channel);
839 rdma->channel = NULL;
845 * Create protection domain and completion queues
847 static int qemu_rdma_alloc_pd_cq(RDMAContext *rdma)
850 rdma->pd = ibv_alloc_pd(rdma->verbs);
852 fprintf(stderr, "failed to allocate protection domain\n");
856 /* create completion channel */
857 rdma->comp_channel = ibv_create_comp_channel(rdma->verbs);
858 if (!rdma->comp_channel) {
859 fprintf(stderr, "failed to allocate completion channel\n");
860 goto err_alloc_pd_cq;
864 * Completion queue can be filled by both read and write work requests,
865 * so must reflect the sum of both possible queue sizes.
867 rdma->cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3),
868 NULL, rdma->comp_channel, 0);
870 fprintf(stderr, "failed to allocate completion queue\n");
871 goto err_alloc_pd_cq;
878 ibv_dealloc_pd(rdma->pd);
880 if (rdma->comp_channel) {
881 ibv_destroy_comp_channel(rdma->comp_channel);
884 rdma->comp_channel = NULL;
890 * Create queue pairs.
892 static int qemu_rdma_alloc_qp(RDMAContext *rdma)
894 struct ibv_qp_init_attr attr = { 0 };
897 attr.cap.max_send_wr = RDMA_SIGNALED_SEND_MAX;
898 attr.cap.max_recv_wr = 3;
899 attr.cap.max_send_sge = 1;
900 attr.cap.max_recv_sge = 1;
901 attr.send_cq = rdma->cq;
902 attr.recv_cq = rdma->cq;
903 attr.qp_type = IBV_QPT_RC;
905 ret = rdma_create_qp(rdma->cm_id, rdma->pd, &attr);
910 rdma->qp = rdma->cm_id->qp;
914 static int qemu_rdma_reg_whole_ram_blocks(RDMAContext *rdma)
917 RDMALocalBlocks *local = &rdma->local_ram_blocks;
919 for (i = 0; i < local->nb_blocks; i++) {
922 local->block[i].local_host_addr,
923 local->block[i].length,
924 IBV_ACCESS_LOCAL_WRITE |
925 IBV_ACCESS_REMOTE_WRITE
927 if (!local->block[i].mr) {
928 perror("Failed to register local dest ram block!\n");
931 rdma->total_registrations++;
934 if (i >= local->nb_blocks) {
938 for (i--; i >= 0; i--) {
939 ibv_dereg_mr(local->block[i].mr);
940 rdma->total_registrations--;
948 * Find the ram block that corresponds to the page requested to be
949 * transmitted by QEMU.
951 * Once the block is found, also identify which 'chunk' within that
952 * block that the page belongs to.
954 * This search cannot fail or the migration will fail.
956 static int qemu_rdma_search_ram_block(RDMAContext *rdma,
957 uint64_t block_offset,
960 uint64_t *block_index,
961 uint64_t *chunk_index)
963 uint64_t current_addr = block_offset + offset;
964 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
965 (void *) block_offset);
967 assert(current_addr >= block->offset);
968 assert((current_addr + length) <= (block->offset + block->length));
970 *block_index = block->index;
971 *chunk_index = ram_chunk_index(block->local_host_addr,
972 block->local_host_addr + (current_addr - block->offset));
978 * Register a chunk with IB. If the chunk was already registered
979 * previously, then skip.
981 * Also return the keys associated with the registration needed
982 * to perform the actual RDMA operation.
984 static int qemu_rdma_register_and_get_keys(RDMAContext *rdma,
985 RDMALocalBlock *block, uint8_t *host_addr,
986 uint32_t *lkey, uint32_t *rkey, int chunk,
987 uint8_t *chunk_start, uint8_t *chunk_end)
991 *lkey = block->mr->lkey;
994 *rkey = block->mr->rkey;
999 /* allocate memory to store chunk MRs */
1001 block->pmr = g_malloc0(block->nb_chunks * sizeof(struct ibv_mr *));
1008 * If 'rkey', then we're the destination, so grant access to the source.
1010 * If 'lkey', then we're the source VM, so grant access only to ourselves.
1012 if (!block->pmr[chunk]) {
1013 uint64_t len = chunk_end - chunk_start;
1015 DDPRINTF("Registering %" PRIu64 " bytes @ %p\n",
1018 block->pmr[chunk] = ibv_reg_mr(rdma->pd,
1020 (rkey ? (IBV_ACCESS_LOCAL_WRITE |
1021 IBV_ACCESS_REMOTE_WRITE) : 0));
1023 if (!block->pmr[chunk]) {
1024 perror("Failed to register chunk!");
1025 fprintf(stderr, "Chunk details: block: %d chunk index %d"
1026 " start %" PRIu64 " end %" PRIu64 " host %" PRIu64
1027 " local %" PRIu64 " registrations: %d\n",
1028 block->index, chunk, (uint64_t) chunk_start,
1029 (uint64_t) chunk_end, (uint64_t) host_addr,
1030 (uint64_t) block->local_host_addr,
1031 rdma->total_registrations);
1034 rdma->total_registrations++;
1038 *lkey = block->pmr[chunk]->lkey;
1041 *rkey = block->pmr[chunk]->rkey;
1047 * Register (at connection time) the memory used for control
1050 static int qemu_rdma_reg_control(RDMAContext *rdma, int idx)
1052 rdma->wr_data[idx].control_mr = ibv_reg_mr(rdma->pd,
1053 rdma->wr_data[idx].control, RDMA_CONTROL_MAX_BUFFER,
1054 IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
1055 if (rdma->wr_data[idx].control_mr) {
1056 rdma->total_registrations++;
1059 fprintf(stderr, "qemu_rdma_reg_control failed!\n");
1063 const char *print_wrid(int wrid)
1065 if (wrid >= RDMA_WRID_RECV_CONTROL) {
1066 return wrid_desc[RDMA_WRID_RECV_CONTROL];
1068 return wrid_desc[wrid];
1072 * RDMA requires memory registration (mlock/pinning), but this is not good for
1075 * In preparation for the future where LRU information or workload-specific
1076 * writable writable working set memory access behavior is available to QEMU
1077 * it would be nice to have in place the ability to UN-register/UN-pin
1078 * particular memory regions from the RDMA hardware when it is determine that
1079 * those regions of memory will likely not be accessed again in the near future.
1081 * While we do not yet have such information right now, the following
1082 * compile-time option allows us to perform a non-optimized version of this
1085 * By uncommenting this option, you will cause *all* RDMA transfers to be
1086 * unregistered immediately after the transfer completes on both sides of the
1087 * connection. This has no effect in 'rdma-pin-all' mode, only regular mode.
1089 * This will have a terrible impact on migration performance, so until future
1090 * workload information or LRU information is available, do not attempt to use
1091 * this feature except for basic testing.
1093 //#define RDMA_UNREGISTRATION_EXAMPLE
1096 * Perform a non-optimized memory unregistration after every transfer
1097 * for demonsration purposes, only if pin-all is not requested.
1099 * Potential optimizations:
1100 * 1. Start a new thread to run this function continuously
1102 - and for receipt of unregister messages
1104 * 3. Use workload hints.
1106 static int qemu_rdma_unregister_waiting(RDMAContext *rdma)
1108 while (rdma->unregistrations[rdma->unregister_current]) {
1110 uint64_t wr_id = rdma->unregistrations[rdma->unregister_current];
1112 (wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1114 (wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1115 RDMALocalBlock *block =
1116 &(rdma->local_ram_blocks.block[index]);
1117 RDMARegister reg = { .current_index = index };
1118 RDMAControlHeader resp = { .type = RDMA_CONTROL_UNREGISTER_FINISHED,
1120 RDMAControlHeader head = { .len = sizeof(RDMARegister),
1121 .type = RDMA_CONTROL_UNREGISTER_REQUEST,
1125 DDPRINTF("Processing unregister for chunk: %" PRIu64
1126 " at position %d\n", chunk, rdma->unregister_current);
1128 rdma->unregistrations[rdma->unregister_current] = 0;
1129 rdma->unregister_current++;
1131 if (rdma->unregister_current == RDMA_SIGNALED_SEND_MAX) {
1132 rdma->unregister_current = 0;
1137 * Unregistration is speculative (because migration is single-threaded
1138 * and we cannot break the protocol's inifinband message ordering).
1139 * Thus, if the memory is currently being used for transmission,
1140 * then abort the attempt to unregister and try again
1141 * later the next time a completion is received for this memory.
1143 clear_bit(chunk, block->unregister_bitmap);
1145 if (test_bit(chunk, block->transit_bitmap)) {
1146 DDPRINTF("Cannot unregister inflight chunk: %" PRIu64 "\n", chunk);
1150 DDPRINTF("Sending unregister for chunk: %" PRIu64 "\n", chunk);
1152 ret = ibv_dereg_mr(block->pmr[chunk]);
1153 block->pmr[chunk] = NULL;
1154 block->remote_keys[chunk] = 0;
1157 perror("unregistration chunk failed");
1160 rdma->total_registrations--;
1162 reg.key.chunk = chunk;
1163 register_to_network(®);
1164 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) ®,
1170 DDPRINTF("Unregister for chunk: %" PRIu64 " complete.\n", chunk);
1176 static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index,
1179 uint64_t result = wr_id & RDMA_WRID_TYPE_MASK;
1181 result |= (index << RDMA_WRID_BLOCK_SHIFT);
1182 result |= (chunk << RDMA_WRID_CHUNK_SHIFT);
1188 * Set bit for unregistration in the next iteration.
1189 * We cannot transmit right here, but will unpin later.
1191 static void qemu_rdma_signal_unregister(RDMAContext *rdma, uint64_t index,
1192 uint64_t chunk, uint64_t wr_id)
1194 if (rdma->unregistrations[rdma->unregister_next] != 0) {
1195 fprintf(stderr, "rdma migration: queue is full!\n");
1197 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1199 if (!test_and_set_bit(chunk, block->unregister_bitmap)) {
1200 DDPRINTF("Appending unregister chunk %" PRIu64
1201 " at position %d\n", chunk, rdma->unregister_next);
1203 rdma->unregistrations[rdma->unregister_next++] =
1204 qemu_rdma_make_wrid(wr_id, index, chunk);
1206 if (rdma->unregister_next == RDMA_SIGNALED_SEND_MAX) {
1207 rdma->unregister_next = 0;
1210 DDPRINTF("Unregister chunk %" PRIu64 " already in queue.\n",
1217 * Consult the connection manager to see a work request
1218 * (of any kind) has completed.
1219 * Return the work request ID that completed.
1221 static uint64_t qemu_rdma_poll(RDMAContext *rdma, uint64_t *wr_id_out,
1228 ret = ibv_poll_cq(rdma->cq, 1, &wc);
1231 *wr_id_out = RDMA_WRID_NONE;
1236 fprintf(stderr, "ibv_poll_cq return %d!\n", ret);
1240 wr_id = wc.wr_id & RDMA_WRID_TYPE_MASK;
1242 if (wc.status != IBV_WC_SUCCESS) {
1243 fprintf(stderr, "ibv_poll_cq wc.status=%d %s!\n",
1244 wc.status, ibv_wc_status_str(wc.status));
1245 fprintf(stderr, "ibv_poll_cq wrid=%s!\n", wrid_desc[wr_id]);
1250 if (rdma->control_ready_expected &&
1251 (wr_id >= RDMA_WRID_RECV_CONTROL)) {
1252 DDDPRINTF("completion %s #%" PRId64 " received (%" PRId64 ")"
1253 " left %d\n", wrid_desc[RDMA_WRID_RECV_CONTROL],
1254 wr_id - RDMA_WRID_RECV_CONTROL, wr_id, rdma->nb_sent);
1255 rdma->control_ready_expected = 0;
1258 if (wr_id == RDMA_WRID_RDMA_WRITE) {
1260 (wc.wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1262 (wc.wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1263 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1265 DDDPRINTF("completions %s (%" PRId64 ") left %d, "
1266 "block %" PRIu64 ", chunk: %" PRIu64 " %p %p\n",
1267 print_wrid(wr_id), wr_id, rdma->nb_sent, index, chunk,
1268 block->local_host_addr, (void *)block->remote_host_addr);
1270 clear_bit(chunk, block->transit_bitmap);
1272 if (rdma->nb_sent > 0) {
1276 if (!rdma->pin_all) {
1278 * FYI: If one wanted to signal a specific chunk to be unregistered
1279 * using LRU or workload-specific information, this is the function
1280 * you would call to do so. That chunk would then get asynchronously
1281 * unregistered later.
1283 #ifdef RDMA_UNREGISTRATION_EXAMPLE
1284 qemu_rdma_signal_unregister(rdma, index, chunk, wc.wr_id);
1288 DDDPRINTF("other completion %s (%" PRId64 ") received left %d\n",
1289 print_wrid(wr_id), wr_id, rdma->nb_sent);
1292 *wr_id_out = wc.wr_id;
1294 *byte_len = wc.byte_len;
1301 * Block until the next work request has completed.
1303 * First poll to see if a work request has already completed,
1306 * If we encounter completed work requests for IDs other than
1307 * the one we're interested in, then that's generally an error.
1309 * The only exception is actual RDMA Write completions. These
1310 * completions only need to be recorded, but do not actually
1311 * need further processing.
1313 static int qemu_rdma_block_for_wrid(RDMAContext *rdma, int wrid_requested,
1316 int num_cq_events = 0, ret = 0;
1319 uint64_t wr_id = RDMA_WRID_NONE, wr_id_in;
1321 if (ibv_req_notify_cq(rdma->cq, 0)) {
1325 while (wr_id != wrid_requested) {
1326 ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
1331 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1333 if (wr_id == RDMA_WRID_NONE) {
1336 if (wr_id != wrid_requested) {
1337 DDDPRINTF("A Wanted wrid %s (%d) but got %s (%" PRIu64 ")\n",
1338 print_wrid(wrid_requested),
1339 wrid_requested, print_wrid(wr_id), wr_id);
1343 if (wr_id == wrid_requested) {
1349 * Coroutine doesn't start until process_incoming_migration()
1350 * so don't yield unless we know we're running inside of a coroutine.
1352 if (rdma->migration_started_on_destination) {
1353 yield_until_fd_readable(rdma->comp_channel->fd);
1356 if (ibv_get_cq_event(rdma->comp_channel, &cq, &cq_ctx)) {
1357 perror("ibv_get_cq_event");
1358 goto err_block_for_wrid;
1363 if (ibv_req_notify_cq(cq, 0)) {
1364 goto err_block_for_wrid;
1367 while (wr_id != wrid_requested) {
1368 ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
1370 goto err_block_for_wrid;
1373 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1375 if (wr_id == RDMA_WRID_NONE) {
1378 if (wr_id != wrid_requested) {
1379 DDDPRINTF("B Wanted wrid %s (%d) but got %s (%" PRIu64 ")\n",
1380 print_wrid(wrid_requested), wrid_requested,
1381 print_wrid(wr_id), wr_id);
1385 if (wr_id == wrid_requested) {
1386 goto success_block_for_wrid;
1390 success_block_for_wrid:
1391 if (num_cq_events) {
1392 ibv_ack_cq_events(cq, num_cq_events);
1397 if (num_cq_events) {
1398 ibv_ack_cq_events(cq, num_cq_events);
1404 * Post a SEND message work request for the control channel
1405 * containing some data and block until the post completes.
1407 static int qemu_rdma_post_send_control(RDMAContext *rdma, uint8_t *buf,
1408 RDMAControlHeader *head)
1411 RDMAWorkRequestData *wr = &rdma->wr_data[RDMA_WRID_CONTROL];
1412 struct ibv_send_wr *bad_wr;
1413 struct ibv_sge sge = {
1414 .addr = (uint64_t)(wr->control),
1415 .length = head->len + sizeof(RDMAControlHeader),
1416 .lkey = wr->control_mr->lkey,
1418 struct ibv_send_wr send_wr = {
1419 .wr_id = RDMA_WRID_SEND_CONTROL,
1420 .opcode = IBV_WR_SEND,
1421 .send_flags = IBV_SEND_SIGNALED,
1426 DDDPRINTF("CONTROL: sending %s..\n", control_desc[head->type]);
1429 * We don't actually need to do a memcpy() in here if we used
1430 * the "sge" properly, but since we're only sending control messages
1431 * (not RAM in a performance-critical path), then its OK for now.
1433 * The copy makes the RDMAControlHeader simpler to manipulate
1434 * for the time being.
1436 assert(head->len <= RDMA_CONTROL_MAX_BUFFER - sizeof(*head));
1437 memcpy(wr->control, head, sizeof(RDMAControlHeader));
1438 control_to_network((void *) wr->control);
1441 memcpy(wr->control + sizeof(RDMAControlHeader), buf, head->len);
1445 if (ibv_post_send(rdma->qp, &send_wr, &bad_wr)) {
1450 fprintf(stderr, "Failed to use post IB SEND for control!\n");
1454 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_SEND_CONTROL, NULL);
1456 fprintf(stderr, "rdma migration: send polling control error!\n");
1463 * Post a RECV work request in anticipation of some future receipt
1464 * of data on the control channel.
1466 static int qemu_rdma_post_recv_control(RDMAContext *rdma, int idx)
1468 struct ibv_recv_wr *bad_wr;
1469 struct ibv_sge sge = {
1470 .addr = (uint64_t)(rdma->wr_data[idx].control),
1471 .length = RDMA_CONTROL_MAX_BUFFER,
1472 .lkey = rdma->wr_data[idx].control_mr->lkey,
1475 struct ibv_recv_wr recv_wr = {
1476 .wr_id = RDMA_WRID_RECV_CONTROL + idx,
1482 if (ibv_post_recv(rdma->qp, &recv_wr, &bad_wr)) {
1490 * Block and wait for a RECV control channel message to arrive.
1492 static int qemu_rdma_exchange_get_response(RDMAContext *rdma,
1493 RDMAControlHeader *head, int expecting, int idx)
1496 int ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RECV_CONTROL + idx,
1500 fprintf(stderr, "rdma migration: recv polling control error!\n");
1504 network_to_control((void *) rdma->wr_data[idx].control);
1505 memcpy(head, rdma->wr_data[idx].control, sizeof(RDMAControlHeader));
1507 DDDPRINTF("CONTROL: %s receiving...\n", control_desc[expecting]);
1509 if (expecting == RDMA_CONTROL_NONE) {
1510 DDDPRINTF("Surprise: got %s (%d)\n",
1511 control_desc[head->type], head->type);
1512 } else if (head->type != expecting || head->type == RDMA_CONTROL_ERROR) {
1513 fprintf(stderr, "Was expecting a %s (%d) control message"
1514 ", but got: %s (%d), length: %d\n",
1515 control_desc[expecting], expecting,
1516 control_desc[head->type], head->type, head->len);
1519 if (head->len > RDMA_CONTROL_MAX_BUFFER - sizeof(*head)) {
1520 fprintf(stderr, "too long length: %d\n", head->len);
1523 if (sizeof(*head) + head->len != byte_len) {
1524 fprintf(stderr, "Malformed length: %d byte_len %d\n",
1525 head->len, byte_len);
1533 * When a RECV work request has completed, the work request's
1534 * buffer is pointed at the header.
1536 * This will advance the pointer to the data portion
1537 * of the control message of the work request's buffer that
1538 * was populated after the work request finished.
1540 static void qemu_rdma_move_header(RDMAContext *rdma, int idx,
1541 RDMAControlHeader *head)
1543 rdma->wr_data[idx].control_len = head->len;
1544 rdma->wr_data[idx].control_curr =
1545 rdma->wr_data[idx].control + sizeof(RDMAControlHeader);
1549 * This is an 'atomic' high-level operation to deliver a single, unified
1550 * control-channel message.
1552 * Additionally, if the user is expecting some kind of reply to this message,
1553 * they can request a 'resp' response message be filled in by posting an
1554 * additional work request on behalf of the user and waiting for an additional
1557 * The extra (optional) response is used during registration to us from having
1558 * to perform an *additional* exchange of message just to provide a response by
1559 * instead piggy-backing on the acknowledgement.
1561 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
1562 uint8_t *data, RDMAControlHeader *resp,
1564 int (*callback)(RDMAContext *rdma))
1569 * Wait until the dest is ready before attempting to deliver the message
1570 * by waiting for a READY message.
1572 if (rdma->control_ready_expected) {
1573 RDMAControlHeader resp;
1574 ret = qemu_rdma_exchange_get_response(rdma,
1575 &resp, RDMA_CONTROL_READY, RDMA_WRID_READY);
1582 * If the user is expecting a response, post a WR in anticipation of it.
1585 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_DATA);
1587 fprintf(stderr, "rdma migration: error posting"
1588 " extra control recv for anticipated result!");
1594 * Post a WR to replace the one we just consumed for the READY message.
1596 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
1598 fprintf(stderr, "rdma migration: error posting first control recv!");
1603 * Deliver the control message that was requested.
1605 ret = qemu_rdma_post_send_control(rdma, data, head);
1608 fprintf(stderr, "Failed to send control buffer!\n");
1613 * If we're expecting a response, block and wait for it.
1617 DDPRINTF("Issuing callback before receiving response...\n");
1618 ret = callback(rdma);
1624 DDPRINTF("Waiting for response %s\n", control_desc[resp->type]);
1625 ret = qemu_rdma_exchange_get_response(rdma, resp,
1626 resp->type, RDMA_WRID_DATA);
1632 qemu_rdma_move_header(rdma, RDMA_WRID_DATA, resp);
1634 *resp_idx = RDMA_WRID_DATA;
1636 DDPRINTF("Response %s received.\n", control_desc[resp->type]);
1639 rdma->control_ready_expected = 1;
1645 * This is an 'atomic' high-level operation to receive a single, unified
1646 * control-channel message.
1648 static int qemu_rdma_exchange_recv(RDMAContext *rdma, RDMAControlHeader *head,
1651 RDMAControlHeader ready = {
1653 .type = RDMA_CONTROL_READY,
1659 * Inform the source that we're ready to receive a message.
1661 ret = qemu_rdma_post_send_control(rdma, NULL, &ready);
1664 fprintf(stderr, "Failed to send control buffer!\n");
1669 * Block and wait for the message.
1671 ret = qemu_rdma_exchange_get_response(rdma, head,
1672 expecting, RDMA_WRID_READY);
1678 qemu_rdma_move_header(rdma, RDMA_WRID_READY, head);
1681 * Post a new RECV work request to replace the one we just consumed.
1683 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
1685 fprintf(stderr, "rdma migration: error posting second control recv!");
1693 * Write an actual chunk of memory using RDMA.
1695 * If we're using dynamic registration on the dest-side, we have to
1696 * send a registration command first.
1698 static int qemu_rdma_write_one(QEMUFile *f, RDMAContext *rdma,
1699 int current_index, uint64_t current_addr,
1703 struct ibv_send_wr send_wr = { 0 };
1704 struct ibv_send_wr *bad_wr;
1705 int reg_result_idx, ret, count = 0;
1706 uint64_t chunk, chunks;
1707 uint8_t *chunk_start, *chunk_end;
1708 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[current_index]);
1710 RDMARegisterResult *reg_result;
1711 RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT };
1712 RDMAControlHeader head = { .len = sizeof(RDMARegister),
1713 .type = RDMA_CONTROL_REGISTER_REQUEST,
1718 sge.addr = (uint64_t)(block->local_host_addr +
1719 (current_addr - block->offset));
1720 sge.length = length;
1722 chunk = ram_chunk_index(block->local_host_addr, (uint8_t *) sge.addr);
1723 chunk_start = ram_chunk_start(block, chunk);
1725 if (block->is_ram_block) {
1726 chunks = length / (1UL << RDMA_REG_CHUNK_SHIFT);
1728 if (chunks && ((length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
1732 chunks = block->length / (1UL << RDMA_REG_CHUNK_SHIFT);
1734 if (chunks && ((block->length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
1739 DDPRINTF("Writing %" PRIu64 " chunks, (%" PRIu64 " MB)\n",
1740 chunks + 1, (chunks + 1) * (1UL << RDMA_REG_CHUNK_SHIFT) / 1024 / 1024);
1742 chunk_end = ram_chunk_end(block, chunk + chunks);
1744 if (!rdma->pin_all) {
1745 #ifdef RDMA_UNREGISTRATION_EXAMPLE
1746 qemu_rdma_unregister_waiting(rdma);
1750 while (test_bit(chunk, block->transit_bitmap)) {
1752 DDPRINTF("(%d) Not clobbering: block: %d chunk %" PRIu64
1753 " current %" PRIu64 " len %" PRIu64 " %d %d\n",
1754 count++, current_index, chunk,
1755 sge.addr, length, rdma->nb_sent, block->nb_chunks);
1757 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
1760 fprintf(stderr, "Failed to Wait for previous write to complete "
1761 "block %d chunk %" PRIu64
1762 " current %" PRIu64 " len %" PRIu64 " %d\n",
1763 current_index, chunk, sge.addr, length, rdma->nb_sent);
1768 if (!rdma->pin_all || !block->is_ram_block) {
1769 if (!block->remote_keys[chunk]) {
1771 * This chunk has not yet been registered, so first check to see
1772 * if the entire chunk is zero. If so, tell the other size to
1773 * memset() + madvise() the entire chunk without RDMA.
1776 if (can_use_buffer_find_nonzero_offset((void *)sge.addr, length)
1777 && buffer_find_nonzero_offset((void *)sge.addr,
1778 length) == length) {
1779 RDMACompress comp = {
1780 .offset = current_addr,
1782 .block_idx = current_index,
1786 head.len = sizeof(comp);
1787 head.type = RDMA_CONTROL_COMPRESS;
1789 DDPRINTF("Entire chunk is zero, sending compress: %"
1791 "bytes, index: %d, offset: %" PRId64 "...\n",
1792 chunk, sge.length, current_index, current_addr);
1794 compress_to_network(&comp);
1795 ret = qemu_rdma_exchange_send(rdma, &head,
1796 (uint8_t *) &comp, NULL, NULL, NULL);
1802 acct_update_position(f, sge.length, true);
1808 * Otherwise, tell other side to register.
1810 reg.current_index = current_index;
1811 if (block->is_ram_block) {
1812 reg.key.current_addr = current_addr;
1814 reg.key.chunk = chunk;
1816 reg.chunks = chunks;
1818 DDPRINTF("Sending registration request chunk %" PRIu64 " for %d "
1819 "bytes, index: %d, offset: %" PRId64 "...\n",
1820 chunk, sge.length, current_index, current_addr);
1822 register_to_network(®);
1823 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) ®,
1824 &resp, ®_result_idx, NULL);
1829 /* try to overlap this single registration with the one we sent. */
1830 if (qemu_rdma_register_and_get_keys(rdma, block,
1831 (uint8_t *) sge.addr,
1832 &sge.lkey, NULL, chunk,
1833 chunk_start, chunk_end)) {
1834 fprintf(stderr, "cannot get lkey!\n");
1838 reg_result = (RDMARegisterResult *)
1839 rdma->wr_data[reg_result_idx].control_curr;
1841 network_to_result(reg_result);
1843 DDPRINTF("Received registration result:"
1844 " my key: %x their key %x, chunk %" PRIu64 "\n",
1845 block->remote_keys[chunk], reg_result->rkey, chunk);
1847 block->remote_keys[chunk] = reg_result->rkey;
1848 block->remote_host_addr = reg_result->host_addr;
1850 /* already registered before */
1851 if (qemu_rdma_register_and_get_keys(rdma, block,
1852 (uint8_t *)sge.addr,
1853 &sge.lkey, NULL, chunk,
1854 chunk_start, chunk_end)) {
1855 fprintf(stderr, "cannot get lkey!\n");
1860 send_wr.wr.rdma.rkey = block->remote_keys[chunk];
1862 send_wr.wr.rdma.rkey = block->remote_rkey;
1864 if (qemu_rdma_register_and_get_keys(rdma, block, (uint8_t *)sge.addr,
1865 &sge.lkey, NULL, chunk,
1866 chunk_start, chunk_end)) {
1867 fprintf(stderr, "cannot get lkey!\n");
1873 * Encode the ram block index and chunk within this wrid.
1874 * We will use this information at the time of completion
1875 * to figure out which bitmap to check against and then which
1876 * chunk in the bitmap to look for.
1878 send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE,
1879 current_index, chunk);
1881 send_wr.opcode = IBV_WR_RDMA_WRITE;
1882 send_wr.send_flags = IBV_SEND_SIGNALED;
1883 send_wr.sg_list = &sge;
1884 send_wr.num_sge = 1;
1885 send_wr.wr.rdma.remote_addr = block->remote_host_addr +
1886 (current_addr - block->offset);
1888 DDDPRINTF("Posting chunk: %" PRIu64 ", addr: %lx"
1889 " remote: %lx, bytes %" PRIu32 "\n",
1890 chunk, sge.addr, send_wr.wr.rdma.remote_addr,
1894 * ibv_post_send() does not return negative error numbers,
1895 * per the specification they are positive - no idea why.
1897 ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
1899 if (ret == ENOMEM) {
1900 DDPRINTF("send queue is full. wait a little....\n");
1901 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
1903 fprintf(stderr, "rdma migration: failed to make "
1904 "room in full send queue! %d\n", ret);
1910 } else if (ret > 0) {
1911 perror("rdma migration: post rdma write failed");
1915 set_bit(chunk, block->transit_bitmap);
1916 acct_update_position(f, sge.length, false);
1917 rdma->total_writes++;
1923 * Push out any unwritten RDMA operations.
1925 * We support sending out multiple chunks at the same time.
1926 * Not all of them need to get signaled in the completion queue.
1928 static int qemu_rdma_write_flush(QEMUFile *f, RDMAContext *rdma)
1932 if (!rdma->current_length) {
1936 ret = qemu_rdma_write_one(f, rdma,
1937 rdma->current_index, rdma->current_addr, rdma->current_length);
1945 DDDPRINTF("sent total: %d\n", rdma->nb_sent);
1948 rdma->current_length = 0;
1949 rdma->current_addr = 0;
1954 static inline int qemu_rdma_buffer_mergable(RDMAContext *rdma,
1955 uint64_t offset, uint64_t len)
1957 RDMALocalBlock *block;
1961 if (rdma->current_index < 0) {
1965 if (rdma->current_chunk < 0) {
1969 block = &(rdma->local_ram_blocks.block[rdma->current_index]);
1970 host_addr = block->local_host_addr + (offset - block->offset);
1971 chunk_end = ram_chunk_end(block, rdma->current_chunk);
1973 if (rdma->current_length == 0) {
1978 * Only merge into chunk sequentially.
1980 if (offset != (rdma->current_addr + rdma->current_length)) {
1984 if (offset < block->offset) {
1988 if ((offset + len) > (block->offset + block->length)) {
1992 if ((host_addr + len) > chunk_end) {
2000 * We're not actually writing here, but doing three things:
2002 * 1. Identify the chunk the buffer belongs to.
2003 * 2. If the chunk is full or the buffer doesn't belong to the current
2004 * chunk, then start a new chunk and flush() the old chunk.
2005 * 3. To keep the hardware busy, we also group chunks into batches
2006 * and only require that a batch gets acknowledged in the completion
2007 * qeueue instead of each individual chunk.
2009 static int qemu_rdma_write(QEMUFile *f, RDMAContext *rdma,
2010 uint64_t block_offset, uint64_t offset,
2013 uint64_t current_addr = block_offset + offset;
2014 uint64_t index = rdma->current_index;
2015 uint64_t chunk = rdma->current_chunk;
2018 /* If we cannot merge it, we flush the current buffer first. */
2019 if (!qemu_rdma_buffer_mergable(rdma, current_addr, len)) {
2020 ret = qemu_rdma_write_flush(f, rdma);
2024 rdma->current_length = 0;
2025 rdma->current_addr = current_addr;
2027 ret = qemu_rdma_search_ram_block(rdma, block_offset,
2028 offset, len, &index, &chunk);
2030 fprintf(stderr, "ram block search failed\n");
2033 rdma->current_index = index;
2034 rdma->current_chunk = chunk;
2038 rdma->current_length += len;
2040 /* flush it if buffer is too large */
2041 if (rdma->current_length >= RDMA_MERGE_MAX) {
2042 return qemu_rdma_write_flush(f, rdma);
2048 static void qemu_rdma_cleanup(RDMAContext *rdma)
2050 struct rdma_cm_event *cm_event;
2054 if (rdma->error_state) {
2055 RDMAControlHeader head = { .len = 0,
2056 .type = RDMA_CONTROL_ERROR,
2059 fprintf(stderr, "Early error. Sending error.\n");
2060 qemu_rdma_post_send_control(rdma, NULL, &head);
2063 ret = rdma_disconnect(rdma->cm_id);
2065 DDPRINTF("waiting for disconnect\n");
2066 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2068 rdma_ack_cm_event(cm_event);
2071 DDPRINTF("Disconnected.\n");
2075 g_free(rdma->block);
2078 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2079 if (rdma->wr_data[idx].control_mr) {
2080 rdma->total_registrations--;
2081 ibv_dereg_mr(rdma->wr_data[idx].control_mr);
2083 rdma->wr_data[idx].control_mr = NULL;
2086 if (rdma->local_ram_blocks.block) {
2087 while (rdma->local_ram_blocks.nb_blocks) {
2088 __qemu_rdma_delete_block(rdma,
2089 rdma->local_ram_blocks.block->offset);
2094 ibv_destroy_qp(rdma->qp);
2098 ibv_destroy_cq(rdma->cq);
2101 if (rdma->comp_channel) {
2102 ibv_destroy_comp_channel(rdma->comp_channel);
2103 rdma->comp_channel = NULL;
2106 ibv_dealloc_pd(rdma->pd);
2109 if (rdma->listen_id) {
2110 rdma_destroy_id(rdma->listen_id);
2111 rdma->listen_id = NULL;
2114 rdma_destroy_id(rdma->cm_id);
2117 if (rdma->channel) {
2118 rdma_destroy_event_channel(rdma->channel);
2119 rdma->channel = NULL;
2126 static int qemu_rdma_source_init(RDMAContext *rdma, Error **errp, bool pin_all)
2129 Error *local_err = NULL, **temp = &local_err;
2132 * Will be validated against destination's actual capabilities
2133 * after the connect() completes.
2135 rdma->pin_all = pin_all;
2137 ret = qemu_rdma_resolve_host(rdma, temp);
2139 goto err_rdma_source_init;
2142 ret = qemu_rdma_alloc_pd_cq(rdma);
2144 ERROR(temp, "rdma migration: error allocating pd and cq! Your mlock()"
2145 " limits may be too low. Please check $ ulimit -a # and "
2146 "search for 'ulimit -l' in the output");
2147 goto err_rdma_source_init;
2150 ret = qemu_rdma_alloc_qp(rdma);
2152 ERROR(temp, "rdma migration: error allocating qp!");
2153 goto err_rdma_source_init;
2156 ret = qemu_rdma_init_ram_blocks(rdma);
2158 ERROR(temp, "rdma migration: error initializing ram blocks!");
2159 goto err_rdma_source_init;
2162 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2163 ret = qemu_rdma_reg_control(rdma, idx);
2165 ERROR(temp, "rdma migration: error registering %d control!",
2167 goto err_rdma_source_init;
2173 err_rdma_source_init:
2174 error_propagate(errp, local_err);
2175 qemu_rdma_cleanup(rdma);
2179 static int qemu_rdma_connect(RDMAContext *rdma, Error **errp)
2181 RDMACapabilities cap = {
2182 .version = RDMA_CONTROL_VERSION_CURRENT,
2185 struct rdma_conn_param conn_param = { .initiator_depth = 2,
2187 .private_data = &cap,
2188 .private_data_len = sizeof(cap),
2190 struct rdma_cm_event *cm_event;
2194 * Only negotiate the capability with destination if the user
2195 * on the source first requested the capability.
2197 if (rdma->pin_all) {
2198 DPRINTF("Server pin-all memory requested.\n");
2199 cap.flags |= RDMA_CAPABILITY_PIN_ALL;
2202 caps_to_network(&cap);
2204 ret = rdma_connect(rdma->cm_id, &conn_param);
2206 perror("rdma_connect");
2207 ERROR(errp, "connecting to destination!");
2208 rdma_destroy_id(rdma->cm_id);
2210 goto err_rdma_source_connect;
2213 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2215 perror("rdma_get_cm_event after rdma_connect");
2216 ERROR(errp, "connecting to destination!");
2217 rdma_ack_cm_event(cm_event);
2218 rdma_destroy_id(rdma->cm_id);
2220 goto err_rdma_source_connect;
2223 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
2224 perror("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect");
2225 ERROR(errp, "connecting to destination!");
2226 rdma_ack_cm_event(cm_event);
2227 rdma_destroy_id(rdma->cm_id);
2229 goto err_rdma_source_connect;
2232 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
2233 network_to_caps(&cap);
2236 * Verify that the *requested* capabilities are supported by the destination
2237 * and disable them otherwise.
2239 if (rdma->pin_all && !(cap.flags & RDMA_CAPABILITY_PIN_ALL)) {
2240 ERROR(errp, "Server cannot support pinning all memory. "
2241 "Will register memory dynamically.");
2242 rdma->pin_all = false;
2245 DPRINTF("Pin all memory: %s\n", rdma->pin_all ? "enabled" : "disabled");
2247 rdma_ack_cm_event(cm_event);
2249 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
2251 ERROR(errp, "posting second control recv!");
2252 goto err_rdma_source_connect;
2255 rdma->control_ready_expected = 1;
2259 err_rdma_source_connect:
2260 qemu_rdma_cleanup(rdma);
2264 static int qemu_rdma_dest_init(RDMAContext *rdma, Error **errp)
2266 int ret = -EINVAL, idx;
2267 struct rdma_cm_id *listen_id;
2268 char ip[40] = "unknown";
2269 struct addrinfo *res;
2272 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2273 rdma->wr_data[idx].control_len = 0;
2274 rdma->wr_data[idx].control_curr = NULL;
2277 if (rdma->host == NULL) {
2278 ERROR(errp, "RDMA host is not set!");
2279 rdma->error_state = -EINVAL;
2282 /* create CM channel */
2283 rdma->channel = rdma_create_event_channel();
2284 if (!rdma->channel) {
2285 ERROR(errp, "could not create rdma event channel");
2286 rdma->error_state = -EINVAL;
2291 ret = rdma_create_id(rdma->channel, &listen_id, NULL, RDMA_PS_TCP);
2293 ERROR(errp, "could not create cm_id!");
2294 goto err_dest_init_create_listen_id;
2297 snprintf(port_str, 16, "%d", rdma->port);
2298 port_str[15] = '\0';
2300 if (rdma->host && strcmp("", rdma->host)) {
2303 ret = getaddrinfo(rdma->host, port_str, NULL, &res);
2305 ERROR(errp, "could not getaddrinfo address %s", rdma->host);
2306 goto err_dest_init_bind_addr;
2309 for (e = res; e != NULL; e = e->ai_next) {
2310 inet_ntop(e->ai_family,
2311 &((struct sockaddr_in *) e->ai_addr)->sin_addr, ip, sizeof ip);
2312 DPRINTF("Trying %s => %s\n", rdma->host, ip);
2313 ret = rdma_bind_addr(listen_id, e->ai_addr);
2319 ERROR(errp, "Error: could not rdma_bind_addr!");
2320 goto err_dest_init_bind_addr;
2322 ERROR(errp, "migration host and port not specified!");
2324 goto err_dest_init_bind_addr;
2328 rdma->listen_id = listen_id;
2329 qemu_rdma_dump_gid("dest_init", listen_id);
2332 err_dest_init_bind_addr:
2333 rdma_destroy_id(listen_id);
2334 err_dest_init_create_listen_id:
2335 rdma_destroy_event_channel(rdma->channel);
2336 rdma->channel = NULL;
2337 rdma->error_state = ret;
2342 static void *qemu_rdma_data_init(const char *host_port, Error **errp)
2344 RDMAContext *rdma = NULL;
2345 InetSocketAddress *addr;
2348 rdma = g_malloc0(sizeof(RDMAContext));
2349 memset(rdma, 0, sizeof(RDMAContext));
2350 rdma->current_index = -1;
2351 rdma->current_chunk = -1;
2353 addr = inet_parse(host_port, NULL);
2355 rdma->port = atoi(addr->port);
2356 rdma->host = g_strdup(addr->host);
2358 ERROR(errp, "bad RDMA migration address '%s'", host_port);
2368 * QEMUFile interface to the control channel.
2369 * SEND messages for control only.
2370 * pc.ram is handled with regular RDMA messages.
2372 static int qemu_rdma_put_buffer(void *opaque, const uint8_t *buf,
2373 int64_t pos, int size)
2375 QEMUFileRDMA *r = opaque;
2376 QEMUFile *f = r->file;
2377 RDMAContext *rdma = r->rdma;
2378 size_t remaining = size;
2379 uint8_t * data = (void *) buf;
2382 CHECK_ERROR_STATE();
2385 * Push out any writes that
2386 * we're queued up for pc.ram.
2388 ret = qemu_rdma_write_flush(f, rdma);
2390 rdma->error_state = ret;
2395 RDMAControlHeader head;
2397 r->len = MIN(remaining, RDMA_SEND_INCREMENT);
2398 remaining -= r->len;
2401 head.type = RDMA_CONTROL_QEMU_FILE;
2403 ret = qemu_rdma_exchange_send(rdma, &head, data, NULL, NULL, NULL);
2406 rdma->error_state = ret;
2416 static size_t qemu_rdma_fill(RDMAContext *rdma, uint8_t *buf,
2421 if (rdma->wr_data[idx].control_len) {
2422 DDDPRINTF("RDMA %" PRId64 " of %d bytes already in buffer\n",
2423 rdma->wr_data[idx].control_len, size);
2425 len = MIN(size, rdma->wr_data[idx].control_len);
2426 memcpy(buf, rdma->wr_data[idx].control_curr, len);
2427 rdma->wr_data[idx].control_curr += len;
2428 rdma->wr_data[idx].control_len -= len;
2435 * QEMUFile interface to the control channel.
2436 * RDMA links don't use bytestreams, so we have to
2437 * return bytes to QEMUFile opportunistically.
2439 static int qemu_rdma_get_buffer(void *opaque, uint8_t *buf,
2440 int64_t pos, int size)
2442 QEMUFileRDMA *r = opaque;
2443 RDMAContext *rdma = r->rdma;
2444 RDMAControlHeader head;
2447 CHECK_ERROR_STATE();
2450 * First, we hold on to the last SEND message we
2451 * were given and dish out the bytes until we run
2454 r->len = qemu_rdma_fill(r->rdma, buf, size, 0);
2460 * Once we run out, we block and wait for another
2461 * SEND message to arrive.
2463 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_QEMU_FILE);
2466 rdma->error_state = ret;
2471 * SEND was received with new bytes, now try again.
2473 return qemu_rdma_fill(r->rdma, buf, size, 0);
2477 * Block until all the outstanding chunks have been delivered by the hardware.
2479 static int qemu_rdma_drain_cq(QEMUFile *f, RDMAContext *rdma)
2483 if (qemu_rdma_write_flush(f, rdma) < 0) {
2487 while (rdma->nb_sent) {
2488 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2490 fprintf(stderr, "rdma migration: complete polling error!\n");
2495 qemu_rdma_unregister_waiting(rdma);
2500 static int qemu_rdma_close(void *opaque)
2502 DPRINTF("Shutting down connection.\n");
2503 QEMUFileRDMA *r = opaque;
2505 qemu_rdma_cleanup(r->rdma);
2515 * This means that 'block_offset' is a full virtual address that does not
2516 * belong to a RAMBlock of the virtual machine and instead
2517 * represents a private malloc'd memory area that the caller wishes to
2521 * Offset is an offset to be added to block_offset and used
2522 * to also lookup the corresponding RAMBlock.
2525 * Initiate an transfer this size.
2528 * A 'hint' or 'advice' that means that we wish to speculatively
2529 * and asynchronously unregister this memory. In this case, there is no
2530 * guarantee that the unregister will actually happen, for example,
2531 * if the memory is being actively transmitted. Additionally, the memory
2532 * may be re-registered at any future time if a write within the same
2533 * chunk was requested again, even if you attempted to unregister it
2536 * @size < 0 : TODO, not yet supported
2537 * Unregister the memory NOW. This means that the caller does not
2538 * expect there to be any future RDMA transfers and we just want to clean
2539 * things up. This is used in case the upper layer owns the memory and
2540 * cannot wait for qemu_fclose() to occur.
2542 * @bytes_sent : User-specificed pointer to indicate how many bytes were
2543 * sent. Usually, this will not be more than a few bytes of
2544 * the protocol because most transfers are sent asynchronously.
2546 static size_t qemu_rdma_save_page(QEMUFile *f, void *opaque,
2547 ram_addr_t block_offset, ram_addr_t offset,
2548 size_t size, int *bytes_sent)
2550 QEMUFileRDMA *rfile = opaque;
2551 RDMAContext *rdma = rfile->rdma;
2554 CHECK_ERROR_STATE();
2560 * Add this page to the current 'chunk'. If the chunk
2561 * is full, or the page doen't belong to the current chunk,
2562 * an actual RDMA write will occur and a new chunk will be formed.
2564 ret = qemu_rdma_write(f, rdma, block_offset, offset, size);
2566 fprintf(stderr, "rdma migration: write error! %d\n", ret);
2571 * We always return 1 bytes because the RDMA
2572 * protocol is completely asynchronous. We do not yet know
2573 * whether an identified chunk is zero or not because we're
2574 * waiting for other pages to potentially be merged with
2575 * the current chunk. So, we have to call qemu_update_position()
2576 * later on when the actual write occurs.
2582 uint64_t index, chunk;
2584 /* TODO: Change QEMUFileOps prototype to be signed: size_t => long
2586 ret = qemu_rdma_drain_cq(f, rdma);
2588 fprintf(stderr, "rdma: failed to synchronously drain"
2589 " completion queue before unregistration.\n");
2595 ret = qemu_rdma_search_ram_block(rdma, block_offset,
2596 offset, size, &index, &chunk);
2599 fprintf(stderr, "ram block search failed\n");
2603 qemu_rdma_signal_unregister(rdma, index, chunk, 0);
2606 * TODO: Synchronous, guaranteed unregistration (should not occur during
2607 * fast-path). Otherwise, unregisters will process on the next call to
2608 * qemu_rdma_drain_cq()
2610 qemu_rdma_unregister_waiting(rdma);
2616 * Drain the Completion Queue if possible, but do not block,
2619 * If nothing to poll, the end of the iteration will do this
2620 * again to make sure we don't overflow the request queue.
2623 uint64_t wr_id, wr_id_in;
2624 int ret = qemu_rdma_poll(rdma, &wr_id_in, NULL);
2626 fprintf(stderr, "rdma migration: polling error! %d\n", ret);
2630 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
2632 if (wr_id == RDMA_WRID_NONE) {
2637 return RAM_SAVE_CONTROL_DELAYED;
2639 rdma->error_state = ret;
2643 static int qemu_rdma_accept(RDMAContext *rdma)
2645 RDMACapabilities cap;
2646 struct rdma_conn_param conn_param = {
2647 .responder_resources = 2,
2648 .private_data = &cap,
2649 .private_data_len = sizeof(cap),
2651 struct rdma_cm_event *cm_event;
2652 struct ibv_context *verbs;
2656 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2658 goto err_rdma_dest_wait;
2661 if (cm_event->event != RDMA_CM_EVENT_CONNECT_REQUEST) {
2662 rdma_ack_cm_event(cm_event);
2663 goto err_rdma_dest_wait;
2666 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
2668 network_to_caps(&cap);
2670 if (cap.version < 1 || cap.version > RDMA_CONTROL_VERSION_CURRENT) {
2671 fprintf(stderr, "Unknown source RDMA version: %d, bailing...\n",
2673 rdma_ack_cm_event(cm_event);
2674 goto err_rdma_dest_wait;
2678 * Respond with only the capabilities this version of QEMU knows about.
2680 cap.flags &= known_capabilities;
2683 * Enable the ones that we do know about.
2684 * Add other checks here as new ones are introduced.
2686 if (cap.flags & RDMA_CAPABILITY_PIN_ALL) {
2687 rdma->pin_all = true;
2690 rdma->cm_id = cm_event->id;
2691 verbs = cm_event->id->verbs;
2693 rdma_ack_cm_event(cm_event);
2695 DPRINTF("Memory pin all: %s\n", rdma->pin_all ? "enabled" : "disabled");
2697 caps_to_network(&cap);
2699 DPRINTF("verbs context after listen: %p\n", verbs);
2702 rdma->verbs = verbs;
2703 } else if (rdma->verbs != verbs) {
2704 fprintf(stderr, "ibv context not matching %p, %p!\n",
2705 rdma->verbs, verbs);
2706 goto err_rdma_dest_wait;
2709 qemu_rdma_dump_id("dest_init", verbs);
2711 ret = qemu_rdma_alloc_pd_cq(rdma);
2713 fprintf(stderr, "rdma migration: error allocating pd and cq!\n");
2714 goto err_rdma_dest_wait;
2717 ret = qemu_rdma_alloc_qp(rdma);
2719 fprintf(stderr, "rdma migration: error allocating qp!\n");
2720 goto err_rdma_dest_wait;
2723 ret = qemu_rdma_init_ram_blocks(rdma);
2725 fprintf(stderr, "rdma migration: error initializing ram blocks!\n");
2726 goto err_rdma_dest_wait;
2729 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2730 ret = qemu_rdma_reg_control(rdma, idx);
2732 fprintf(stderr, "rdma: error registering %d control!\n", idx);
2733 goto err_rdma_dest_wait;
2737 qemu_set_fd_handler2(rdma->channel->fd, NULL, NULL, NULL, NULL);
2739 ret = rdma_accept(rdma->cm_id, &conn_param);
2741 fprintf(stderr, "rdma_accept returns %d!\n", ret);
2742 goto err_rdma_dest_wait;
2745 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2747 fprintf(stderr, "rdma_accept get_cm_event failed %d!\n", ret);
2748 goto err_rdma_dest_wait;
2751 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
2752 fprintf(stderr, "rdma_accept not event established!\n");
2753 rdma_ack_cm_event(cm_event);
2754 goto err_rdma_dest_wait;
2757 rdma_ack_cm_event(cm_event);
2759 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
2761 fprintf(stderr, "rdma migration: error posting second control recv!\n");
2762 goto err_rdma_dest_wait;
2765 qemu_rdma_dump_gid("dest_connect", rdma->cm_id);
2770 rdma->error_state = ret;
2771 qemu_rdma_cleanup(rdma);
2776 * During each iteration of the migration, we listen for instructions
2777 * by the source VM to perform dynamic page registrations before they
2778 * can perform RDMA operations.
2780 * We respond with the 'rkey'.
2782 * Keep doing this until the source tells us to stop.
2784 static int qemu_rdma_registration_handle(QEMUFile *f, void *opaque,
2787 RDMAControlHeader reg_resp = { .len = sizeof(RDMARegisterResult),
2788 .type = RDMA_CONTROL_REGISTER_RESULT,
2791 RDMAControlHeader unreg_resp = { .len = 0,
2792 .type = RDMA_CONTROL_UNREGISTER_FINISHED,
2795 RDMAControlHeader blocks = { .type = RDMA_CONTROL_RAM_BLOCKS_RESULT,
2797 QEMUFileRDMA *rfile = opaque;
2798 RDMAContext *rdma = rfile->rdma;
2799 RDMALocalBlocks *local = &rdma->local_ram_blocks;
2800 RDMAControlHeader head;
2801 RDMARegister *reg, *registers;
2803 RDMARegisterResult *reg_result;
2804 static RDMARegisterResult results[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE];
2805 RDMALocalBlock *block;
2812 CHECK_ERROR_STATE();
2815 DDDPRINTF("Waiting for next request %" PRIu64 "...\n", flags);
2817 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_NONE);
2823 if (head.repeat > RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE) {
2824 fprintf(stderr, "rdma: Too many requests in this message (%d)."
2825 "Bailing.\n", head.repeat);
2830 switch (head.type) {
2831 case RDMA_CONTROL_COMPRESS:
2832 comp = (RDMACompress *) rdma->wr_data[idx].control_curr;
2833 network_to_compress(comp);
2835 DDPRINTF("Zapping zero chunk: %" PRId64
2836 " bytes, index %d, offset %" PRId64 "\n",
2837 comp->length, comp->block_idx, comp->offset);
2838 block = &(rdma->local_ram_blocks.block[comp->block_idx]);
2840 host_addr = block->local_host_addr +
2841 (comp->offset - block->offset);
2843 ram_handle_compressed(host_addr, comp->value, comp->length);
2846 case RDMA_CONTROL_REGISTER_FINISHED:
2847 DDDPRINTF("Current registrations complete.\n");
2850 case RDMA_CONTROL_RAM_BLOCKS_REQUEST:
2851 DPRINTF("Initial setup info requested.\n");
2853 if (rdma->pin_all) {
2854 ret = qemu_rdma_reg_whole_ram_blocks(rdma);
2856 fprintf(stderr, "rdma migration: error dest "
2857 "registering ram blocks!\n");
2863 * Dest uses this to prepare to transmit the RAMBlock descriptions
2864 * to the source VM after connection setup.
2865 * Both sides use the "remote" structure to communicate and update
2866 * their "local" descriptions with what was sent.
2868 for (i = 0; i < local->nb_blocks; i++) {
2869 rdma->block[i].remote_host_addr =
2870 (uint64_t)(local->block[i].local_host_addr);
2872 if (rdma->pin_all) {
2873 rdma->block[i].remote_rkey = local->block[i].mr->rkey;
2876 rdma->block[i].offset = local->block[i].offset;
2877 rdma->block[i].length = local->block[i].length;
2879 remote_block_to_network(&rdma->block[i]);
2882 blocks.len = rdma->local_ram_blocks.nb_blocks
2883 * sizeof(RDMARemoteBlock);
2886 ret = qemu_rdma_post_send_control(rdma,
2887 (uint8_t *) rdma->block, &blocks);
2890 fprintf(stderr, "rdma migration: error sending remote info!\n");
2895 case RDMA_CONTROL_REGISTER_REQUEST:
2896 DDPRINTF("There are %d registration requests\n", head.repeat);
2898 reg_resp.repeat = head.repeat;
2899 registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
2901 for (count = 0; count < head.repeat; count++) {
2903 uint8_t *chunk_start, *chunk_end;
2905 reg = ®isters[count];
2906 network_to_register(reg);
2908 reg_result = &results[count];
2910 DDPRINTF("Registration request (%d): index %d, current_addr %"
2911 PRIu64 " chunks: %" PRIu64 "\n", count,
2912 reg->current_index, reg->key.current_addr, reg->chunks);
2914 block = &(rdma->local_ram_blocks.block[reg->current_index]);
2915 if (block->is_ram_block) {
2916 host_addr = (block->local_host_addr +
2917 (reg->key.current_addr - block->offset));
2918 chunk = ram_chunk_index(block->local_host_addr,
2919 (uint8_t *) host_addr);
2921 chunk = reg->key.chunk;
2922 host_addr = block->local_host_addr +
2923 (reg->key.chunk * (1UL << RDMA_REG_CHUNK_SHIFT));
2925 chunk_start = ram_chunk_start(block, chunk);
2926 chunk_end = ram_chunk_end(block, chunk + reg->chunks);
2927 if (qemu_rdma_register_and_get_keys(rdma, block,
2928 (uint8_t *)host_addr, NULL, ®_result->rkey,
2929 chunk, chunk_start, chunk_end)) {
2930 fprintf(stderr, "cannot get rkey!\n");
2935 reg_result->host_addr = (uint64_t) block->local_host_addr;
2937 DDPRINTF("Registered rkey for this request: %x\n",
2940 result_to_network(reg_result);
2943 ret = qemu_rdma_post_send_control(rdma,
2944 (uint8_t *) results, ®_resp);
2947 fprintf(stderr, "Failed to send control buffer!\n");
2951 case RDMA_CONTROL_UNREGISTER_REQUEST:
2952 DDPRINTF("There are %d unregistration requests\n", head.repeat);
2953 unreg_resp.repeat = head.repeat;
2954 registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
2956 for (count = 0; count < head.repeat; count++) {
2957 reg = ®isters[count];
2958 network_to_register(reg);
2960 DDPRINTF("Unregistration request (%d): "
2961 " index %d, chunk %" PRIu64 "\n",
2962 count, reg->current_index, reg->key.chunk);
2964 block = &(rdma->local_ram_blocks.block[reg->current_index]);
2966 ret = ibv_dereg_mr(block->pmr[reg->key.chunk]);
2967 block->pmr[reg->key.chunk] = NULL;
2970 perror("rdma unregistration chunk failed");
2975 rdma->total_registrations--;
2977 DDPRINTF("Unregistered chunk %" PRIu64 " successfully.\n",
2981 ret = qemu_rdma_post_send_control(rdma, NULL, &unreg_resp);
2984 fprintf(stderr, "Failed to send control buffer!\n");
2988 case RDMA_CONTROL_REGISTER_RESULT:
2989 fprintf(stderr, "Invalid RESULT message at dest.\n");
2993 fprintf(stderr, "Unknown control message %s\n",
2994 control_desc[head.type]);
3001 rdma->error_state = ret;
3006 static int qemu_rdma_registration_start(QEMUFile *f, void *opaque,
3009 QEMUFileRDMA *rfile = opaque;
3010 RDMAContext *rdma = rfile->rdma;
3012 CHECK_ERROR_STATE();
3014 DDDPRINTF("start section: %" PRIu64 "\n", flags);
3015 qemu_put_be64(f, RAM_SAVE_FLAG_HOOK);
3022 * Inform dest that dynamic registrations are done for now.
3023 * First, flush writes, if any.
3025 static int qemu_rdma_registration_stop(QEMUFile *f, void *opaque,
3028 Error *local_err = NULL, **errp = &local_err;
3029 QEMUFileRDMA *rfile = opaque;
3030 RDMAContext *rdma = rfile->rdma;
3031 RDMAControlHeader head = { .len = 0, .repeat = 1 };
3034 CHECK_ERROR_STATE();
3037 ret = qemu_rdma_drain_cq(f, rdma);
3043 if (flags == RAM_CONTROL_SETUP) {
3044 RDMAControlHeader resp = {.type = RDMA_CONTROL_RAM_BLOCKS_RESULT };
3045 RDMALocalBlocks *local = &rdma->local_ram_blocks;
3046 int reg_result_idx, i, j, nb_remote_blocks;
3048 head.type = RDMA_CONTROL_RAM_BLOCKS_REQUEST;
3049 DPRINTF("Sending registration setup for ram blocks...\n");
3052 * Make sure that we parallelize the pinning on both sides.
3053 * For very large guests, doing this serially takes a really
3054 * long time, so we have to 'interleave' the pinning locally
3055 * with the control messages by performing the pinning on this
3056 * side before we receive the control response from the other
3057 * side that the pinning has completed.
3059 ret = qemu_rdma_exchange_send(rdma, &head, NULL, &resp,
3060 ®_result_idx, rdma->pin_all ?
3061 qemu_rdma_reg_whole_ram_blocks : NULL);
3063 ERROR(errp, "receiving remote info!");
3067 nb_remote_blocks = resp.len / sizeof(RDMARemoteBlock);
3070 * The protocol uses two different sets of rkeys (mutually exclusive):
3071 * 1. One key to represent the virtual address of the entire ram block.
3072 * (dynamic chunk registration disabled - pin everything with one rkey.)
3073 * 2. One to represent individual chunks within a ram block.
3074 * (dynamic chunk registration enabled - pin individual chunks.)
3076 * Once the capability is successfully negotiated, the destination transmits
3077 * the keys to use (or sends them later) including the virtual addresses
3078 * and then propagates the remote ram block descriptions to his local copy.
3081 if (local->nb_blocks != nb_remote_blocks) {
3082 ERROR(errp, "ram blocks mismatch #1! "
3083 "Your QEMU command line parameters are probably "
3084 "not identical on both the source and destination.");
3088 qemu_rdma_move_header(rdma, reg_result_idx, &resp);
3090 rdma->wr_data[reg_result_idx].control_curr, resp.len);
3091 for (i = 0; i < nb_remote_blocks; i++) {
3092 network_to_remote_block(&rdma->block[i]);
3094 /* search local ram blocks */
3095 for (j = 0; j < local->nb_blocks; j++) {
3096 if (rdma->block[i].offset != local->block[j].offset) {
3100 if (rdma->block[i].length != local->block[j].length) {
3101 ERROR(errp, "ram blocks mismatch #2! "
3102 "Your QEMU command line parameters are probably "
3103 "not identical on both the source and destination.");
3106 local->block[j].remote_host_addr =
3107 rdma->block[i].remote_host_addr;
3108 local->block[j].remote_rkey = rdma->block[i].remote_rkey;
3112 if (j >= local->nb_blocks) {
3113 ERROR(errp, "ram blocks mismatch #3! "
3114 "Your QEMU command line parameters are probably "
3115 "not identical on both the source and destination.");
3121 DDDPRINTF("Sending registration finish %" PRIu64 "...\n", flags);
3123 head.type = RDMA_CONTROL_REGISTER_FINISHED;
3124 ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL);
3132 rdma->error_state = ret;
3136 static int qemu_rdma_get_fd(void *opaque)
3138 QEMUFileRDMA *rfile = opaque;
3139 RDMAContext *rdma = rfile->rdma;
3141 return rdma->comp_channel->fd;
3144 const QEMUFileOps rdma_read_ops = {
3145 .get_buffer = qemu_rdma_get_buffer,
3146 .get_fd = qemu_rdma_get_fd,
3147 .close = qemu_rdma_close,
3148 .hook_ram_load = qemu_rdma_registration_handle,
3151 const QEMUFileOps rdma_write_ops = {
3152 .put_buffer = qemu_rdma_put_buffer,
3153 .close = qemu_rdma_close,
3154 .before_ram_iterate = qemu_rdma_registration_start,
3155 .after_ram_iterate = qemu_rdma_registration_stop,
3156 .save_page = qemu_rdma_save_page,
3159 static void *qemu_fopen_rdma(RDMAContext *rdma, const char *mode)
3161 QEMUFileRDMA *r = g_malloc0(sizeof(QEMUFileRDMA));
3163 if (qemu_file_mode_is_not_valid(mode)) {
3169 if (mode[0] == 'w') {
3170 r->file = qemu_fopen_ops(r, &rdma_write_ops);
3172 r->file = qemu_fopen_ops(r, &rdma_read_ops);
3178 static void rdma_accept_incoming_migration(void *opaque)
3180 RDMAContext *rdma = opaque;
3183 Error *local_err = NULL, **errp = &local_err;
3185 DPRINTF("Accepting rdma connection...\n");
3186 ret = qemu_rdma_accept(rdma);
3189 ERROR(errp, "RDMA Migration initialization failed!");
3193 DPRINTF("Accepted migration\n");
3195 f = qemu_fopen_rdma(rdma, "rb");
3197 ERROR(errp, "could not qemu_fopen_rdma!");
3198 qemu_rdma_cleanup(rdma);
3202 rdma->migration_started_on_destination = 1;
3203 process_incoming_migration(f);
3206 void rdma_start_incoming_migration(const char *host_port, Error **errp)
3210 Error *local_err = NULL;
3212 DPRINTF("Starting RDMA-based incoming migration\n");
3213 rdma = qemu_rdma_data_init(host_port, &local_err);
3219 ret = qemu_rdma_dest_init(rdma, &local_err);
3225 DPRINTF("qemu_rdma_dest_init success\n");
3227 ret = rdma_listen(rdma->listen_id, 5);
3230 ERROR(errp, "listening on socket!");
3234 DPRINTF("rdma_listen success\n");
3236 qemu_set_fd_handler2(rdma->channel->fd, NULL,
3237 rdma_accept_incoming_migration, NULL,
3238 (void *)(intptr_t) rdma);
3241 error_propagate(errp, local_err);
3245 void rdma_start_outgoing_migration(void *opaque,
3246 const char *host_port, Error **errp)
3248 MigrationState *s = opaque;
3249 Error *local_err = NULL, **temp = &local_err;
3250 RDMAContext *rdma = qemu_rdma_data_init(host_port, &local_err);
3254 ERROR(temp, "Failed to initialize RDMA data structures! %d", ret);
3258 ret = qemu_rdma_source_init(rdma, &local_err,
3259 s->enabled_capabilities[MIGRATION_CAPABILITY_X_RDMA_PIN_ALL]);
3265 DPRINTF("qemu_rdma_source_init success\n");
3266 ret = qemu_rdma_connect(rdma, &local_err);
3272 DPRINTF("qemu_rdma_source_connect success\n");
3274 s->file = qemu_fopen_rdma(rdma, "wb");
3275 migrate_fd_connect(s);
3278 error_propagate(errp, local_err);
3280 migrate_fd_error(s);