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, ## __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
+ 1];
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
;
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 printf("%s RDMA Device opened: kernel name %s "
712 "uverbs device name %s, "
713 "infiniband_verbs class device path %s,"
714 " infiniband class device path %s\n",
717 verbs
->device
->dev_name
,
718 verbs
->device
->dev_path
,
719 verbs
->device
->ibdev_path
);
723 * Put in the log file the RDMA gid addressing information,
724 * useful for folks who have trouble understanding the
725 * RDMA device hierarchy in the kernel.
727 static void qemu_rdma_dump_gid(const char *who
, struct rdma_cm_id
*id
)
731 inet_ntop(AF_INET6
, &id
->route
.addr
.addr
.ibaddr
.sgid
, sgid
, sizeof sgid
);
732 inet_ntop(AF_INET6
, &id
->route
.addr
.addr
.ibaddr
.dgid
, dgid
, sizeof dgid
);
733 DPRINTF("%s Source GID: %s, Dest GID: %s\n", who
, sgid
, dgid
);
737 * Figure out which RDMA device corresponds to the requested IP hostname
738 * Also create the initial connection manager identifiers for opening
741 static int qemu_rdma_resolve_host(RDMAContext
*rdma
, Error
**errp
)
744 struct addrinfo
*res
;
746 struct rdma_cm_event
*cm_event
;
747 char ip
[40] = "unknown";
748 int af
= rdma
->ipv6
? PF_INET6
: PF_INET
;
750 if (rdma
->host
== NULL
|| !strcmp(rdma
->host
, "")) {
751 ERROR(errp
, "RDMA hostname has not been set\n");
755 /* create CM channel */
756 rdma
->channel
= rdma_create_event_channel();
757 if (!rdma
->channel
) {
758 ERROR(errp
, "could not create CM channel\n");
763 ret
= rdma_create_id(rdma
->channel
, &rdma
->cm_id
, NULL
, RDMA_PS_TCP
);
765 ERROR(errp
, "could not create channel id\n");
766 goto err_resolve_create_id
;
769 snprintf(port_str
, 16, "%d", rdma
->port
);
772 ret
= getaddrinfo(rdma
->host
, port_str
, NULL
, &res
);
774 ERROR(errp
, "could not getaddrinfo address %s\n", rdma
->host
);
775 goto err_resolve_get_addr
;
778 inet_ntop(af
, &((struct sockaddr_in
*) res
->ai_addr
)->sin_addr
,
780 DPRINTF("%s => %s\n", rdma
->host
, ip
);
782 /* resolve the first address */
783 ret
= rdma_resolve_addr(rdma
->cm_id
, NULL
, res
->ai_addr
,
784 RDMA_RESOLVE_TIMEOUT_MS
);
786 ERROR(errp
, "could not resolve address %s\n", rdma
->host
);
787 goto err_resolve_get_addr
;
790 qemu_rdma_dump_gid("source_resolve_addr", rdma
->cm_id
);
792 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
794 ERROR(errp
, "could not perform event_addr_resolved\n");
795 goto err_resolve_get_addr
;
798 if (cm_event
->event
!= RDMA_CM_EVENT_ADDR_RESOLVED
) {
799 ERROR(errp
, "result not equal to event_addr_resolved %s\n",
800 rdma_event_str(cm_event
->event
));
801 perror("rdma_resolve_addr");
802 goto err_resolve_get_addr
;
804 rdma_ack_cm_event(cm_event
);
807 ret
= rdma_resolve_route(rdma
->cm_id
, RDMA_RESOLVE_TIMEOUT_MS
);
809 ERROR(errp
, "could not resolve rdma route\n");
810 goto err_resolve_get_addr
;
813 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
815 ERROR(errp
, "could not perform event_route_resolved\n");
816 goto err_resolve_get_addr
;
818 if (cm_event
->event
!= RDMA_CM_EVENT_ROUTE_RESOLVED
) {
819 ERROR(errp
, "result not equal to event_route_resolved: %s\n",
820 rdma_event_str(cm_event
->event
));
821 rdma_ack_cm_event(cm_event
);
822 goto err_resolve_get_addr
;
824 rdma_ack_cm_event(cm_event
);
825 rdma
->verbs
= rdma
->cm_id
->verbs
;
826 qemu_rdma_dump_id("source_resolve_host", rdma
->cm_id
->verbs
);
827 qemu_rdma_dump_gid("source_resolve_host", rdma
->cm_id
);
830 err_resolve_get_addr
:
831 rdma_destroy_id(rdma
->cm_id
);
833 err_resolve_create_id
:
834 rdma_destroy_event_channel(rdma
->channel
);
835 rdma
->channel
= NULL
;
841 * Create protection domain and completion queues
843 static int qemu_rdma_alloc_pd_cq(RDMAContext
*rdma
)
846 rdma
->pd
= ibv_alloc_pd(rdma
->verbs
);
848 fprintf(stderr
, "failed to allocate protection domain\n");
852 /* create completion channel */
853 rdma
->comp_channel
= ibv_create_comp_channel(rdma
->verbs
);
854 if (!rdma
->comp_channel
) {
855 fprintf(stderr
, "failed to allocate completion channel\n");
856 goto err_alloc_pd_cq
;
860 * Completion queue can be filled by both read and write work requests,
861 * so must reflect the sum of both possible queue sizes.
863 rdma
->cq
= ibv_create_cq(rdma
->verbs
, (RDMA_SIGNALED_SEND_MAX
* 3),
864 NULL
, rdma
->comp_channel
, 0);
866 fprintf(stderr
, "failed to allocate completion queue\n");
867 goto err_alloc_pd_cq
;
874 ibv_dealloc_pd(rdma
->pd
);
876 if (rdma
->comp_channel
) {
877 ibv_destroy_comp_channel(rdma
->comp_channel
);
880 rdma
->comp_channel
= NULL
;
886 * Create queue pairs.
888 static int qemu_rdma_alloc_qp(RDMAContext
*rdma
)
890 struct ibv_qp_init_attr attr
= { 0 };
893 attr
.cap
.max_send_wr
= RDMA_SIGNALED_SEND_MAX
;
894 attr
.cap
.max_recv_wr
= 3;
895 attr
.cap
.max_send_sge
= 1;
896 attr
.cap
.max_recv_sge
= 1;
897 attr
.send_cq
= rdma
->cq
;
898 attr
.recv_cq
= rdma
->cq
;
899 attr
.qp_type
= IBV_QPT_RC
;
901 ret
= rdma_create_qp(rdma
->cm_id
, rdma
->pd
, &attr
);
906 rdma
->qp
= rdma
->cm_id
->qp
;
910 static int qemu_rdma_reg_whole_ram_blocks(RDMAContext
*rdma
)
913 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
915 for (i
= 0; i
< local
->nb_blocks
; i
++) {
918 local
->block
[i
].local_host_addr
,
919 local
->block
[i
].length
,
920 IBV_ACCESS_LOCAL_WRITE
|
921 IBV_ACCESS_REMOTE_WRITE
923 if (!local
->block
[i
].mr
) {
924 perror("Failed to register local dest ram block!\n");
927 rdma
->total_registrations
++;
930 if (i
>= local
->nb_blocks
) {
934 for (i
--; i
>= 0; i
--) {
935 ibv_dereg_mr(local
->block
[i
].mr
);
936 rdma
->total_registrations
--;
944 * Find the ram block that corresponds to the page requested to be
945 * transmitted by QEMU.
947 * Once the block is found, also identify which 'chunk' within that
948 * block that the page belongs to.
950 * This search cannot fail or the migration will fail.
952 static int qemu_rdma_search_ram_block(RDMAContext
*rdma
,
953 uint64_t block_offset
,
956 uint64_t *block_index
,
957 uint64_t *chunk_index
)
959 uint64_t current_addr
= block_offset
+ offset
;
960 RDMALocalBlock
*block
= g_hash_table_lookup(rdma
->blockmap
,
961 (void *) block_offset
);
963 assert(current_addr
>= block
->offset
);
964 assert((current_addr
+ length
) <= (block
->offset
+ block
->length
));
966 *block_index
= block
->index
;
967 *chunk_index
= ram_chunk_index(block
->local_host_addr
,
968 block
->local_host_addr
+ (current_addr
- block
->offset
));
974 * Register a chunk with IB. If the chunk was already registered
975 * previously, then skip.
977 * Also return the keys associated with the registration needed
978 * to perform the actual RDMA operation.
980 static int qemu_rdma_register_and_get_keys(RDMAContext
*rdma
,
981 RDMALocalBlock
*block
, uint8_t *host_addr
,
982 uint32_t *lkey
, uint32_t *rkey
, int chunk
,
983 uint8_t *chunk_start
, uint8_t *chunk_end
)
987 *lkey
= block
->mr
->lkey
;
990 *rkey
= block
->mr
->rkey
;
995 /* allocate memory to store chunk MRs */
997 block
->pmr
= g_malloc0(block
->nb_chunks
* sizeof(struct ibv_mr
*));
1004 * If 'rkey', then we're the destination, so grant access to the source.
1006 * If 'lkey', then we're the source VM, so grant access only to ourselves.
1008 if (!block
->pmr
[chunk
]) {
1009 uint64_t len
= chunk_end
- chunk_start
;
1011 DDPRINTF("Registering %" PRIu64
" bytes @ %p\n",
1014 block
->pmr
[chunk
] = ibv_reg_mr(rdma
->pd
,
1016 (rkey
? (IBV_ACCESS_LOCAL_WRITE
|
1017 IBV_ACCESS_REMOTE_WRITE
) : 0));
1019 if (!block
->pmr
[chunk
]) {
1020 perror("Failed to register chunk!");
1021 fprintf(stderr
, "Chunk details: block: %d chunk index %d"
1022 " start %" PRIu64
" end %" PRIu64
" host %" PRIu64
1023 " local %" PRIu64
" registrations: %d\n",
1024 block
->index
, chunk
, (uint64_t) chunk_start
,
1025 (uint64_t) chunk_end
, (uint64_t) host_addr
,
1026 (uint64_t) block
->local_host_addr
,
1027 rdma
->total_registrations
);
1030 rdma
->total_registrations
++;
1034 *lkey
= block
->pmr
[chunk
]->lkey
;
1037 *rkey
= block
->pmr
[chunk
]->rkey
;
1043 * Register (at connection time) the memory used for control
1046 static int qemu_rdma_reg_control(RDMAContext
*rdma
, int idx
)
1048 rdma
->wr_data
[idx
].control_mr
= ibv_reg_mr(rdma
->pd
,
1049 rdma
->wr_data
[idx
].control
, RDMA_CONTROL_MAX_BUFFER
,
1050 IBV_ACCESS_LOCAL_WRITE
| IBV_ACCESS_REMOTE_WRITE
);
1051 if (rdma
->wr_data
[idx
].control_mr
) {
1052 rdma
->total_registrations
++;
1055 fprintf(stderr
, "qemu_rdma_reg_control failed!\n");
1059 const char *print_wrid(int wrid
)
1061 if (wrid
>= RDMA_WRID_RECV_CONTROL
) {
1062 return wrid_desc
[RDMA_WRID_RECV_CONTROL
];
1064 return wrid_desc
[wrid
];
1068 * RDMA requires memory registration (mlock/pinning), but this is not good for
1071 * In preparation for the future where LRU information or workload-specific
1072 * writable writable working set memory access behavior is available to QEMU
1073 * it would be nice to have in place the ability to UN-register/UN-pin
1074 * particular memory regions from the RDMA hardware when it is determine that
1075 * those regions of memory will likely not be accessed again in the near future.
1077 * While we do not yet have such information right now, the following
1078 * compile-time option allows us to perform a non-optimized version of this
1081 * By uncommenting this option, you will cause *all* RDMA transfers to be
1082 * unregistered immediately after the transfer completes on both sides of the
1083 * connection. This has no effect in 'rdma-pin-all' mode, only regular mode.
1085 * This will have a terrible impact on migration performance, so until future
1086 * workload information or LRU information is available, do not attempt to use
1087 * this feature except for basic testing.
1089 //#define RDMA_UNREGISTRATION_EXAMPLE
1092 * Perform a non-optimized memory unregistration after every transfer
1093 * for demonsration purposes, only if pin-all is not requested.
1095 * Potential optimizations:
1096 * 1. Start a new thread to run this function continuously
1098 - and for receipt of unregister messages
1100 * 3. Use workload hints.
1102 static int qemu_rdma_unregister_waiting(RDMAContext
*rdma
)
1104 while (rdma
->unregistrations
[rdma
->unregister_current
]) {
1106 uint64_t wr_id
= rdma
->unregistrations
[rdma
->unregister_current
];
1108 (wr_id
& RDMA_WRID_CHUNK_MASK
) >> RDMA_WRID_CHUNK_SHIFT
;
1110 (wr_id
& RDMA_WRID_BLOCK_MASK
) >> RDMA_WRID_BLOCK_SHIFT
;
1111 RDMALocalBlock
*block
=
1112 &(rdma
->local_ram_blocks
.block
[index
]);
1113 RDMARegister reg
= { .current_index
= index
};
1114 RDMAControlHeader resp
= { .type
= RDMA_CONTROL_UNREGISTER_FINISHED
,
1116 RDMAControlHeader head
= { .len
= sizeof(RDMARegister
),
1117 .type
= RDMA_CONTROL_UNREGISTER_REQUEST
,
1121 DDPRINTF("Processing unregister for chunk: %" PRIu64
1122 " at position %d\n", chunk
, rdma
->unregister_current
);
1124 rdma
->unregistrations
[rdma
->unregister_current
] = 0;
1125 rdma
->unregister_current
++;
1127 if (rdma
->unregister_current
== RDMA_SIGNALED_SEND_MAX
) {
1128 rdma
->unregister_current
= 0;
1133 * Unregistration is speculative (because migration is single-threaded
1134 * and we cannot break the protocol's inifinband message ordering).
1135 * Thus, if the memory is currently being used for transmission,
1136 * then abort the attempt to unregister and try again
1137 * later the next time a completion is received for this memory.
1139 clear_bit(chunk
, block
->unregister_bitmap
);
1141 if (test_bit(chunk
, block
->transit_bitmap
)) {
1142 DDPRINTF("Cannot unregister inflight chunk: %" PRIu64
"\n", chunk
);
1146 DDPRINTF("Sending unregister for chunk: %" PRIu64
"\n", chunk
);
1148 ret
= ibv_dereg_mr(block
->pmr
[chunk
]);
1149 block
->pmr
[chunk
] = NULL
;
1150 block
->remote_keys
[chunk
] = 0;
1153 perror("unregistration chunk failed");
1156 rdma
->total_registrations
--;
1158 reg
.key
.chunk
= chunk
;
1159 register_to_network(®
);
1160 ret
= qemu_rdma_exchange_send(rdma
, &head
, (uint8_t *) ®
,
1166 DDPRINTF("Unregister for chunk: %" PRIu64
" complete.\n", chunk
);
1172 static uint64_t qemu_rdma_make_wrid(uint64_t wr_id
, uint64_t index
,
1175 uint64_t result
= wr_id
& RDMA_WRID_TYPE_MASK
;
1177 result
|= (index
<< RDMA_WRID_BLOCK_SHIFT
);
1178 result
|= (chunk
<< RDMA_WRID_CHUNK_SHIFT
);
1184 * Set bit for unregistration in the next iteration.
1185 * We cannot transmit right here, but will unpin later.
1187 static void qemu_rdma_signal_unregister(RDMAContext
*rdma
, uint64_t index
,
1188 uint64_t chunk
, uint64_t wr_id
)
1190 if (rdma
->unregistrations
[rdma
->unregister_next
] != 0) {
1191 fprintf(stderr
, "rdma migration: queue is full!\n");
1193 RDMALocalBlock
*block
= &(rdma
->local_ram_blocks
.block
[index
]);
1195 if (!test_and_set_bit(chunk
, block
->unregister_bitmap
)) {
1196 DDPRINTF("Appending unregister chunk %" PRIu64
1197 " at position %d\n", chunk
, rdma
->unregister_next
);
1199 rdma
->unregistrations
[rdma
->unregister_next
++] =
1200 qemu_rdma_make_wrid(wr_id
, index
, chunk
);
1202 if (rdma
->unregister_next
== RDMA_SIGNALED_SEND_MAX
) {
1203 rdma
->unregister_next
= 0;
1206 DDPRINTF("Unregister chunk %" PRIu64
" already in queue.\n",
1213 * Consult the connection manager to see a work request
1214 * (of any kind) has completed.
1215 * Return the work request ID that completed.
1217 static uint64_t qemu_rdma_poll(RDMAContext
*rdma
, uint64_t *wr_id_out
)
1223 ret
= ibv_poll_cq(rdma
->cq
, 1, &wc
);
1226 *wr_id_out
= RDMA_WRID_NONE
;
1231 fprintf(stderr
, "ibv_poll_cq return %d!\n", ret
);
1235 wr_id
= wc
.wr_id
& RDMA_WRID_TYPE_MASK
;
1237 if (wc
.status
!= IBV_WC_SUCCESS
) {
1238 fprintf(stderr
, "ibv_poll_cq wc.status=%d %s!\n",
1239 wc
.status
, ibv_wc_status_str(wc
.status
));
1240 fprintf(stderr
, "ibv_poll_cq wrid=%s!\n", wrid_desc
[wr_id
]);
1245 if (rdma
->control_ready_expected
&&
1246 (wr_id
>= RDMA_WRID_RECV_CONTROL
)) {
1247 DDDPRINTF("completion %s #%" PRId64
" received (%" PRId64
")"
1248 " left %d\n", wrid_desc
[RDMA_WRID_RECV_CONTROL
],
1249 wr_id
- RDMA_WRID_RECV_CONTROL
, wr_id
, rdma
->nb_sent
);
1250 rdma
->control_ready_expected
= 0;
1253 if (wr_id
== RDMA_WRID_RDMA_WRITE
) {
1255 (wc
.wr_id
& RDMA_WRID_CHUNK_MASK
) >> RDMA_WRID_CHUNK_SHIFT
;
1257 (wc
.wr_id
& RDMA_WRID_BLOCK_MASK
) >> RDMA_WRID_BLOCK_SHIFT
;
1258 RDMALocalBlock
*block
= &(rdma
->local_ram_blocks
.block
[index
]);
1260 DDDPRINTF("completions %s (%" PRId64
") left %d, "
1261 "block %" PRIu64
", chunk: %" PRIu64
" %p %p\n",
1262 print_wrid(wr_id
), wr_id
, rdma
->nb_sent
, index
, chunk
,
1263 block
->local_host_addr
, (void *)block
->remote_host_addr
);
1265 clear_bit(chunk
, block
->transit_bitmap
);
1267 if (rdma
->nb_sent
> 0) {
1271 if (!rdma
->pin_all
) {
1273 * FYI: If one wanted to signal a specific chunk to be unregistered
1274 * using LRU or workload-specific information, this is the function
1275 * you would call to do so. That chunk would then get asynchronously
1276 * unregistered later.
1278 #ifdef RDMA_UNREGISTRATION_EXAMPLE
1279 qemu_rdma_signal_unregister(rdma
, index
, chunk
, wc
.wr_id
);
1283 DDDPRINTF("other completion %s (%" PRId64
") received left %d\n",
1284 print_wrid(wr_id
), wr_id
, rdma
->nb_sent
);
1287 *wr_id_out
= wc
.wr_id
;
1293 * Block until the next work request has completed.
1295 * First poll to see if a work request has already completed,
1298 * If we encounter completed work requests for IDs other than
1299 * the one we're interested in, then that's generally an error.
1301 * The only exception is actual RDMA Write completions. These
1302 * completions only need to be recorded, but do not actually
1303 * need further processing.
1305 static int qemu_rdma_block_for_wrid(RDMAContext
*rdma
, int wrid_requested
)
1307 int num_cq_events
= 0, ret
= 0;
1310 uint64_t wr_id
= RDMA_WRID_NONE
, wr_id_in
;
1312 if (ibv_req_notify_cq(rdma
->cq
, 0)) {
1316 while (wr_id
!= wrid_requested
) {
1317 ret
= qemu_rdma_poll(rdma
, &wr_id_in
);
1322 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
1324 if (wr_id
== RDMA_WRID_NONE
) {
1327 if (wr_id
!= wrid_requested
) {
1328 DDDPRINTF("A Wanted wrid %s (%d) but got %s (%" PRIu64
")\n",
1329 print_wrid(wrid_requested
),
1330 wrid_requested
, print_wrid(wr_id
), wr_id
);
1334 if (wr_id
== wrid_requested
) {
1340 * Coroutine doesn't start until process_incoming_migration()
1341 * so don't yield unless we know we're running inside of a coroutine.
1343 if (rdma
->migration_started_on_destination
) {
1344 yield_until_fd_readable(rdma
->comp_channel
->fd
);
1347 if (ibv_get_cq_event(rdma
->comp_channel
, &cq
, &cq_ctx
)) {
1348 perror("ibv_get_cq_event");
1349 goto err_block_for_wrid
;
1354 if (ibv_req_notify_cq(cq
, 0)) {
1355 goto err_block_for_wrid
;
1358 while (wr_id
!= wrid_requested
) {
1359 ret
= qemu_rdma_poll(rdma
, &wr_id_in
);
1361 goto err_block_for_wrid
;
1364 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
1366 if (wr_id
== RDMA_WRID_NONE
) {
1369 if (wr_id
!= wrid_requested
) {
1370 DDDPRINTF("B Wanted wrid %s (%d) but got %s (%" PRIu64
")\n",
1371 print_wrid(wrid_requested
), wrid_requested
,
1372 print_wrid(wr_id
), wr_id
);
1376 if (wr_id
== wrid_requested
) {
1377 goto success_block_for_wrid
;
1381 success_block_for_wrid
:
1382 if (num_cq_events
) {
1383 ibv_ack_cq_events(cq
, num_cq_events
);
1388 if (num_cq_events
) {
1389 ibv_ack_cq_events(cq
, num_cq_events
);
1395 * Post a SEND message work request for the control channel
1396 * containing some data and block until the post completes.
1398 static int qemu_rdma_post_send_control(RDMAContext
*rdma
, uint8_t *buf
,
1399 RDMAControlHeader
*head
)
1402 RDMAWorkRequestData
*wr
= &rdma
->wr_data
[RDMA_WRID_MAX
];
1403 struct ibv_send_wr
*bad_wr
;
1404 struct ibv_sge sge
= {
1405 .addr
= (uint64_t)(wr
->control
),
1406 .length
= head
->len
+ sizeof(RDMAControlHeader
),
1407 .lkey
= wr
->control_mr
->lkey
,
1409 struct ibv_send_wr send_wr
= {
1410 .wr_id
= RDMA_WRID_SEND_CONTROL
,
1411 .opcode
= IBV_WR_SEND
,
1412 .send_flags
= IBV_SEND_SIGNALED
,
1417 DDDPRINTF("CONTROL: sending %s..\n", control_desc
[head
->type
]);
1420 * We don't actually need to do a memcpy() in here if we used
1421 * the "sge" properly, but since we're only sending control messages
1422 * (not RAM in a performance-critical path), then its OK for now.
1424 * The copy makes the RDMAControlHeader simpler to manipulate
1425 * for the time being.
1427 memcpy(wr
->control
, head
, sizeof(RDMAControlHeader
));
1428 control_to_network((void *) wr
->control
);
1431 memcpy(wr
->control
+ sizeof(RDMAControlHeader
), buf
, head
->len
);
1435 if (ibv_post_send(rdma
->qp
, &send_wr
, &bad_wr
)) {
1440 fprintf(stderr
, "Failed to use post IB SEND for control!\n");
1444 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_SEND_CONTROL
);
1446 fprintf(stderr
, "rdma migration: send polling control error!\n");
1453 * Post a RECV work request in anticipation of some future receipt
1454 * of data on the control channel.
1456 static int qemu_rdma_post_recv_control(RDMAContext
*rdma
, int idx
)
1458 struct ibv_recv_wr
*bad_wr
;
1459 struct ibv_sge sge
= {
1460 .addr
= (uint64_t)(rdma
->wr_data
[idx
].control
),
1461 .length
= RDMA_CONTROL_MAX_BUFFER
,
1462 .lkey
= rdma
->wr_data
[idx
].control_mr
->lkey
,
1465 struct ibv_recv_wr recv_wr
= {
1466 .wr_id
= RDMA_WRID_RECV_CONTROL
+ idx
,
1472 if (ibv_post_recv(rdma
->qp
, &recv_wr
, &bad_wr
)) {
1480 * Block and wait for a RECV control channel message to arrive.
1482 static int qemu_rdma_exchange_get_response(RDMAContext
*rdma
,
1483 RDMAControlHeader
*head
, int expecting
, int idx
)
1485 int ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RECV_CONTROL
+ idx
);
1488 fprintf(stderr
, "rdma migration: recv polling control error!\n");
1492 network_to_control((void *) rdma
->wr_data
[idx
].control
);
1493 memcpy(head
, rdma
->wr_data
[idx
].control
, sizeof(RDMAControlHeader
));
1495 DDDPRINTF("CONTROL: %s receiving...\n", control_desc
[expecting
]);
1497 if (expecting
== RDMA_CONTROL_NONE
) {
1498 DDDPRINTF("Surprise: got %s (%d)\n",
1499 control_desc
[head
->type
], head
->type
);
1500 } else if (head
->type
!= expecting
|| head
->type
== RDMA_CONTROL_ERROR
) {
1501 fprintf(stderr
, "Was expecting a %s (%d) control message"
1502 ", but got: %s (%d), length: %d\n",
1503 control_desc
[expecting
], expecting
,
1504 control_desc
[head
->type
], head
->type
, head
->len
);
1512 * When a RECV work request has completed, the work request's
1513 * buffer is pointed at the header.
1515 * This will advance the pointer to the data portion
1516 * of the control message of the work request's buffer that
1517 * was populated after the work request finished.
1519 static void qemu_rdma_move_header(RDMAContext
*rdma
, int idx
,
1520 RDMAControlHeader
*head
)
1522 rdma
->wr_data
[idx
].control_len
= head
->len
;
1523 rdma
->wr_data
[idx
].control_curr
=
1524 rdma
->wr_data
[idx
].control
+ sizeof(RDMAControlHeader
);
1528 * This is an 'atomic' high-level operation to deliver a single, unified
1529 * control-channel message.
1531 * Additionally, if the user is expecting some kind of reply to this message,
1532 * they can request a 'resp' response message be filled in by posting an
1533 * additional work request on behalf of the user and waiting for an additional
1536 * The extra (optional) response is used during registration to us from having
1537 * to perform an *additional* exchange of message just to provide a response by
1538 * instead piggy-backing on the acknowledgement.
1540 static int qemu_rdma_exchange_send(RDMAContext
*rdma
, RDMAControlHeader
*head
,
1541 uint8_t *data
, RDMAControlHeader
*resp
,
1543 int (*callback
)(RDMAContext
*rdma
))
1548 * Wait until the dest is ready before attempting to deliver the message
1549 * by waiting for a READY message.
1551 if (rdma
->control_ready_expected
) {
1552 RDMAControlHeader resp
;
1553 ret
= qemu_rdma_exchange_get_response(rdma
,
1554 &resp
, RDMA_CONTROL_READY
, RDMA_WRID_READY
);
1561 * If the user is expecting a response, post a WR in anticipation of it.
1564 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_DATA
);
1566 fprintf(stderr
, "rdma migration: error posting"
1567 " extra control recv for anticipated result!");
1573 * Post a WR to replace the one we just consumed for the READY message.
1575 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
1577 fprintf(stderr
, "rdma migration: error posting first control recv!");
1582 * Deliver the control message that was requested.
1584 ret
= qemu_rdma_post_send_control(rdma
, data
, head
);
1587 fprintf(stderr
, "Failed to send control buffer!\n");
1592 * If we're expecting a response, block and wait for it.
1596 DDPRINTF("Issuing callback before receiving response...\n");
1597 ret
= callback(rdma
);
1603 DDPRINTF("Waiting for response %s\n", control_desc
[resp
->type
]);
1604 ret
= qemu_rdma_exchange_get_response(rdma
, resp
,
1605 resp
->type
, RDMA_WRID_DATA
);
1611 qemu_rdma_move_header(rdma
, RDMA_WRID_DATA
, resp
);
1613 *resp_idx
= RDMA_WRID_DATA
;
1615 DDPRINTF("Response %s received.\n", control_desc
[resp
->type
]);
1618 rdma
->control_ready_expected
= 1;
1624 * This is an 'atomic' high-level operation to receive a single, unified
1625 * control-channel message.
1627 static int qemu_rdma_exchange_recv(RDMAContext
*rdma
, RDMAControlHeader
*head
,
1630 RDMAControlHeader ready
= {
1632 .type
= RDMA_CONTROL_READY
,
1638 * Inform the source that we're ready to receive a message.
1640 ret
= qemu_rdma_post_send_control(rdma
, NULL
, &ready
);
1643 fprintf(stderr
, "Failed to send control buffer!\n");
1648 * Block and wait for the message.
1650 ret
= qemu_rdma_exchange_get_response(rdma
, head
,
1651 expecting
, RDMA_WRID_READY
);
1657 qemu_rdma_move_header(rdma
, RDMA_WRID_READY
, head
);
1660 * Post a new RECV work request to replace the one we just consumed.
1662 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
1664 fprintf(stderr
, "rdma migration: error posting second control recv!");
1672 * Write an actual chunk of memory using RDMA.
1674 * If we're using dynamic registration on the dest-side, we have to
1675 * send a registration command first.
1677 static int qemu_rdma_write_one(QEMUFile
*f
, RDMAContext
*rdma
,
1678 int current_index
, uint64_t current_addr
,
1682 struct ibv_send_wr send_wr
= { 0 };
1683 struct ibv_send_wr
*bad_wr
;
1684 int reg_result_idx
, ret
, count
= 0;
1685 uint64_t chunk
, chunks
;
1686 uint8_t *chunk_start
, *chunk_end
;
1687 RDMALocalBlock
*block
= &(rdma
->local_ram_blocks
.block
[current_index
]);
1689 RDMARegisterResult
*reg_result
;
1690 RDMAControlHeader resp
= { .type
= RDMA_CONTROL_REGISTER_RESULT
};
1691 RDMAControlHeader head
= { .len
= sizeof(RDMARegister
),
1692 .type
= RDMA_CONTROL_REGISTER_REQUEST
,
1697 sge
.addr
= (uint64_t)(block
->local_host_addr
+
1698 (current_addr
- block
->offset
));
1699 sge
.length
= length
;
1701 chunk
= ram_chunk_index(block
->local_host_addr
, (uint8_t *) sge
.addr
);
1702 chunk_start
= ram_chunk_start(block
, chunk
);
1704 if (block
->is_ram_block
) {
1705 chunks
= length
/ (1UL << RDMA_REG_CHUNK_SHIFT
);
1707 if (chunks
&& ((length
% (1UL << RDMA_REG_CHUNK_SHIFT
)) == 0)) {
1711 chunks
= block
->length
/ (1UL << RDMA_REG_CHUNK_SHIFT
);
1713 if (chunks
&& ((block
->length
% (1UL << RDMA_REG_CHUNK_SHIFT
)) == 0)) {
1718 DDPRINTF("Writing %" PRIu64
" chunks, (%" PRIu64
" MB)\n",
1719 chunks
+ 1, (chunks
+ 1) * (1UL << RDMA_REG_CHUNK_SHIFT
) / 1024 / 1024);
1721 chunk_end
= ram_chunk_end(block
, chunk
+ chunks
);
1723 if (!rdma
->pin_all
) {
1724 #ifdef RDMA_UNREGISTRATION_EXAMPLE
1725 qemu_rdma_unregister_waiting(rdma
);
1729 while (test_bit(chunk
, block
->transit_bitmap
)) {
1731 DDPRINTF("(%d) Not clobbering: block: %d chunk %" PRIu64
1732 " current %" PRIu64
" len %" PRIu64
" %d %d\n",
1733 count
++, current_index
, chunk
,
1734 sge
.addr
, length
, rdma
->nb_sent
, block
->nb_chunks
);
1736 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RDMA_WRITE
);
1739 fprintf(stderr
, "Failed to Wait for previous write to complete "
1740 "block %d chunk %" PRIu64
1741 " current %" PRIu64
" len %" PRIu64
" %d\n",
1742 current_index
, chunk
, sge
.addr
, length
, rdma
->nb_sent
);
1747 if (!rdma
->pin_all
|| !block
->is_ram_block
) {
1748 if (!block
->remote_keys
[chunk
]) {
1750 * This chunk has not yet been registered, so first check to see
1751 * if the entire chunk is zero. If so, tell the other size to
1752 * memset() + madvise() the entire chunk without RDMA.
1755 if (can_use_buffer_find_nonzero_offset((void *)sge
.addr
, length
)
1756 && buffer_find_nonzero_offset((void *)sge
.addr
,
1757 length
) == length
) {
1758 RDMACompress comp
= {
1759 .offset
= current_addr
,
1761 .block_idx
= current_index
,
1765 head
.len
= sizeof(comp
);
1766 head
.type
= RDMA_CONTROL_COMPRESS
;
1768 DDPRINTF("Entire chunk is zero, sending compress: %"
1770 "bytes, index: %d, offset: %" PRId64
"...\n",
1771 chunk
, sge
.length
, current_index
, current_addr
);
1773 compress_to_network(&comp
);
1774 ret
= qemu_rdma_exchange_send(rdma
, &head
,
1775 (uint8_t *) &comp
, NULL
, NULL
, NULL
);
1781 acct_update_position(f
, sge
.length
, true);
1787 * Otherwise, tell other side to register.
1789 reg
.current_index
= current_index
;
1790 if (block
->is_ram_block
) {
1791 reg
.key
.current_addr
= current_addr
;
1793 reg
.key
.chunk
= chunk
;
1795 reg
.chunks
= chunks
;
1797 DDPRINTF("Sending registration request chunk %" PRIu64
" for %d "
1798 "bytes, index: %d, offset: %" PRId64
"...\n",
1799 chunk
, sge
.length
, current_index
, current_addr
);
1801 register_to_network(®
);
1802 ret
= qemu_rdma_exchange_send(rdma
, &head
, (uint8_t *) ®
,
1803 &resp
, ®_result_idx
, NULL
);
1808 /* try to overlap this single registration with the one we sent. */
1809 if (qemu_rdma_register_and_get_keys(rdma
, block
,
1810 (uint8_t *) sge
.addr
,
1811 &sge
.lkey
, NULL
, chunk
,
1812 chunk_start
, chunk_end
)) {
1813 fprintf(stderr
, "cannot get lkey!\n");
1817 reg_result
= (RDMARegisterResult
*)
1818 rdma
->wr_data
[reg_result_idx
].control_curr
;
1820 network_to_result(reg_result
);
1822 DDPRINTF("Received registration result:"
1823 " my key: %x their key %x, chunk %" PRIu64
"\n",
1824 block
->remote_keys
[chunk
], reg_result
->rkey
, chunk
);
1826 block
->remote_keys
[chunk
] = reg_result
->rkey
;
1827 block
->remote_host_addr
= reg_result
->host_addr
;
1829 /* already registered before */
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");
1839 send_wr
.wr
.rdma
.rkey
= block
->remote_keys
[chunk
];
1841 send_wr
.wr
.rdma
.rkey
= block
->remote_rkey
;
1843 if (qemu_rdma_register_and_get_keys(rdma
, block
, (uint8_t *)sge
.addr
,
1844 &sge
.lkey
, NULL
, chunk
,
1845 chunk_start
, chunk_end
)) {
1846 fprintf(stderr
, "cannot get lkey!\n");
1852 * Encode the ram block index and chunk within this wrid.
1853 * We will use this information at the time of completion
1854 * to figure out which bitmap to check against and then which
1855 * chunk in the bitmap to look for.
1857 send_wr
.wr_id
= qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE
,
1858 current_index
, chunk
);
1860 send_wr
.opcode
= IBV_WR_RDMA_WRITE
;
1861 send_wr
.send_flags
= IBV_SEND_SIGNALED
;
1862 send_wr
.sg_list
= &sge
;
1863 send_wr
.num_sge
= 1;
1864 send_wr
.wr
.rdma
.remote_addr
= block
->remote_host_addr
+
1865 (current_addr
- block
->offset
);
1867 DDDPRINTF("Posting chunk: %" PRIu64
", addr: %lx"
1868 " remote: %lx, bytes %" PRIu32
"\n",
1869 chunk
, sge
.addr
, send_wr
.wr
.rdma
.remote_addr
,
1873 * ibv_post_send() does not return negative error numbers,
1874 * per the specification they are positive - no idea why.
1876 ret
= ibv_post_send(rdma
->qp
, &send_wr
, &bad_wr
);
1878 if (ret
== ENOMEM
) {
1879 DDPRINTF("send queue is full. wait a little....\n");
1880 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RDMA_WRITE
);
1882 fprintf(stderr
, "rdma migration: failed to make "
1883 "room in full send queue! %d\n", ret
);
1889 } else if (ret
> 0) {
1890 perror("rdma migration: post rdma write failed");
1894 set_bit(chunk
, block
->transit_bitmap
);
1895 acct_update_position(f
, sge
.length
, false);
1896 rdma
->total_writes
++;
1902 * Push out any unwritten RDMA operations.
1904 * We support sending out multiple chunks at the same time.
1905 * Not all of them need to get signaled in the completion queue.
1907 static int qemu_rdma_write_flush(QEMUFile
*f
, RDMAContext
*rdma
)
1911 if (!rdma
->current_length
) {
1915 ret
= qemu_rdma_write_one(f
, rdma
,
1916 rdma
->current_index
, rdma
->current_addr
, rdma
->current_length
);
1924 DDDPRINTF("sent total: %d\n", rdma
->nb_sent
);
1927 rdma
->current_length
= 0;
1928 rdma
->current_addr
= 0;
1933 static inline int qemu_rdma_buffer_mergable(RDMAContext
*rdma
,
1934 uint64_t offset
, uint64_t len
)
1936 RDMALocalBlock
*block
=
1937 &(rdma
->local_ram_blocks
.block
[rdma
->current_index
]);
1938 uint8_t *host_addr
= block
->local_host_addr
+ (offset
- block
->offset
);
1939 uint8_t *chunk_end
= ram_chunk_end(block
, rdma
->current_chunk
);
1941 if (rdma
->current_length
== 0) {
1946 * Only merge into chunk sequentially.
1948 if (offset
!= (rdma
->current_addr
+ rdma
->current_length
)) {
1952 if (rdma
->current_index
< 0) {
1956 if (offset
< block
->offset
) {
1960 if ((offset
+ len
) > (block
->offset
+ block
->length
)) {
1964 if (rdma
->current_chunk
< 0) {
1968 if ((host_addr
+ len
) > chunk_end
) {
1976 * We're not actually writing here, but doing three things:
1978 * 1. Identify the chunk the buffer belongs to.
1979 * 2. If the chunk is full or the buffer doesn't belong to the current
1980 * chunk, then start a new chunk and flush() the old chunk.
1981 * 3. To keep the hardware busy, we also group chunks into batches
1982 * and only require that a batch gets acknowledged in the completion
1983 * qeueue instead of each individual chunk.
1985 static int qemu_rdma_write(QEMUFile
*f
, RDMAContext
*rdma
,
1986 uint64_t block_offset
, uint64_t offset
,
1989 uint64_t current_addr
= block_offset
+ offset
;
1990 uint64_t index
= rdma
->current_index
;
1991 uint64_t chunk
= rdma
->current_chunk
;
1994 /* If we cannot merge it, we flush the current buffer first. */
1995 if (!qemu_rdma_buffer_mergable(rdma
, current_addr
, len
)) {
1996 ret
= qemu_rdma_write_flush(f
, rdma
);
2000 rdma
->current_length
= 0;
2001 rdma
->current_addr
= current_addr
;
2003 ret
= qemu_rdma_search_ram_block(rdma
, block_offset
,
2004 offset
, len
, &index
, &chunk
);
2006 fprintf(stderr
, "ram block search failed\n");
2009 rdma
->current_index
= index
;
2010 rdma
->current_chunk
= chunk
;
2014 rdma
->current_length
+= len
;
2016 /* flush it if buffer is too large */
2017 if (rdma
->current_length
>= RDMA_MERGE_MAX
) {
2018 return qemu_rdma_write_flush(f
, rdma
);
2024 static void qemu_rdma_cleanup(RDMAContext
*rdma
)
2026 struct rdma_cm_event
*cm_event
;
2030 if (rdma
->error_state
) {
2031 RDMAControlHeader head
= { .len
= 0,
2032 .type
= RDMA_CONTROL_ERROR
,
2035 fprintf(stderr
, "Early error. Sending error.\n");
2036 qemu_rdma_post_send_control(rdma
, NULL
, &head
);
2039 ret
= rdma_disconnect(rdma
->cm_id
);
2041 DDPRINTF("waiting for disconnect\n");
2042 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
2044 rdma_ack_cm_event(cm_event
);
2047 DDPRINTF("Disconnected.\n");
2051 g_free(rdma
->block
);
2054 for (idx
= 0; idx
<= RDMA_WRID_MAX
; idx
++) {
2055 if (rdma
->wr_data
[idx
].control_mr
) {
2056 rdma
->total_registrations
--;
2057 ibv_dereg_mr(rdma
->wr_data
[idx
].control_mr
);
2059 rdma
->wr_data
[idx
].control_mr
= NULL
;
2062 if (rdma
->local_ram_blocks
.block
) {
2063 while (rdma
->local_ram_blocks
.nb_blocks
) {
2064 __qemu_rdma_delete_block(rdma
,
2065 rdma
->local_ram_blocks
.block
->offset
);
2070 ibv_destroy_qp(rdma
->qp
);
2074 ibv_destroy_cq(rdma
->cq
);
2077 if (rdma
->comp_channel
) {
2078 ibv_destroy_comp_channel(rdma
->comp_channel
);
2079 rdma
->comp_channel
= NULL
;
2082 ibv_dealloc_pd(rdma
->pd
);
2085 if (rdma
->listen_id
) {
2086 rdma_destroy_id(rdma
->listen_id
);
2087 rdma
->listen_id
= NULL
;
2090 rdma_destroy_id(rdma
->cm_id
);
2093 if (rdma
->channel
) {
2094 rdma_destroy_event_channel(rdma
->channel
);
2095 rdma
->channel
= NULL
;
2100 static int qemu_rdma_source_init(RDMAContext
*rdma
, Error
**errp
, bool pin_all
)
2103 Error
*local_err
= NULL
, **temp
= &local_err
;
2106 * Will be validated against destination's actual capabilities
2107 * after the connect() completes.
2109 rdma
->pin_all
= pin_all
;
2111 ret
= qemu_rdma_resolve_host(rdma
, temp
);
2113 goto err_rdma_source_init
;
2116 ret
= qemu_rdma_alloc_pd_cq(rdma
);
2118 ERROR(temp
, "rdma migration: error allocating pd and cq! Your mlock()"
2119 " limits may be too low. Please check $ ulimit -a # and "
2120 "search for 'ulimit -l' in the output\n");
2121 goto err_rdma_source_init
;
2124 ret
= qemu_rdma_alloc_qp(rdma
);
2126 ERROR(temp
, "rdma migration: error allocating qp!\n");
2127 goto err_rdma_source_init
;
2130 ret
= qemu_rdma_init_ram_blocks(rdma
);
2132 ERROR(temp
, "rdma migration: error initializing ram blocks!\n");
2133 goto err_rdma_source_init
;
2136 for (idx
= 0; idx
<= RDMA_WRID_MAX
; idx
++) {
2137 ret
= qemu_rdma_reg_control(rdma
, idx
);
2139 ERROR(temp
, "rdma migration: error registering %d control!\n",
2141 goto err_rdma_source_init
;
2147 err_rdma_source_init
:
2148 error_propagate(errp
, local_err
);
2149 qemu_rdma_cleanup(rdma
);
2153 static int qemu_rdma_connect(RDMAContext
*rdma
, Error
**errp
)
2155 RDMACapabilities cap
= {
2156 .version
= RDMA_CONTROL_VERSION_CURRENT
,
2159 struct rdma_conn_param conn_param
= { .initiator_depth
= 2,
2161 .private_data
= &cap
,
2162 .private_data_len
= sizeof(cap
),
2164 struct rdma_cm_event
*cm_event
;
2168 * Only negotiate the capability with destination if the user
2169 * on the source first requested the capability.
2171 if (rdma
->pin_all
) {
2172 DPRINTF("Server pin-all memory requested.\n");
2173 cap
.flags
|= RDMA_CAPABILITY_PIN_ALL
;
2176 caps_to_network(&cap
);
2178 ret
= rdma_connect(rdma
->cm_id
, &conn_param
);
2180 perror("rdma_connect");
2181 ERROR(errp
, "connecting to destination!\n");
2182 rdma_destroy_id(rdma
->cm_id
);
2184 goto err_rdma_source_connect
;
2187 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
2189 perror("rdma_get_cm_event after rdma_connect");
2190 ERROR(errp
, "connecting to destination!\n");
2191 rdma_ack_cm_event(cm_event
);
2192 rdma_destroy_id(rdma
->cm_id
);
2194 goto err_rdma_source_connect
;
2197 if (cm_event
->event
!= RDMA_CM_EVENT_ESTABLISHED
) {
2198 perror("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect");
2199 ERROR(errp
, "connecting to destination!\n");
2200 rdma_ack_cm_event(cm_event
);
2201 rdma_destroy_id(rdma
->cm_id
);
2203 goto err_rdma_source_connect
;
2206 memcpy(&cap
, cm_event
->param
.conn
.private_data
, sizeof(cap
));
2207 network_to_caps(&cap
);
2210 * Verify that the *requested* capabilities are supported by the destination
2211 * and disable them otherwise.
2213 if (rdma
->pin_all
&& !(cap
.flags
& RDMA_CAPABILITY_PIN_ALL
)) {
2214 ERROR(errp
, "Server cannot support pinning all memory. "
2215 "Will register memory dynamically.\n");
2216 rdma
->pin_all
= false;
2219 DPRINTF("Pin all memory: %s\n", rdma
->pin_all
? "enabled" : "disabled");
2221 rdma_ack_cm_event(cm_event
);
2223 ret
= qemu_rdma_post_recv_control(rdma
, 0);
2225 ERROR(errp
, "posting second control recv!\n");
2226 goto err_rdma_source_connect
;
2229 rdma
->control_ready_expected
= 1;
2233 err_rdma_source_connect
:
2234 qemu_rdma_cleanup(rdma
);
2238 static int qemu_rdma_dest_init(RDMAContext
*rdma
, Error
**errp
)
2240 int ret
= -EINVAL
, idx
;
2241 int af
= rdma
->ipv6
? PF_INET6
: PF_INET
;
2242 struct sockaddr_in sin
;
2243 struct rdma_cm_id
*listen_id
;
2244 char ip
[40] = "unknown";
2245 struct addrinfo
*res
;
2248 for (idx
= 0; idx
<= RDMA_WRID_MAX
; idx
++) {
2249 rdma
->wr_data
[idx
].control_len
= 0;
2250 rdma
->wr_data
[idx
].control_curr
= NULL
;
2253 if (rdma
->host
== NULL
) {
2254 ERROR(errp
, "RDMA host is not set!\n");
2255 rdma
->error_state
= -EINVAL
;
2258 /* create CM channel */
2259 rdma
->channel
= rdma_create_event_channel();
2260 if (!rdma
->channel
) {
2261 ERROR(errp
, "could not create rdma event channel\n");
2262 rdma
->error_state
= -EINVAL
;
2267 ret
= rdma_create_id(rdma
->channel
, &listen_id
, NULL
, RDMA_PS_TCP
);
2269 ERROR(errp
, "could not create cm_id!\n");
2270 goto err_dest_init_create_listen_id
;
2273 memset(&sin
, 0, sizeof(sin
));
2274 sin
.sin_family
= af
;
2275 sin
.sin_port
= htons(rdma
->port
);
2276 snprintf(port_str
, 16, "%d", rdma
->port
);
2277 port_str
[15] = '\0';
2279 if (rdma
->host
&& strcmp("", rdma
->host
)) {
2280 ret
= getaddrinfo(rdma
->host
, port_str
, NULL
, &res
);
2282 ERROR(errp
, "could not getaddrinfo address %s\n", rdma
->host
);
2283 goto err_dest_init_bind_addr
;
2287 inet_ntop(af
, &((struct sockaddr_in
*) res
->ai_addr
)->sin_addr
,
2290 ERROR(errp
, "migration host and port not specified!\n");
2292 goto err_dest_init_bind_addr
;
2295 DPRINTF("%s => %s\n", rdma
->host
, ip
);
2297 ret
= rdma_bind_addr(listen_id
, res
->ai_addr
);
2299 ERROR(errp
, "Error: could not rdma_bind_addr!\n");
2300 goto err_dest_init_bind_addr
;
2303 rdma
->listen_id
= listen_id
;
2304 qemu_rdma_dump_gid("dest_init", listen_id
);
2307 err_dest_init_bind_addr
:
2308 rdma_destroy_id(listen_id
);
2309 err_dest_init_create_listen_id
:
2310 rdma_destroy_event_channel(rdma
->channel
);
2311 rdma
->channel
= NULL
;
2312 rdma
->error_state
= ret
;
2317 static void *qemu_rdma_data_init(const char *host_port
, Error
**errp
)
2319 RDMAContext
*rdma
= NULL
;
2320 InetSocketAddress
*addr
;
2323 rdma
= g_malloc0(sizeof(RDMAContext
));
2324 memset(rdma
, 0, sizeof(RDMAContext
));
2325 rdma
->current_index
= -1;
2326 rdma
->current_chunk
= -1;
2328 addr
= inet_parse(host_port
, NULL
);
2330 rdma
->port
= atoi(addr
->port
);
2331 rdma
->host
= g_strdup(addr
->host
);
2332 rdma
->ipv6
= addr
->ipv6
;
2334 ERROR(errp
, "bad RDMA migration address '%s'", host_port
);
2344 * QEMUFile interface to the control channel.
2345 * SEND messages for control only.
2346 * pc.ram is handled with regular RDMA messages.
2348 static int qemu_rdma_put_buffer(void *opaque
, const uint8_t *buf
,
2349 int64_t pos
, int size
)
2351 QEMUFileRDMA
*r
= opaque
;
2352 QEMUFile
*f
= r
->file
;
2353 RDMAContext
*rdma
= r
->rdma
;
2354 size_t remaining
= size
;
2355 uint8_t * data
= (void *) buf
;
2358 CHECK_ERROR_STATE();
2361 * Push out any writes that
2362 * we're queued up for pc.ram.
2364 ret
= qemu_rdma_write_flush(f
, rdma
);
2366 rdma
->error_state
= ret
;
2371 RDMAControlHeader head
;
2373 r
->len
= MIN(remaining
, RDMA_SEND_INCREMENT
);
2374 remaining
-= r
->len
;
2377 head
.type
= RDMA_CONTROL_QEMU_FILE
;
2379 ret
= qemu_rdma_exchange_send(rdma
, &head
, data
, NULL
, NULL
, NULL
);
2382 rdma
->error_state
= ret
;
2392 static size_t qemu_rdma_fill(RDMAContext
*rdma
, uint8_t *buf
,
2397 if (rdma
->wr_data
[idx
].control_len
) {
2398 DDDPRINTF("RDMA %" PRId64
" of %d bytes already in buffer\n",
2399 rdma
->wr_data
[idx
].control_len
, size
);
2401 len
= MIN(size
, rdma
->wr_data
[idx
].control_len
);
2402 memcpy(buf
, rdma
->wr_data
[idx
].control_curr
, len
);
2403 rdma
->wr_data
[idx
].control_curr
+= len
;
2404 rdma
->wr_data
[idx
].control_len
-= len
;
2411 * QEMUFile interface to the control channel.
2412 * RDMA links don't use bytestreams, so we have to
2413 * return bytes to QEMUFile opportunistically.
2415 static int qemu_rdma_get_buffer(void *opaque
, uint8_t *buf
,
2416 int64_t pos
, int size
)
2418 QEMUFileRDMA
*r
= opaque
;
2419 RDMAContext
*rdma
= r
->rdma
;
2420 RDMAControlHeader head
;
2423 CHECK_ERROR_STATE();
2426 * First, we hold on to the last SEND message we
2427 * were given and dish out the bytes until we run
2430 r
->len
= qemu_rdma_fill(r
->rdma
, buf
, size
, 0);
2436 * Once we run out, we block and wait for another
2437 * SEND message to arrive.
2439 ret
= qemu_rdma_exchange_recv(rdma
, &head
, RDMA_CONTROL_QEMU_FILE
);
2442 rdma
->error_state
= ret
;
2447 * SEND was received with new bytes, now try again.
2449 return qemu_rdma_fill(r
->rdma
, buf
, size
, 0);
2453 * Block until all the outstanding chunks have been delivered by the hardware.
2455 static int qemu_rdma_drain_cq(QEMUFile
*f
, RDMAContext
*rdma
)
2459 if (qemu_rdma_write_flush(f
, rdma
) < 0) {
2463 while (rdma
->nb_sent
) {
2464 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RDMA_WRITE
);
2466 fprintf(stderr
, "rdma migration: complete polling error!\n");
2471 qemu_rdma_unregister_waiting(rdma
);
2476 static int qemu_rdma_close(void *opaque
)
2478 DPRINTF("Shutting down connection.\n");
2479 QEMUFileRDMA
*r
= opaque
;
2481 qemu_rdma_cleanup(r
->rdma
);
2491 * This means that 'block_offset' is a full virtual address that does not
2492 * belong to a RAMBlock of the virtual machine and instead
2493 * represents a private malloc'd memory area that the caller wishes to
2497 * Offset is an offset to be added to block_offset and used
2498 * to also lookup the corresponding RAMBlock.
2501 * Initiate an transfer this size.
2504 * A 'hint' or 'advice' that means that we wish to speculatively
2505 * and asynchronously unregister this memory. In this case, there is no
2506 * guarantee that the unregister will actually happen, for example,
2507 * if the memory is being actively transmitted. Additionally, the memory
2508 * may be re-registered at any future time if a write within the same
2509 * chunk was requested again, even if you attempted to unregister it
2512 * @size < 0 : TODO, not yet supported
2513 * Unregister the memory NOW. This means that the caller does not
2514 * expect there to be any future RDMA transfers and we just want to clean
2515 * things up. This is used in case the upper layer owns the memory and
2516 * cannot wait for qemu_fclose() to occur.
2518 * @bytes_sent : User-specificed pointer to indicate how many bytes were
2519 * sent. Usually, this will not be more than a few bytes of
2520 * the protocol because most transfers are sent asynchronously.
2522 static size_t qemu_rdma_save_page(QEMUFile
*f
, void *opaque
,
2523 ram_addr_t block_offset
, ram_addr_t offset
,
2524 size_t size
, int *bytes_sent
)
2526 QEMUFileRDMA
*rfile
= opaque
;
2527 RDMAContext
*rdma
= rfile
->rdma
;
2530 CHECK_ERROR_STATE();
2536 * Add this page to the current 'chunk'. If the chunk
2537 * is full, or the page doen't belong to the current chunk,
2538 * an actual RDMA write will occur and a new chunk will be formed.
2540 ret
= qemu_rdma_write(f
, rdma
, block_offset
, offset
, size
);
2542 fprintf(stderr
, "rdma migration: write error! %d\n", ret
);
2547 * We always return 1 bytes because the RDMA
2548 * protocol is completely asynchronous. We do not yet know
2549 * whether an identified chunk is zero or not because we're
2550 * waiting for other pages to potentially be merged with
2551 * the current chunk. So, we have to call qemu_update_position()
2552 * later on when the actual write occurs.
2558 uint64_t index
, chunk
;
2560 /* TODO: Change QEMUFileOps prototype to be signed: size_t => long
2562 ret = qemu_rdma_drain_cq(f, rdma);
2564 fprintf(stderr, "rdma: failed to synchronously drain"
2565 " completion queue before unregistration.\n");
2571 ret
= qemu_rdma_search_ram_block(rdma
, block_offset
,
2572 offset
, size
, &index
, &chunk
);
2575 fprintf(stderr
, "ram block search failed\n");
2579 qemu_rdma_signal_unregister(rdma
, index
, chunk
, 0);
2582 * TODO: Synchronous, guaranteed unregistration (should not occur during
2583 * fast-path). Otherwise, unregisters will process on the next call to
2584 * qemu_rdma_drain_cq()
2586 qemu_rdma_unregister_waiting(rdma);
2592 * Drain the Completion Queue if possible, but do not block,
2595 * If nothing to poll, the end of the iteration will do this
2596 * again to make sure we don't overflow the request queue.
2599 uint64_t wr_id
, wr_id_in
;
2600 int ret
= qemu_rdma_poll(rdma
, &wr_id_in
);
2602 fprintf(stderr
, "rdma migration: polling error! %d\n", ret
);
2606 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
2608 if (wr_id
== RDMA_WRID_NONE
) {
2613 return RAM_SAVE_CONTROL_DELAYED
;
2615 rdma
->error_state
= ret
;
2619 static int qemu_rdma_accept(RDMAContext
*rdma
)
2621 RDMACapabilities cap
;
2622 struct rdma_conn_param conn_param
= {
2623 .responder_resources
= 2,
2624 .private_data
= &cap
,
2625 .private_data_len
= sizeof(cap
),
2627 struct rdma_cm_event
*cm_event
;
2628 struct ibv_context
*verbs
;
2632 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
2634 goto err_rdma_dest_wait
;
2637 if (cm_event
->event
!= RDMA_CM_EVENT_CONNECT_REQUEST
) {
2638 rdma_ack_cm_event(cm_event
);
2639 goto err_rdma_dest_wait
;
2642 memcpy(&cap
, cm_event
->param
.conn
.private_data
, sizeof(cap
));
2644 network_to_caps(&cap
);
2646 if (cap
.version
< 1 || cap
.version
> RDMA_CONTROL_VERSION_CURRENT
) {
2647 fprintf(stderr
, "Unknown source RDMA version: %d, bailing...\n",
2649 rdma_ack_cm_event(cm_event
);
2650 goto err_rdma_dest_wait
;
2654 * Respond with only the capabilities this version of QEMU knows about.
2656 cap
.flags
&= known_capabilities
;
2659 * Enable the ones that we do know about.
2660 * Add other checks here as new ones are introduced.
2662 if (cap
.flags
& RDMA_CAPABILITY_PIN_ALL
) {
2663 rdma
->pin_all
= true;
2666 rdma
->cm_id
= cm_event
->id
;
2667 verbs
= cm_event
->id
->verbs
;
2669 rdma_ack_cm_event(cm_event
);
2671 DPRINTF("Memory pin all: %s\n", rdma
->pin_all
? "enabled" : "disabled");
2673 caps_to_network(&cap
);
2675 DPRINTF("verbs context after listen: %p\n", verbs
);
2678 rdma
->verbs
= verbs
;
2679 } else if (rdma
->verbs
!= verbs
) {
2680 fprintf(stderr
, "ibv context not matching %p, %p!\n",
2681 rdma
->verbs
, verbs
);
2682 goto err_rdma_dest_wait
;
2685 qemu_rdma_dump_id("dest_init", verbs
);
2687 ret
= qemu_rdma_alloc_pd_cq(rdma
);
2689 fprintf(stderr
, "rdma migration: error allocating pd and cq!\n");
2690 goto err_rdma_dest_wait
;
2693 ret
= qemu_rdma_alloc_qp(rdma
);
2695 fprintf(stderr
, "rdma migration: error allocating qp!\n");
2696 goto err_rdma_dest_wait
;
2699 ret
= qemu_rdma_init_ram_blocks(rdma
);
2701 fprintf(stderr
, "rdma migration: error initializing ram blocks!\n");
2702 goto err_rdma_dest_wait
;
2705 for (idx
= 0; idx
<= RDMA_WRID_MAX
; idx
++) {
2706 ret
= qemu_rdma_reg_control(rdma
, idx
);
2708 fprintf(stderr
, "rdma: error registering %d control!\n", idx
);
2709 goto err_rdma_dest_wait
;
2713 qemu_set_fd_handler2(rdma
->channel
->fd
, NULL
, NULL
, NULL
, NULL
);
2715 ret
= rdma_accept(rdma
->cm_id
, &conn_param
);
2717 fprintf(stderr
, "rdma_accept returns %d!\n", ret
);
2718 goto err_rdma_dest_wait
;
2721 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
2723 fprintf(stderr
, "rdma_accept get_cm_event failed %d!\n", ret
);
2724 goto err_rdma_dest_wait
;
2727 if (cm_event
->event
!= RDMA_CM_EVENT_ESTABLISHED
) {
2728 fprintf(stderr
, "rdma_accept not event established!\n");
2729 rdma_ack_cm_event(cm_event
);
2730 goto err_rdma_dest_wait
;
2733 rdma_ack_cm_event(cm_event
);
2735 ret
= qemu_rdma_post_recv_control(rdma
, 0);
2737 fprintf(stderr
, "rdma migration: error posting second control recv!\n");
2738 goto err_rdma_dest_wait
;
2741 qemu_rdma_dump_gid("dest_connect", rdma
->cm_id
);
2746 rdma
->error_state
= ret
;
2747 qemu_rdma_cleanup(rdma
);
2752 * During each iteration of the migration, we listen for instructions
2753 * by the source VM to perform dynamic page registrations before they
2754 * can perform RDMA operations.
2756 * We respond with the 'rkey'.
2758 * Keep doing this until the source tells us to stop.
2760 static int qemu_rdma_registration_handle(QEMUFile
*f
, void *opaque
,
2763 RDMAControlHeader reg_resp
= { .len
= sizeof(RDMARegisterResult
),
2764 .type
= RDMA_CONTROL_REGISTER_RESULT
,
2767 RDMAControlHeader unreg_resp
= { .len
= 0,
2768 .type
= RDMA_CONTROL_UNREGISTER_FINISHED
,
2771 RDMAControlHeader blocks
= { .type
= RDMA_CONTROL_RAM_BLOCKS_RESULT
,
2773 QEMUFileRDMA
*rfile
= opaque
;
2774 RDMAContext
*rdma
= rfile
->rdma
;
2775 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
2776 RDMAControlHeader head
;
2777 RDMARegister
*reg
, *registers
;
2779 RDMARegisterResult
*reg_result
;
2780 static RDMARegisterResult results
[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE
];
2781 RDMALocalBlock
*block
;
2788 CHECK_ERROR_STATE();
2791 DDDPRINTF("Waiting for next request %" PRIu64
"...\n", flags
);
2793 ret
= qemu_rdma_exchange_recv(rdma
, &head
, RDMA_CONTROL_NONE
);
2799 if (head
.repeat
> RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE
) {
2800 fprintf(stderr
, "rdma: Too many requests in this message (%d)."
2801 "Bailing.\n", head
.repeat
);
2806 switch (head
.type
) {
2807 case RDMA_CONTROL_COMPRESS
:
2808 comp
= (RDMACompress
*) rdma
->wr_data
[idx
].control_curr
;
2809 network_to_compress(comp
);
2811 DDPRINTF("Zapping zero chunk: %" PRId64
2812 " bytes, index %d, offset %" PRId64
"\n",
2813 comp
->length
, comp
->block_idx
, comp
->offset
);
2814 block
= &(rdma
->local_ram_blocks
.block
[comp
->block_idx
]);
2816 host_addr
= block
->local_host_addr
+
2817 (comp
->offset
- block
->offset
);
2819 ram_handle_compressed(host_addr
, comp
->value
, comp
->length
);
2822 case RDMA_CONTROL_REGISTER_FINISHED
:
2823 DDDPRINTF("Current registrations complete.\n");
2826 case RDMA_CONTROL_RAM_BLOCKS_REQUEST
:
2827 DPRINTF("Initial setup info requested.\n");
2829 if (rdma
->pin_all
) {
2830 ret
= qemu_rdma_reg_whole_ram_blocks(rdma
);
2832 fprintf(stderr
, "rdma migration: error dest "
2833 "registering ram blocks!\n");
2839 * Dest uses this to prepare to transmit the RAMBlock descriptions
2840 * to the source VM after connection setup.
2841 * Both sides use the "remote" structure to communicate and update
2842 * their "local" descriptions with what was sent.
2844 for (i
= 0; i
< local
->nb_blocks
; i
++) {
2845 rdma
->block
[i
].remote_host_addr
=
2846 (uint64_t)(local
->block
[i
].local_host_addr
);
2848 if (rdma
->pin_all
) {
2849 rdma
->block
[i
].remote_rkey
= local
->block
[i
].mr
->rkey
;
2852 rdma
->block
[i
].offset
= local
->block
[i
].offset
;
2853 rdma
->block
[i
].length
= local
->block
[i
].length
;
2855 remote_block_to_network(&rdma
->block
[i
]);
2858 blocks
.len
= rdma
->local_ram_blocks
.nb_blocks
2859 * sizeof(RDMARemoteBlock
);
2862 ret
= qemu_rdma_post_send_control(rdma
,
2863 (uint8_t *) rdma
->block
, &blocks
);
2866 fprintf(stderr
, "rdma migration: error sending remote info!\n");
2871 case RDMA_CONTROL_REGISTER_REQUEST
:
2872 DDPRINTF("There are %d registration requests\n", head
.repeat
);
2874 reg_resp
.repeat
= head
.repeat
;
2875 registers
= (RDMARegister
*) rdma
->wr_data
[idx
].control_curr
;
2877 for (count
= 0; count
< head
.repeat
; count
++) {
2879 uint8_t *chunk_start
, *chunk_end
;
2881 reg
= ®isters
[count
];
2882 network_to_register(reg
);
2884 reg_result
= &results
[count
];
2886 DDPRINTF("Registration request (%d): index %d, current_addr %"
2887 PRIu64
" chunks: %" PRIu64
"\n", count
,
2888 reg
->current_index
, reg
->key
.current_addr
, reg
->chunks
);
2890 block
= &(rdma
->local_ram_blocks
.block
[reg
->current_index
]);
2891 if (block
->is_ram_block
) {
2892 host_addr
= (block
->local_host_addr
+
2893 (reg
->key
.current_addr
- block
->offset
));
2894 chunk
= ram_chunk_index(block
->local_host_addr
,
2895 (uint8_t *) host_addr
);
2897 chunk
= reg
->key
.chunk
;
2898 host_addr
= block
->local_host_addr
+
2899 (reg
->key
.chunk
* (1UL << RDMA_REG_CHUNK_SHIFT
));
2901 chunk_start
= ram_chunk_start(block
, chunk
);
2902 chunk_end
= ram_chunk_end(block
, chunk
+ reg
->chunks
);
2903 if (qemu_rdma_register_and_get_keys(rdma
, block
,
2904 (uint8_t *)host_addr
, NULL
, ®_result
->rkey
,
2905 chunk
, chunk_start
, chunk_end
)) {
2906 fprintf(stderr
, "cannot get rkey!\n");
2911 reg_result
->host_addr
= (uint64_t) block
->local_host_addr
;
2913 DDPRINTF("Registered rkey for this request: %x\n",
2916 result_to_network(reg_result
);
2919 ret
= qemu_rdma_post_send_control(rdma
,
2920 (uint8_t *) results
, ®_resp
);
2923 fprintf(stderr
, "Failed to send control buffer!\n");
2927 case RDMA_CONTROL_UNREGISTER_REQUEST
:
2928 DDPRINTF("There are %d unregistration requests\n", head
.repeat
);
2929 unreg_resp
.repeat
= head
.repeat
;
2930 registers
= (RDMARegister
*) rdma
->wr_data
[idx
].control_curr
;
2932 for (count
= 0; count
< head
.repeat
; count
++) {
2933 reg
= ®isters
[count
];
2934 network_to_register(reg
);
2936 DDPRINTF("Unregistration request (%d): "
2937 " index %d, chunk %" PRIu64
"\n",
2938 count
, reg
->current_index
, reg
->key
.chunk
);
2940 block
= &(rdma
->local_ram_blocks
.block
[reg
->current_index
]);
2942 ret
= ibv_dereg_mr(block
->pmr
[reg
->key
.chunk
]);
2943 block
->pmr
[reg
->key
.chunk
] = NULL
;
2946 perror("rdma unregistration chunk failed");
2951 rdma
->total_registrations
--;
2953 DDPRINTF("Unregistered chunk %" PRIu64
" successfully.\n",
2957 ret
= qemu_rdma_post_send_control(rdma
, NULL
, &unreg_resp
);
2960 fprintf(stderr
, "Failed to send control buffer!\n");
2964 case RDMA_CONTROL_REGISTER_RESULT
:
2965 fprintf(stderr
, "Invalid RESULT message at dest.\n");
2969 fprintf(stderr
, "Unknown control message %s\n",
2970 control_desc
[head
.type
]);
2977 rdma
->error_state
= ret
;
2982 static int qemu_rdma_registration_start(QEMUFile
*f
, void *opaque
,
2985 QEMUFileRDMA
*rfile
= opaque
;
2986 RDMAContext
*rdma
= rfile
->rdma
;
2988 CHECK_ERROR_STATE();
2990 DDDPRINTF("start section: %" PRIu64
"\n", flags
);
2991 qemu_put_be64(f
, RAM_SAVE_FLAG_HOOK
);
2998 * Inform dest that dynamic registrations are done for now.
2999 * First, flush writes, if any.
3001 static int qemu_rdma_registration_stop(QEMUFile
*f
, void *opaque
,
3004 Error
*local_err
= NULL
, **errp
= &local_err
;
3005 QEMUFileRDMA
*rfile
= opaque
;
3006 RDMAContext
*rdma
= rfile
->rdma
;
3007 RDMAControlHeader head
= { .len
= 0, .repeat
= 1 };
3010 CHECK_ERROR_STATE();
3013 ret
= qemu_rdma_drain_cq(f
, rdma
);
3019 if (flags
== RAM_CONTROL_SETUP
) {
3020 RDMAControlHeader resp
= {.type
= RDMA_CONTROL_RAM_BLOCKS_RESULT
};
3021 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
3022 int reg_result_idx
, i
, j
, nb_remote_blocks
;
3024 head
.type
= RDMA_CONTROL_RAM_BLOCKS_REQUEST
;
3025 DPRINTF("Sending registration setup for ram blocks...\n");
3028 * Make sure that we parallelize the pinning on both sides.
3029 * For very large guests, doing this serially takes a really
3030 * long time, so we have to 'interleave' the pinning locally
3031 * with the control messages by performing the pinning on this
3032 * side before we receive the control response from the other
3033 * side that the pinning has completed.
3035 ret
= qemu_rdma_exchange_send(rdma
, &head
, NULL
, &resp
,
3036 ®_result_idx
, rdma
->pin_all
?
3037 qemu_rdma_reg_whole_ram_blocks
: NULL
);
3039 ERROR(errp
, "receiving remote info!\n");
3043 qemu_rdma_move_header(rdma
, reg_result_idx
, &resp
);
3045 rdma
->wr_data
[reg_result_idx
].control_curr
, resp
.len
);
3047 nb_remote_blocks
= resp
.len
/ sizeof(RDMARemoteBlock
);
3050 * The protocol uses two different sets of rkeys (mutually exclusive):
3051 * 1. One key to represent the virtual address of the entire ram block.
3052 * (dynamic chunk registration disabled - pin everything with one rkey.)
3053 * 2. One to represent individual chunks within a ram block.
3054 * (dynamic chunk registration enabled - pin individual chunks.)
3056 * Once the capability is successfully negotiated, the destination transmits
3057 * the keys to use (or sends them later) including the virtual addresses
3058 * and then propagates the remote ram block descriptions to his local copy.
3061 if (local
->nb_blocks
!= nb_remote_blocks
) {
3062 ERROR(errp
, "ram blocks mismatch #1! "
3063 "Your QEMU command line parameters are probably "
3064 "not identical on both the source and destination.\n");
3068 for (i
= 0; i
< nb_remote_blocks
; i
++) {
3069 network_to_remote_block(&rdma
->block
[i
]);
3071 /* search local ram blocks */
3072 for (j
= 0; j
< local
->nb_blocks
; j
++) {
3073 if (rdma
->block
[i
].offset
!= local
->block
[j
].offset
) {
3077 if (rdma
->block
[i
].length
!= local
->block
[j
].length
) {
3078 ERROR(errp
, "ram blocks mismatch #2! "
3079 "Your QEMU command line parameters are probably "
3080 "not identical on both the source and destination.\n");
3083 local
->block
[j
].remote_host_addr
=
3084 rdma
->block
[i
].remote_host_addr
;
3085 local
->block
[j
].remote_rkey
= rdma
->block
[i
].remote_rkey
;
3089 if (j
>= local
->nb_blocks
) {
3090 ERROR(errp
, "ram blocks mismatch #3! "
3091 "Your QEMU command line parameters are probably "
3092 "not identical on both the source and destination.\n");
3098 DDDPRINTF("Sending registration finish %" PRIu64
"...\n", flags
);
3100 head
.type
= RDMA_CONTROL_REGISTER_FINISHED
;
3101 ret
= qemu_rdma_exchange_send(rdma
, &head
, NULL
, NULL
, NULL
, NULL
);
3109 rdma
->error_state
= ret
;
3113 static int qemu_rdma_get_fd(void *opaque
)
3115 QEMUFileRDMA
*rfile
= opaque
;
3116 RDMAContext
*rdma
= rfile
->rdma
;
3118 return rdma
->comp_channel
->fd
;
3121 const QEMUFileOps rdma_read_ops
= {
3122 .get_buffer
= qemu_rdma_get_buffer
,
3123 .get_fd
= qemu_rdma_get_fd
,
3124 .close
= qemu_rdma_close
,
3125 .hook_ram_load
= qemu_rdma_registration_handle
,
3128 const QEMUFileOps rdma_write_ops
= {
3129 .put_buffer
= qemu_rdma_put_buffer
,
3130 .close
= qemu_rdma_close
,
3131 .before_ram_iterate
= qemu_rdma_registration_start
,
3132 .after_ram_iterate
= qemu_rdma_registration_stop
,
3133 .save_page
= qemu_rdma_save_page
,
3136 static void *qemu_fopen_rdma(RDMAContext
*rdma
, const char *mode
)
3138 QEMUFileRDMA
*r
= g_malloc0(sizeof(QEMUFileRDMA
));
3140 if (qemu_file_mode_is_not_valid(mode
)) {
3146 if (mode
[0] == 'w') {
3147 r
->file
= qemu_fopen_ops(r
, &rdma_write_ops
);
3149 r
->file
= qemu_fopen_ops(r
, &rdma_read_ops
);
3155 static void rdma_accept_incoming_migration(void *opaque
)
3157 RDMAContext
*rdma
= opaque
;
3160 Error
*local_err
= NULL
, **errp
= &local_err
;
3162 DPRINTF("Accepting rdma connection...\n");
3163 ret
= qemu_rdma_accept(rdma
);
3166 ERROR(errp
, "RDMA Migration initialization failed!\n");
3170 DPRINTF("Accepted migration\n");
3172 f
= qemu_fopen_rdma(rdma
, "rb");
3174 ERROR(errp
, "could not qemu_fopen_rdma!\n");
3175 qemu_rdma_cleanup(rdma
);
3179 rdma
->migration_started_on_destination
= 1;
3180 process_incoming_migration(f
);
3183 void rdma_start_incoming_migration(const char *host_port
, Error
**errp
)
3187 Error
*local_err
= NULL
;
3189 DPRINTF("Starting RDMA-based incoming migration\n");
3190 rdma
= qemu_rdma_data_init(host_port
, &local_err
);
3196 ret
= qemu_rdma_dest_init(rdma
, &local_err
);
3202 DPRINTF("qemu_rdma_dest_init success\n");
3204 ret
= rdma_listen(rdma
->listen_id
, 5);
3207 ERROR(errp
, "listening on socket!\n");
3211 DPRINTF("rdma_listen success\n");
3213 qemu_set_fd_handler2(rdma
->channel
->fd
, NULL
,
3214 rdma_accept_incoming_migration
, NULL
,
3215 (void *)(intptr_t) rdma
);
3218 error_propagate(errp
, local_err
);
3222 void rdma_start_outgoing_migration(void *opaque
,
3223 const char *host_port
, Error
**errp
)
3225 MigrationState
*s
= opaque
;
3226 Error
*local_err
= NULL
, **temp
= &local_err
;
3227 RDMAContext
*rdma
= qemu_rdma_data_init(host_port
, &local_err
);
3231 ERROR(temp
, "Failed to initialize RDMA data structures! %d\n", ret
);
3235 ret
= qemu_rdma_source_init(rdma
, &local_err
,
3236 s
->enabled_capabilities
[MIGRATION_CAPABILITY_X_RDMA_PIN_ALL
]);
3242 DPRINTF("qemu_rdma_source_init success\n");
3243 ret
= qemu_rdma_connect(rdma
, &local_err
);
3249 DPRINTF("qemu_rdma_source_connect success\n");
3251 s
->file
= qemu_fopen_rdma(rdma
, "wb");
3252 migrate_fd_connect(s
);
3255 error_propagate(errp
, local_err
);
3257 migrate_fd_error(s
);