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;
361 struct ibv_context *verbs;
362 struct rdma_event_channel *channel;
363 struct ibv_qp *qp; /* queue pair */
364 struct ibv_comp_channel *comp_channel; /* completion channel */
365 struct ibv_pd *pd; /* protection domain */
366 struct ibv_cq *cq; /* completion queue */
369 * If a previous write failed (perhaps because of a failed
370 * memory registration, then do not attempt any future work
371 * and remember the error state.
377 * Description of ram blocks used throughout the code.
379 RDMALocalBlocks local_ram_blocks;
380 RDMARemoteBlock *block;
383 * Migration on *destination* started.
384 * Then use coroutine yield function.
385 * Source runs in a thread, so we don't care.
387 int migration_started_on_destination;
389 int total_registrations;
392 int unregister_current, unregister_next;
393 uint64_t unregistrations[RDMA_SIGNALED_SEND_MAX];
395 GHashTable *blockmap;
399 * Interface to the rest of the migration call stack.
401 typedef struct QEMUFileRDMA {
408 * Main structure for IB Send/Recv control messages.
409 * This gets prepended at the beginning of every Send/Recv.
411 typedef struct QEMU_PACKED {
412 uint32_t len; /* Total length of data portion */
413 uint32_t type; /* which control command to perform */
414 uint32_t repeat; /* number of commands in data portion of same type */
418 static void control_to_network(RDMAControlHeader *control)
420 control->type = htonl(control->type);
421 control->len = htonl(control->len);
422 control->repeat = htonl(control->repeat);
425 static void network_to_control(RDMAControlHeader *control)
427 control->type = ntohl(control->type);
428 control->len = ntohl(control->len);
429 control->repeat = ntohl(control->repeat);
433 * Register a single Chunk.
434 * Information sent by the source VM to inform the dest
435 * to register an single chunk of memory before we can perform
436 * the actual RDMA operation.
438 typedef struct QEMU_PACKED {
440 uint64_t current_addr; /* offset into the ramblock of the chunk */
441 uint64_t chunk; /* chunk to lookup if unregistering */
443 uint32_t current_index; /* which ramblock the chunk belongs to */
445 uint64_t chunks; /* how many sequential chunks to register */
448 static void register_to_network(RDMARegister *reg)
450 reg->key.current_addr = htonll(reg->key.current_addr);
451 reg->current_index = htonl(reg->current_index);
452 reg->chunks = htonll(reg->chunks);
455 static void network_to_register(RDMARegister *reg)
457 reg->key.current_addr = ntohll(reg->key.current_addr);
458 reg->current_index = ntohl(reg->current_index);
459 reg->chunks = ntohll(reg->chunks);
462 typedef struct QEMU_PACKED {
463 uint32_t value; /* if zero, we will madvise() */
464 uint32_t block_idx; /* which ram block index */
465 uint64_t offset; /* where in the remote ramblock this chunk */
466 uint64_t length; /* length of the chunk */
469 static void compress_to_network(RDMACompress *comp)
471 comp->value = htonl(comp->value);
472 comp->block_idx = htonl(comp->block_idx);
473 comp->offset = htonll(comp->offset);
474 comp->length = htonll(comp->length);
477 static void network_to_compress(RDMACompress *comp)
479 comp->value = ntohl(comp->value);
480 comp->block_idx = ntohl(comp->block_idx);
481 comp->offset = ntohll(comp->offset);
482 comp->length = ntohll(comp->length);
486 * The result of the dest's memory registration produces an "rkey"
487 * which the source VM must reference in order to perform
488 * the RDMA operation.
490 typedef struct QEMU_PACKED {
494 } RDMARegisterResult;
496 static void result_to_network(RDMARegisterResult *result)
498 result->rkey = htonl(result->rkey);
499 result->host_addr = htonll(result->host_addr);
502 static void network_to_result(RDMARegisterResult *result)
504 result->rkey = ntohl(result->rkey);
505 result->host_addr = ntohll(result->host_addr);
508 const char *print_wrid(int wrid);
509 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
510 uint8_t *data, RDMAControlHeader *resp,
512 int (*callback)(RDMAContext *rdma));
514 static inline uint64_t ram_chunk_index(uint8_t *start, uint8_t *host)
516 return ((uintptr_t) host - (uintptr_t) start) >> RDMA_REG_CHUNK_SHIFT;
519 static inline uint8_t *ram_chunk_start(RDMALocalBlock *rdma_ram_block,
522 return (uint8_t *) (((uintptr_t) rdma_ram_block->local_host_addr)
523 + (i << RDMA_REG_CHUNK_SHIFT));
526 static inline uint8_t *ram_chunk_end(RDMALocalBlock *rdma_ram_block, uint64_t i)
528 uint8_t *result = ram_chunk_start(rdma_ram_block, i) +
529 (1UL << RDMA_REG_CHUNK_SHIFT);
531 if (result > (rdma_ram_block->local_host_addr + rdma_ram_block->length)) {
532 result = rdma_ram_block->local_host_addr + rdma_ram_block->length;
538 static int __qemu_rdma_add_block(RDMAContext *rdma, void *host_addr,
539 ram_addr_t block_offset, uint64_t length)
541 RDMALocalBlocks *local = &rdma->local_ram_blocks;
542 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
543 (void *) block_offset);
544 RDMALocalBlock *old = local->block;
546 assert(block == NULL);
548 local->block = g_malloc0(sizeof(RDMALocalBlock) * (local->nb_blocks + 1));
550 if (local->nb_blocks) {
553 for (x = 0; x < local->nb_blocks; x++) {
554 g_hash_table_remove(rdma->blockmap, (void *)old[x].offset);
555 g_hash_table_insert(rdma->blockmap, (void *)old[x].offset,
558 memcpy(local->block, old, sizeof(RDMALocalBlock) * local->nb_blocks);
562 block = &local->block[local->nb_blocks];
564 block->local_host_addr = host_addr;
565 block->offset = block_offset;
566 block->length = length;
567 block->index = local->nb_blocks;
568 block->nb_chunks = ram_chunk_index(host_addr, host_addr + length) + 1UL;
569 block->transit_bitmap = bitmap_new(block->nb_chunks);
570 bitmap_clear(block->transit_bitmap, 0, block->nb_chunks);
571 block->unregister_bitmap = bitmap_new(block->nb_chunks);
572 bitmap_clear(block->unregister_bitmap, 0, block->nb_chunks);
573 block->remote_keys = g_malloc0(block->nb_chunks * sizeof(uint32_t));
575 block->is_ram_block = local->init ? false : true;
577 g_hash_table_insert(rdma->blockmap, (void *) block_offset, block);
579 DDPRINTF("Added Block: %d, addr: %" PRIu64 ", offset: %" PRIu64
580 " length: %" PRIu64 " end: %" PRIu64 " bits %" PRIu64 " chunks %d\n",
581 local->nb_blocks, (uint64_t) block->local_host_addr, block->offset,
582 block->length, (uint64_t) (block->local_host_addr + block->length),
583 BITS_TO_LONGS(block->nb_chunks) *
584 sizeof(unsigned long) * 8, block->nb_chunks);
592 * Memory regions need to be registered with the device and queue pairs setup
593 * in advanced before the migration starts. This tells us where the RAM blocks
594 * are so that we can register them individually.
596 static void qemu_rdma_init_one_block(void *host_addr,
597 ram_addr_t block_offset, ram_addr_t length, void *opaque)
599 __qemu_rdma_add_block(opaque, host_addr, block_offset, length);
603 * Identify the RAMBlocks and their quantity. They will be references to
604 * identify chunk boundaries inside each RAMBlock and also be referenced
605 * during dynamic page registration.
607 static int qemu_rdma_init_ram_blocks(RDMAContext *rdma)
609 RDMALocalBlocks *local = &rdma->local_ram_blocks;
611 assert(rdma->blockmap == NULL);
612 rdma->blockmap = g_hash_table_new(g_direct_hash, g_direct_equal);
613 memset(local, 0, sizeof *local);
614 qemu_ram_foreach_block(qemu_rdma_init_one_block, rdma);
615 DPRINTF("Allocated %d local ram block structures\n", local->nb_blocks);
616 rdma->block = (RDMARemoteBlock *) g_malloc0(sizeof(RDMARemoteBlock) *
617 rdma->local_ram_blocks.nb_blocks);
622 static int __qemu_rdma_delete_block(RDMAContext *rdma, ram_addr_t block_offset)
624 RDMALocalBlocks *local = &rdma->local_ram_blocks;
625 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
626 (void *) block_offset);
627 RDMALocalBlock *old = local->block;
635 for (j = 0; j < block->nb_chunks; j++) {
636 if (!block->pmr[j]) {
639 ibv_dereg_mr(block->pmr[j]);
640 rdma->total_registrations--;
647 ibv_dereg_mr(block->mr);
648 rdma->total_registrations--;
652 g_free(block->transit_bitmap);
653 block->transit_bitmap = NULL;
655 g_free(block->unregister_bitmap);
656 block->unregister_bitmap = NULL;
658 g_free(block->remote_keys);
659 block->remote_keys = NULL;
661 for (x = 0; x < local->nb_blocks; x++) {
662 g_hash_table_remove(rdma->blockmap, (void *)old[x].offset);
665 if (local->nb_blocks > 1) {
667 local->block = g_malloc0(sizeof(RDMALocalBlock) *
668 (local->nb_blocks - 1));
671 memcpy(local->block, old, sizeof(RDMALocalBlock) * block->index);
674 if (block->index < (local->nb_blocks - 1)) {
675 memcpy(local->block + block->index, old + (block->index + 1),
676 sizeof(RDMALocalBlock) *
677 (local->nb_blocks - (block->index + 1)));
680 assert(block == local->block);
684 DDPRINTF("Deleted Block: %d, addr: %" PRIu64 ", offset: %" PRIu64
685 " length: %" PRIu64 " end: %" PRIu64 " bits %" PRIu64 " chunks %d\n",
686 local->nb_blocks, (uint64_t) block->local_host_addr, block->offset,
687 block->length, (uint64_t) (block->local_host_addr + block->length),
688 BITS_TO_LONGS(block->nb_chunks) *
689 sizeof(unsigned long) * 8, block->nb_chunks);
695 if (local->nb_blocks) {
696 for (x = 0; x < local->nb_blocks; x++) {
697 g_hash_table_insert(rdma->blockmap, (void *)local->block[x].offset,
706 * Put in the log file which RDMA device was opened and the details
707 * associated with that device.
709 static void qemu_rdma_dump_id(const char *who, struct ibv_context *verbs)
711 struct ibv_port_attr port;
713 if (ibv_query_port(verbs, 1, &port)) {
714 fprintf(stderr, "FAILED TO QUERY PORT INFORMATION!\n");
718 printf("%s RDMA Device opened: kernel name %s "
719 "uverbs device name %s, "
720 "infiniband_verbs class device path %s, "
721 "infiniband class device path %s, "
722 "transport: (%d) %s\n",
725 verbs->device->dev_name,
726 verbs->device->dev_path,
727 verbs->device->ibdev_path,
729 (port.link_layer == IBV_LINK_LAYER_INFINIBAND) ? "Infiniband" :
730 ((port.link_layer == IBV_LINK_LAYER_ETHERNET)
731 ? "Ethernet" : "Unknown"));
735 * Put in the log file the RDMA gid addressing information,
736 * useful for folks who have trouble understanding the
737 * RDMA device hierarchy in the kernel.
739 static void qemu_rdma_dump_gid(const char *who, struct rdma_cm_id *id)
743 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.sgid, sgid, sizeof sgid);
744 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.dgid, dgid, sizeof dgid);
745 DPRINTF("%s Source GID: %s, Dest GID: %s\n", who, sgid, dgid);
749 * As of now, IPv6 over RoCE / iWARP is not supported by linux.
750 * We will try the next addrinfo struct, and fail if there are
751 * no other valid addresses to bind against.
753 * If user is listening on '[::]', then we will not have a opened a device
754 * yet and have no way of verifying if the device is RoCE or not.
756 * In this case, the source VM will throw an error for ALL types of
757 * connections (both IPv4 and IPv6) if the destination machine does not have
758 * a regular infiniband network available for use.
760 * The only way to guarantee that an error is thrown for broken kernels is
761 * for the management software to choose a *specific* interface at bind time
762 * and validate what time of hardware it is.
764 * Unfortunately, this puts the user in a fix:
766 * If the source VM connects with an IPv4 address without knowing that the
767 * destination has bound to '[::]' the migration will unconditionally fail
768 * unless the management software is explicitly listening on the the IPv4
769 * address while using a RoCE-based device.
771 * If the source VM connects with an IPv6 address, then we're OK because we can
772 * throw an error on the source (and similarly on the destination).
774 * But in mixed environments, this will be broken for a while until it is fixed
777 * We do provide a *tiny* bit of help in this function: We can list all of the
778 * devices in the system and check to see if all the devices are RoCE or
781 * If we detect that we have a *pure* RoCE environment, then we can safely
782 * thrown an error even if the management software has specified '[::]' as the
785 * However, if there is are multiple hetergeneous devices, then we cannot make
786 * this assumption and the user just has to be sure they know what they are
789 * Patches are being reviewed on linux-rdma.
791 static int qemu_rdma_broken_ipv6_kernel(Error **errp, struct ibv_context *verbs)
793 struct ibv_port_attr port_attr;
795 /* This bug only exists in linux, to our knowledge. */
799 * Verbs are only NULL if management has bound to '[::]'.
801 * Let's iterate through all the devices and see if there any pure IB
802 * devices (non-ethernet).
804 * If not, then we can safely proceed with the migration.
805 * Otherwise, there are no guarantees until the bug is fixed in linux.
809 struct ibv_device ** dev_list = ibv_get_device_list(&num_devices);
810 bool roce_found = false;
811 bool ib_found = false;
813 for (x = 0; x < num_devices; x++) {
814 verbs = ibv_open_device(dev_list[x]);
816 if (ibv_query_port(verbs, 1, &port_attr)) {
817 ibv_close_device(verbs);
818 ERROR(errp, "Could not query initial IB port");
822 if (port_attr.link_layer == IBV_LINK_LAYER_INFINIBAND) {
824 } else if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
828 ibv_close_device(verbs);
834 fprintf(stderr, "WARN: migrations may fail:"
835 " IPv6 over RoCE / iWARP in linux"
836 " is broken. But since you appear to have a"
837 " mixed RoCE / IB environment, be sure to only"
838 " migrate over the IB fabric until the kernel "
839 " fixes the bug.\n");
841 ERROR(errp, "You only have RoCE / iWARP devices in your systems"
842 " and your management software has specified '[::]'"
843 ", but IPv6 over RoCE / iWARP is not supported in Linux.");
852 * If we have a verbs context, that means that some other than '[::]' was
853 * used by the management software for binding. In which case we can actually
854 * warn the user about a potential broken kernel;
857 /* IB ports start with 1, not 0 */
858 if (ibv_query_port(verbs, 1, &port_attr)) {
859 ERROR(errp, "Could not query initial IB port");
863 if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
864 ERROR(errp, "Linux kernel's RoCE / iWARP does not support IPv6 "
865 "(but patches on linux-rdma in progress)");
875 * Figure out which RDMA device corresponds to the requested IP hostname
876 * Also create the initial connection manager identifiers for opening
879 static int qemu_rdma_resolve_host(RDMAContext *rdma, Error **errp)
882 struct rdma_addrinfo *res;
884 struct rdma_cm_event *cm_event;
885 char ip[40] = "unknown";
886 struct rdma_addrinfo *e;
888 if (rdma->host == NULL || !strcmp(rdma->host, "")) {
889 ERROR(errp, "RDMA hostname has not been set");
893 /* create CM channel */
894 rdma->channel = rdma_create_event_channel();
895 if (!rdma->channel) {
896 ERROR(errp, "could not create CM channel");
901 ret = rdma_create_id(rdma->channel, &rdma->cm_id, NULL, RDMA_PS_TCP);
903 ERROR(errp, "could not create channel id");
904 goto err_resolve_create_id;
907 snprintf(port_str, 16, "%d", rdma->port);
910 ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
912 ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
913 goto err_resolve_get_addr;
916 for (e = res; e != NULL; e = e->ai_next) {
917 inet_ntop(e->ai_family,
918 &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
919 DPRINTF("Trying %s => %s\n", rdma->host, ip);
921 ret = rdma_resolve_addr(rdma->cm_id, NULL, e->ai_dst_addr,
922 RDMA_RESOLVE_TIMEOUT_MS);
924 if (e->ai_family == AF_INET6) {
925 ret = qemu_rdma_broken_ipv6_kernel(errp, rdma->cm_id->verbs);
934 ERROR(errp, "could not resolve address %s", rdma->host);
935 goto err_resolve_get_addr;
938 qemu_rdma_dump_gid("source_resolve_addr", rdma->cm_id);
940 ret = rdma_get_cm_event(rdma->channel, &cm_event);
942 ERROR(errp, "could not perform event_addr_resolved");
943 goto err_resolve_get_addr;
946 if (cm_event->event != RDMA_CM_EVENT_ADDR_RESOLVED) {
947 ERROR(errp, "result not equal to event_addr_resolved %s",
948 rdma_event_str(cm_event->event));
949 perror("rdma_resolve_addr");
951 goto err_resolve_get_addr;
953 rdma_ack_cm_event(cm_event);
956 ret = rdma_resolve_route(rdma->cm_id, RDMA_RESOLVE_TIMEOUT_MS);
958 ERROR(errp, "could not resolve rdma route");
959 goto err_resolve_get_addr;
962 ret = rdma_get_cm_event(rdma->channel, &cm_event);
964 ERROR(errp, "could not perform event_route_resolved");
965 goto err_resolve_get_addr;
967 if (cm_event->event != RDMA_CM_EVENT_ROUTE_RESOLVED) {
968 ERROR(errp, "result not equal to event_route_resolved: %s",
969 rdma_event_str(cm_event->event));
970 rdma_ack_cm_event(cm_event);
972 goto err_resolve_get_addr;
974 rdma_ack_cm_event(cm_event);
975 rdma->verbs = rdma->cm_id->verbs;
976 qemu_rdma_dump_id("source_resolve_host", rdma->cm_id->verbs);
977 qemu_rdma_dump_gid("source_resolve_host", rdma->cm_id);
980 err_resolve_get_addr:
981 rdma_destroy_id(rdma->cm_id);
983 err_resolve_create_id:
984 rdma_destroy_event_channel(rdma->channel);
985 rdma->channel = NULL;
990 * Create protection domain and completion queues
992 static int qemu_rdma_alloc_pd_cq(RDMAContext *rdma)
995 rdma->pd = ibv_alloc_pd(rdma->verbs);
997 fprintf(stderr, "failed to allocate protection domain\n");
1001 /* create completion channel */
1002 rdma->comp_channel = ibv_create_comp_channel(rdma->verbs);
1003 if (!rdma->comp_channel) {
1004 fprintf(stderr, "failed to allocate completion channel\n");
1005 goto err_alloc_pd_cq;
1009 * Completion queue can be filled by both read and write work requests,
1010 * so must reflect the sum of both possible queue sizes.
1012 rdma->cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3),
1013 NULL, rdma->comp_channel, 0);
1015 fprintf(stderr, "failed to allocate completion queue\n");
1016 goto err_alloc_pd_cq;
1023 ibv_dealloc_pd(rdma->pd);
1025 if (rdma->comp_channel) {
1026 ibv_destroy_comp_channel(rdma->comp_channel);
1029 rdma->comp_channel = NULL;
1035 * Create queue pairs.
1037 static int qemu_rdma_alloc_qp(RDMAContext *rdma)
1039 struct ibv_qp_init_attr attr = { 0 };
1042 attr.cap.max_send_wr = RDMA_SIGNALED_SEND_MAX;
1043 attr.cap.max_recv_wr = 3;
1044 attr.cap.max_send_sge = 1;
1045 attr.cap.max_recv_sge = 1;
1046 attr.send_cq = rdma->cq;
1047 attr.recv_cq = rdma->cq;
1048 attr.qp_type = IBV_QPT_RC;
1050 ret = rdma_create_qp(rdma->cm_id, rdma->pd, &attr);
1055 rdma->qp = rdma->cm_id->qp;
1059 static int qemu_rdma_reg_whole_ram_blocks(RDMAContext *rdma)
1062 RDMALocalBlocks *local = &rdma->local_ram_blocks;
1064 for (i = 0; i < local->nb_blocks; i++) {
1065 local->block[i].mr =
1066 ibv_reg_mr(rdma->pd,
1067 local->block[i].local_host_addr,
1068 local->block[i].length,
1069 IBV_ACCESS_LOCAL_WRITE |
1070 IBV_ACCESS_REMOTE_WRITE
1072 if (!local->block[i].mr) {
1073 perror("Failed to register local dest ram block!\n");
1076 rdma->total_registrations++;
1079 if (i >= local->nb_blocks) {
1083 for (i--; i >= 0; i--) {
1084 ibv_dereg_mr(local->block[i].mr);
1085 rdma->total_registrations--;
1093 * Find the ram block that corresponds to the page requested to be
1094 * transmitted by QEMU.
1096 * Once the block is found, also identify which 'chunk' within that
1097 * block that the page belongs to.
1099 * This search cannot fail or the migration will fail.
1101 static int qemu_rdma_search_ram_block(RDMAContext *rdma,
1102 uint64_t block_offset,
1105 uint64_t *block_index,
1106 uint64_t *chunk_index)
1108 uint64_t current_addr = block_offset + offset;
1109 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
1110 (void *) block_offset);
1112 assert(current_addr >= block->offset);
1113 assert((current_addr + length) <= (block->offset + block->length));
1115 *block_index = block->index;
1116 *chunk_index = ram_chunk_index(block->local_host_addr,
1117 block->local_host_addr + (current_addr - block->offset));
1123 * Register a chunk with IB. If the chunk was already registered
1124 * previously, then skip.
1126 * Also return the keys associated with the registration needed
1127 * to perform the actual RDMA operation.
1129 static int qemu_rdma_register_and_get_keys(RDMAContext *rdma,
1130 RDMALocalBlock *block, uint8_t *host_addr,
1131 uint32_t *lkey, uint32_t *rkey, int chunk,
1132 uint8_t *chunk_start, uint8_t *chunk_end)
1136 *lkey = block->mr->lkey;
1139 *rkey = block->mr->rkey;
1144 /* allocate memory to store chunk MRs */
1146 block->pmr = g_malloc0(block->nb_chunks * sizeof(struct ibv_mr *));
1153 * If 'rkey', then we're the destination, so grant access to the source.
1155 * If 'lkey', then we're the source VM, so grant access only to ourselves.
1157 if (!block->pmr[chunk]) {
1158 uint64_t len = chunk_end - chunk_start;
1160 DDPRINTF("Registering %" PRIu64 " bytes @ %p\n",
1163 block->pmr[chunk] = ibv_reg_mr(rdma->pd,
1165 (rkey ? (IBV_ACCESS_LOCAL_WRITE |
1166 IBV_ACCESS_REMOTE_WRITE) : 0));
1168 if (!block->pmr[chunk]) {
1169 perror("Failed to register chunk!");
1170 fprintf(stderr, "Chunk details: block: %d chunk index %d"
1171 " start %" PRIu64 " end %" PRIu64 " host %" PRIu64
1172 " local %" PRIu64 " registrations: %d\n",
1173 block->index, chunk, (uint64_t) chunk_start,
1174 (uint64_t) chunk_end, (uint64_t) host_addr,
1175 (uint64_t) block->local_host_addr,
1176 rdma->total_registrations);
1179 rdma->total_registrations++;
1183 *lkey = block->pmr[chunk]->lkey;
1186 *rkey = block->pmr[chunk]->rkey;
1192 * Register (at connection time) the memory used for control
1195 static int qemu_rdma_reg_control(RDMAContext *rdma, int idx)
1197 rdma->wr_data[idx].control_mr = ibv_reg_mr(rdma->pd,
1198 rdma->wr_data[idx].control, RDMA_CONTROL_MAX_BUFFER,
1199 IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
1200 if (rdma->wr_data[idx].control_mr) {
1201 rdma->total_registrations++;
1204 fprintf(stderr, "qemu_rdma_reg_control failed!\n");
1208 const char *print_wrid(int wrid)
1210 if (wrid >= RDMA_WRID_RECV_CONTROL) {
1211 return wrid_desc[RDMA_WRID_RECV_CONTROL];
1213 return wrid_desc[wrid];
1217 * RDMA requires memory registration (mlock/pinning), but this is not good for
1220 * In preparation for the future where LRU information or workload-specific
1221 * writable writable working set memory access behavior is available to QEMU
1222 * it would be nice to have in place the ability to UN-register/UN-pin
1223 * particular memory regions from the RDMA hardware when it is determine that
1224 * those regions of memory will likely not be accessed again in the near future.
1226 * While we do not yet have such information right now, the following
1227 * compile-time option allows us to perform a non-optimized version of this
1230 * By uncommenting this option, you will cause *all* RDMA transfers to be
1231 * unregistered immediately after the transfer completes on both sides of the
1232 * connection. This has no effect in 'rdma-pin-all' mode, only regular mode.
1234 * This will have a terrible impact on migration performance, so until future
1235 * workload information or LRU information is available, do not attempt to use
1236 * this feature except for basic testing.
1238 //#define RDMA_UNREGISTRATION_EXAMPLE
1241 * Perform a non-optimized memory unregistration after every transfer
1242 * for demonsration purposes, only if pin-all is not requested.
1244 * Potential optimizations:
1245 * 1. Start a new thread to run this function continuously
1247 - and for receipt of unregister messages
1249 * 3. Use workload hints.
1251 static int qemu_rdma_unregister_waiting(RDMAContext *rdma)
1253 while (rdma->unregistrations[rdma->unregister_current]) {
1255 uint64_t wr_id = rdma->unregistrations[rdma->unregister_current];
1257 (wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1259 (wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1260 RDMALocalBlock *block =
1261 &(rdma->local_ram_blocks.block[index]);
1262 RDMARegister reg = { .current_index = index };
1263 RDMAControlHeader resp = { .type = RDMA_CONTROL_UNREGISTER_FINISHED,
1265 RDMAControlHeader head = { .len = sizeof(RDMARegister),
1266 .type = RDMA_CONTROL_UNREGISTER_REQUEST,
1270 DDPRINTF("Processing unregister for chunk: %" PRIu64
1271 " at position %d\n", chunk, rdma->unregister_current);
1273 rdma->unregistrations[rdma->unregister_current] = 0;
1274 rdma->unregister_current++;
1276 if (rdma->unregister_current == RDMA_SIGNALED_SEND_MAX) {
1277 rdma->unregister_current = 0;
1282 * Unregistration is speculative (because migration is single-threaded
1283 * and we cannot break the protocol's inifinband message ordering).
1284 * Thus, if the memory is currently being used for transmission,
1285 * then abort the attempt to unregister and try again
1286 * later the next time a completion is received for this memory.
1288 clear_bit(chunk, block->unregister_bitmap);
1290 if (test_bit(chunk, block->transit_bitmap)) {
1291 DDPRINTF("Cannot unregister inflight chunk: %" PRIu64 "\n", chunk);
1295 DDPRINTF("Sending unregister for chunk: %" PRIu64 "\n", chunk);
1297 ret = ibv_dereg_mr(block->pmr[chunk]);
1298 block->pmr[chunk] = NULL;
1299 block->remote_keys[chunk] = 0;
1302 perror("unregistration chunk failed");
1305 rdma->total_registrations--;
1307 reg.key.chunk = chunk;
1308 register_to_network(®);
1309 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) ®,
1315 DDPRINTF("Unregister for chunk: %" PRIu64 " complete.\n", chunk);
1321 static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index,
1324 uint64_t result = wr_id & RDMA_WRID_TYPE_MASK;
1326 result |= (index << RDMA_WRID_BLOCK_SHIFT);
1327 result |= (chunk << RDMA_WRID_CHUNK_SHIFT);
1333 * Set bit for unregistration in the next iteration.
1334 * We cannot transmit right here, but will unpin later.
1336 static void qemu_rdma_signal_unregister(RDMAContext *rdma, uint64_t index,
1337 uint64_t chunk, uint64_t wr_id)
1339 if (rdma->unregistrations[rdma->unregister_next] != 0) {
1340 fprintf(stderr, "rdma migration: queue is full!\n");
1342 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1344 if (!test_and_set_bit(chunk, block->unregister_bitmap)) {
1345 DDPRINTF("Appending unregister chunk %" PRIu64
1346 " at position %d\n", chunk, rdma->unregister_next);
1348 rdma->unregistrations[rdma->unregister_next++] =
1349 qemu_rdma_make_wrid(wr_id, index, chunk);
1351 if (rdma->unregister_next == RDMA_SIGNALED_SEND_MAX) {
1352 rdma->unregister_next = 0;
1355 DDPRINTF("Unregister chunk %" PRIu64 " already in queue.\n",
1362 * Consult the connection manager to see a work request
1363 * (of any kind) has completed.
1364 * Return the work request ID that completed.
1366 static uint64_t qemu_rdma_poll(RDMAContext *rdma, uint64_t *wr_id_out,
1373 ret = ibv_poll_cq(rdma->cq, 1, &wc);
1376 *wr_id_out = RDMA_WRID_NONE;
1381 fprintf(stderr, "ibv_poll_cq return %d!\n", ret);
1385 wr_id = wc.wr_id & RDMA_WRID_TYPE_MASK;
1387 if (wc.status != IBV_WC_SUCCESS) {
1388 fprintf(stderr, "ibv_poll_cq wc.status=%d %s!\n",
1389 wc.status, ibv_wc_status_str(wc.status));
1390 fprintf(stderr, "ibv_poll_cq wrid=%s!\n", wrid_desc[wr_id]);
1395 if (rdma->control_ready_expected &&
1396 (wr_id >= RDMA_WRID_RECV_CONTROL)) {
1397 DDDPRINTF("completion %s #%" PRId64 " received (%" PRId64 ")"
1398 " left %d\n", wrid_desc[RDMA_WRID_RECV_CONTROL],
1399 wr_id - RDMA_WRID_RECV_CONTROL, wr_id, rdma->nb_sent);
1400 rdma->control_ready_expected = 0;
1403 if (wr_id == RDMA_WRID_RDMA_WRITE) {
1405 (wc.wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1407 (wc.wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1408 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1410 DDDPRINTF("completions %s (%" PRId64 ") left %d, "
1411 "block %" PRIu64 ", chunk: %" PRIu64 " %p %p\n",
1412 print_wrid(wr_id), wr_id, rdma->nb_sent, index, chunk,
1413 block->local_host_addr, (void *)block->remote_host_addr);
1415 clear_bit(chunk, block->transit_bitmap);
1417 if (rdma->nb_sent > 0) {
1421 if (!rdma->pin_all) {
1423 * FYI: If one wanted to signal a specific chunk to be unregistered
1424 * using LRU or workload-specific information, this is the function
1425 * you would call to do so. That chunk would then get asynchronously
1426 * unregistered later.
1428 #ifdef RDMA_UNREGISTRATION_EXAMPLE
1429 qemu_rdma_signal_unregister(rdma, index, chunk, wc.wr_id);
1433 DDDPRINTF("other completion %s (%" PRId64 ") received left %d\n",
1434 print_wrid(wr_id), wr_id, rdma->nb_sent);
1437 *wr_id_out = wc.wr_id;
1439 *byte_len = wc.byte_len;
1446 * Block until the next work request has completed.
1448 * First poll to see if a work request has already completed,
1451 * If we encounter completed work requests for IDs other than
1452 * the one we're interested in, then that's generally an error.
1454 * The only exception is actual RDMA Write completions. These
1455 * completions only need to be recorded, but do not actually
1456 * need further processing.
1458 static int qemu_rdma_block_for_wrid(RDMAContext *rdma, int wrid_requested,
1461 int num_cq_events = 0, ret = 0;
1464 uint64_t wr_id = RDMA_WRID_NONE, wr_id_in;
1466 if (ibv_req_notify_cq(rdma->cq, 0)) {
1470 while (wr_id != wrid_requested) {
1471 ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
1476 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1478 if (wr_id == RDMA_WRID_NONE) {
1481 if (wr_id != wrid_requested) {
1482 DDDPRINTF("A Wanted wrid %s (%d) but got %s (%" PRIu64 ")\n",
1483 print_wrid(wrid_requested),
1484 wrid_requested, print_wrid(wr_id), wr_id);
1488 if (wr_id == wrid_requested) {
1494 * Coroutine doesn't start until process_incoming_migration()
1495 * so don't yield unless we know we're running inside of a coroutine.
1497 if (rdma->migration_started_on_destination) {
1498 yield_until_fd_readable(rdma->comp_channel->fd);
1501 if (ibv_get_cq_event(rdma->comp_channel, &cq, &cq_ctx)) {
1502 perror("ibv_get_cq_event");
1503 goto err_block_for_wrid;
1508 if (ibv_req_notify_cq(cq, 0)) {
1509 goto err_block_for_wrid;
1512 while (wr_id != wrid_requested) {
1513 ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
1515 goto err_block_for_wrid;
1518 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1520 if (wr_id == RDMA_WRID_NONE) {
1523 if (wr_id != wrid_requested) {
1524 DDDPRINTF("B Wanted wrid %s (%d) but got %s (%" PRIu64 ")\n",
1525 print_wrid(wrid_requested), wrid_requested,
1526 print_wrid(wr_id), wr_id);
1530 if (wr_id == wrid_requested) {
1531 goto success_block_for_wrid;
1535 success_block_for_wrid:
1536 if (num_cq_events) {
1537 ibv_ack_cq_events(cq, num_cq_events);
1542 if (num_cq_events) {
1543 ibv_ack_cq_events(cq, num_cq_events);
1549 * Post a SEND message work request for the control channel
1550 * containing some data and block until the post completes.
1552 static int qemu_rdma_post_send_control(RDMAContext *rdma, uint8_t *buf,
1553 RDMAControlHeader *head)
1556 RDMAWorkRequestData *wr = &rdma->wr_data[RDMA_WRID_CONTROL];
1557 struct ibv_send_wr *bad_wr;
1558 struct ibv_sge sge = {
1559 .addr = (uint64_t)(wr->control),
1560 .length = head->len + sizeof(RDMAControlHeader),
1561 .lkey = wr->control_mr->lkey,
1563 struct ibv_send_wr send_wr = {
1564 .wr_id = RDMA_WRID_SEND_CONTROL,
1565 .opcode = IBV_WR_SEND,
1566 .send_flags = IBV_SEND_SIGNALED,
1571 DDDPRINTF("CONTROL: sending %s..\n", control_desc[head->type]);
1574 * We don't actually need to do a memcpy() in here if we used
1575 * the "sge" properly, but since we're only sending control messages
1576 * (not RAM in a performance-critical path), then its OK for now.
1578 * The copy makes the RDMAControlHeader simpler to manipulate
1579 * for the time being.
1581 assert(head->len <= RDMA_CONTROL_MAX_BUFFER - sizeof(*head));
1582 memcpy(wr->control, head, sizeof(RDMAControlHeader));
1583 control_to_network((void *) wr->control);
1586 memcpy(wr->control + sizeof(RDMAControlHeader), buf, head->len);
1590 if (ibv_post_send(rdma->qp, &send_wr, &bad_wr)) {
1595 fprintf(stderr, "Failed to use post IB SEND for control!\n");
1599 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_SEND_CONTROL, NULL);
1601 fprintf(stderr, "rdma migration: send polling control error!\n");
1608 * Post a RECV work request in anticipation of some future receipt
1609 * of data on the control channel.
1611 static int qemu_rdma_post_recv_control(RDMAContext *rdma, int idx)
1613 struct ibv_recv_wr *bad_wr;
1614 struct ibv_sge sge = {
1615 .addr = (uint64_t)(rdma->wr_data[idx].control),
1616 .length = RDMA_CONTROL_MAX_BUFFER,
1617 .lkey = rdma->wr_data[idx].control_mr->lkey,
1620 struct ibv_recv_wr recv_wr = {
1621 .wr_id = RDMA_WRID_RECV_CONTROL + idx,
1627 if (ibv_post_recv(rdma->qp, &recv_wr, &bad_wr)) {
1635 * Block and wait for a RECV control channel message to arrive.
1637 static int qemu_rdma_exchange_get_response(RDMAContext *rdma,
1638 RDMAControlHeader *head, int expecting, int idx)
1641 int ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RECV_CONTROL + idx,
1645 fprintf(stderr, "rdma migration: recv polling control error!\n");
1649 network_to_control((void *) rdma->wr_data[idx].control);
1650 memcpy(head, rdma->wr_data[idx].control, sizeof(RDMAControlHeader));
1652 DDDPRINTF("CONTROL: %s receiving...\n", control_desc[expecting]);
1654 if (expecting == RDMA_CONTROL_NONE) {
1655 DDDPRINTF("Surprise: got %s (%d)\n",
1656 control_desc[head->type], head->type);
1657 } else if (head->type != expecting || head->type == RDMA_CONTROL_ERROR) {
1658 fprintf(stderr, "Was expecting a %s (%d) control message"
1659 ", but got: %s (%d), length: %d\n",
1660 control_desc[expecting], expecting,
1661 control_desc[head->type], head->type, head->len);
1664 if (head->len > RDMA_CONTROL_MAX_BUFFER - sizeof(*head)) {
1665 fprintf(stderr, "too long length: %d\n", head->len);
1668 if (sizeof(*head) + head->len != byte_len) {
1669 fprintf(stderr, "Malformed length: %d byte_len %d\n",
1670 head->len, byte_len);
1678 * When a RECV work request has completed, the work request's
1679 * buffer is pointed at the header.
1681 * This will advance the pointer to the data portion
1682 * of the control message of the work request's buffer that
1683 * was populated after the work request finished.
1685 static void qemu_rdma_move_header(RDMAContext *rdma, int idx,
1686 RDMAControlHeader *head)
1688 rdma->wr_data[idx].control_len = head->len;
1689 rdma->wr_data[idx].control_curr =
1690 rdma->wr_data[idx].control + sizeof(RDMAControlHeader);
1694 * This is an 'atomic' high-level operation to deliver a single, unified
1695 * control-channel message.
1697 * Additionally, if the user is expecting some kind of reply to this message,
1698 * they can request a 'resp' response message be filled in by posting an
1699 * additional work request on behalf of the user and waiting for an additional
1702 * The extra (optional) response is used during registration to us from having
1703 * to perform an *additional* exchange of message just to provide a response by
1704 * instead piggy-backing on the acknowledgement.
1706 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
1707 uint8_t *data, RDMAControlHeader *resp,
1709 int (*callback)(RDMAContext *rdma))
1714 * Wait until the dest is ready before attempting to deliver the message
1715 * by waiting for a READY message.
1717 if (rdma->control_ready_expected) {
1718 RDMAControlHeader resp;
1719 ret = qemu_rdma_exchange_get_response(rdma,
1720 &resp, RDMA_CONTROL_READY, RDMA_WRID_READY);
1727 * If the user is expecting a response, post a WR in anticipation of it.
1730 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_DATA);
1732 fprintf(stderr, "rdma migration: error posting"
1733 " extra control recv for anticipated result!");
1739 * Post a WR to replace the one we just consumed for the READY message.
1741 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
1743 fprintf(stderr, "rdma migration: error posting first control recv!");
1748 * Deliver the control message that was requested.
1750 ret = qemu_rdma_post_send_control(rdma, data, head);
1753 fprintf(stderr, "Failed to send control buffer!\n");
1758 * If we're expecting a response, block and wait for it.
1762 DDPRINTF("Issuing callback before receiving response...\n");
1763 ret = callback(rdma);
1769 DDPRINTF("Waiting for response %s\n", control_desc[resp->type]);
1770 ret = qemu_rdma_exchange_get_response(rdma, resp,
1771 resp->type, RDMA_WRID_DATA);
1777 qemu_rdma_move_header(rdma, RDMA_WRID_DATA, resp);
1779 *resp_idx = RDMA_WRID_DATA;
1781 DDPRINTF("Response %s received.\n", control_desc[resp->type]);
1784 rdma->control_ready_expected = 1;
1790 * This is an 'atomic' high-level operation to receive a single, unified
1791 * control-channel message.
1793 static int qemu_rdma_exchange_recv(RDMAContext *rdma, RDMAControlHeader *head,
1796 RDMAControlHeader ready = {
1798 .type = RDMA_CONTROL_READY,
1804 * Inform the source that we're ready to receive a message.
1806 ret = qemu_rdma_post_send_control(rdma, NULL, &ready);
1809 fprintf(stderr, "Failed to send control buffer!\n");
1814 * Block and wait for the message.
1816 ret = qemu_rdma_exchange_get_response(rdma, head,
1817 expecting, RDMA_WRID_READY);
1823 qemu_rdma_move_header(rdma, RDMA_WRID_READY, head);
1826 * Post a new RECV work request to replace the one we just consumed.
1828 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
1830 fprintf(stderr, "rdma migration: error posting second control recv!");
1838 * Write an actual chunk of memory using RDMA.
1840 * If we're using dynamic registration on the dest-side, we have to
1841 * send a registration command first.
1843 static int qemu_rdma_write_one(QEMUFile *f, RDMAContext *rdma,
1844 int current_index, uint64_t current_addr,
1848 struct ibv_send_wr send_wr = { 0 };
1849 struct ibv_send_wr *bad_wr;
1850 int reg_result_idx, ret, count = 0;
1851 uint64_t chunk, chunks;
1852 uint8_t *chunk_start, *chunk_end;
1853 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[current_index]);
1855 RDMARegisterResult *reg_result;
1856 RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT };
1857 RDMAControlHeader head = { .len = sizeof(RDMARegister),
1858 .type = RDMA_CONTROL_REGISTER_REQUEST,
1863 sge.addr = (uint64_t)(block->local_host_addr +
1864 (current_addr - block->offset));
1865 sge.length = length;
1867 chunk = ram_chunk_index(block->local_host_addr, (uint8_t *) sge.addr);
1868 chunk_start = ram_chunk_start(block, chunk);
1870 if (block->is_ram_block) {
1871 chunks = length / (1UL << RDMA_REG_CHUNK_SHIFT);
1873 if (chunks && ((length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
1877 chunks = block->length / (1UL << RDMA_REG_CHUNK_SHIFT);
1879 if (chunks && ((block->length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
1884 DDPRINTF("Writing %" PRIu64 " chunks, (%" PRIu64 " MB)\n",
1885 chunks + 1, (chunks + 1) * (1UL << RDMA_REG_CHUNK_SHIFT) / 1024 / 1024);
1887 chunk_end = ram_chunk_end(block, chunk + chunks);
1889 if (!rdma->pin_all) {
1890 #ifdef RDMA_UNREGISTRATION_EXAMPLE
1891 qemu_rdma_unregister_waiting(rdma);
1895 while (test_bit(chunk, block->transit_bitmap)) {
1897 DDPRINTF("(%d) Not clobbering: block: %d chunk %" PRIu64
1898 " current %" PRIu64 " len %" PRIu64 " %d %d\n",
1899 count++, current_index, chunk,
1900 sge.addr, length, rdma->nb_sent, block->nb_chunks);
1902 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
1905 fprintf(stderr, "Failed to Wait for previous write to complete "
1906 "block %d chunk %" PRIu64
1907 " current %" PRIu64 " len %" PRIu64 " %d\n",
1908 current_index, chunk, sge.addr, length, rdma->nb_sent);
1913 if (!rdma->pin_all || !block->is_ram_block) {
1914 if (!block->remote_keys[chunk]) {
1916 * This chunk has not yet been registered, so first check to see
1917 * if the entire chunk is zero. If so, tell the other size to
1918 * memset() + madvise() the entire chunk without RDMA.
1921 if (can_use_buffer_find_nonzero_offset((void *)sge.addr, length)
1922 && buffer_find_nonzero_offset((void *)sge.addr,
1923 length) == length) {
1924 RDMACompress comp = {
1925 .offset = current_addr,
1927 .block_idx = current_index,
1931 head.len = sizeof(comp);
1932 head.type = RDMA_CONTROL_COMPRESS;
1934 DDPRINTF("Entire chunk is zero, sending compress: %"
1936 "bytes, index: %d, offset: %" PRId64 "...\n",
1937 chunk, sge.length, current_index, current_addr);
1939 compress_to_network(&comp);
1940 ret = qemu_rdma_exchange_send(rdma, &head,
1941 (uint8_t *) &comp, NULL, NULL, NULL);
1947 acct_update_position(f, sge.length, true);
1953 * Otherwise, tell other side to register.
1955 reg.current_index = current_index;
1956 if (block->is_ram_block) {
1957 reg.key.current_addr = current_addr;
1959 reg.key.chunk = chunk;
1961 reg.chunks = chunks;
1963 DDPRINTF("Sending registration request chunk %" PRIu64 " for %d "
1964 "bytes, index: %d, offset: %" PRId64 "...\n",
1965 chunk, sge.length, current_index, current_addr);
1967 register_to_network(®);
1968 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) ®,
1969 &resp, ®_result_idx, NULL);
1974 /* try to overlap this single registration with the one we sent. */
1975 if (qemu_rdma_register_and_get_keys(rdma, block,
1976 (uint8_t *) sge.addr,
1977 &sge.lkey, NULL, chunk,
1978 chunk_start, chunk_end)) {
1979 fprintf(stderr, "cannot get lkey!\n");
1983 reg_result = (RDMARegisterResult *)
1984 rdma->wr_data[reg_result_idx].control_curr;
1986 network_to_result(reg_result);
1988 DDPRINTF("Received registration result:"
1989 " my key: %x their key %x, chunk %" PRIu64 "\n",
1990 block->remote_keys[chunk], reg_result->rkey, chunk);
1992 block->remote_keys[chunk] = reg_result->rkey;
1993 block->remote_host_addr = reg_result->host_addr;
1995 /* already registered before */
1996 if (qemu_rdma_register_and_get_keys(rdma, block,
1997 (uint8_t *)sge.addr,
1998 &sge.lkey, NULL, chunk,
1999 chunk_start, chunk_end)) {
2000 fprintf(stderr, "cannot get lkey!\n");
2005 send_wr.wr.rdma.rkey = block->remote_keys[chunk];
2007 send_wr.wr.rdma.rkey = block->remote_rkey;
2009 if (qemu_rdma_register_and_get_keys(rdma, block, (uint8_t *)sge.addr,
2010 &sge.lkey, NULL, chunk,
2011 chunk_start, chunk_end)) {
2012 fprintf(stderr, "cannot get lkey!\n");
2018 * Encode the ram block index and chunk within this wrid.
2019 * We will use this information at the time of completion
2020 * to figure out which bitmap to check against and then which
2021 * chunk in the bitmap to look for.
2023 send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE,
2024 current_index, chunk);
2026 send_wr.opcode = IBV_WR_RDMA_WRITE;
2027 send_wr.send_flags = IBV_SEND_SIGNALED;
2028 send_wr.sg_list = &sge;
2029 send_wr.num_sge = 1;
2030 send_wr.wr.rdma.remote_addr = block->remote_host_addr +
2031 (current_addr - block->offset);
2033 DDDPRINTF("Posting chunk: %" PRIu64 ", addr: %lx"
2034 " remote: %lx, bytes %" PRIu32 "\n",
2035 chunk, sge.addr, send_wr.wr.rdma.remote_addr,
2039 * ibv_post_send() does not return negative error numbers,
2040 * per the specification they are positive - no idea why.
2042 ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
2044 if (ret == ENOMEM) {
2045 DDPRINTF("send queue is full. wait a little....\n");
2046 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2048 fprintf(stderr, "rdma migration: failed to make "
2049 "room in full send queue! %d\n", ret);
2055 } else if (ret > 0) {
2056 perror("rdma migration: post rdma write failed");
2060 set_bit(chunk, block->transit_bitmap);
2061 acct_update_position(f, sge.length, false);
2062 rdma->total_writes++;
2068 * Push out any unwritten RDMA operations.
2070 * We support sending out multiple chunks at the same time.
2071 * Not all of them need to get signaled in the completion queue.
2073 static int qemu_rdma_write_flush(QEMUFile *f, RDMAContext *rdma)
2077 if (!rdma->current_length) {
2081 ret = qemu_rdma_write_one(f, rdma,
2082 rdma->current_index, rdma->current_addr, rdma->current_length);
2090 DDDPRINTF("sent total: %d\n", rdma->nb_sent);
2093 rdma->current_length = 0;
2094 rdma->current_addr = 0;
2099 static inline int qemu_rdma_buffer_mergable(RDMAContext *rdma,
2100 uint64_t offset, uint64_t len)
2102 RDMALocalBlock *block;
2106 if (rdma->current_index < 0) {
2110 if (rdma->current_chunk < 0) {
2114 block = &(rdma->local_ram_blocks.block[rdma->current_index]);
2115 host_addr = block->local_host_addr + (offset - block->offset);
2116 chunk_end = ram_chunk_end(block, rdma->current_chunk);
2118 if (rdma->current_length == 0) {
2123 * Only merge into chunk sequentially.
2125 if (offset != (rdma->current_addr + rdma->current_length)) {
2129 if (offset < block->offset) {
2133 if ((offset + len) > (block->offset + block->length)) {
2137 if ((host_addr + len) > chunk_end) {
2145 * We're not actually writing here, but doing three things:
2147 * 1. Identify the chunk the buffer belongs to.
2148 * 2. If the chunk is full or the buffer doesn't belong to the current
2149 * chunk, then start a new chunk and flush() the old chunk.
2150 * 3. To keep the hardware busy, we also group chunks into batches
2151 * and only require that a batch gets acknowledged in the completion
2152 * qeueue instead of each individual chunk.
2154 static int qemu_rdma_write(QEMUFile *f, RDMAContext *rdma,
2155 uint64_t block_offset, uint64_t offset,
2158 uint64_t current_addr = block_offset + offset;
2159 uint64_t index = rdma->current_index;
2160 uint64_t chunk = rdma->current_chunk;
2163 /* If we cannot merge it, we flush the current buffer first. */
2164 if (!qemu_rdma_buffer_mergable(rdma, current_addr, len)) {
2165 ret = qemu_rdma_write_flush(f, rdma);
2169 rdma->current_length = 0;
2170 rdma->current_addr = current_addr;
2172 ret = qemu_rdma_search_ram_block(rdma, block_offset,
2173 offset, len, &index, &chunk);
2175 fprintf(stderr, "ram block search failed\n");
2178 rdma->current_index = index;
2179 rdma->current_chunk = chunk;
2183 rdma->current_length += len;
2185 /* flush it if buffer is too large */
2186 if (rdma->current_length >= RDMA_MERGE_MAX) {
2187 return qemu_rdma_write_flush(f, rdma);
2193 static void qemu_rdma_cleanup(RDMAContext *rdma)
2195 struct rdma_cm_event *cm_event;
2198 if (rdma->cm_id && rdma->connected) {
2199 if (rdma->error_state) {
2200 RDMAControlHeader head = { .len = 0,
2201 .type = RDMA_CONTROL_ERROR,
2204 fprintf(stderr, "Early error. Sending error.\n");
2205 qemu_rdma_post_send_control(rdma, NULL, &head);
2208 ret = rdma_disconnect(rdma->cm_id);
2210 DDPRINTF("waiting for disconnect\n");
2211 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2213 rdma_ack_cm_event(cm_event);
2216 DDPRINTF("Disconnected.\n");
2217 rdma->connected = false;
2220 g_free(rdma->block);
2223 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2224 if (rdma->wr_data[idx].control_mr) {
2225 rdma->total_registrations--;
2226 ibv_dereg_mr(rdma->wr_data[idx].control_mr);
2228 rdma->wr_data[idx].control_mr = NULL;
2231 if (rdma->local_ram_blocks.block) {
2232 while (rdma->local_ram_blocks.nb_blocks) {
2233 __qemu_rdma_delete_block(rdma,
2234 rdma->local_ram_blocks.block->offset);
2239 rdma_destroy_qp(rdma->cm_id);
2243 ibv_destroy_cq(rdma->cq);
2246 if (rdma->comp_channel) {
2247 ibv_destroy_comp_channel(rdma->comp_channel);
2248 rdma->comp_channel = NULL;
2251 ibv_dealloc_pd(rdma->pd);
2254 if (rdma->listen_id) {
2255 rdma_destroy_id(rdma->listen_id);
2256 rdma->listen_id = NULL;
2259 rdma_destroy_id(rdma->cm_id);
2262 if (rdma->channel) {
2263 rdma_destroy_event_channel(rdma->channel);
2264 rdma->channel = NULL;
2271 static int qemu_rdma_source_init(RDMAContext *rdma, Error **errp, bool pin_all)
2274 Error *local_err = NULL, **temp = &local_err;
2277 * Will be validated against destination's actual capabilities
2278 * after the connect() completes.
2280 rdma->pin_all = pin_all;
2282 ret = qemu_rdma_resolve_host(rdma, temp);
2284 goto err_rdma_source_init;
2287 ret = qemu_rdma_alloc_pd_cq(rdma);
2289 ERROR(temp, "rdma migration: error allocating pd and cq! Your mlock()"
2290 " limits may be too low. Please check $ ulimit -a # and "
2291 "search for 'ulimit -l' in the output");
2292 goto err_rdma_source_init;
2295 ret = qemu_rdma_alloc_qp(rdma);
2297 ERROR(temp, "rdma migration: error allocating qp!");
2298 goto err_rdma_source_init;
2301 ret = qemu_rdma_init_ram_blocks(rdma);
2303 ERROR(temp, "rdma migration: error initializing ram blocks!");
2304 goto err_rdma_source_init;
2307 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2308 ret = qemu_rdma_reg_control(rdma, idx);
2310 ERROR(temp, "rdma migration: error registering %d control!",
2312 goto err_rdma_source_init;
2318 err_rdma_source_init:
2319 error_propagate(errp, local_err);
2320 qemu_rdma_cleanup(rdma);
2324 static int qemu_rdma_connect(RDMAContext *rdma, Error **errp)
2326 RDMACapabilities cap = {
2327 .version = RDMA_CONTROL_VERSION_CURRENT,
2330 struct rdma_conn_param conn_param = { .initiator_depth = 2,
2332 .private_data = &cap,
2333 .private_data_len = sizeof(cap),
2335 struct rdma_cm_event *cm_event;
2339 * Only negotiate the capability with destination if the user
2340 * on the source first requested the capability.
2342 if (rdma->pin_all) {
2343 DPRINTF("Server pin-all memory requested.\n");
2344 cap.flags |= RDMA_CAPABILITY_PIN_ALL;
2347 caps_to_network(&cap);
2349 ret = rdma_connect(rdma->cm_id, &conn_param);
2351 perror("rdma_connect");
2352 ERROR(errp, "connecting to destination!");
2353 rdma_destroy_id(rdma->cm_id);
2355 goto err_rdma_source_connect;
2358 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2360 perror("rdma_get_cm_event after rdma_connect");
2361 ERROR(errp, "connecting to destination!");
2362 rdma_ack_cm_event(cm_event);
2363 rdma_destroy_id(rdma->cm_id);
2365 goto err_rdma_source_connect;
2368 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
2369 perror("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect");
2370 ERROR(errp, "connecting to destination!");
2371 rdma_ack_cm_event(cm_event);
2372 rdma_destroy_id(rdma->cm_id);
2374 goto err_rdma_source_connect;
2376 rdma->connected = true;
2378 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
2379 network_to_caps(&cap);
2382 * Verify that the *requested* capabilities are supported by the destination
2383 * and disable them otherwise.
2385 if (rdma->pin_all && !(cap.flags & RDMA_CAPABILITY_PIN_ALL)) {
2386 ERROR(errp, "Server cannot support pinning all memory. "
2387 "Will register memory dynamically.");
2388 rdma->pin_all = false;
2391 DPRINTF("Pin all memory: %s\n", rdma->pin_all ? "enabled" : "disabled");
2393 rdma_ack_cm_event(cm_event);
2395 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
2397 ERROR(errp, "posting second control recv!");
2398 goto err_rdma_source_connect;
2401 rdma->control_ready_expected = 1;
2405 err_rdma_source_connect:
2406 qemu_rdma_cleanup(rdma);
2410 static int qemu_rdma_dest_init(RDMAContext *rdma, Error **errp)
2412 int ret = -EINVAL, idx;
2413 struct rdma_cm_id *listen_id;
2414 char ip[40] = "unknown";
2415 struct rdma_addrinfo *res;
2418 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2419 rdma->wr_data[idx].control_len = 0;
2420 rdma->wr_data[idx].control_curr = NULL;
2423 if (rdma->host == NULL) {
2424 ERROR(errp, "RDMA host is not set!");
2425 rdma->error_state = -EINVAL;
2428 /* create CM channel */
2429 rdma->channel = rdma_create_event_channel();
2430 if (!rdma->channel) {
2431 ERROR(errp, "could not create rdma event channel");
2432 rdma->error_state = -EINVAL;
2437 ret = rdma_create_id(rdma->channel, &listen_id, NULL, RDMA_PS_TCP);
2439 ERROR(errp, "could not create cm_id!");
2440 goto err_dest_init_create_listen_id;
2443 snprintf(port_str, 16, "%d", rdma->port);
2444 port_str[15] = '\0';
2446 if (rdma->host && strcmp("", rdma->host)) {
2447 struct rdma_addrinfo *e;
2449 ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
2451 ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
2452 goto err_dest_init_bind_addr;
2455 for (e = res; e != NULL; e = e->ai_next) {
2456 inet_ntop(e->ai_family,
2457 &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
2458 DPRINTF("Trying %s => %s\n", rdma->host, ip);
2459 ret = rdma_bind_addr(listen_id, e->ai_dst_addr);
2461 if (e->ai_family == AF_INET6) {
2462 ret = qemu_rdma_broken_ipv6_kernel(errp, listen_id->verbs);
2472 ERROR(errp, "Error: could not rdma_bind_addr!");
2473 goto err_dest_init_bind_addr;
2475 ERROR(errp, "migration host and port not specified!");
2477 goto err_dest_init_bind_addr;
2481 rdma->listen_id = listen_id;
2482 qemu_rdma_dump_gid("dest_init", listen_id);
2485 err_dest_init_bind_addr:
2486 rdma_destroy_id(listen_id);
2487 err_dest_init_create_listen_id:
2488 rdma_destroy_event_channel(rdma->channel);
2489 rdma->channel = NULL;
2490 rdma->error_state = ret;
2495 static void *qemu_rdma_data_init(const char *host_port, Error **errp)
2497 RDMAContext *rdma = NULL;
2498 InetSocketAddress *addr;
2501 rdma = g_malloc0(sizeof(RDMAContext));
2502 memset(rdma, 0, sizeof(RDMAContext));
2503 rdma->current_index = -1;
2504 rdma->current_chunk = -1;
2506 addr = inet_parse(host_port, NULL);
2508 rdma->port = atoi(addr->port);
2509 rdma->host = g_strdup(addr->host);
2511 ERROR(errp, "bad RDMA migration address '%s'", host_port);
2521 * QEMUFile interface to the control channel.
2522 * SEND messages for control only.
2523 * pc.ram is handled with regular RDMA messages.
2525 static int qemu_rdma_put_buffer(void *opaque, const uint8_t *buf,
2526 int64_t pos, int size)
2528 QEMUFileRDMA *r = opaque;
2529 QEMUFile *f = r->file;
2530 RDMAContext *rdma = r->rdma;
2531 size_t remaining = size;
2532 uint8_t * data = (void *) buf;
2535 CHECK_ERROR_STATE();
2538 * Push out any writes that
2539 * we're queued up for pc.ram.
2541 ret = qemu_rdma_write_flush(f, rdma);
2543 rdma->error_state = ret;
2548 RDMAControlHeader head;
2550 r->len = MIN(remaining, RDMA_SEND_INCREMENT);
2551 remaining -= r->len;
2554 head.type = RDMA_CONTROL_QEMU_FILE;
2556 ret = qemu_rdma_exchange_send(rdma, &head, data, NULL, NULL, NULL);
2559 rdma->error_state = ret;
2569 static size_t qemu_rdma_fill(RDMAContext *rdma, uint8_t *buf,
2574 if (rdma->wr_data[idx].control_len) {
2575 DDDPRINTF("RDMA %" PRId64 " of %d bytes already in buffer\n",
2576 rdma->wr_data[idx].control_len, size);
2578 len = MIN(size, rdma->wr_data[idx].control_len);
2579 memcpy(buf, rdma->wr_data[idx].control_curr, len);
2580 rdma->wr_data[idx].control_curr += len;
2581 rdma->wr_data[idx].control_len -= len;
2588 * QEMUFile interface to the control channel.
2589 * RDMA links don't use bytestreams, so we have to
2590 * return bytes to QEMUFile opportunistically.
2592 static int qemu_rdma_get_buffer(void *opaque, uint8_t *buf,
2593 int64_t pos, int size)
2595 QEMUFileRDMA *r = opaque;
2596 RDMAContext *rdma = r->rdma;
2597 RDMAControlHeader head;
2600 CHECK_ERROR_STATE();
2603 * First, we hold on to the last SEND message we
2604 * were given and dish out the bytes until we run
2607 r->len = qemu_rdma_fill(r->rdma, buf, size, 0);
2613 * Once we run out, we block and wait for another
2614 * SEND message to arrive.
2616 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_QEMU_FILE);
2619 rdma->error_state = ret;
2624 * SEND was received with new bytes, now try again.
2626 return qemu_rdma_fill(r->rdma, buf, size, 0);
2630 * Block until all the outstanding chunks have been delivered by the hardware.
2632 static int qemu_rdma_drain_cq(QEMUFile *f, RDMAContext *rdma)
2636 if (qemu_rdma_write_flush(f, rdma) < 0) {
2640 while (rdma->nb_sent) {
2641 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
2643 fprintf(stderr, "rdma migration: complete polling error!\n");
2648 qemu_rdma_unregister_waiting(rdma);
2653 static int qemu_rdma_close(void *opaque)
2655 DPRINTF("Shutting down connection.\n");
2656 QEMUFileRDMA *r = opaque;
2658 qemu_rdma_cleanup(r->rdma);
2668 * This means that 'block_offset' is a full virtual address that does not
2669 * belong to a RAMBlock of the virtual machine and instead
2670 * represents a private malloc'd memory area that the caller wishes to
2674 * Offset is an offset to be added to block_offset and used
2675 * to also lookup the corresponding RAMBlock.
2678 * Initiate an transfer this size.
2681 * A 'hint' or 'advice' that means that we wish to speculatively
2682 * and asynchronously unregister this memory. In this case, there is no
2683 * guarantee that the unregister will actually happen, for example,
2684 * if the memory is being actively transmitted. Additionally, the memory
2685 * may be re-registered at any future time if a write within the same
2686 * chunk was requested again, even if you attempted to unregister it
2689 * @size < 0 : TODO, not yet supported
2690 * Unregister the memory NOW. This means that the caller does not
2691 * expect there to be any future RDMA transfers and we just want to clean
2692 * things up. This is used in case the upper layer owns the memory and
2693 * cannot wait for qemu_fclose() to occur.
2695 * @bytes_sent : User-specificed pointer to indicate how many bytes were
2696 * sent. Usually, this will not be more than a few bytes of
2697 * the protocol because most transfers are sent asynchronously.
2699 static size_t qemu_rdma_save_page(QEMUFile *f, void *opaque,
2700 ram_addr_t block_offset, ram_addr_t offset,
2701 size_t size, int *bytes_sent)
2703 QEMUFileRDMA *rfile = opaque;
2704 RDMAContext *rdma = rfile->rdma;
2707 CHECK_ERROR_STATE();
2713 * Add this page to the current 'chunk'. If the chunk
2714 * is full, or the page doen't belong to the current chunk,
2715 * an actual RDMA write will occur and a new chunk will be formed.
2717 ret = qemu_rdma_write(f, rdma, block_offset, offset, size);
2719 fprintf(stderr, "rdma migration: write error! %d\n", ret);
2724 * We always return 1 bytes because the RDMA
2725 * protocol is completely asynchronous. We do not yet know
2726 * whether an identified chunk is zero or not because we're
2727 * waiting for other pages to potentially be merged with
2728 * the current chunk. So, we have to call qemu_update_position()
2729 * later on when the actual write occurs.
2735 uint64_t index, chunk;
2737 /* TODO: Change QEMUFileOps prototype to be signed: size_t => long
2739 ret = qemu_rdma_drain_cq(f, rdma);
2741 fprintf(stderr, "rdma: failed to synchronously drain"
2742 " completion queue before unregistration.\n");
2748 ret = qemu_rdma_search_ram_block(rdma, block_offset,
2749 offset, size, &index, &chunk);
2752 fprintf(stderr, "ram block search failed\n");
2756 qemu_rdma_signal_unregister(rdma, index, chunk, 0);
2759 * TODO: Synchronous, guaranteed unregistration (should not occur during
2760 * fast-path). Otherwise, unregisters will process on the next call to
2761 * qemu_rdma_drain_cq()
2763 qemu_rdma_unregister_waiting(rdma);
2769 * Drain the Completion Queue if possible, but do not block,
2772 * If nothing to poll, the end of the iteration will do this
2773 * again to make sure we don't overflow the request queue.
2776 uint64_t wr_id, wr_id_in;
2777 int ret = qemu_rdma_poll(rdma, &wr_id_in, NULL);
2779 fprintf(stderr, "rdma migration: polling error! %d\n", ret);
2783 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
2785 if (wr_id == RDMA_WRID_NONE) {
2790 return RAM_SAVE_CONTROL_DELAYED;
2792 rdma->error_state = ret;
2796 static int qemu_rdma_accept(RDMAContext *rdma)
2798 RDMACapabilities cap;
2799 struct rdma_conn_param conn_param = {
2800 .responder_resources = 2,
2801 .private_data = &cap,
2802 .private_data_len = sizeof(cap),
2804 struct rdma_cm_event *cm_event;
2805 struct ibv_context *verbs;
2809 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2811 goto err_rdma_dest_wait;
2814 if (cm_event->event != RDMA_CM_EVENT_CONNECT_REQUEST) {
2815 rdma_ack_cm_event(cm_event);
2816 goto err_rdma_dest_wait;
2819 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
2821 network_to_caps(&cap);
2823 if (cap.version < 1 || cap.version > RDMA_CONTROL_VERSION_CURRENT) {
2824 fprintf(stderr, "Unknown source RDMA version: %d, bailing...\n",
2826 rdma_ack_cm_event(cm_event);
2827 goto err_rdma_dest_wait;
2831 * Respond with only the capabilities this version of QEMU knows about.
2833 cap.flags &= known_capabilities;
2836 * Enable the ones that we do know about.
2837 * Add other checks here as new ones are introduced.
2839 if (cap.flags & RDMA_CAPABILITY_PIN_ALL) {
2840 rdma->pin_all = true;
2843 rdma->cm_id = cm_event->id;
2844 verbs = cm_event->id->verbs;
2846 rdma_ack_cm_event(cm_event);
2848 DPRINTF("Memory pin all: %s\n", rdma->pin_all ? "enabled" : "disabled");
2850 caps_to_network(&cap);
2852 DPRINTF("verbs context after listen: %p\n", verbs);
2855 rdma->verbs = verbs;
2856 } else if (rdma->verbs != verbs) {
2857 fprintf(stderr, "ibv context not matching %p, %p!\n",
2858 rdma->verbs, verbs);
2859 goto err_rdma_dest_wait;
2862 qemu_rdma_dump_id("dest_init", verbs);
2864 ret = qemu_rdma_alloc_pd_cq(rdma);
2866 fprintf(stderr, "rdma migration: error allocating pd and cq!\n");
2867 goto err_rdma_dest_wait;
2870 ret = qemu_rdma_alloc_qp(rdma);
2872 fprintf(stderr, "rdma migration: error allocating qp!\n");
2873 goto err_rdma_dest_wait;
2876 ret = qemu_rdma_init_ram_blocks(rdma);
2878 fprintf(stderr, "rdma migration: error initializing ram blocks!\n");
2879 goto err_rdma_dest_wait;
2882 for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
2883 ret = qemu_rdma_reg_control(rdma, idx);
2885 fprintf(stderr, "rdma: error registering %d control!\n", idx);
2886 goto err_rdma_dest_wait;
2890 qemu_set_fd_handler2(rdma->channel->fd, NULL, NULL, NULL, NULL);
2892 ret = rdma_accept(rdma->cm_id, &conn_param);
2894 fprintf(stderr, "rdma_accept returns %d!\n", ret);
2895 goto err_rdma_dest_wait;
2898 ret = rdma_get_cm_event(rdma->channel, &cm_event);
2900 fprintf(stderr, "rdma_accept get_cm_event failed %d!\n", ret);
2901 goto err_rdma_dest_wait;
2904 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
2905 fprintf(stderr, "rdma_accept not event established!\n");
2906 rdma_ack_cm_event(cm_event);
2907 goto err_rdma_dest_wait;
2910 rdma_ack_cm_event(cm_event);
2911 rdma->connected = true;
2913 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
2915 fprintf(stderr, "rdma migration: error posting second control recv!\n");
2916 goto err_rdma_dest_wait;
2919 qemu_rdma_dump_gid("dest_connect", rdma->cm_id);
2924 rdma->error_state = ret;
2925 qemu_rdma_cleanup(rdma);
2930 * During each iteration of the migration, we listen for instructions
2931 * by the source VM to perform dynamic page registrations before they
2932 * can perform RDMA operations.
2934 * We respond with the 'rkey'.
2936 * Keep doing this until the source tells us to stop.
2938 static int qemu_rdma_registration_handle(QEMUFile *f, void *opaque,
2941 RDMAControlHeader reg_resp = { .len = sizeof(RDMARegisterResult),
2942 .type = RDMA_CONTROL_REGISTER_RESULT,
2945 RDMAControlHeader unreg_resp = { .len = 0,
2946 .type = RDMA_CONTROL_UNREGISTER_FINISHED,
2949 RDMAControlHeader blocks = { .type = RDMA_CONTROL_RAM_BLOCKS_RESULT,
2951 QEMUFileRDMA *rfile = opaque;
2952 RDMAContext *rdma = rfile->rdma;
2953 RDMALocalBlocks *local = &rdma->local_ram_blocks;
2954 RDMAControlHeader head;
2955 RDMARegister *reg, *registers;
2957 RDMARegisterResult *reg_result;
2958 static RDMARegisterResult results[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE];
2959 RDMALocalBlock *block;
2966 CHECK_ERROR_STATE();
2969 DDDPRINTF("Waiting for next request %" PRIu64 "...\n", flags);
2971 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_NONE);
2977 if (head.repeat > RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE) {
2978 fprintf(stderr, "rdma: Too many requests in this message (%d)."
2979 "Bailing.\n", head.repeat);
2984 switch (head.type) {
2985 case RDMA_CONTROL_COMPRESS:
2986 comp = (RDMACompress *) rdma->wr_data[idx].control_curr;
2987 network_to_compress(comp);
2989 DDPRINTF("Zapping zero chunk: %" PRId64
2990 " bytes, index %d, offset %" PRId64 "\n",
2991 comp->length, comp->block_idx, comp->offset);
2992 block = &(rdma->local_ram_blocks.block[comp->block_idx]);
2994 host_addr = block->local_host_addr +
2995 (comp->offset - block->offset);
2997 ram_handle_compressed(host_addr, comp->value, comp->length);
3000 case RDMA_CONTROL_REGISTER_FINISHED:
3001 DDDPRINTF("Current registrations complete.\n");
3004 case RDMA_CONTROL_RAM_BLOCKS_REQUEST:
3005 DPRINTF("Initial setup info requested.\n");
3007 if (rdma->pin_all) {
3008 ret = qemu_rdma_reg_whole_ram_blocks(rdma);
3010 fprintf(stderr, "rdma migration: error dest "
3011 "registering ram blocks!\n");
3017 * Dest uses this to prepare to transmit the RAMBlock descriptions
3018 * to the source VM after connection setup.
3019 * Both sides use the "remote" structure to communicate and update
3020 * their "local" descriptions with what was sent.
3022 for (i = 0; i < local->nb_blocks; i++) {
3023 rdma->block[i].remote_host_addr =
3024 (uint64_t)(local->block[i].local_host_addr);
3026 if (rdma->pin_all) {
3027 rdma->block[i].remote_rkey = local->block[i].mr->rkey;
3030 rdma->block[i].offset = local->block[i].offset;
3031 rdma->block[i].length = local->block[i].length;
3033 remote_block_to_network(&rdma->block[i]);
3036 blocks.len = rdma->local_ram_blocks.nb_blocks
3037 * sizeof(RDMARemoteBlock);
3040 ret = qemu_rdma_post_send_control(rdma,
3041 (uint8_t *) rdma->block, &blocks);
3044 fprintf(stderr, "rdma migration: error sending remote info!\n");
3049 case RDMA_CONTROL_REGISTER_REQUEST:
3050 DDPRINTF("There are %d registration requests\n", head.repeat);
3052 reg_resp.repeat = head.repeat;
3053 registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
3055 for (count = 0; count < head.repeat; count++) {
3057 uint8_t *chunk_start, *chunk_end;
3059 reg = ®isters[count];
3060 network_to_register(reg);
3062 reg_result = &results[count];
3064 DDPRINTF("Registration request (%d): index %d, current_addr %"
3065 PRIu64 " chunks: %" PRIu64 "\n", count,
3066 reg->current_index, reg->key.current_addr, reg->chunks);
3068 block = &(rdma->local_ram_blocks.block[reg->current_index]);
3069 if (block->is_ram_block) {
3070 host_addr = (block->local_host_addr +
3071 (reg->key.current_addr - block->offset));
3072 chunk = ram_chunk_index(block->local_host_addr,
3073 (uint8_t *) host_addr);
3075 chunk = reg->key.chunk;
3076 host_addr = block->local_host_addr +
3077 (reg->key.chunk * (1UL << RDMA_REG_CHUNK_SHIFT));
3079 chunk_start = ram_chunk_start(block, chunk);
3080 chunk_end = ram_chunk_end(block, chunk + reg->chunks);
3081 if (qemu_rdma_register_and_get_keys(rdma, block,
3082 (uint8_t *)host_addr, NULL, ®_result->rkey,
3083 chunk, chunk_start, chunk_end)) {
3084 fprintf(stderr, "cannot get rkey!\n");
3089 reg_result->host_addr = (uint64_t) block->local_host_addr;
3091 DDPRINTF("Registered rkey for this request: %x\n",
3094 result_to_network(reg_result);
3097 ret = qemu_rdma_post_send_control(rdma,
3098 (uint8_t *) results, ®_resp);
3101 fprintf(stderr, "Failed to send control buffer!\n");
3105 case RDMA_CONTROL_UNREGISTER_REQUEST:
3106 DDPRINTF("There are %d unregistration requests\n", head.repeat);
3107 unreg_resp.repeat = head.repeat;
3108 registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
3110 for (count = 0; count < head.repeat; count++) {
3111 reg = ®isters[count];
3112 network_to_register(reg);
3114 DDPRINTF("Unregistration request (%d): "
3115 " index %d, chunk %" PRIu64 "\n",
3116 count, reg->current_index, reg->key.chunk);
3118 block = &(rdma->local_ram_blocks.block[reg->current_index]);
3120 ret = ibv_dereg_mr(block->pmr[reg->key.chunk]);
3121 block->pmr[reg->key.chunk] = NULL;
3124 perror("rdma unregistration chunk failed");
3129 rdma->total_registrations--;
3131 DDPRINTF("Unregistered chunk %" PRIu64 " successfully.\n",
3135 ret = qemu_rdma_post_send_control(rdma, NULL, &unreg_resp);
3138 fprintf(stderr, "Failed to send control buffer!\n");
3142 case RDMA_CONTROL_REGISTER_RESULT:
3143 fprintf(stderr, "Invalid RESULT message at dest.\n");
3147 fprintf(stderr, "Unknown control message %s\n",
3148 control_desc[head.type]);
3155 rdma->error_state = ret;
3160 static int qemu_rdma_registration_start(QEMUFile *f, void *opaque,
3163 QEMUFileRDMA *rfile = opaque;
3164 RDMAContext *rdma = rfile->rdma;
3166 CHECK_ERROR_STATE();
3168 DDDPRINTF("start section: %" PRIu64 "\n", flags);
3169 qemu_put_be64(f, RAM_SAVE_FLAG_HOOK);
3176 * Inform dest that dynamic registrations are done for now.
3177 * First, flush writes, if any.
3179 static int qemu_rdma_registration_stop(QEMUFile *f, void *opaque,
3182 Error *local_err = NULL, **errp = &local_err;
3183 QEMUFileRDMA *rfile = opaque;
3184 RDMAContext *rdma = rfile->rdma;
3185 RDMAControlHeader head = { .len = 0, .repeat = 1 };
3188 CHECK_ERROR_STATE();
3191 ret = qemu_rdma_drain_cq(f, rdma);
3197 if (flags == RAM_CONTROL_SETUP) {
3198 RDMAControlHeader resp = {.type = RDMA_CONTROL_RAM_BLOCKS_RESULT };
3199 RDMALocalBlocks *local = &rdma->local_ram_blocks;
3200 int reg_result_idx, i, j, nb_remote_blocks;
3202 head.type = RDMA_CONTROL_RAM_BLOCKS_REQUEST;
3203 DPRINTF("Sending registration setup for ram blocks...\n");
3206 * Make sure that we parallelize the pinning on both sides.
3207 * For very large guests, doing this serially takes a really
3208 * long time, so we have to 'interleave' the pinning locally
3209 * with the control messages by performing the pinning on this
3210 * side before we receive the control response from the other
3211 * side that the pinning has completed.
3213 ret = qemu_rdma_exchange_send(rdma, &head, NULL, &resp,
3214 ®_result_idx, rdma->pin_all ?
3215 qemu_rdma_reg_whole_ram_blocks : NULL);
3217 ERROR(errp, "receiving remote info!");
3221 nb_remote_blocks = resp.len / sizeof(RDMARemoteBlock);
3224 * The protocol uses two different sets of rkeys (mutually exclusive):
3225 * 1. One key to represent the virtual address of the entire ram block.
3226 * (dynamic chunk registration disabled - pin everything with one rkey.)
3227 * 2. One to represent individual chunks within a ram block.
3228 * (dynamic chunk registration enabled - pin individual chunks.)
3230 * Once the capability is successfully negotiated, the destination transmits
3231 * the keys to use (or sends them later) including the virtual addresses
3232 * and then propagates the remote ram block descriptions to his local copy.
3235 if (local->nb_blocks != nb_remote_blocks) {
3236 ERROR(errp, "ram blocks mismatch #1! "
3237 "Your QEMU command line parameters are probably "
3238 "not identical on both the source and destination.");
3242 qemu_rdma_move_header(rdma, reg_result_idx, &resp);
3244 rdma->wr_data[reg_result_idx].control_curr, resp.len);
3245 for (i = 0; i < nb_remote_blocks; i++) {
3246 network_to_remote_block(&rdma->block[i]);
3248 /* search local ram blocks */
3249 for (j = 0; j < local->nb_blocks; j++) {
3250 if (rdma->block[i].offset != local->block[j].offset) {
3254 if (rdma->block[i].length != local->block[j].length) {
3255 ERROR(errp, "ram blocks mismatch #2! "
3256 "Your QEMU command line parameters are probably "
3257 "not identical on both the source and destination.");
3260 local->block[j].remote_host_addr =
3261 rdma->block[i].remote_host_addr;
3262 local->block[j].remote_rkey = rdma->block[i].remote_rkey;
3266 if (j >= local->nb_blocks) {
3267 ERROR(errp, "ram blocks mismatch #3! "
3268 "Your QEMU command line parameters are probably "
3269 "not identical on both the source and destination.");
3275 DDDPRINTF("Sending registration finish %" PRIu64 "...\n", flags);
3277 head.type = RDMA_CONTROL_REGISTER_FINISHED;
3278 ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL);
3286 rdma->error_state = ret;
3290 static int qemu_rdma_get_fd(void *opaque)
3292 QEMUFileRDMA *rfile = opaque;
3293 RDMAContext *rdma = rfile->rdma;
3295 return rdma->comp_channel->fd;
3298 const QEMUFileOps rdma_read_ops = {
3299 .get_buffer = qemu_rdma_get_buffer,
3300 .get_fd = qemu_rdma_get_fd,
3301 .close = qemu_rdma_close,
3302 .hook_ram_load = qemu_rdma_registration_handle,
3305 const QEMUFileOps rdma_write_ops = {
3306 .put_buffer = qemu_rdma_put_buffer,
3307 .close = qemu_rdma_close,
3308 .before_ram_iterate = qemu_rdma_registration_start,
3309 .after_ram_iterate = qemu_rdma_registration_stop,
3310 .save_page = qemu_rdma_save_page,
3313 static void *qemu_fopen_rdma(RDMAContext *rdma, const char *mode)
3315 QEMUFileRDMA *r = g_malloc0(sizeof(QEMUFileRDMA));
3317 if (qemu_file_mode_is_not_valid(mode)) {
3323 if (mode[0] == 'w') {
3324 r->file = qemu_fopen_ops(r, &rdma_write_ops);
3326 r->file = qemu_fopen_ops(r, &rdma_read_ops);
3332 static void rdma_accept_incoming_migration(void *opaque)
3334 RDMAContext *rdma = opaque;
3337 Error *local_err = NULL, **errp = &local_err;
3339 DPRINTF("Accepting rdma connection...\n");
3340 ret = qemu_rdma_accept(rdma);
3343 ERROR(errp, "RDMA Migration initialization failed!");
3347 DPRINTF("Accepted migration\n");
3349 f = qemu_fopen_rdma(rdma, "rb");
3351 ERROR(errp, "could not qemu_fopen_rdma!");
3352 qemu_rdma_cleanup(rdma);
3356 rdma->migration_started_on_destination = 1;
3357 process_incoming_migration(f);
3360 void rdma_start_incoming_migration(const char *host_port, Error **errp)
3364 Error *local_err = NULL;
3366 DPRINTF("Starting RDMA-based incoming migration\n");
3367 rdma = qemu_rdma_data_init(host_port, &local_err);
3373 ret = qemu_rdma_dest_init(rdma, &local_err);
3379 DPRINTF("qemu_rdma_dest_init success\n");
3381 ret = rdma_listen(rdma->listen_id, 5);
3384 ERROR(errp, "listening on socket!");
3388 DPRINTF("rdma_listen success\n");
3390 qemu_set_fd_handler2(rdma->channel->fd, NULL,
3391 rdma_accept_incoming_migration, NULL,
3392 (void *)(intptr_t) rdma);
3395 error_propagate(errp, local_err);
3399 void rdma_start_outgoing_migration(void *opaque,
3400 const char *host_port, Error **errp)
3402 MigrationState *s = opaque;
3403 Error *local_err = NULL, **temp = &local_err;
3404 RDMAContext *rdma = qemu_rdma_data_init(host_port, &local_err);
3408 ERROR(temp, "Failed to initialize RDMA data structures! %d", ret);
3412 ret = qemu_rdma_source_init(rdma, &local_err,
3413 s->enabled_capabilities[MIGRATION_CAPABILITY_X_RDMA_PIN_ALL]);
3419 DPRINTF("qemu_rdma_source_init success\n");
3420 ret = qemu_rdma_connect(rdma, &local_err);
3426 DPRINTF("qemu_rdma_source_connect success\n");
3428 s->file = qemu_fopen_rdma(rdma, "wb");
3429 migrate_fd_connect(s);
3432 error_propagate(errp, local_err);
3434 migrate_fd_error(s);