2 * RDMA protocol and interfaces
4 * Copyright IBM, Corp. 2010-2013
5 * Copyright Red Hat, Inc. 2015-2016
8 * Michael R. Hines <mrhines@us.ibm.com>
9 * Jiuxing Liu <jl@us.ibm.com>
10 * Daniel P. Berrange <berrange@redhat.com>
12 * This work is licensed under the terms of the GNU GPL, version 2 or
13 * later. See the COPYING file in the top-level directory.
17 #include "qemu/osdep.h"
18 #include "qapi/error.h"
19 #include "qemu/cutils.h"
20 #include "exec/target_page.h"
22 #include "migration.h"
23 #include "migration-stats.h"
24 #include "qemu-file.h"
26 #include "qemu/error-report.h"
27 #include "qemu/main-loop.h"
28 #include "qemu/module.h"
30 #include "qemu/sockets.h"
31 #include "qemu/bitmap.h"
32 #include "qemu/coroutine.h"
33 #include "exec/memory.h"
34 #include <sys/socket.h>
36 #include <arpa/inet.h>
37 #include <rdma/rdma_cma.h>
39 #include "qom/object.h"
44 * Print and error on both the Monitor and the Log file.
46 #define ERROR(errp, fmt, ...) \
48 fprintf(stderr, "RDMA ERROR: " fmt "\n", ## __VA_ARGS__); \
49 if (errp && (*(errp) == NULL)) { \
50 error_setg(errp, "RDMA ERROR: " fmt, ## __VA_ARGS__); \
54 #define RDMA_RESOLVE_TIMEOUT_MS 10000
56 /* Do not merge data if larger than this. */
57 #define RDMA_MERGE_MAX (2 * 1024 * 1024)
58 #define RDMA_SIGNALED_SEND_MAX (RDMA_MERGE_MAX / 4096)
60 #define RDMA_REG_CHUNK_SHIFT 20 /* 1 MB */
63 * This is only for non-live state being migrated.
64 * Instead of RDMA_WRITE messages, we use RDMA_SEND
65 * messages for that state, which requires a different
66 * delivery design than main memory.
68 #define RDMA_SEND_INCREMENT 32768
71 * Maximum size infiniband SEND message
73 #define RDMA_CONTROL_MAX_BUFFER (512 * 1024)
74 #define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096
76 #define RDMA_CONTROL_VERSION_CURRENT 1
78 * Capabilities for negotiation.
80 #define RDMA_CAPABILITY_PIN_ALL 0x01
83 * Add the other flags above to this list of known capabilities
84 * as they are introduced.
86 static uint32_t known_capabilities
= RDMA_CAPABILITY_PIN_ALL
;
88 #define CHECK_ERROR_STATE() \
90 if (rdma->error_state) { \
91 if (!rdma->error_reported) { \
92 error_report("RDMA is in an error state waiting migration" \
94 rdma->error_reported = 1; \
96 return rdma->error_state; \
101 * A work request ID is 64-bits and we split up these bits
104 * bits 0-15 : type of control message, 2^16
105 * bits 16-29: ram block index, 2^14
106 * bits 30-63: ram block chunk number, 2^34
108 * The last two bit ranges are only used for RDMA writes,
109 * in order to track their completion and potentially
110 * also track unregistration status of the message.
112 #define RDMA_WRID_TYPE_SHIFT 0UL
113 #define RDMA_WRID_BLOCK_SHIFT 16UL
114 #define RDMA_WRID_CHUNK_SHIFT 30UL
116 #define RDMA_WRID_TYPE_MASK \
117 ((1UL << RDMA_WRID_BLOCK_SHIFT) - 1UL)
119 #define RDMA_WRID_BLOCK_MASK \
120 (~RDMA_WRID_TYPE_MASK & ((1UL << RDMA_WRID_CHUNK_SHIFT) - 1UL))
122 #define RDMA_WRID_CHUNK_MASK (~RDMA_WRID_BLOCK_MASK & ~RDMA_WRID_TYPE_MASK)
125 * RDMA migration protocol:
126 * 1. RDMA Writes (data messages, i.e. RAM)
127 * 2. IB Send/Recv (control channel messages)
131 RDMA_WRID_RDMA_WRITE
= 1,
132 RDMA_WRID_SEND_CONTROL
= 2000,
133 RDMA_WRID_RECV_CONTROL
= 4000,
136 static const char *wrid_desc
[] = {
137 [RDMA_WRID_NONE
] = "NONE",
138 [RDMA_WRID_RDMA_WRITE
] = "WRITE RDMA",
139 [RDMA_WRID_SEND_CONTROL
] = "CONTROL SEND",
140 [RDMA_WRID_RECV_CONTROL
] = "CONTROL RECV",
144 * Work request IDs for IB SEND messages only (not RDMA writes).
145 * This is used by the migration protocol to transmit
146 * control messages (such as device state and registration commands)
148 * We could use more WRs, but we have enough for now.
158 * SEND/RECV IB Control Messages.
161 RDMA_CONTROL_NONE
= 0,
163 RDMA_CONTROL_READY
, /* ready to receive */
164 RDMA_CONTROL_QEMU_FILE
, /* QEMUFile-transmitted bytes */
165 RDMA_CONTROL_RAM_BLOCKS_REQUEST
, /* RAMBlock synchronization */
166 RDMA_CONTROL_RAM_BLOCKS_RESULT
, /* RAMBlock synchronization */
167 RDMA_CONTROL_COMPRESS
, /* page contains repeat values */
168 RDMA_CONTROL_REGISTER_REQUEST
, /* dynamic page registration */
169 RDMA_CONTROL_REGISTER_RESULT
, /* key to use after registration */
170 RDMA_CONTROL_REGISTER_FINISHED
, /* current iteration finished */
171 RDMA_CONTROL_UNREGISTER_REQUEST
, /* dynamic UN-registration */
172 RDMA_CONTROL_UNREGISTER_FINISHED
, /* unpinning finished */
177 * Memory and MR structures used to represent an IB Send/Recv work request.
178 * This is *not* used for RDMA writes, only IB Send/Recv.
181 uint8_t control
[RDMA_CONTROL_MAX_BUFFER
]; /* actual buffer to register */
182 struct ibv_mr
*control_mr
; /* registration metadata */
183 size_t control_len
; /* length of the message */
184 uint8_t *control_curr
; /* start of unconsumed bytes */
185 } RDMAWorkRequestData
;
188 * Negotiate RDMA capabilities during connection-setup time.
195 static void caps_to_network(RDMACapabilities
*cap
)
197 cap
->version
= htonl(cap
->version
);
198 cap
->flags
= htonl(cap
->flags
);
201 static void network_to_caps(RDMACapabilities
*cap
)
203 cap
->version
= ntohl(cap
->version
);
204 cap
->flags
= ntohl(cap
->flags
);
208 * Representation of a RAMBlock from an RDMA perspective.
209 * This is not transmitted, only local.
210 * This and subsequent structures cannot be linked lists
211 * because we're using a single IB message to transmit
212 * the information. It's small anyway, so a list is overkill.
214 typedef struct RDMALocalBlock
{
216 uint8_t *local_host_addr
; /* local virtual address */
217 uint64_t remote_host_addr
; /* remote virtual address */
220 struct ibv_mr
**pmr
; /* MRs for chunk-level registration */
221 struct ibv_mr
*mr
; /* MR for non-chunk-level registration */
222 uint32_t *remote_keys
; /* rkeys for chunk-level registration */
223 uint32_t remote_rkey
; /* rkeys for non-chunk-level registration */
224 int index
; /* which block are we */
225 unsigned int src_index
; /* (Only used on dest) */
228 unsigned long *transit_bitmap
;
229 unsigned long *unregister_bitmap
;
233 * Also represents a RAMblock, but only on the dest.
234 * This gets transmitted by the dest during connection-time
235 * to the source VM and then is used to populate the
236 * corresponding RDMALocalBlock with
237 * the information needed to perform the actual RDMA.
239 typedef struct QEMU_PACKED RDMADestBlock
{
240 uint64_t remote_host_addr
;
243 uint32_t remote_rkey
;
247 static const char *control_desc(unsigned int rdma_control
)
249 static const char *strs
[] = {
250 [RDMA_CONTROL_NONE
] = "NONE",
251 [RDMA_CONTROL_ERROR
] = "ERROR",
252 [RDMA_CONTROL_READY
] = "READY",
253 [RDMA_CONTROL_QEMU_FILE
] = "QEMU FILE",
254 [RDMA_CONTROL_RAM_BLOCKS_REQUEST
] = "RAM BLOCKS REQUEST",
255 [RDMA_CONTROL_RAM_BLOCKS_RESULT
] = "RAM BLOCKS RESULT",
256 [RDMA_CONTROL_COMPRESS
] = "COMPRESS",
257 [RDMA_CONTROL_REGISTER_REQUEST
] = "REGISTER REQUEST",
258 [RDMA_CONTROL_REGISTER_RESULT
] = "REGISTER RESULT",
259 [RDMA_CONTROL_REGISTER_FINISHED
] = "REGISTER FINISHED",
260 [RDMA_CONTROL_UNREGISTER_REQUEST
] = "UNREGISTER REQUEST",
261 [RDMA_CONTROL_UNREGISTER_FINISHED
] = "UNREGISTER FINISHED",
264 if (rdma_control
> RDMA_CONTROL_UNREGISTER_FINISHED
) {
265 return "??BAD CONTROL VALUE??";
268 return strs
[rdma_control
];
271 static uint64_t htonll(uint64_t v
)
273 union { uint32_t lv
[2]; uint64_t llv
; } u
;
274 u
.lv
[0] = htonl(v
>> 32);
275 u
.lv
[1] = htonl(v
& 0xFFFFFFFFULL
);
279 static uint64_t ntohll(uint64_t v
)
281 union { uint32_t lv
[2]; uint64_t llv
; } u
;
283 return ((uint64_t)ntohl(u
.lv
[0]) << 32) | (uint64_t) ntohl(u
.lv
[1]);
286 static void dest_block_to_network(RDMADestBlock
*db
)
288 db
->remote_host_addr
= htonll(db
->remote_host_addr
);
289 db
->offset
= htonll(db
->offset
);
290 db
->length
= htonll(db
->length
);
291 db
->remote_rkey
= htonl(db
->remote_rkey
);
294 static void network_to_dest_block(RDMADestBlock
*db
)
296 db
->remote_host_addr
= ntohll(db
->remote_host_addr
);
297 db
->offset
= ntohll(db
->offset
);
298 db
->length
= ntohll(db
->length
);
299 db
->remote_rkey
= ntohl(db
->remote_rkey
);
303 * Virtual address of the above structures used for transmitting
304 * the RAMBlock descriptions at connection-time.
305 * This structure is *not* transmitted.
307 typedef struct RDMALocalBlocks
{
309 bool init
; /* main memory init complete */
310 RDMALocalBlock
*block
;
314 * Main data structure for RDMA state.
315 * While there is only one copy of this structure being allocated right now,
316 * this is the place where one would start if you wanted to consider
317 * having more than one RDMA connection open at the same time.
319 typedef struct RDMAContext
{
324 RDMAWorkRequestData wr_data
[RDMA_WRID_MAX
];
327 * This is used by *_exchange_send() to figure out whether or not
328 * the initial "READY" message has already been received or not.
329 * This is because other functions may potentially poll() and detect
330 * the READY message before send() does, in which case we need to
331 * know if it completed.
333 int control_ready_expected
;
335 /* number of outstanding writes */
338 /* store info about current buffer so that we can
339 merge it with future sends */
340 uint64_t current_addr
;
341 uint64_t current_length
;
342 /* index of ram block the current buffer belongs to */
344 /* index of the chunk in the current ram block */
350 * infiniband-specific variables for opening the device
351 * and maintaining connection state and so forth.
353 * cm_id also has ibv_context, rdma_event_channel, and ibv_qp in
354 * cm_id->verbs, cm_id->channel, and cm_id->qp.
356 struct rdma_cm_id
*cm_id
; /* connection manager ID */
357 struct rdma_cm_id
*listen_id
;
360 struct ibv_context
*verbs
;
361 struct rdma_event_channel
*channel
;
362 struct ibv_qp
*qp
; /* queue pair */
363 struct ibv_comp_channel
*recv_comp_channel
; /* recv completion channel */
364 struct ibv_comp_channel
*send_comp_channel
; /* send completion channel */
365 struct ibv_pd
*pd
; /* protection domain */
366 struct ibv_cq
*recv_cq
; /* recvieve completion queue */
367 struct ibv_cq
*send_cq
; /* send completion queue */
370 * If a previous write failed (perhaps because of a failed
371 * memory registration, then do not attempt any future work
372 * and remember the error state.
379 * Description of ram blocks used throughout the code.
381 RDMALocalBlocks local_ram_blocks
;
382 RDMADestBlock
*dest_blocks
;
384 /* Index of the next RAMBlock received during block registration */
385 unsigned int next_src_index
;
388 * Migration on *destination* started.
389 * Then use coroutine yield function.
390 * Source runs in a thread, so we don't care.
392 int migration_started_on_destination
;
394 int total_registrations
;
397 int unregister_current
, unregister_next
;
398 uint64_t unregistrations
[RDMA_SIGNALED_SEND_MAX
];
400 GHashTable
*blockmap
;
402 /* the RDMAContext for return path */
403 struct RDMAContext
*return_path
;
407 #define TYPE_QIO_CHANNEL_RDMA "qio-channel-rdma"
408 OBJECT_DECLARE_SIMPLE_TYPE(QIOChannelRDMA
, QIO_CHANNEL_RDMA
)
412 struct QIOChannelRDMA
{
415 RDMAContext
*rdmaout
;
417 bool blocking
; /* XXX we don't actually honour this yet */
421 * Main structure for IB Send/Recv control messages.
422 * This gets prepended at the beginning of every Send/Recv.
424 typedef struct QEMU_PACKED
{
425 uint32_t len
; /* Total length of data portion */
426 uint32_t type
; /* which control command to perform */
427 uint32_t repeat
; /* number of commands in data portion of same type */
431 static void control_to_network(RDMAControlHeader
*control
)
433 control
->type
= htonl(control
->type
);
434 control
->len
= htonl(control
->len
);
435 control
->repeat
= htonl(control
->repeat
);
438 static void network_to_control(RDMAControlHeader
*control
)
440 control
->type
= ntohl(control
->type
);
441 control
->len
= ntohl(control
->len
);
442 control
->repeat
= ntohl(control
->repeat
);
446 * Register a single Chunk.
447 * Information sent by the source VM to inform the dest
448 * to register an single chunk of memory before we can perform
449 * the actual RDMA operation.
451 typedef struct QEMU_PACKED
{
453 uint64_t current_addr
; /* offset into the ram_addr_t space */
454 uint64_t chunk
; /* chunk to lookup if unregistering */
456 uint32_t current_index
; /* which ramblock the chunk belongs to */
458 uint64_t chunks
; /* how many sequential chunks to register */
461 static void register_to_network(RDMAContext
*rdma
, RDMARegister
*reg
)
463 RDMALocalBlock
*local_block
;
464 local_block
= &rdma
->local_ram_blocks
.block
[reg
->current_index
];
466 if (local_block
->is_ram_block
) {
468 * current_addr as passed in is an address in the local ram_addr_t
469 * space, we need to translate this for the destination
471 reg
->key
.current_addr
-= local_block
->offset
;
472 reg
->key
.current_addr
+= rdma
->dest_blocks
[reg
->current_index
].offset
;
474 reg
->key
.current_addr
= htonll(reg
->key
.current_addr
);
475 reg
->current_index
= htonl(reg
->current_index
);
476 reg
->chunks
= htonll(reg
->chunks
);
479 static void network_to_register(RDMARegister
*reg
)
481 reg
->key
.current_addr
= ntohll(reg
->key
.current_addr
);
482 reg
->current_index
= ntohl(reg
->current_index
);
483 reg
->chunks
= ntohll(reg
->chunks
);
486 typedef struct QEMU_PACKED
{
487 uint32_t value
; /* if zero, we will madvise() */
488 uint32_t block_idx
; /* which ram block index */
489 uint64_t offset
; /* Address in remote ram_addr_t space */
490 uint64_t length
; /* length of the chunk */
493 static void compress_to_network(RDMAContext
*rdma
, RDMACompress
*comp
)
495 comp
->value
= htonl(comp
->value
);
497 * comp->offset as passed in is an address in the local ram_addr_t
498 * space, we need to translate this for the destination
500 comp
->offset
-= rdma
->local_ram_blocks
.block
[comp
->block_idx
].offset
;
501 comp
->offset
+= rdma
->dest_blocks
[comp
->block_idx
].offset
;
502 comp
->block_idx
= htonl(comp
->block_idx
);
503 comp
->offset
= htonll(comp
->offset
);
504 comp
->length
= htonll(comp
->length
);
507 static void network_to_compress(RDMACompress
*comp
)
509 comp
->value
= ntohl(comp
->value
);
510 comp
->block_idx
= ntohl(comp
->block_idx
);
511 comp
->offset
= ntohll(comp
->offset
);
512 comp
->length
= ntohll(comp
->length
);
516 * The result of the dest's memory registration produces an "rkey"
517 * which the source VM must reference in order to perform
518 * the RDMA operation.
520 typedef struct QEMU_PACKED
{
524 } RDMARegisterResult
;
526 static void result_to_network(RDMARegisterResult
*result
)
528 result
->rkey
= htonl(result
->rkey
);
529 result
->host_addr
= htonll(result
->host_addr
);
532 static void network_to_result(RDMARegisterResult
*result
)
534 result
->rkey
= ntohl(result
->rkey
);
535 result
->host_addr
= ntohll(result
->host_addr
);
538 const char *print_wrid(int wrid
);
539 static int qemu_rdma_exchange_send(RDMAContext
*rdma
, RDMAControlHeader
*head
,
540 uint8_t *data
, RDMAControlHeader
*resp
,
542 int (*callback
)(RDMAContext
*rdma
));
544 static inline uint64_t ram_chunk_index(const uint8_t *start
,
547 return ((uintptr_t) host
- (uintptr_t) start
) >> RDMA_REG_CHUNK_SHIFT
;
550 static inline uint8_t *ram_chunk_start(const RDMALocalBlock
*rdma_ram_block
,
553 return (uint8_t *)(uintptr_t)(rdma_ram_block
->local_host_addr
+
554 (i
<< RDMA_REG_CHUNK_SHIFT
));
557 static inline uint8_t *ram_chunk_end(const RDMALocalBlock
*rdma_ram_block
,
560 uint8_t *result
= ram_chunk_start(rdma_ram_block
, i
) +
561 (1UL << RDMA_REG_CHUNK_SHIFT
);
563 if (result
> (rdma_ram_block
->local_host_addr
+ rdma_ram_block
->length
)) {
564 result
= rdma_ram_block
->local_host_addr
+ rdma_ram_block
->length
;
570 static int rdma_add_block(RDMAContext
*rdma
, const char *block_name
,
572 ram_addr_t block_offset
, uint64_t length
)
574 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
575 RDMALocalBlock
*block
;
576 RDMALocalBlock
*old
= local
->block
;
578 local
->block
= g_new0(RDMALocalBlock
, local
->nb_blocks
+ 1);
580 if (local
->nb_blocks
) {
583 if (rdma
->blockmap
) {
584 for (x
= 0; x
< local
->nb_blocks
; x
++) {
585 g_hash_table_remove(rdma
->blockmap
,
586 (void *)(uintptr_t)old
[x
].offset
);
587 g_hash_table_insert(rdma
->blockmap
,
588 (void *)(uintptr_t)old
[x
].offset
,
592 memcpy(local
->block
, old
, sizeof(RDMALocalBlock
) * local
->nb_blocks
);
596 block
= &local
->block
[local
->nb_blocks
];
598 block
->block_name
= g_strdup(block_name
);
599 block
->local_host_addr
= host_addr
;
600 block
->offset
= block_offset
;
601 block
->length
= length
;
602 block
->index
= local
->nb_blocks
;
603 block
->src_index
= ~0U; /* Filled in by the receipt of the block list */
604 block
->nb_chunks
= ram_chunk_index(host_addr
, host_addr
+ length
) + 1UL;
605 block
->transit_bitmap
= bitmap_new(block
->nb_chunks
);
606 bitmap_clear(block
->transit_bitmap
, 0, block
->nb_chunks
);
607 block
->unregister_bitmap
= bitmap_new(block
->nb_chunks
);
608 bitmap_clear(block
->unregister_bitmap
, 0, block
->nb_chunks
);
609 block
->remote_keys
= g_new0(uint32_t, block
->nb_chunks
);
611 block
->is_ram_block
= local
->init
? false : true;
613 if (rdma
->blockmap
) {
614 g_hash_table_insert(rdma
->blockmap
, (void *)(uintptr_t)block_offset
, block
);
617 trace_rdma_add_block(block_name
, local
->nb_blocks
,
618 (uintptr_t) block
->local_host_addr
,
619 block
->offset
, block
->length
,
620 (uintptr_t) (block
->local_host_addr
+ block
->length
),
621 BITS_TO_LONGS(block
->nb_chunks
) *
622 sizeof(unsigned long) * 8,
631 * Memory regions need to be registered with the device and queue pairs setup
632 * in advanced before the migration starts. This tells us where the RAM blocks
633 * are so that we can register them individually.
635 static int qemu_rdma_init_one_block(RAMBlock
*rb
, void *opaque
)
637 const char *block_name
= qemu_ram_get_idstr(rb
);
638 void *host_addr
= qemu_ram_get_host_addr(rb
);
639 ram_addr_t block_offset
= qemu_ram_get_offset(rb
);
640 ram_addr_t length
= qemu_ram_get_used_length(rb
);
641 return rdma_add_block(opaque
, block_name
, host_addr
, block_offset
, length
);
645 * Identify the RAMBlocks and their quantity. They will be references to
646 * identify chunk boundaries inside each RAMBlock and also be referenced
647 * during dynamic page registration.
649 static int qemu_rdma_init_ram_blocks(RDMAContext
*rdma
)
651 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
654 assert(rdma
->blockmap
== NULL
);
655 memset(local
, 0, sizeof *local
);
656 ret
= foreach_not_ignored_block(qemu_rdma_init_one_block
, rdma
);
660 trace_qemu_rdma_init_ram_blocks(local
->nb_blocks
);
661 rdma
->dest_blocks
= g_new0(RDMADestBlock
,
662 rdma
->local_ram_blocks
.nb_blocks
);
668 * Note: If used outside of cleanup, the caller must ensure that the destination
669 * block structures are also updated
671 static int rdma_delete_block(RDMAContext
*rdma
, RDMALocalBlock
*block
)
673 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
674 RDMALocalBlock
*old
= local
->block
;
677 if (rdma
->blockmap
) {
678 g_hash_table_remove(rdma
->blockmap
, (void *)(uintptr_t)block
->offset
);
683 for (j
= 0; j
< block
->nb_chunks
; j
++) {
684 if (!block
->pmr
[j
]) {
687 ibv_dereg_mr(block
->pmr
[j
]);
688 rdma
->total_registrations
--;
695 ibv_dereg_mr(block
->mr
);
696 rdma
->total_registrations
--;
700 g_free(block
->transit_bitmap
);
701 block
->transit_bitmap
= NULL
;
703 g_free(block
->unregister_bitmap
);
704 block
->unregister_bitmap
= NULL
;
706 g_free(block
->remote_keys
);
707 block
->remote_keys
= NULL
;
709 g_free(block
->block_name
);
710 block
->block_name
= NULL
;
712 if (rdma
->blockmap
) {
713 for (x
= 0; x
< local
->nb_blocks
; x
++) {
714 g_hash_table_remove(rdma
->blockmap
,
715 (void *)(uintptr_t)old
[x
].offset
);
719 if (local
->nb_blocks
> 1) {
721 local
->block
= g_new0(RDMALocalBlock
, local
->nb_blocks
- 1);
724 memcpy(local
->block
, old
, sizeof(RDMALocalBlock
) * block
->index
);
727 if (block
->index
< (local
->nb_blocks
- 1)) {
728 memcpy(local
->block
+ block
->index
, old
+ (block
->index
+ 1),
729 sizeof(RDMALocalBlock
) *
730 (local
->nb_blocks
- (block
->index
+ 1)));
731 for (x
= block
->index
; x
< local
->nb_blocks
- 1; x
++) {
732 local
->block
[x
].index
--;
736 assert(block
== local
->block
);
740 trace_rdma_delete_block(block
, (uintptr_t)block
->local_host_addr
,
741 block
->offset
, block
->length
,
742 (uintptr_t)(block
->local_host_addr
+ block
->length
),
743 BITS_TO_LONGS(block
->nb_chunks
) *
744 sizeof(unsigned long) * 8, block
->nb_chunks
);
750 if (local
->nb_blocks
&& rdma
->blockmap
) {
751 for (x
= 0; x
< local
->nb_blocks
; x
++) {
752 g_hash_table_insert(rdma
->blockmap
,
753 (void *)(uintptr_t)local
->block
[x
].offset
,
762 * Put in the log file which RDMA device was opened and the details
763 * associated with that device.
765 static void qemu_rdma_dump_id(const char *who
, struct ibv_context
*verbs
)
767 struct ibv_port_attr port
;
769 if (ibv_query_port(verbs
, 1, &port
)) {
770 error_report("Failed to query port information");
774 printf("%s RDMA Device opened: kernel name %s "
775 "uverbs device name %s, "
776 "infiniband_verbs class device path %s, "
777 "infiniband class device path %s, "
778 "transport: (%d) %s\n",
781 verbs
->device
->dev_name
,
782 verbs
->device
->dev_path
,
783 verbs
->device
->ibdev_path
,
785 (port
.link_layer
== IBV_LINK_LAYER_INFINIBAND
) ? "Infiniband" :
786 ((port
.link_layer
== IBV_LINK_LAYER_ETHERNET
)
787 ? "Ethernet" : "Unknown"));
791 * Put in the log file the RDMA gid addressing information,
792 * useful for folks who have trouble understanding the
793 * RDMA device hierarchy in the kernel.
795 static void qemu_rdma_dump_gid(const char *who
, struct rdma_cm_id
*id
)
799 inet_ntop(AF_INET6
, &id
->route
.addr
.addr
.ibaddr
.sgid
, sgid
, sizeof sgid
);
800 inet_ntop(AF_INET6
, &id
->route
.addr
.addr
.ibaddr
.dgid
, dgid
, sizeof dgid
);
801 trace_qemu_rdma_dump_gid(who
, sgid
, dgid
);
805 * As of now, IPv6 over RoCE / iWARP is not supported by linux.
806 * We will try the next addrinfo struct, and fail if there are
807 * no other valid addresses to bind against.
809 * If user is listening on '[::]', then we will not have a opened a device
810 * yet and have no way of verifying if the device is RoCE or not.
812 * In this case, the source VM will throw an error for ALL types of
813 * connections (both IPv4 and IPv6) if the destination machine does not have
814 * a regular infiniband network available for use.
816 * The only way to guarantee that an error is thrown for broken kernels is
817 * for the management software to choose a *specific* interface at bind time
818 * and validate what time of hardware it is.
820 * Unfortunately, this puts the user in a fix:
822 * If the source VM connects with an IPv4 address without knowing that the
823 * destination has bound to '[::]' the migration will unconditionally fail
824 * unless the management software is explicitly listening on the IPv4
825 * address while using a RoCE-based device.
827 * If the source VM connects with an IPv6 address, then we're OK because we can
828 * throw an error on the source (and similarly on the destination).
830 * But in mixed environments, this will be broken for a while until it is fixed
833 * We do provide a *tiny* bit of help in this function: We can list all of the
834 * devices in the system and check to see if all the devices are RoCE or
837 * If we detect that we have a *pure* RoCE environment, then we can safely
838 * thrown an error even if the management software has specified '[::]' as the
841 * However, if there is are multiple hetergeneous devices, then we cannot make
842 * this assumption and the user just has to be sure they know what they are
845 * Patches are being reviewed on linux-rdma.
847 static int qemu_rdma_broken_ipv6_kernel(struct ibv_context
*verbs
, Error
**errp
)
849 /* This bug only exists in linux, to our knowledge. */
851 struct ibv_port_attr port_attr
;
854 * Verbs are only NULL if management has bound to '[::]'.
856 * Let's iterate through all the devices and see if there any pure IB
857 * devices (non-ethernet).
859 * If not, then we can safely proceed with the migration.
860 * Otherwise, there are no guarantees until the bug is fixed in linux.
864 struct ibv_device
**dev_list
= ibv_get_device_list(&num_devices
);
865 bool roce_found
= false;
866 bool ib_found
= false;
868 for (x
= 0; x
< num_devices
; x
++) {
869 verbs
= ibv_open_device(dev_list
[x
]);
871 if (errno
== EPERM
) {
878 if (ibv_query_port(verbs
, 1, &port_attr
)) {
879 ibv_close_device(verbs
);
880 ERROR(errp
, "Could not query initial IB port");
884 if (port_attr
.link_layer
== IBV_LINK_LAYER_INFINIBAND
) {
886 } else if (port_attr
.link_layer
== IBV_LINK_LAYER_ETHERNET
) {
890 ibv_close_device(verbs
);
896 fprintf(stderr
, "WARN: migrations may fail:"
897 " IPv6 over RoCE / iWARP in linux"
898 " is broken. But since you appear to have a"
899 " mixed RoCE / IB environment, be sure to only"
900 " migrate over the IB fabric until the kernel "
901 " fixes the bug.\n");
903 ERROR(errp
, "You only have RoCE / iWARP devices in your systems"
904 " and your management software has specified '[::]'"
905 ", but IPv6 over RoCE / iWARP is not supported in Linux.");
914 * If we have a verbs context, that means that some other than '[::]' was
915 * used by the management software for binding. In which case we can
916 * actually warn the user about a potentially broken kernel.
919 /* IB ports start with 1, not 0 */
920 if (ibv_query_port(verbs
, 1, &port_attr
)) {
921 ERROR(errp
, "Could not query initial IB port");
925 if (port_attr
.link_layer
== IBV_LINK_LAYER_ETHERNET
) {
926 ERROR(errp
, "Linux kernel's RoCE / iWARP does not support IPv6 "
927 "(but patches on linux-rdma in progress)");
937 * Figure out which RDMA device corresponds to the requested IP hostname
938 * Also create the initial connection manager identifiers for opening
941 static int qemu_rdma_resolve_host(RDMAContext
*rdma
, Error
**errp
)
944 struct rdma_addrinfo
*res
;
946 struct rdma_cm_event
*cm_event
;
947 char ip
[40] = "unknown";
948 struct rdma_addrinfo
*e
;
950 if (rdma
->host
== NULL
|| !strcmp(rdma
->host
, "")) {
951 ERROR(errp
, "RDMA hostname has not been set");
955 /* create CM channel */
956 rdma
->channel
= rdma_create_event_channel();
957 if (!rdma
->channel
) {
958 ERROR(errp
, "could not create CM channel");
963 ret
= rdma_create_id(rdma
->channel
, &rdma
->cm_id
, NULL
, RDMA_PS_TCP
);
965 ERROR(errp
, "could not create channel id");
966 goto err_resolve_create_id
;
969 snprintf(port_str
, 16, "%d", rdma
->port
);
972 ret
= rdma_getaddrinfo(rdma
->host
, port_str
, NULL
, &res
);
974 ERROR(errp
, "could not rdma_getaddrinfo address %s", rdma
->host
);
975 goto err_resolve_get_addr
;
978 for (e
= res
; e
!= NULL
; e
= e
->ai_next
) {
979 inet_ntop(e
->ai_family
,
980 &((struct sockaddr_in
*) e
->ai_dst_addr
)->sin_addr
, ip
, sizeof ip
);
981 trace_qemu_rdma_resolve_host_trying(rdma
->host
, ip
);
983 ret
= rdma_resolve_addr(rdma
->cm_id
, NULL
, e
->ai_dst_addr
,
984 RDMA_RESOLVE_TIMEOUT_MS
);
986 if (e
->ai_family
== AF_INET6
) {
987 ret
= qemu_rdma_broken_ipv6_kernel(rdma
->cm_id
->verbs
, errp
);
996 rdma_freeaddrinfo(res
);
997 ERROR(errp
, "could not resolve address %s", rdma
->host
);
998 goto err_resolve_get_addr
;
1001 rdma_freeaddrinfo(res
);
1002 qemu_rdma_dump_gid("source_resolve_addr", rdma
->cm_id
);
1004 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
1006 ERROR(errp
, "could not perform event_addr_resolved");
1007 goto err_resolve_get_addr
;
1010 if (cm_event
->event
!= RDMA_CM_EVENT_ADDR_RESOLVED
) {
1011 ERROR(errp
, "result not equal to event_addr_resolved %s",
1012 rdma_event_str(cm_event
->event
));
1013 error_report("rdma_resolve_addr");
1014 rdma_ack_cm_event(cm_event
);
1016 goto err_resolve_get_addr
;
1018 rdma_ack_cm_event(cm_event
);
1021 ret
= rdma_resolve_route(rdma
->cm_id
, RDMA_RESOLVE_TIMEOUT_MS
);
1023 ERROR(errp
, "could not resolve rdma route");
1024 goto err_resolve_get_addr
;
1027 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
1029 ERROR(errp
, "could not perform event_route_resolved");
1030 goto err_resolve_get_addr
;
1032 if (cm_event
->event
!= RDMA_CM_EVENT_ROUTE_RESOLVED
) {
1033 ERROR(errp
, "result not equal to event_route_resolved: %s",
1034 rdma_event_str(cm_event
->event
));
1035 rdma_ack_cm_event(cm_event
);
1037 goto err_resolve_get_addr
;
1039 rdma_ack_cm_event(cm_event
);
1040 rdma
->verbs
= rdma
->cm_id
->verbs
;
1041 qemu_rdma_dump_id("source_resolve_host", rdma
->cm_id
->verbs
);
1042 qemu_rdma_dump_gid("source_resolve_host", rdma
->cm_id
);
1045 err_resolve_get_addr
:
1046 rdma_destroy_id(rdma
->cm_id
);
1048 err_resolve_create_id
:
1049 rdma_destroy_event_channel(rdma
->channel
);
1050 rdma
->channel
= NULL
;
1055 * Create protection domain and completion queues
1057 static int qemu_rdma_alloc_pd_cq(RDMAContext
*rdma
)
1060 rdma
->pd
= ibv_alloc_pd(rdma
->verbs
);
1062 error_report("failed to allocate protection domain");
1066 /* create receive completion channel */
1067 rdma
->recv_comp_channel
= ibv_create_comp_channel(rdma
->verbs
);
1068 if (!rdma
->recv_comp_channel
) {
1069 error_report("failed to allocate receive completion channel");
1070 goto err_alloc_pd_cq
;
1074 * Completion queue can be filled by read work requests.
1076 rdma
->recv_cq
= ibv_create_cq(rdma
->verbs
, (RDMA_SIGNALED_SEND_MAX
* 3),
1077 NULL
, rdma
->recv_comp_channel
, 0);
1078 if (!rdma
->recv_cq
) {
1079 error_report("failed to allocate receive completion queue");
1080 goto err_alloc_pd_cq
;
1083 /* create send completion channel */
1084 rdma
->send_comp_channel
= ibv_create_comp_channel(rdma
->verbs
);
1085 if (!rdma
->send_comp_channel
) {
1086 error_report("failed to allocate send completion channel");
1087 goto err_alloc_pd_cq
;
1090 rdma
->send_cq
= ibv_create_cq(rdma
->verbs
, (RDMA_SIGNALED_SEND_MAX
* 3),
1091 NULL
, rdma
->send_comp_channel
, 0);
1092 if (!rdma
->send_cq
) {
1093 error_report("failed to allocate send completion queue");
1094 goto err_alloc_pd_cq
;
1101 ibv_dealloc_pd(rdma
->pd
);
1103 if (rdma
->recv_comp_channel
) {
1104 ibv_destroy_comp_channel(rdma
->recv_comp_channel
);
1106 if (rdma
->send_comp_channel
) {
1107 ibv_destroy_comp_channel(rdma
->send_comp_channel
);
1109 if (rdma
->recv_cq
) {
1110 ibv_destroy_cq(rdma
->recv_cq
);
1111 rdma
->recv_cq
= NULL
;
1114 rdma
->recv_comp_channel
= NULL
;
1115 rdma
->send_comp_channel
= NULL
;
1121 * Create queue pairs.
1123 static int qemu_rdma_alloc_qp(RDMAContext
*rdma
)
1125 struct ibv_qp_init_attr attr
= { 0 };
1128 attr
.cap
.max_send_wr
= RDMA_SIGNALED_SEND_MAX
;
1129 attr
.cap
.max_recv_wr
= 3;
1130 attr
.cap
.max_send_sge
= 1;
1131 attr
.cap
.max_recv_sge
= 1;
1132 attr
.send_cq
= rdma
->send_cq
;
1133 attr
.recv_cq
= rdma
->recv_cq
;
1134 attr
.qp_type
= IBV_QPT_RC
;
1136 ret
= rdma_create_qp(rdma
->cm_id
, rdma
->pd
, &attr
);
1141 rdma
->qp
= rdma
->cm_id
->qp
;
1145 /* Check whether On-Demand Paging is supported by RDAM device */
1146 static bool rdma_support_odp(struct ibv_context
*dev
)
1148 struct ibv_device_attr_ex attr
= {0};
1149 int ret
= ibv_query_device_ex(dev
, NULL
, &attr
);
1154 if (attr
.odp_caps
.general_caps
& IBV_ODP_SUPPORT
) {
1162 * ibv_advise_mr to avoid RNR NAK error as far as possible.
1163 * The responder mr registering with ODP will sent RNR NAK back to
1164 * the requester in the face of the page fault.
1166 static void qemu_rdma_advise_prefetch_mr(struct ibv_pd
*pd
, uint64_t addr
,
1167 uint32_t len
, uint32_t lkey
,
1168 const char *name
, bool wr
)
1170 #ifdef HAVE_IBV_ADVISE_MR
1172 int advice
= wr
? IBV_ADVISE_MR_ADVICE_PREFETCH_WRITE
:
1173 IBV_ADVISE_MR_ADVICE_PREFETCH
;
1174 struct ibv_sge sg_list
= {.lkey
= lkey
, .addr
= addr
, .length
= len
};
1176 ret
= ibv_advise_mr(pd
, advice
,
1177 IBV_ADVISE_MR_FLAG_FLUSH
, &sg_list
, 1);
1178 /* ignore the error */
1180 trace_qemu_rdma_advise_mr(name
, len
, addr
, strerror(errno
));
1182 trace_qemu_rdma_advise_mr(name
, len
, addr
, "successed");
1187 static int qemu_rdma_reg_whole_ram_blocks(RDMAContext
*rdma
)
1190 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
1192 for (i
= 0; i
< local
->nb_blocks
; i
++) {
1193 int access
= IBV_ACCESS_LOCAL_WRITE
| IBV_ACCESS_REMOTE_WRITE
;
1195 local
->block
[i
].mr
=
1196 ibv_reg_mr(rdma
->pd
,
1197 local
->block
[i
].local_host_addr
,
1198 local
->block
[i
].length
, access
1201 if (!local
->block
[i
].mr
&&
1202 errno
== ENOTSUP
&& rdma_support_odp(rdma
->verbs
)) {
1203 access
|= IBV_ACCESS_ON_DEMAND
;
1204 /* register ODP mr */
1205 local
->block
[i
].mr
=
1206 ibv_reg_mr(rdma
->pd
,
1207 local
->block
[i
].local_host_addr
,
1208 local
->block
[i
].length
, access
);
1209 trace_qemu_rdma_register_odp_mr(local
->block
[i
].block_name
);
1211 if (local
->block
[i
].mr
) {
1212 qemu_rdma_advise_prefetch_mr(rdma
->pd
,
1213 (uintptr_t)local
->block
[i
].local_host_addr
,
1214 local
->block
[i
].length
,
1215 local
->block
[i
].mr
->lkey
,
1216 local
->block
[i
].block_name
,
1221 if (!local
->block
[i
].mr
) {
1222 perror("Failed to register local dest ram block!");
1225 rdma
->total_registrations
++;
1228 if (i
>= local
->nb_blocks
) {
1232 for (i
--; i
>= 0; i
--) {
1233 ibv_dereg_mr(local
->block
[i
].mr
);
1234 local
->block
[i
].mr
= NULL
;
1235 rdma
->total_registrations
--;
1243 * Find the ram block that corresponds to the page requested to be
1244 * transmitted by QEMU.
1246 * Once the block is found, also identify which 'chunk' within that
1247 * block that the page belongs to.
1249 * This search cannot fail or the migration will fail.
1251 static int qemu_rdma_search_ram_block(RDMAContext
*rdma
,
1252 uintptr_t block_offset
,
1255 uint64_t *block_index
,
1256 uint64_t *chunk_index
)
1258 uint64_t current_addr
= block_offset
+ offset
;
1259 RDMALocalBlock
*block
= g_hash_table_lookup(rdma
->blockmap
,
1260 (void *) block_offset
);
1262 assert(current_addr
>= block
->offset
);
1263 assert((current_addr
+ length
) <= (block
->offset
+ block
->length
));
1265 *block_index
= block
->index
;
1266 *chunk_index
= ram_chunk_index(block
->local_host_addr
,
1267 block
->local_host_addr
+ (current_addr
- block
->offset
));
1273 * Register a chunk with IB. If the chunk was already registered
1274 * previously, then skip.
1276 * Also return the keys associated with the registration needed
1277 * to perform the actual RDMA operation.
1279 static int qemu_rdma_register_and_get_keys(RDMAContext
*rdma
,
1280 RDMALocalBlock
*block
, uintptr_t host_addr
,
1281 uint32_t *lkey
, uint32_t *rkey
, int chunk
,
1282 uint8_t *chunk_start
, uint8_t *chunk_end
)
1286 *lkey
= block
->mr
->lkey
;
1289 *rkey
= block
->mr
->rkey
;
1294 /* allocate memory to store chunk MRs */
1296 block
->pmr
= g_new0(struct ibv_mr
*, block
->nb_chunks
);
1300 * If 'rkey', then we're the destination, so grant access to the source.
1302 * If 'lkey', then we're the source VM, so grant access only to ourselves.
1304 if (!block
->pmr
[chunk
]) {
1305 uint64_t len
= chunk_end
- chunk_start
;
1306 int access
= rkey
? IBV_ACCESS_LOCAL_WRITE
| IBV_ACCESS_REMOTE_WRITE
:
1309 trace_qemu_rdma_register_and_get_keys(len
, chunk_start
);
1311 block
->pmr
[chunk
] = ibv_reg_mr(rdma
->pd
, chunk_start
, len
, access
);
1312 if (!block
->pmr
[chunk
] &&
1313 errno
== ENOTSUP
&& rdma_support_odp(rdma
->verbs
)) {
1314 access
|= IBV_ACCESS_ON_DEMAND
;
1315 /* register ODP mr */
1316 block
->pmr
[chunk
] = ibv_reg_mr(rdma
->pd
, chunk_start
, len
, access
);
1317 trace_qemu_rdma_register_odp_mr(block
->block_name
);
1319 if (block
->pmr
[chunk
]) {
1320 qemu_rdma_advise_prefetch_mr(rdma
->pd
, (uintptr_t)chunk_start
,
1321 len
, block
->pmr
[chunk
]->lkey
,
1322 block
->block_name
, rkey
);
1327 if (!block
->pmr
[chunk
]) {
1328 perror("Failed to register chunk!");
1329 fprintf(stderr
, "Chunk details: block: %d chunk index %d"
1330 " start %" PRIuPTR
" end %" PRIuPTR
1332 " local %" PRIuPTR
" registrations: %d\n",
1333 block
->index
, chunk
, (uintptr_t)chunk_start
,
1334 (uintptr_t)chunk_end
, host_addr
,
1335 (uintptr_t)block
->local_host_addr
,
1336 rdma
->total_registrations
);
1339 rdma
->total_registrations
++;
1342 *lkey
= block
->pmr
[chunk
]->lkey
;
1345 *rkey
= block
->pmr
[chunk
]->rkey
;
1351 * Register (at connection time) the memory used for control
1354 static int qemu_rdma_reg_control(RDMAContext
*rdma
, int idx
)
1356 rdma
->wr_data
[idx
].control_mr
= ibv_reg_mr(rdma
->pd
,
1357 rdma
->wr_data
[idx
].control
, RDMA_CONTROL_MAX_BUFFER
,
1358 IBV_ACCESS_LOCAL_WRITE
| IBV_ACCESS_REMOTE_WRITE
);
1359 if (rdma
->wr_data
[idx
].control_mr
) {
1360 rdma
->total_registrations
++;
1363 error_report("qemu_rdma_reg_control failed");
1367 const char *print_wrid(int wrid
)
1369 if (wrid
>= RDMA_WRID_RECV_CONTROL
) {
1370 return wrid_desc
[RDMA_WRID_RECV_CONTROL
];
1372 return wrid_desc
[wrid
];
1376 * Perform a non-optimized memory unregistration after every transfer
1377 * for demonstration purposes, only if pin-all is not requested.
1379 * Potential optimizations:
1380 * 1. Start a new thread to run this function continuously
1382 - and for receipt of unregister messages
1384 * 3. Use workload hints.
1386 static int qemu_rdma_unregister_waiting(RDMAContext
*rdma
)
1388 while (rdma
->unregistrations
[rdma
->unregister_current
]) {
1390 uint64_t wr_id
= rdma
->unregistrations
[rdma
->unregister_current
];
1392 (wr_id
& RDMA_WRID_CHUNK_MASK
) >> RDMA_WRID_CHUNK_SHIFT
;
1394 (wr_id
& RDMA_WRID_BLOCK_MASK
) >> RDMA_WRID_BLOCK_SHIFT
;
1395 RDMALocalBlock
*block
=
1396 &(rdma
->local_ram_blocks
.block
[index
]);
1397 RDMARegister reg
= { .current_index
= index
};
1398 RDMAControlHeader resp
= { .type
= RDMA_CONTROL_UNREGISTER_FINISHED
,
1400 RDMAControlHeader head
= { .len
= sizeof(RDMARegister
),
1401 .type
= RDMA_CONTROL_UNREGISTER_REQUEST
,
1405 trace_qemu_rdma_unregister_waiting_proc(chunk
,
1406 rdma
->unregister_current
);
1408 rdma
->unregistrations
[rdma
->unregister_current
] = 0;
1409 rdma
->unregister_current
++;
1411 if (rdma
->unregister_current
== RDMA_SIGNALED_SEND_MAX
) {
1412 rdma
->unregister_current
= 0;
1417 * Unregistration is speculative (because migration is single-threaded
1418 * and we cannot break the protocol's inifinband message ordering).
1419 * Thus, if the memory is currently being used for transmission,
1420 * then abort the attempt to unregister and try again
1421 * later the next time a completion is received for this memory.
1423 clear_bit(chunk
, block
->unregister_bitmap
);
1425 if (test_bit(chunk
, block
->transit_bitmap
)) {
1426 trace_qemu_rdma_unregister_waiting_inflight(chunk
);
1430 trace_qemu_rdma_unregister_waiting_send(chunk
);
1432 ret
= ibv_dereg_mr(block
->pmr
[chunk
]);
1433 block
->pmr
[chunk
] = NULL
;
1434 block
->remote_keys
[chunk
] = 0;
1437 perror("unregistration chunk failed");
1440 rdma
->total_registrations
--;
1442 reg
.key
.chunk
= chunk
;
1443 register_to_network(rdma
, ®
);
1444 ret
= qemu_rdma_exchange_send(rdma
, &head
, (uint8_t *) ®
,
1450 trace_qemu_rdma_unregister_waiting_complete(chunk
);
1456 static uint64_t qemu_rdma_make_wrid(uint64_t wr_id
, uint64_t index
,
1459 uint64_t result
= wr_id
& RDMA_WRID_TYPE_MASK
;
1461 result
|= (index
<< RDMA_WRID_BLOCK_SHIFT
);
1462 result
|= (chunk
<< RDMA_WRID_CHUNK_SHIFT
);
1468 * Consult the connection manager to see a work request
1469 * (of any kind) has completed.
1470 * Return the work request ID that completed.
1472 static uint64_t qemu_rdma_poll(RDMAContext
*rdma
, struct ibv_cq
*cq
,
1473 uint64_t *wr_id_out
, uint32_t *byte_len
)
1479 ret
= ibv_poll_cq(cq
, 1, &wc
);
1482 *wr_id_out
= RDMA_WRID_NONE
;
1487 error_report("ibv_poll_cq return %d", ret
);
1491 wr_id
= wc
.wr_id
& RDMA_WRID_TYPE_MASK
;
1493 if (wc
.status
!= IBV_WC_SUCCESS
) {
1494 fprintf(stderr
, "ibv_poll_cq wc.status=%d %s!\n",
1495 wc
.status
, ibv_wc_status_str(wc
.status
));
1496 fprintf(stderr
, "ibv_poll_cq wrid=%s!\n", wrid_desc
[wr_id
]);
1501 if (rdma
->control_ready_expected
&&
1502 (wr_id
>= RDMA_WRID_RECV_CONTROL
)) {
1503 trace_qemu_rdma_poll_recv(wrid_desc
[RDMA_WRID_RECV_CONTROL
],
1504 wr_id
- RDMA_WRID_RECV_CONTROL
, wr_id
, rdma
->nb_sent
);
1505 rdma
->control_ready_expected
= 0;
1508 if (wr_id
== RDMA_WRID_RDMA_WRITE
) {
1510 (wc
.wr_id
& RDMA_WRID_CHUNK_MASK
) >> RDMA_WRID_CHUNK_SHIFT
;
1512 (wc
.wr_id
& RDMA_WRID_BLOCK_MASK
) >> RDMA_WRID_BLOCK_SHIFT
;
1513 RDMALocalBlock
*block
= &(rdma
->local_ram_blocks
.block
[index
]);
1515 trace_qemu_rdma_poll_write(print_wrid(wr_id
), wr_id
, rdma
->nb_sent
,
1516 index
, chunk
, block
->local_host_addr
,
1517 (void *)(uintptr_t)block
->remote_host_addr
);
1519 clear_bit(chunk
, block
->transit_bitmap
);
1521 if (rdma
->nb_sent
> 0) {
1525 trace_qemu_rdma_poll_other(print_wrid(wr_id
), wr_id
, rdma
->nb_sent
);
1528 *wr_id_out
= wc
.wr_id
;
1530 *byte_len
= wc
.byte_len
;
1536 /* Wait for activity on the completion channel.
1537 * Returns 0 on success, none-0 on error.
1539 static int qemu_rdma_wait_comp_channel(RDMAContext
*rdma
,
1540 struct ibv_comp_channel
*comp_channel
)
1542 struct rdma_cm_event
*cm_event
;
1546 * Coroutine doesn't start until migration_fd_process_incoming()
1547 * so don't yield unless we know we're running inside of a coroutine.
1549 if (rdma
->migration_started_on_destination
&&
1550 migration_incoming_get_current()->state
== MIGRATION_STATUS_ACTIVE
) {
1551 yield_until_fd_readable(comp_channel
->fd
);
1553 /* This is the source side, we're in a separate thread
1554 * or destination prior to migration_fd_process_incoming()
1555 * after postcopy, the destination also in a separate thread.
1556 * we can't yield; so we have to poll the fd.
1557 * But we need to be able to handle 'cancel' or an error
1558 * without hanging forever.
1560 while (!rdma
->error_state
&& !rdma
->received_error
) {
1562 pfds
[0].fd
= comp_channel
->fd
;
1563 pfds
[0].events
= G_IO_IN
| G_IO_HUP
| G_IO_ERR
;
1564 pfds
[0].revents
= 0;
1566 pfds
[1].fd
= rdma
->channel
->fd
;
1567 pfds
[1].events
= G_IO_IN
| G_IO_HUP
| G_IO_ERR
;
1568 pfds
[1].revents
= 0;
1570 /* 0.1s timeout, should be fine for a 'cancel' */
1571 switch (qemu_poll_ns(pfds
, 2, 100 * 1000 * 1000)) {
1573 case 1: /* fd active */
1574 if (pfds
[0].revents
) {
1578 if (pfds
[1].revents
) {
1579 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
1581 error_report("failed to get cm event while wait "
1582 "completion channel");
1586 error_report("receive cm event while wait comp channel,"
1587 "cm event is %d", cm_event
->event
);
1588 if (cm_event
->event
== RDMA_CM_EVENT_DISCONNECTED
||
1589 cm_event
->event
== RDMA_CM_EVENT_DEVICE_REMOVAL
) {
1590 rdma_ack_cm_event(cm_event
);
1593 rdma_ack_cm_event(cm_event
);
1597 case 0: /* Timeout, go around again */
1600 default: /* Error of some type -
1601 * I don't trust errno from qemu_poll_ns
1603 error_report("%s: poll failed", __func__
);
1607 if (migrate_get_current()->state
== MIGRATION_STATUS_CANCELLING
) {
1608 /* Bail out and let the cancellation happen */
1614 if (rdma
->received_error
) {
1617 return rdma
->error_state
;
1620 static struct ibv_comp_channel
*to_channel(RDMAContext
*rdma
, int wrid
)
1622 return wrid
< RDMA_WRID_RECV_CONTROL
? rdma
->send_comp_channel
:
1623 rdma
->recv_comp_channel
;
1626 static struct ibv_cq
*to_cq(RDMAContext
*rdma
, int wrid
)
1628 return wrid
< RDMA_WRID_RECV_CONTROL
? rdma
->send_cq
: rdma
->recv_cq
;
1632 * Block until the next work request has completed.
1634 * First poll to see if a work request has already completed,
1637 * If we encounter completed work requests for IDs other than
1638 * the one we're interested in, then that's generally an error.
1640 * The only exception is actual RDMA Write completions. These
1641 * completions only need to be recorded, but do not actually
1642 * need further processing.
1644 static int qemu_rdma_block_for_wrid(RDMAContext
*rdma
, int wrid_requested
,
1647 int num_cq_events
= 0, ret
= 0;
1650 uint64_t wr_id
= RDMA_WRID_NONE
, wr_id_in
;
1651 struct ibv_comp_channel
*ch
= to_channel(rdma
, wrid_requested
);
1652 struct ibv_cq
*poll_cq
= to_cq(rdma
, wrid_requested
);
1654 if (ibv_req_notify_cq(poll_cq
, 0)) {
1658 while (wr_id
!= wrid_requested
) {
1659 ret
= qemu_rdma_poll(rdma
, poll_cq
, &wr_id_in
, byte_len
);
1664 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
1666 if (wr_id
== RDMA_WRID_NONE
) {
1669 if (wr_id
!= wrid_requested
) {
1670 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested
),
1671 wrid_requested
, print_wrid(wr_id
), wr_id
);
1675 if (wr_id
== wrid_requested
) {
1680 ret
= qemu_rdma_wait_comp_channel(rdma
, ch
);
1682 goto err_block_for_wrid
;
1685 ret
= ibv_get_cq_event(ch
, &cq
, &cq_ctx
);
1687 perror("ibv_get_cq_event");
1688 goto err_block_for_wrid
;
1693 ret
= -ibv_req_notify_cq(cq
, 0);
1695 goto err_block_for_wrid
;
1698 while (wr_id
!= wrid_requested
) {
1699 ret
= qemu_rdma_poll(rdma
, poll_cq
, &wr_id_in
, byte_len
);
1701 goto err_block_for_wrid
;
1704 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
1706 if (wr_id
== RDMA_WRID_NONE
) {
1709 if (wr_id
!= wrid_requested
) {
1710 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested
),
1711 wrid_requested
, print_wrid(wr_id
), wr_id
);
1715 if (wr_id
== wrid_requested
) {
1716 goto success_block_for_wrid
;
1720 success_block_for_wrid
:
1721 if (num_cq_events
) {
1722 ibv_ack_cq_events(cq
, num_cq_events
);
1727 if (num_cq_events
) {
1728 ibv_ack_cq_events(cq
, num_cq_events
);
1731 rdma
->error_state
= ret
;
1736 * Post a SEND message work request for the control channel
1737 * containing some data and block until the post completes.
1739 static int qemu_rdma_post_send_control(RDMAContext
*rdma
, uint8_t *buf
,
1740 RDMAControlHeader
*head
)
1743 RDMAWorkRequestData
*wr
= &rdma
->wr_data
[RDMA_WRID_CONTROL
];
1744 struct ibv_send_wr
*bad_wr
;
1745 struct ibv_sge sge
= {
1746 .addr
= (uintptr_t)(wr
->control
),
1747 .length
= head
->len
+ sizeof(RDMAControlHeader
),
1748 .lkey
= wr
->control_mr
->lkey
,
1750 struct ibv_send_wr send_wr
= {
1751 .wr_id
= RDMA_WRID_SEND_CONTROL
,
1752 .opcode
= IBV_WR_SEND
,
1753 .send_flags
= IBV_SEND_SIGNALED
,
1758 trace_qemu_rdma_post_send_control(control_desc(head
->type
));
1761 * We don't actually need to do a memcpy() in here if we used
1762 * the "sge" properly, but since we're only sending control messages
1763 * (not RAM in a performance-critical path), then its OK for now.
1765 * The copy makes the RDMAControlHeader simpler to manipulate
1766 * for the time being.
1768 assert(head
->len
<= RDMA_CONTROL_MAX_BUFFER
- sizeof(*head
));
1769 memcpy(wr
->control
, head
, sizeof(RDMAControlHeader
));
1770 control_to_network((void *) wr
->control
);
1773 memcpy(wr
->control
+ sizeof(RDMAControlHeader
), buf
, head
->len
);
1777 ret
= ibv_post_send(rdma
->qp
, &send_wr
, &bad_wr
);
1780 error_report("Failed to use post IB SEND for control");
1784 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_SEND_CONTROL
, NULL
);
1786 error_report("rdma migration: send polling control error");
1793 * Post a RECV work request in anticipation of some future receipt
1794 * of data on the control channel.
1796 static int qemu_rdma_post_recv_control(RDMAContext
*rdma
, int idx
)
1798 struct ibv_recv_wr
*bad_wr
;
1799 struct ibv_sge sge
= {
1800 .addr
= (uintptr_t)(rdma
->wr_data
[idx
].control
),
1801 .length
= RDMA_CONTROL_MAX_BUFFER
,
1802 .lkey
= rdma
->wr_data
[idx
].control_mr
->lkey
,
1805 struct ibv_recv_wr recv_wr
= {
1806 .wr_id
= RDMA_WRID_RECV_CONTROL
+ idx
,
1812 if (ibv_post_recv(rdma
->qp
, &recv_wr
, &bad_wr
)) {
1820 * Block and wait for a RECV control channel message to arrive.
1822 static int qemu_rdma_exchange_get_response(RDMAContext
*rdma
,
1823 RDMAControlHeader
*head
, int expecting
, int idx
)
1826 int ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RECV_CONTROL
+ idx
,
1830 error_report("rdma migration: recv polling control error!");
1834 network_to_control((void *) rdma
->wr_data
[idx
].control
);
1835 memcpy(head
, rdma
->wr_data
[idx
].control
, sizeof(RDMAControlHeader
));
1837 trace_qemu_rdma_exchange_get_response_start(control_desc(expecting
));
1839 if (expecting
== RDMA_CONTROL_NONE
) {
1840 trace_qemu_rdma_exchange_get_response_none(control_desc(head
->type
),
1842 } else if (head
->type
!= expecting
|| head
->type
== RDMA_CONTROL_ERROR
) {
1843 error_report("Was expecting a %s (%d) control message"
1844 ", but got: %s (%d), length: %d",
1845 control_desc(expecting
), expecting
,
1846 control_desc(head
->type
), head
->type
, head
->len
);
1847 if (head
->type
== RDMA_CONTROL_ERROR
) {
1848 rdma
->received_error
= true;
1852 if (head
->len
> RDMA_CONTROL_MAX_BUFFER
- sizeof(*head
)) {
1853 error_report("too long length: %d", head
->len
);
1856 if (sizeof(*head
) + head
->len
!= byte_len
) {
1857 error_report("Malformed length: %d byte_len %d", head
->len
, byte_len
);
1865 * When a RECV work request has completed, the work request's
1866 * buffer is pointed at the header.
1868 * This will advance the pointer to the data portion
1869 * of the control message of the work request's buffer that
1870 * was populated after the work request finished.
1872 static void qemu_rdma_move_header(RDMAContext
*rdma
, int idx
,
1873 RDMAControlHeader
*head
)
1875 rdma
->wr_data
[idx
].control_len
= head
->len
;
1876 rdma
->wr_data
[idx
].control_curr
=
1877 rdma
->wr_data
[idx
].control
+ sizeof(RDMAControlHeader
);
1881 * This is an 'atomic' high-level operation to deliver a single, unified
1882 * control-channel message.
1884 * Additionally, if the user is expecting some kind of reply to this message,
1885 * they can request a 'resp' response message be filled in by posting an
1886 * additional work request on behalf of the user and waiting for an additional
1889 * The extra (optional) response is used during registration to us from having
1890 * to perform an *additional* exchange of message just to provide a response by
1891 * instead piggy-backing on the acknowledgement.
1893 static int qemu_rdma_exchange_send(RDMAContext
*rdma
, RDMAControlHeader
*head
,
1894 uint8_t *data
, RDMAControlHeader
*resp
,
1896 int (*callback
)(RDMAContext
*rdma
))
1901 * Wait until the dest is ready before attempting to deliver the message
1902 * by waiting for a READY message.
1904 if (rdma
->control_ready_expected
) {
1905 RDMAControlHeader resp
;
1906 ret
= qemu_rdma_exchange_get_response(rdma
,
1907 &resp
, RDMA_CONTROL_READY
, RDMA_WRID_READY
);
1914 * If the user is expecting a response, post a WR in anticipation of it.
1917 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_DATA
);
1919 error_report("rdma migration: error posting"
1920 " extra control recv for anticipated result!");
1926 * Post a WR to replace the one we just consumed for the READY message.
1928 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
1930 error_report("rdma migration: error posting first control recv!");
1935 * Deliver the control message that was requested.
1937 ret
= qemu_rdma_post_send_control(rdma
, data
, head
);
1940 error_report("Failed to send control buffer!");
1945 * If we're expecting a response, block and wait for it.
1949 trace_qemu_rdma_exchange_send_issue_callback();
1950 ret
= callback(rdma
);
1956 trace_qemu_rdma_exchange_send_waiting(control_desc(resp
->type
));
1957 ret
= qemu_rdma_exchange_get_response(rdma
, resp
,
1958 resp
->type
, RDMA_WRID_DATA
);
1964 qemu_rdma_move_header(rdma
, RDMA_WRID_DATA
, resp
);
1966 *resp_idx
= RDMA_WRID_DATA
;
1968 trace_qemu_rdma_exchange_send_received(control_desc(resp
->type
));
1971 rdma
->control_ready_expected
= 1;
1977 * This is an 'atomic' high-level operation to receive a single, unified
1978 * control-channel message.
1980 static int qemu_rdma_exchange_recv(RDMAContext
*rdma
, RDMAControlHeader
*head
,
1983 RDMAControlHeader ready
= {
1985 .type
= RDMA_CONTROL_READY
,
1991 * Inform the source that we're ready to receive a message.
1993 ret
= qemu_rdma_post_send_control(rdma
, NULL
, &ready
);
1996 error_report("Failed to send control buffer!");
2001 * Block and wait for the message.
2003 ret
= qemu_rdma_exchange_get_response(rdma
, head
,
2004 expecting
, RDMA_WRID_READY
);
2010 qemu_rdma_move_header(rdma
, RDMA_WRID_READY
, head
);
2013 * Post a new RECV work request to replace the one we just consumed.
2015 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
2017 error_report("rdma migration: error posting second control recv!");
2025 * Write an actual chunk of memory using RDMA.
2027 * If we're using dynamic registration on the dest-side, we have to
2028 * send a registration command first.
2030 static int qemu_rdma_write_one(QEMUFile
*f
, RDMAContext
*rdma
,
2031 int current_index
, uint64_t current_addr
,
2035 struct ibv_send_wr send_wr
= { 0 };
2036 struct ibv_send_wr
*bad_wr
;
2037 int reg_result_idx
, ret
, count
= 0;
2038 uint64_t chunk
, chunks
;
2039 uint8_t *chunk_start
, *chunk_end
;
2040 RDMALocalBlock
*block
= &(rdma
->local_ram_blocks
.block
[current_index
]);
2042 RDMARegisterResult
*reg_result
;
2043 RDMAControlHeader resp
= { .type
= RDMA_CONTROL_REGISTER_RESULT
};
2044 RDMAControlHeader head
= { .len
= sizeof(RDMARegister
),
2045 .type
= RDMA_CONTROL_REGISTER_REQUEST
,
2050 sge
.addr
= (uintptr_t)(block
->local_host_addr
+
2051 (current_addr
- block
->offset
));
2052 sge
.length
= length
;
2054 chunk
= ram_chunk_index(block
->local_host_addr
,
2055 (uint8_t *)(uintptr_t)sge
.addr
);
2056 chunk_start
= ram_chunk_start(block
, chunk
);
2058 if (block
->is_ram_block
) {
2059 chunks
= length
/ (1UL << RDMA_REG_CHUNK_SHIFT
);
2061 if (chunks
&& ((length
% (1UL << RDMA_REG_CHUNK_SHIFT
)) == 0)) {
2065 chunks
= block
->length
/ (1UL << RDMA_REG_CHUNK_SHIFT
);
2067 if (chunks
&& ((block
->length
% (1UL << RDMA_REG_CHUNK_SHIFT
)) == 0)) {
2072 trace_qemu_rdma_write_one_top(chunks
+ 1,
2074 (1UL << RDMA_REG_CHUNK_SHIFT
) / 1024 / 1024);
2076 chunk_end
= ram_chunk_end(block
, chunk
+ chunks
);
2079 while (test_bit(chunk
, block
->transit_bitmap
)) {
2081 trace_qemu_rdma_write_one_block(count
++, current_index
, chunk
,
2082 sge
.addr
, length
, rdma
->nb_sent
, block
->nb_chunks
);
2084 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RDMA_WRITE
, NULL
);
2087 error_report("Failed to Wait for previous write to complete "
2088 "block %d chunk %" PRIu64
2089 " current %" PRIu64
" len %" PRIu64
" %d",
2090 current_index
, chunk
, sge
.addr
, length
, rdma
->nb_sent
);
2095 if (!rdma
->pin_all
|| !block
->is_ram_block
) {
2096 if (!block
->remote_keys
[chunk
]) {
2098 * This chunk has not yet been registered, so first check to see
2099 * if the entire chunk is zero. If so, tell the other size to
2100 * memset() + madvise() the entire chunk without RDMA.
2103 if (buffer_is_zero((void *)(uintptr_t)sge
.addr
, length
)) {
2104 RDMACompress comp
= {
2105 .offset
= current_addr
,
2107 .block_idx
= current_index
,
2111 head
.len
= sizeof(comp
);
2112 head
.type
= RDMA_CONTROL_COMPRESS
;
2114 trace_qemu_rdma_write_one_zero(chunk
, sge
.length
,
2115 current_index
, current_addr
);
2117 compress_to_network(rdma
, &comp
);
2118 ret
= qemu_rdma_exchange_send(rdma
, &head
,
2119 (uint8_t *) &comp
, NULL
, NULL
, NULL
);
2126 * TODO: Here we are sending something, but we are not
2127 * accounting for anything transferred. The following is wrong:
2129 * stat64_add(&mig_stats.rdma_bytes, sge.length);
2131 * because we are using some kind of compression. I
2132 * would think that head.len would be the more similar
2133 * thing to a correct value.
2135 stat64_add(&mig_stats
.zero_pages
,
2136 sge
.length
/ qemu_target_page_size());
2141 * Otherwise, tell other side to register.
2143 reg
.current_index
= current_index
;
2144 if (block
->is_ram_block
) {
2145 reg
.key
.current_addr
= current_addr
;
2147 reg
.key
.chunk
= chunk
;
2149 reg
.chunks
= chunks
;
2151 trace_qemu_rdma_write_one_sendreg(chunk
, sge
.length
, current_index
,
2154 register_to_network(rdma
, ®
);
2155 ret
= qemu_rdma_exchange_send(rdma
, &head
, (uint8_t *) ®
,
2156 &resp
, ®_result_idx
, NULL
);
2161 /* try to overlap this single registration with the one we sent. */
2162 if (qemu_rdma_register_and_get_keys(rdma
, block
, sge
.addr
,
2163 &sge
.lkey
, NULL
, chunk
,
2164 chunk_start
, chunk_end
)) {
2165 error_report("cannot get lkey");
2169 reg_result
= (RDMARegisterResult
*)
2170 rdma
->wr_data
[reg_result_idx
].control_curr
;
2172 network_to_result(reg_result
);
2174 trace_qemu_rdma_write_one_recvregres(block
->remote_keys
[chunk
],
2175 reg_result
->rkey
, chunk
);
2177 block
->remote_keys
[chunk
] = reg_result
->rkey
;
2178 block
->remote_host_addr
= reg_result
->host_addr
;
2180 /* already registered before */
2181 if (qemu_rdma_register_and_get_keys(rdma
, block
, sge
.addr
,
2182 &sge
.lkey
, NULL
, chunk
,
2183 chunk_start
, chunk_end
)) {
2184 error_report("cannot get lkey!");
2189 send_wr
.wr
.rdma
.rkey
= block
->remote_keys
[chunk
];
2191 send_wr
.wr
.rdma
.rkey
= block
->remote_rkey
;
2193 if (qemu_rdma_register_and_get_keys(rdma
, block
, sge
.addr
,
2194 &sge
.lkey
, NULL
, chunk
,
2195 chunk_start
, chunk_end
)) {
2196 error_report("cannot get lkey!");
2202 * Encode the ram block index and chunk within this wrid.
2203 * We will use this information at the time of completion
2204 * to figure out which bitmap to check against and then which
2205 * chunk in the bitmap to look for.
2207 send_wr
.wr_id
= qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE
,
2208 current_index
, chunk
);
2210 send_wr
.opcode
= IBV_WR_RDMA_WRITE
;
2211 send_wr
.send_flags
= IBV_SEND_SIGNALED
;
2212 send_wr
.sg_list
= &sge
;
2213 send_wr
.num_sge
= 1;
2214 send_wr
.wr
.rdma
.remote_addr
= block
->remote_host_addr
+
2215 (current_addr
- block
->offset
);
2217 trace_qemu_rdma_write_one_post(chunk
, sge
.addr
, send_wr
.wr
.rdma
.remote_addr
,
2221 * ibv_post_send() does not return negative error numbers,
2222 * per the specification they are positive - no idea why.
2224 ret
= ibv_post_send(rdma
->qp
, &send_wr
, &bad_wr
);
2226 if (ret
== ENOMEM
) {
2227 trace_qemu_rdma_write_one_queue_full();
2228 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RDMA_WRITE
, NULL
);
2230 error_report("rdma migration: failed to make "
2231 "room in full send queue! %d", ret
);
2237 } else if (ret
> 0) {
2238 perror("rdma migration: post rdma write failed");
2242 set_bit(chunk
, block
->transit_bitmap
);
2243 stat64_add(&mig_stats
.normal_pages
, sge
.length
/ qemu_target_page_size());
2245 * We are adding to transferred the amount of data written, but no
2246 * overhead at all. I will asume that RDMA is magicaly and don't
2247 * need to transfer (at least) the addresses where it wants to
2248 * write the pages. Here it looks like it should be something
2250 * sizeof(send_wr) + sge.length
2251 * but this being RDMA, who knows.
2253 stat64_add(&mig_stats
.rdma_bytes
, sge
.length
);
2254 ram_transferred_add(sge
.length
);
2255 rdma
->total_writes
++;
2261 * Push out any unwritten RDMA operations.
2263 * We support sending out multiple chunks at the same time.
2264 * Not all of them need to get signaled in the completion queue.
2266 static int qemu_rdma_write_flush(QEMUFile
*f
, RDMAContext
*rdma
)
2270 if (!rdma
->current_length
) {
2274 ret
= qemu_rdma_write_one(f
, rdma
,
2275 rdma
->current_index
, rdma
->current_addr
, rdma
->current_length
);
2283 trace_qemu_rdma_write_flush(rdma
->nb_sent
);
2286 rdma
->current_length
= 0;
2287 rdma
->current_addr
= 0;
2292 static inline int qemu_rdma_buffer_mergable(RDMAContext
*rdma
,
2293 uint64_t offset
, uint64_t len
)
2295 RDMALocalBlock
*block
;
2299 if (rdma
->current_index
< 0) {
2303 if (rdma
->current_chunk
< 0) {
2307 block
= &(rdma
->local_ram_blocks
.block
[rdma
->current_index
]);
2308 host_addr
= block
->local_host_addr
+ (offset
- block
->offset
);
2309 chunk_end
= ram_chunk_end(block
, rdma
->current_chunk
);
2311 if (rdma
->current_length
== 0) {
2316 * Only merge into chunk sequentially.
2318 if (offset
!= (rdma
->current_addr
+ rdma
->current_length
)) {
2322 if (offset
< block
->offset
) {
2326 if ((offset
+ len
) > (block
->offset
+ block
->length
)) {
2330 if ((host_addr
+ len
) > chunk_end
) {
2338 * We're not actually writing here, but doing three things:
2340 * 1. Identify the chunk the buffer belongs to.
2341 * 2. If the chunk is full or the buffer doesn't belong to the current
2342 * chunk, then start a new chunk and flush() the old chunk.
2343 * 3. To keep the hardware busy, we also group chunks into batches
2344 * and only require that a batch gets acknowledged in the completion
2345 * queue instead of each individual chunk.
2347 static int qemu_rdma_write(QEMUFile
*f
, RDMAContext
*rdma
,
2348 uint64_t block_offset
, uint64_t offset
,
2351 uint64_t current_addr
= block_offset
+ offset
;
2352 uint64_t index
= rdma
->current_index
;
2353 uint64_t chunk
= rdma
->current_chunk
;
2356 /* If we cannot merge it, we flush the current buffer first. */
2357 if (!qemu_rdma_buffer_mergable(rdma
, current_addr
, len
)) {
2358 ret
= qemu_rdma_write_flush(f
, rdma
);
2362 rdma
->current_length
= 0;
2363 rdma
->current_addr
= current_addr
;
2365 ret
= qemu_rdma_search_ram_block(rdma
, block_offset
,
2366 offset
, len
, &index
, &chunk
);
2368 error_report("ram block search failed");
2371 rdma
->current_index
= index
;
2372 rdma
->current_chunk
= chunk
;
2376 rdma
->current_length
+= len
;
2378 /* flush it if buffer is too large */
2379 if (rdma
->current_length
>= RDMA_MERGE_MAX
) {
2380 return qemu_rdma_write_flush(f
, rdma
);
2386 static void qemu_rdma_cleanup(RDMAContext
*rdma
)
2390 if (rdma
->cm_id
&& rdma
->connected
) {
2391 if ((rdma
->error_state
||
2392 migrate_get_current()->state
== MIGRATION_STATUS_CANCELLING
) &&
2393 !rdma
->received_error
) {
2394 RDMAControlHeader head
= { .len
= 0,
2395 .type
= RDMA_CONTROL_ERROR
,
2398 error_report("Early error. Sending error.");
2399 qemu_rdma_post_send_control(rdma
, NULL
, &head
);
2402 rdma_disconnect(rdma
->cm_id
);
2403 trace_qemu_rdma_cleanup_disconnect();
2404 rdma
->connected
= false;
2407 if (rdma
->channel
) {
2408 qemu_set_fd_handler(rdma
->channel
->fd
, NULL
, NULL
, NULL
);
2410 g_free(rdma
->dest_blocks
);
2411 rdma
->dest_blocks
= NULL
;
2413 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
2414 if (rdma
->wr_data
[idx
].control_mr
) {
2415 rdma
->total_registrations
--;
2416 ibv_dereg_mr(rdma
->wr_data
[idx
].control_mr
);
2418 rdma
->wr_data
[idx
].control_mr
= NULL
;
2421 if (rdma
->local_ram_blocks
.block
) {
2422 while (rdma
->local_ram_blocks
.nb_blocks
) {
2423 rdma_delete_block(rdma
, &rdma
->local_ram_blocks
.block
[0]);
2428 rdma_destroy_qp(rdma
->cm_id
);
2431 if (rdma
->recv_cq
) {
2432 ibv_destroy_cq(rdma
->recv_cq
);
2433 rdma
->recv_cq
= NULL
;
2435 if (rdma
->send_cq
) {
2436 ibv_destroy_cq(rdma
->send_cq
);
2437 rdma
->send_cq
= NULL
;
2439 if (rdma
->recv_comp_channel
) {
2440 ibv_destroy_comp_channel(rdma
->recv_comp_channel
);
2441 rdma
->recv_comp_channel
= NULL
;
2443 if (rdma
->send_comp_channel
) {
2444 ibv_destroy_comp_channel(rdma
->send_comp_channel
);
2445 rdma
->send_comp_channel
= NULL
;
2448 ibv_dealloc_pd(rdma
->pd
);
2452 rdma_destroy_id(rdma
->cm_id
);
2456 /* the destination side, listen_id and channel is shared */
2457 if (rdma
->listen_id
) {
2458 if (!rdma
->is_return_path
) {
2459 rdma_destroy_id(rdma
->listen_id
);
2461 rdma
->listen_id
= NULL
;
2463 if (rdma
->channel
) {
2464 if (!rdma
->is_return_path
) {
2465 rdma_destroy_event_channel(rdma
->channel
);
2467 rdma
->channel
= NULL
;
2471 if (rdma
->channel
) {
2472 rdma_destroy_event_channel(rdma
->channel
);
2473 rdma
->channel
= NULL
;
2476 g_free(rdma
->host_port
);
2478 rdma
->host_port
= NULL
;
2482 static int qemu_rdma_source_init(RDMAContext
*rdma
, bool pin_all
, Error
**errp
)
2485 Error
*local_err
= NULL
, **temp
= &local_err
;
2488 * Will be validated against destination's actual capabilities
2489 * after the connect() completes.
2491 rdma
->pin_all
= pin_all
;
2493 ret
= qemu_rdma_resolve_host(rdma
, temp
);
2495 goto err_rdma_source_init
;
2498 ret
= qemu_rdma_alloc_pd_cq(rdma
);
2500 ERROR(temp
, "rdma migration: error allocating pd and cq! Your mlock()"
2501 " limits may be too low. Please check $ ulimit -a # and "
2502 "search for 'ulimit -l' in the output");
2503 goto err_rdma_source_init
;
2506 ret
= qemu_rdma_alloc_qp(rdma
);
2508 ERROR(temp
, "rdma migration: error allocating qp!");
2509 goto err_rdma_source_init
;
2512 ret
= qemu_rdma_init_ram_blocks(rdma
);
2514 ERROR(temp
, "rdma migration: error initializing ram blocks!");
2515 goto err_rdma_source_init
;
2518 /* Build the hash that maps from offset to RAMBlock */
2519 rdma
->blockmap
= g_hash_table_new(g_direct_hash
, g_direct_equal
);
2520 for (idx
= 0; idx
< rdma
->local_ram_blocks
.nb_blocks
; idx
++) {
2521 g_hash_table_insert(rdma
->blockmap
,
2522 (void *)(uintptr_t)rdma
->local_ram_blocks
.block
[idx
].offset
,
2523 &rdma
->local_ram_blocks
.block
[idx
]);
2526 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
2527 ret
= qemu_rdma_reg_control(rdma
, idx
);
2529 ERROR(temp
, "rdma migration: error registering %d control!",
2531 goto err_rdma_source_init
;
2537 err_rdma_source_init
:
2538 error_propagate(errp
, local_err
);
2539 qemu_rdma_cleanup(rdma
);
2543 static int qemu_get_cm_event_timeout(RDMAContext
*rdma
,
2544 struct rdma_cm_event
**cm_event
,
2545 long msec
, Error
**errp
)
2548 struct pollfd poll_fd
= {
2549 .fd
= rdma
->channel
->fd
,
2555 ret
= poll(&poll_fd
, 1, msec
);
2556 } while (ret
< 0 && errno
== EINTR
);
2559 ERROR(errp
, "poll cm event timeout");
2561 } else if (ret
< 0) {
2562 ERROR(errp
, "failed to poll cm event, errno=%i", errno
);
2564 } else if (poll_fd
.revents
& POLLIN
) {
2565 return rdma_get_cm_event(rdma
->channel
, cm_event
);
2567 ERROR(errp
, "no POLLIN event, revent=%x", poll_fd
.revents
);
2572 static int qemu_rdma_connect(RDMAContext
*rdma
, Error
**errp
, bool return_path
)
2574 RDMACapabilities cap
= {
2575 .version
= RDMA_CONTROL_VERSION_CURRENT
,
2578 struct rdma_conn_param conn_param
= { .initiator_depth
= 2,
2580 .private_data
= &cap
,
2581 .private_data_len
= sizeof(cap
),
2583 struct rdma_cm_event
*cm_event
;
2587 * Only negotiate the capability with destination if the user
2588 * on the source first requested the capability.
2590 if (rdma
->pin_all
) {
2591 trace_qemu_rdma_connect_pin_all_requested();
2592 cap
.flags
|= RDMA_CAPABILITY_PIN_ALL
;
2595 caps_to_network(&cap
);
2597 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
2599 ERROR(errp
, "posting second control recv");
2600 goto err_rdma_source_connect
;
2603 ret
= rdma_connect(rdma
->cm_id
, &conn_param
);
2605 perror("rdma_connect");
2606 ERROR(errp
, "connecting to destination!");
2607 goto err_rdma_source_connect
;
2611 ret
= qemu_get_cm_event_timeout(rdma
, &cm_event
, 5000, errp
);
2613 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
2616 perror("rdma_get_cm_event after rdma_connect");
2617 ERROR(errp
, "connecting to destination!");
2618 goto err_rdma_source_connect
;
2621 if (cm_event
->event
!= RDMA_CM_EVENT_ESTABLISHED
) {
2622 error_report("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect");
2623 ERROR(errp
, "connecting to destination!");
2624 rdma_ack_cm_event(cm_event
);
2625 goto err_rdma_source_connect
;
2627 rdma
->connected
= true;
2629 memcpy(&cap
, cm_event
->param
.conn
.private_data
, sizeof(cap
));
2630 network_to_caps(&cap
);
2633 * Verify that the *requested* capabilities are supported by the destination
2634 * and disable them otherwise.
2636 if (rdma
->pin_all
&& !(cap
.flags
& RDMA_CAPABILITY_PIN_ALL
)) {
2637 ERROR(errp
, "Server cannot support pinning all memory. "
2638 "Will register memory dynamically.");
2639 rdma
->pin_all
= false;
2642 trace_qemu_rdma_connect_pin_all_outcome(rdma
->pin_all
);
2644 rdma_ack_cm_event(cm_event
);
2646 rdma
->control_ready_expected
= 1;
2650 err_rdma_source_connect
:
2651 qemu_rdma_cleanup(rdma
);
2655 static int qemu_rdma_dest_init(RDMAContext
*rdma
, Error
**errp
)
2658 struct rdma_cm_id
*listen_id
;
2659 char ip
[40] = "unknown";
2660 struct rdma_addrinfo
*res
, *e
;
2664 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
2665 rdma
->wr_data
[idx
].control_len
= 0;
2666 rdma
->wr_data
[idx
].control_curr
= NULL
;
2669 if (!rdma
->host
|| !rdma
->host
[0]) {
2670 ERROR(errp
, "RDMA host is not set!");
2671 rdma
->error_state
= -EINVAL
;
2674 /* create CM channel */
2675 rdma
->channel
= rdma_create_event_channel();
2676 if (!rdma
->channel
) {
2677 ERROR(errp
, "could not create rdma event channel");
2678 rdma
->error_state
= -EINVAL
;
2683 ret
= rdma_create_id(rdma
->channel
, &listen_id
, NULL
, RDMA_PS_TCP
);
2685 ERROR(errp
, "could not create cm_id!");
2686 goto err_dest_init_create_listen_id
;
2689 snprintf(port_str
, 16, "%d", rdma
->port
);
2690 port_str
[15] = '\0';
2692 ret
= rdma_getaddrinfo(rdma
->host
, port_str
, NULL
, &res
);
2694 ERROR(errp
, "could not rdma_getaddrinfo address %s", rdma
->host
);
2695 goto err_dest_init_bind_addr
;
2698 ret
= rdma_set_option(listen_id
, RDMA_OPTION_ID
, RDMA_OPTION_ID_REUSEADDR
,
2699 &reuse
, sizeof reuse
);
2701 ERROR(errp
, "Error: could not set REUSEADDR option");
2702 goto err_dest_init_bind_addr
;
2704 for (e
= res
; e
!= NULL
; e
= e
->ai_next
) {
2705 inet_ntop(e
->ai_family
,
2706 &((struct sockaddr_in
*) e
->ai_dst_addr
)->sin_addr
, ip
, sizeof ip
);
2707 trace_qemu_rdma_dest_init_trying(rdma
->host
, ip
);
2708 ret
= rdma_bind_addr(listen_id
, e
->ai_dst_addr
);
2712 if (e
->ai_family
== AF_INET6
) {
2713 ret
= qemu_rdma_broken_ipv6_kernel(listen_id
->verbs
, errp
);
2721 rdma_freeaddrinfo(res
);
2723 ERROR(errp
, "Error: could not rdma_bind_addr!");
2724 goto err_dest_init_bind_addr
;
2727 rdma
->listen_id
= listen_id
;
2728 qemu_rdma_dump_gid("dest_init", listen_id
);
2731 err_dest_init_bind_addr
:
2732 rdma_destroy_id(listen_id
);
2733 err_dest_init_create_listen_id
:
2734 rdma_destroy_event_channel(rdma
->channel
);
2735 rdma
->channel
= NULL
;
2736 rdma
->error_state
= ret
;
2741 static void qemu_rdma_return_path_dest_init(RDMAContext
*rdma_return_path
,
2746 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
2747 rdma_return_path
->wr_data
[idx
].control_len
= 0;
2748 rdma_return_path
->wr_data
[idx
].control_curr
= NULL
;
2751 /*the CM channel and CM id is shared*/
2752 rdma_return_path
->channel
= rdma
->channel
;
2753 rdma_return_path
->listen_id
= rdma
->listen_id
;
2755 rdma
->return_path
= rdma_return_path
;
2756 rdma_return_path
->return_path
= rdma
;
2757 rdma_return_path
->is_return_path
= true;
2760 static void *qemu_rdma_data_init(const char *host_port
, Error
**errp
)
2762 RDMAContext
*rdma
= NULL
;
2763 InetSocketAddress
*addr
;
2766 rdma
= g_new0(RDMAContext
, 1);
2767 rdma
->current_index
= -1;
2768 rdma
->current_chunk
= -1;
2770 addr
= g_new(InetSocketAddress
, 1);
2771 if (!inet_parse(addr
, host_port
, NULL
)) {
2772 rdma
->port
= atoi(addr
->port
);
2773 rdma
->host
= g_strdup(addr
->host
);
2774 rdma
->host_port
= g_strdup(host_port
);
2776 ERROR(errp
, "bad RDMA migration address '%s'", host_port
);
2781 qapi_free_InetSocketAddress(addr
);
2788 * QEMUFile interface to the control channel.
2789 * SEND messages for control only.
2790 * VM's ram is handled with regular RDMA messages.
2792 static ssize_t
qio_channel_rdma_writev(QIOChannel
*ioc
,
2793 const struct iovec
*iov
,
2800 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
2801 QEMUFile
*f
= rioc
->file
;
2808 RCU_READ_LOCK_GUARD();
2809 rdma
= qatomic_rcu_read(&rioc
->rdmaout
);
2812 error_setg(errp
, "RDMA control channel output is not set");
2816 CHECK_ERROR_STATE();
2819 * Push out any writes that
2820 * we're queued up for VM's ram.
2822 ret
= qemu_rdma_write_flush(f
, rdma
);
2824 rdma
->error_state
= ret
;
2825 error_setg(errp
, "qemu_rdma_write_flush returned %d", ret
);
2829 for (i
= 0; i
< niov
; i
++) {
2830 size_t remaining
= iov
[i
].iov_len
;
2831 uint8_t * data
= (void *)iov
[i
].iov_base
;
2833 RDMAControlHeader head
;
2835 len
= MIN(remaining
, RDMA_SEND_INCREMENT
);
2839 head
.type
= RDMA_CONTROL_QEMU_FILE
;
2841 ret
= qemu_rdma_exchange_send(rdma
, &head
, data
, NULL
, NULL
, NULL
);
2844 rdma
->error_state
= ret
;
2845 error_setg(errp
, "qemu_rdma_exchange_send returned %d", ret
);
2857 static size_t qemu_rdma_fill(RDMAContext
*rdma
, uint8_t *buf
,
2858 size_t size
, int idx
)
2862 if (rdma
->wr_data
[idx
].control_len
) {
2863 trace_qemu_rdma_fill(rdma
->wr_data
[idx
].control_len
, size
);
2865 len
= MIN(size
, rdma
->wr_data
[idx
].control_len
);
2866 memcpy(buf
, rdma
->wr_data
[idx
].control_curr
, len
);
2867 rdma
->wr_data
[idx
].control_curr
+= len
;
2868 rdma
->wr_data
[idx
].control_len
-= len
;
2875 * QEMUFile interface to the control channel.
2876 * RDMA links don't use bytestreams, so we have to
2877 * return bytes to QEMUFile opportunistically.
2879 static ssize_t
qio_channel_rdma_readv(QIOChannel
*ioc
,
2880 const struct iovec
*iov
,
2887 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
2889 RDMAControlHeader head
;
2894 RCU_READ_LOCK_GUARD();
2895 rdma
= qatomic_rcu_read(&rioc
->rdmain
);
2898 error_setg(errp
, "RDMA control channel input is not set");
2902 CHECK_ERROR_STATE();
2904 for (i
= 0; i
< niov
; i
++) {
2905 size_t want
= iov
[i
].iov_len
;
2906 uint8_t *data
= (void *)iov
[i
].iov_base
;
2909 * First, we hold on to the last SEND message we
2910 * were given and dish out the bytes until we run
2913 ret
= qemu_rdma_fill(rdma
, data
, want
, 0);
2916 /* Got what we needed, so go to next iovec */
2921 /* If we got any data so far, then don't wait
2922 * for more, just return what we have */
2928 /* We've got nothing at all, so lets wait for
2931 ret
= qemu_rdma_exchange_recv(rdma
, &head
, RDMA_CONTROL_QEMU_FILE
);
2934 rdma
->error_state
= ret
;
2935 error_setg(errp
, "qemu_rdma_exchange_recv returned %d", ret
);
2940 * SEND was received with new bytes, now try again.
2942 ret
= qemu_rdma_fill(rdma
, data
, want
, 0);
2946 /* Still didn't get enough, so lets just return */
2949 return QIO_CHANNEL_ERR_BLOCK
;
2959 * Block until all the outstanding chunks have been delivered by the hardware.
2961 static int qemu_rdma_drain_cq(QEMUFile
*f
, RDMAContext
*rdma
)
2965 if (qemu_rdma_write_flush(f
, rdma
) < 0) {
2969 while (rdma
->nb_sent
) {
2970 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RDMA_WRITE
, NULL
);
2972 error_report("rdma migration: complete polling error!");
2977 qemu_rdma_unregister_waiting(rdma
);
2983 static int qio_channel_rdma_set_blocking(QIOChannel
*ioc
,
2987 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
2988 /* XXX we should make readv/writev actually honour this :-) */
2989 rioc
->blocking
= blocking
;
2994 typedef struct QIOChannelRDMASource QIOChannelRDMASource
;
2995 struct QIOChannelRDMASource
{
2997 QIOChannelRDMA
*rioc
;
2998 GIOCondition condition
;
3002 qio_channel_rdma_source_prepare(GSource
*source
,
3005 QIOChannelRDMASource
*rsource
= (QIOChannelRDMASource
*)source
;
3007 GIOCondition cond
= 0;
3010 RCU_READ_LOCK_GUARD();
3011 if (rsource
->condition
== G_IO_IN
) {
3012 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmain
);
3014 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmaout
);
3018 error_report("RDMAContext is NULL when prepare Gsource");
3022 if (rdma
->wr_data
[0].control_len
) {
3027 return cond
& rsource
->condition
;
3031 qio_channel_rdma_source_check(GSource
*source
)
3033 QIOChannelRDMASource
*rsource
= (QIOChannelRDMASource
*)source
;
3035 GIOCondition cond
= 0;
3037 RCU_READ_LOCK_GUARD();
3038 if (rsource
->condition
== G_IO_IN
) {
3039 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmain
);
3041 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmaout
);
3045 error_report("RDMAContext is NULL when check Gsource");
3049 if (rdma
->wr_data
[0].control_len
) {
3054 return cond
& rsource
->condition
;
3058 qio_channel_rdma_source_dispatch(GSource
*source
,
3059 GSourceFunc callback
,
3062 QIOChannelFunc func
= (QIOChannelFunc
)callback
;
3063 QIOChannelRDMASource
*rsource
= (QIOChannelRDMASource
*)source
;
3065 GIOCondition cond
= 0;
3067 RCU_READ_LOCK_GUARD();
3068 if (rsource
->condition
== G_IO_IN
) {
3069 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmain
);
3071 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmaout
);
3075 error_report("RDMAContext is NULL when dispatch Gsource");
3079 if (rdma
->wr_data
[0].control_len
) {
3084 return (*func
)(QIO_CHANNEL(rsource
->rioc
),
3085 (cond
& rsource
->condition
),
3090 qio_channel_rdma_source_finalize(GSource
*source
)
3092 QIOChannelRDMASource
*ssource
= (QIOChannelRDMASource
*)source
;
3094 object_unref(OBJECT(ssource
->rioc
));
3097 GSourceFuncs qio_channel_rdma_source_funcs
= {
3098 qio_channel_rdma_source_prepare
,
3099 qio_channel_rdma_source_check
,
3100 qio_channel_rdma_source_dispatch
,
3101 qio_channel_rdma_source_finalize
3104 static GSource
*qio_channel_rdma_create_watch(QIOChannel
*ioc
,
3105 GIOCondition condition
)
3107 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
3108 QIOChannelRDMASource
*ssource
;
3111 source
= g_source_new(&qio_channel_rdma_source_funcs
,
3112 sizeof(QIOChannelRDMASource
));
3113 ssource
= (QIOChannelRDMASource
*)source
;
3115 ssource
->rioc
= rioc
;
3116 object_ref(OBJECT(rioc
));
3118 ssource
->condition
= condition
;
3123 static void qio_channel_rdma_set_aio_fd_handler(QIOChannel
*ioc
,
3124 AioContext
*read_ctx
,
3126 AioContext
*write_ctx
,
3127 IOHandler
*io_write
,
3130 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
3132 aio_set_fd_handler(read_ctx
, rioc
->rdmain
->recv_comp_channel
->fd
,
3133 io_read
, io_write
, NULL
, NULL
, opaque
);
3134 aio_set_fd_handler(read_ctx
, rioc
->rdmain
->send_comp_channel
->fd
,
3135 io_read
, io_write
, NULL
, NULL
, opaque
);
3137 aio_set_fd_handler(write_ctx
, rioc
->rdmaout
->recv_comp_channel
->fd
,
3138 io_read
, io_write
, NULL
, NULL
, opaque
);
3139 aio_set_fd_handler(write_ctx
, rioc
->rdmaout
->send_comp_channel
->fd
,
3140 io_read
, io_write
, NULL
, NULL
, opaque
);
3144 struct rdma_close_rcu
{
3145 struct rcu_head rcu
;
3146 RDMAContext
*rdmain
;
3147 RDMAContext
*rdmaout
;
3150 /* callback from qio_channel_rdma_close via call_rcu */
3151 static void qio_channel_rdma_close_rcu(struct rdma_close_rcu
*rcu
)
3154 qemu_rdma_cleanup(rcu
->rdmain
);
3158 qemu_rdma_cleanup(rcu
->rdmaout
);
3161 g_free(rcu
->rdmain
);
3162 g_free(rcu
->rdmaout
);
3166 static int qio_channel_rdma_close(QIOChannel
*ioc
,
3169 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
3170 RDMAContext
*rdmain
, *rdmaout
;
3171 struct rdma_close_rcu
*rcu
= g_new(struct rdma_close_rcu
, 1);
3173 trace_qemu_rdma_close();
3175 rdmain
= rioc
->rdmain
;
3177 qatomic_rcu_set(&rioc
->rdmain
, NULL
);
3180 rdmaout
= rioc
->rdmaout
;
3182 qatomic_rcu_set(&rioc
->rdmaout
, NULL
);
3185 rcu
->rdmain
= rdmain
;
3186 rcu
->rdmaout
= rdmaout
;
3187 call_rcu(rcu
, qio_channel_rdma_close_rcu
, rcu
);
3193 qio_channel_rdma_shutdown(QIOChannel
*ioc
,
3194 QIOChannelShutdown how
,
3197 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
3198 RDMAContext
*rdmain
, *rdmaout
;
3200 RCU_READ_LOCK_GUARD();
3202 rdmain
= qatomic_rcu_read(&rioc
->rdmain
);
3203 rdmaout
= qatomic_rcu_read(&rioc
->rdmain
);
3206 case QIO_CHANNEL_SHUTDOWN_READ
:
3208 rdmain
->error_state
= -1;
3211 case QIO_CHANNEL_SHUTDOWN_WRITE
:
3213 rdmaout
->error_state
= -1;
3216 case QIO_CHANNEL_SHUTDOWN_BOTH
:
3219 rdmain
->error_state
= -1;
3222 rdmaout
->error_state
= -1;
3233 * This means that 'block_offset' is a full virtual address that does not
3234 * belong to a RAMBlock of the virtual machine and instead
3235 * represents a private malloc'd memory area that the caller wishes to
3239 * Offset is an offset to be added to block_offset and used
3240 * to also lookup the corresponding RAMBlock.
3242 * @size : Number of bytes to transfer
3244 * @bytes_sent : User-specificed pointer to indicate how many bytes were
3245 * sent. Usually, this will not be more than a few bytes of
3246 * the protocol because most transfers are sent asynchronously.
3248 static size_t qemu_rdma_save_page(QEMUFile
*f
,
3249 ram_addr_t block_offset
, ram_addr_t offset
,
3250 size_t size
, uint64_t *bytes_sent
)
3252 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(qemu_file_get_ioc(f
));
3256 if (migration_in_postcopy()) {
3257 return RAM_SAVE_CONTROL_NOT_SUPP
;
3260 RCU_READ_LOCK_GUARD();
3261 rdma
= qatomic_rcu_read(&rioc
->rdmaout
);
3267 CHECK_ERROR_STATE();
3272 * Add this page to the current 'chunk'. If the chunk
3273 * is full, or the page doesn't belong to the current chunk,
3274 * an actual RDMA write will occur and a new chunk will be formed.
3276 ret
= qemu_rdma_write(f
, rdma
, block_offset
, offset
, size
);
3278 error_report("rdma migration: write error! %d", ret
);
3283 * We always return 1 bytes because the RDMA
3284 * protocol is completely asynchronous. We do not yet know
3285 * whether an identified chunk is zero or not because we're
3286 * waiting for other pages to potentially be merged with
3287 * the current chunk. So, we have to call qemu_update_position()
3288 * later on when the actual write occurs.
3295 * Drain the Completion Queue if possible, but do not block,
3298 * If nothing to poll, the end of the iteration will do this
3299 * again to make sure we don't overflow the request queue.
3302 uint64_t wr_id
, wr_id_in
;
3303 int ret
= qemu_rdma_poll(rdma
, rdma
->recv_cq
, &wr_id_in
, NULL
);
3305 error_report("rdma migration: polling error! %d", ret
);
3309 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
3311 if (wr_id
== RDMA_WRID_NONE
) {
3317 uint64_t wr_id
, wr_id_in
;
3318 int ret
= qemu_rdma_poll(rdma
, rdma
->send_cq
, &wr_id_in
, NULL
);
3320 error_report("rdma migration: polling error! %d", ret
);
3324 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
3326 if (wr_id
== RDMA_WRID_NONE
) {
3331 return RAM_SAVE_CONTROL_DELAYED
;
3333 rdma
->error_state
= ret
;
3337 static void rdma_accept_incoming_migration(void *opaque
);
3339 static void rdma_cm_poll_handler(void *opaque
)
3341 RDMAContext
*rdma
= opaque
;
3343 struct rdma_cm_event
*cm_event
;
3344 MigrationIncomingState
*mis
= migration_incoming_get_current();
3346 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
3348 error_report("get_cm_event failed %d", errno
);
3352 if (cm_event
->event
== RDMA_CM_EVENT_DISCONNECTED
||
3353 cm_event
->event
== RDMA_CM_EVENT_DEVICE_REMOVAL
) {
3354 if (!rdma
->error_state
&&
3355 migration_incoming_get_current()->state
!=
3356 MIGRATION_STATUS_COMPLETED
) {
3357 error_report("receive cm event, cm event is %d", cm_event
->event
);
3358 rdma
->error_state
= -EPIPE
;
3359 if (rdma
->return_path
) {
3360 rdma
->return_path
->error_state
= -EPIPE
;
3363 rdma_ack_cm_event(cm_event
);
3364 if (mis
->loadvm_co
) {
3365 qemu_coroutine_enter(mis
->loadvm_co
);
3369 rdma_ack_cm_event(cm_event
);
3372 static int qemu_rdma_accept(RDMAContext
*rdma
)
3374 RDMACapabilities cap
;
3375 struct rdma_conn_param conn_param
= {
3376 .responder_resources
= 2,
3377 .private_data
= &cap
,
3378 .private_data_len
= sizeof(cap
),
3380 RDMAContext
*rdma_return_path
= NULL
;
3381 struct rdma_cm_event
*cm_event
;
3382 struct ibv_context
*verbs
;
3386 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
3388 goto err_rdma_dest_wait
;
3391 if (cm_event
->event
!= RDMA_CM_EVENT_CONNECT_REQUEST
) {
3392 rdma_ack_cm_event(cm_event
);
3393 goto err_rdma_dest_wait
;
3397 * initialize the RDMAContext for return path for postcopy after first
3398 * connection request reached.
3400 if ((migrate_postcopy() || migrate_return_path())
3401 && !rdma
->is_return_path
) {
3402 rdma_return_path
= qemu_rdma_data_init(rdma
->host_port
, NULL
);
3403 if (rdma_return_path
== NULL
) {
3404 rdma_ack_cm_event(cm_event
);
3405 goto err_rdma_dest_wait
;
3408 qemu_rdma_return_path_dest_init(rdma_return_path
, rdma
);
3411 memcpy(&cap
, cm_event
->param
.conn
.private_data
, sizeof(cap
));
3413 network_to_caps(&cap
);
3415 if (cap
.version
< 1 || cap
.version
> RDMA_CONTROL_VERSION_CURRENT
) {
3416 error_report("Unknown source RDMA version: %d, bailing...",
3418 rdma_ack_cm_event(cm_event
);
3419 goto err_rdma_dest_wait
;
3423 * Respond with only the capabilities this version of QEMU knows about.
3425 cap
.flags
&= known_capabilities
;
3428 * Enable the ones that we do know about.
3429 * Add other checks here as new ones are introduced.
3431 if (cap
.flags
& RDMA_CAPABILITY_PIN_ALL
) {
3432 rdma
->pin_all
= true;
3435 rdma
->cm_id
= cm_event
->id
;
3436 verbs
= cm_event
->id
->verbs
;
3438 rdma_ack_cm_event(cm_event
);
3440 trace_qemu_rdma_accept_pin_state(rdma
->pin_all
);
3442 caps_to_network(&cap
);
3444 trace_qemu_rdma_accept_pin_verbsc(verbs
);
3447 rdma
->verbs
= verbs
;
3448 } else if (rdma
->verbs
!= verbs
) {
3449 error_report("ibv context not matching %p, %p!", rdma
->verbs
,
3451 goto err_rdma_dest_wait
;
3454 qemu_rdma_dump_id("dest_init", verbs
);
3456 ret
= qemu_rdma_alloc_pd_cq(rdma
);
3458 error_report("rdma migration: error allocating pd and cq!");
3459 goto err_rdma_dest_wait
;
3462 ret
= qemu_rdma_alloc_qp(rdma
);
3464 error_report("rdma migration: error allocating qp!");
3465 goto err_rdma_dest_wait
;
3468 ret
= qemu_rdma_init_ram_blocks(rdma
);
3470 error_report("rdma migration: error initializing ram blocks!");
3471 goto err_rdma_dest_wait
;
3474 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
3475 ret
= qemu_rdma_reg_control(rdma
, idx
);
3477 error_report("rdma: error registering %d control", idx
);
3478 goto err_rdma_dest_wait
;
3482 /* Accept the second connection request for return path */
3483 if ((migrate_postcopy() || migrate_return_path())
3484 && !rdma
->is_return_path
) {
3485 qemu_set_fd_handler(rdma
->channel
->fd
, rdma_accept_incoming_migration
,
3487 (void *)(intptr_t)rdma
->return_path
);
3489 qemu_set_fd_handler(rdma
->channel
->fd
, rdma_cm_poll_handler
,
3493 ret
= rdma_accept(rdma
->cm_id
, &conn_param
);
3495 error_report("rdma_accept returns %d", ret
);
3496 goto err_rdma_dest_wait
;
3499 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
3501 error_report("rdma_accept get_cm_event failed %d", ret
);
3502 goto err_rdma_dest_wait
;
3505 if (cm_event
->event
!= RDMA_CM_EVENT_ESTABLISHED
) {
3506 error_report("rdma_accept not event established");
3507 rdma_ack_cm_event(cm_event
);
3508 goto err_rdma_dest_wait
;
3511 rdma_ack_cm_event(cm_event
);
3512 rdma
->connected
= true;
3514 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
3516 error_report("rdma migration: error posting second control recv");
3517 goto err_rdma_dest_wait
;
3520 qemu_rdma_dump_gid("dest_connect", rdma
->cm_id
);
3525 rdma
->error_state
= ret
;
3526 qemu_rdma_cleanup(rdma
);
3527 g_free(rdma_return_path
);
3531 static int dest_ram_sort_func(const void *a
, const void *b
)
3533 unsigned int a_index
= ((const RDMALocalBlock
*)a
)->src_index
;
3534 unsigned int b_index
= ((const RDMALocalBlock
*)b
)->src_index
;
3536 return (a_index
< b_index
) ? -1 : (a_index
!= b_index
);
3540 * During each iteration of the migration, we listen for instructions
3541 * by the source VM to perform dynamic page registrations before they
3542 * can perform RDMA operations.
3544 * We respond with the 'rkey'.
3546 * Keep doing this until the source tells us to stop.
3548 static int qemu_rdma_registration_handle(QEMUFile
*f
)
3550 RDMAControlHeader reg_resp
= { .len
= sizeof(RDMARegisterResult
),
3551 .type
= RDMA_CONTROL_REGISTER_RESULT
,
3554 RDMAControlHeader unreg_resp
= { .len
= 0,
3555 .type
= RDMA_CONTROL_UNREGISTER_FINISHED
,
3558 RDMAControlHeader blocks
= { .type
= RDMA_CONTROL_RAM_BLOCKS_RESULT
,
3560 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(qemu_file_get_ioc(f
));
3562 RDMALocalBlocks
*local
;
3563 RDMAControlHeader head
;
3564 RDMARegister
*reg
, *registers
;
3566 RDMARegisterResult
*reg_result
;
3567 static RDMARegisterResult results
[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE
];
3568 RDMALocalBlock
*block
;
3575 RCU_READ_LOCK_GUARD();
3576 rdma
= qatomic_rcu_read(&rioc
->rdmain
);
3582 CHECK_ERROR_STATE();
3584 local
= &rdma
->local_ram_blocks
;
3586 trace_qemu_rdma_registration_handle_wait();
3588 ret
= qemu_rdma_exchange_recv(rdma
, &head
, RDMA_CONTROL_NONE
);
3594 if (head
.repeat
> RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE
) {
3595 error_report("rdma: Too many requests in this message (%d)."
3596 "Bailing.", head
.repeat
);
3601 switch (head
.type
) {
3602 case RDMA_CONTROL_COMPRESS
:
3603 comp
= (RDMACompress
*) rdma
->wr_data
[idx
].control_curr
;
3604 network_to_compress(comp
);
3606 trace_qemu_rdma_registration_handle_compress(comp
->length
,
3609 if (comp
->block_idx
>= rdma
->local_ram_blocks
.nb_blocks
) {
3610 error_report("rdma: 'compress' bad block index %u (vs %d)",
3611 (unsigned int)comp
->block_idx
,
3612 rdma
->local_ram_blocks
.nb_blocks
);
3616 block
= &(rdma
->local_ram_blocks
.block
[comp
->block_idx
]);
3618 host_addr
= block
->local_host_addr
+
3619 (comp
->offset
- block
->offset
);
3621 ram_handle_compressed(host_addr
, comp
->value
, comp
->length
);
3624 case RDMA_CONTROL_REGISTER_FINISHED
:
3625 trace_qemu_rdma_registration_handle_finished();
3628 case RDMA_CONTROL_RAM_BLOCKS_REQUEST
:
3629 trace_qemu_rdma_registration_handle_ram_blocks();
3631 /* Sort our local RAM Block list so it's the same as the source,
3632 * we can do this since we've filled in a src_index in the list
3633 * as we received the RAMBlock list earlier.
3635 qsort(rdma
->local_ram_blocks
.block
,
3636 rdma
->local_ram_blocks
.nb_blocks
,
3637 sizeof(RDMALocalBlock
), dest_ram_sort_func
);
3638 for (i
= 0; i
< local
->nb_blocks
; i
++) {
3639 local
->block
[i
].index
= i
;
3642 if (rdma
->pin_all
) {
3643 ret
= qemu_rdma_reg_whole_ram_blocks(rdma
);
3645 error_report("rdma migration: error dest "
3646 "registering ram blocks");
3652 * Dest uses this to prepare to transmit the RAMBlock descriptions
3653 * to the source VM after connection setup.
3654 * Both sides use the "remote" structure to communicate and update
3655 * their "local" descriptions with what was sent.
3657 for (i
= 0; i
< local
->nb_blocks
; i
++) {
3658 rdma
->dest_blocks
[i
].remote_host_addr
=
3659 (uintptr_t)(local
->block
[i
].local_host_addr
);
3661 if (rdma
->pin_all
) {
3662 rdma
->dest_blocks
[i
].remote_rkey
= local
->block
[i
].mr
->rkey
;
3665 rdma
->dest_blocks
[i
].offset
= local
->block
[i
].offset
;
3666 rdma
->dest_blocks
[i
].length
= local
->block
[i
].length
;
3668 dest_block_to_network(&rdma
->dest_blocks
[i
]);
3669 trace_qemu_rdma_registration_handle_ram_blocks_loop(
3670 local
->block
[i
].block_name
,
3671 local
->block
[i
].offset
,
3672 local
->block
[i
].length
,
3673 local
->block
[i
].local_host_addr
,
3674 local
->block
[i
].src_index
);
3677 blocks
.len
= rdma
->local_ram_blocks
.nb_blocks
3678 * sizeof(RDMADestBlock
);
3681 ret
= qemu_rdma_post_send_control(rdma
,
3682 (uint8_t *) rdma
->dest_blocks
, &blocks
);
3685 error_report("rdma migration: error sending remote info");
3690 case RDMA_CONTROL_REGISTER_REQUEST
:
3691 trace_qemu_rdma_registration_handle_register(head
.repeat
);
3693 reg_resp
.repeat
= head
.repeat
;
3694 registers
= (RDMARegister
*) rdma
->wr_data
[idx
].control_curr
;
3696 for (count
= 0; count
< head
.repeat
; count
++) {
3698 uint8_t *chunk_start
, *chunk_end
;
3700 reg
= ®isters
[count
];
3701 network_to_register(reg
);
3703 reg_result
= &results
[count
];
3705 trace_qemu_rdma_registration_handle_register_loop(count
,
3706 reg
->current_index
, reg
->key
.current_addr
, reg
->chunks
);
3708 if (reg
->current_index
>= rdma
->local_ram_blocks
.nb_blocks
) {
3709 error_report("rdma: 'register' bad block index %u (vs %d)",
3710 (unsigned int)reg
->current_index
,
3711 rdma
->local_ram_blocks
.nb_blocks
);
3715 block
= &(rdma
->local_ram_blocks
.block
[reg
->current_index
]);
3716 if (block
->is_ram_block
) {
3717 if (block
->offset
> reg
->key
.current_addr
) {
3718 error_report("rdma: bad register address for block %s"
3719 " offset: %" PRIx64
" current_addr: %" PRIx64
,
3720 block
->block_name
, block
->offset
,
3721 reg
->key
.current_addr
);
3725 host_addr
= (block
->local_host_addr
+
3726 (reg
->key
.current_addr
- block
->offset
));
3727 chunk
= ram_chunk_index(block
->local_host_addr
,
3728 (uint8_t *) host_addr
);
3730 chunk
= reg
->key
.chunk
;
3731 host_addr
= block
->local_host_addr
+
3732 (reg
->key
.chunk
* (1UL << RDMA_REG_CHUNK_SHIFT
));
3733 /* Check for particularly bad chunk value */
3734 if (host_addr
< (void *)block
->local_host_addr
) {
3735 error_report("rdma: bad chunk for block %s"
3737 block
->block_name
, reg
->key
.chunk
);
3742 chunk_start
= ram_chunk_start(block
, chunk
);
3743 chunk_end
= ram_chunk_end(block
, chunk
+ reg
->chunks
);
3744 /* avoid "-Waddress-of-packed-member" warning */
3745 uint32_t tmp_rkey
= 0;
3746 if (qemu_rdma_register_and_get_keys(rdma
, block
,
3747 (uintptr_t)host_addr
, NULL
, &tmp_rkey
,
3748 chunk
, chunk_start
, chunk_end
)) {
3749 error_report("cannot get rkey");
3753 reg_result
->rkey
= tmp_rkey
;
3755 reg_result
->host_addr
= (uintptr_t)block
->local_host_addr
;
3757 trace_qemu_rdma_registration_handle_register_rkey(
3760 result_to_network(reg_result
);
3763 ret
= qemu_rdma_post_send_control(rdma
,
3764 (uint8_t *) results
, ®_resp
);
3767 error_report("Failed to send control buffer");
3771 case RDMA_CONTROL_UNREGISTER_REQUEST
:
3772 trace_qemu_rdma_registration_handle_unregister(head
.repeat
);
3773 unreg_resp
.repeat
= head
.repeat
;
3774 registers
= (RDMARegister
*) rdma
->wr_data
[idx
].control_curr
;
3776 for (count
= 0; count
< head
.repeat
; count
++) {
3777 reg
= ®isters
[count
];
3778 network_to_register(reg
);
3780 trace_qemu_rdma_registration_handle_unregister_loop(count
,
3781 reg
->current_index
, reg
->key
.chunk
);
3783 block
= &(rdma
->local_ram_blocks
.block
[reg
->current_index
]);
3785 ret
= ibv_dereg_mr(block
->pmr
[reg
->key
.chunk
]);
3786 block
->pmr
[reg
->key
.chunk
] = NULL
;
3789 perror("rdma unregistration chunk failed");
3794 rdma
->total_registrations
--;
3796 trace_qemu_rdma_registration_handle_unregister_success(
3800 ret
= qemu_rdma_post_send_control(rdma
, NULL
, &unreg_resp
);
3803 error_report("Failed to send control buffer");
3807 case RDMA_CONTROL_REGISTER_RESULT
:
3808 error_report("Invalid RESULT message at dest.");
3812 error_report("Unknown control message %s", control_desc(head
.type
));
3819 rdma
->error_state
= ret
;
3825 * Called via a ram_control_load_hook during the initial RAM load section which
3826 * lists the RAMBlocks by name. This lets us know the order of the RAMBlocks
3828 * We've already built our local RAMBlock list, but not yet sent the list to
3832 rdma_block_notification_handle(QEMUFile
*f
, const char *name
)
3835 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(qemu_file_get_ioc(f
));
3839 RCU_READ_LOCK_GUARD();
3840 rdma
= qatomic_rcu_read(&rioc
->rdmain
);
3846 /* Find the matching RAMBlock in our local list */
3847 for (curr
= 0; curr
< rdma
->local_ram_blocks
.nb_blocks
; curr
++) {
3848 if (!strcmp(rdma
->local_ram_blocks
.block
[curr
].block_name
, name
)) {
3855 error_report("RAMBlock '%s' not found on destination", name
);
3859 rdma
->local_ram_blocks
.block
[curr
].src_index
= rdma
->next_src_index
;
3860 trace_rdma_block_notification_handle(name
, rdma
->next_src_index
);
3861 rdma
->next_src_index
++;
3866 static int rdma_load_hook(QEMUFile
*f
, uint64_t flags
, void *data
)
3869 case RAM_CONTROL_BLOCK_REG
:
3870 return rdma_block_notification_handle(f
, data
);
3872 case RAM_CONTROL_HOOK
:
3873 return qemu_rdma_registration_handle(f
);
3876 /* Shouldn't be called with any other values */
3881 static int qemu_rdma_registration_start(QEMUFile
*f
,
3882 uint64_t flags
, void *data
)
3884 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(qemu_file_get_ioc(f
));
3887 if (migration_in_postcopy()) {
3891 RCU_READ_LOCK_GUARD();
3892 rdma
= qatomic_rcu_read(&rioc
->rdmaout
);
3897 CHECK_ERROR_STATE();
3899 trace_qemu_rdma_registration_start(flags
);
3900 qemu_put_be64(f
, RAM_SAVE_FLAG_HOOK
);
3907 * Inform dest that dynamic registrations are done for now.
3908 * First, flush writes, if any.
3910 static int qemu_rdma_registration_stop(QEMUFile
*f
,
3911 uint64_t flags
, void *data
)
3913 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(qemu_file_get_ioc(f
));
3915 RDMAControlHeader head
= { .len
= 0, .repeat
= 1 };
3918 if (migration_in_postcopy()) {
3922 RCU_READ_LOCK_GUARD();
3923 rdma
= qatomic_rcu_read(&rioc
->rdmaout
);
3928 CHECK_ERROR_STATE();
3931 ret
= qemu_rdma_drain_cq(f
, rdma
);
3937 if (flags
== RAM_CONTROL_SETUP
) {
3938 RDMAControlHeader resp
= {.type
= RDMA_CONTROL_RAM_BLOCKS_RESULT
};
3939 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
3940 int reg_result_idx
, i
, nb_dest_blocks
;
3942 head
.type
= RDMA_CONTROL_RAM_BLOCKS_REQUEST
;
3943 trace_qemu_rdma_registration_stop_ram();
3946 * Make sure that we parallelize the pinning on both sides.
3947 * For very large guests, doing this serially takes a really
3948 * long time, so we have to 'interleave' the pinning locally
3949 * with the control messages by performing the pinning on this
3950 * side before we receive the control response from the other
3951 * side that the pinning has completed.
3953 ret
= qemu_rdma_exchange_send(rdma
, &head
, NULL
, &resp
,
3954 ®_result_idx
, rdma
->pin_all
?
3955 qemu_rdma_reg_whole_ram_blocks
: NULL
);
3957 fprintf(stderr
, "receiving remote info!");
3961 nb_dest_blocks
= resp
.len
/ sizeof(RDMADestBlock
);
3964 * The protocol uses two different sets of rkeys (mutually exclusive):
3965 * 1. One key to represent the virtual address of the entire ram block.
3966 * (dynamic chunk registration disabled - pin everything with one rkey.)
3967 * 2. One to represent individual chunks within a ram block.
3968 * (dynamic chunk registration enabled - pin individual chunks.)
3970 * Once the capability is successfully negotiated, the destination transmits
3971 * the keys to use (or sends them later) including the virtual addresses
3972 * and then propagates the remote ram block descriptions to his local copy.
3975 if (local
->nb_blocks
!= nb_dest_blocks
) {
3976 fprintf(stderr
, "ram blocks mismatch (Number of blocks %d vs %d) "
3977 "Your QEMU command line parameters are probably "
3978 "not identical on both the source and destination.",
3979 local
->nb_blocks
, nb_dest_blocks
);
3980 rdma
->error_state
= -EINVAL
;
3984 qemu_rdma_move_header(rdma
, reg_result_idx
, &resp
);
3985 memcpy(rdma
->dest_blocks
,
3986 rdma
->wr_data
[reg_result_idx
].control_curr
, resp
.len
);
3987 for (i
= 0; i
< nb_dest_blocks
; i
++) {
3988 network_to_dest_block(&rdma
->dest_blocks
[i
]);
3990 /* We require that the blocks are in the same order */
3991 if (rdma
->dest_blocks
[i
].length
!= local
->block
[i
].length
) {
3992 fprintf(stderr
, "Block %s/%d has a different length %" PRIu64
3993 "vs %" PRIu64
, local
->block
[i
].block_name
, i
,
3994 local
->block
[i
].length
,
3995 rdma
->dest_blocks
[i
].length
);
3996 rdma
->error_state
= -EINVAL
;
3999 local
->block
[i
].remote_host_addr
=
4000 rdma
->dest_blocks
[i
].remote_host_addr
;
4001 local
->block
[i
].remote_rkey
= rdma
->dest_blocks
[i
].remote_rkey
;
4005 trace_qemu_rdma_registration_stop(flags
);
4007 head
.type
= RDMA_CONTROL_REGISTER_FINISHED
;
4008 ret
= qemu_rdma_exchange_send(rdma
, &head
, NULL
, NULL
, NULL
, NULL
);
4016 rdma
->error_state
= ret
;
4020 static const QEMUFileHooks rdma_read_hooks
= {
4021 .hook_ram_load
= rdma_load_hook
,
4024 static const QEMUFileHooks rdma_write_hooks
= {
4025 .before_ram_iterate
= qemu_rdma_registration_start
,
4026 .after_ram_iterate
= qemu_rdma_registration_stop
,
4027 .save_page
= qemu_rdma_save_page
,
4031 static void qio_channel_rdma_finalize(Object
*obj
)
4033 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(obj
);
4035 qemu_rdma_cleanup(rioc
->rdmain
);
4036 g_free(rioc
->rdmain
);
4037 rioc
->rdmain
= NULL
;
4039 if (rioc
->rdmaout
) {
4040 qemu_rdma_cleanup(rioc
->rdmaout
);
4041 g_free(rioc
->rdmaout
);
4042 rioc
->rdmaout
= NULL
;
4046 static void qio_channel_rdma_class_init(ObjectClass
*klass
,
4047 void *class_data G_GNUC_UNUSED
)
4049 QIOChannelClass
*ioc_klass
= QIO_CHANNEL_CLASS(klass
);
4051 ioc_klass
->io_writev
= qio_channel_rdma_writev
;
4052 ioc_klass
->io_readv
= qio_channel_rdma_readv
;
4053 ioc_klass
->io_set_blocking
= qio_channel_rdma_set_blocking
;
4054 ioc_klass
->io_close
= qio_channel_rdma_close
;
4055 ioc_klass
->io_create_watch
= qio_channel_rdma_create_watch
;
4056 ioc_klass
->io_set_aio_fd_handler
= qio_channel_rdma_set_aio_fd_handler
;
4057 ioc_klass
->io_shutdown
= qio_channel_rdma_shutdown
;
4060 static const TypeInfo qio_channel_rdma_info
= {
4061 .parent
= TYPE_QIO_CHANNEL
,
4062 .name
= TYPE_QIO_CHANNEL_RDMA
,
4063 .instance_size
= sizeof(QIOChannelRDMA
),
4064 .instance_finalize
= qio_channel_rdma_finalize
,
4065 .class_init
= qio_channel_rdma_class_init
,
4068 static void qio_channel_rdma_register_types(void)
4070 type_register_static(&qio_channel_rdma_info
);
4073 type_init(qio_channel_rdma_register_types
);
4075 static QEMUFile
*rdma_new_input(RDMAContext
*rdma
)
4077 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(object_new(TYPE_QIO_CHANNEL_RDMA
));
4079 rioc
->file
= qemu_file_new_input(QIO_CHANNEL(rioc
));
4080 rioc
->rdmain
= rdma
;
4081 rioc
->rdmaout
= rdma
->return_path
;
4082 qemu_file_set_hooks(rioc
->file
, &rdma_read_hooks
);
4087 static QEMUFile
*rdma_new_output(RDMAContext
*rdma
)
4089 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(object_new(TYPE_QIO_CHANNEL_RDMA
));
4091 rioc
->file
= qemu_file_new_output(QIO_CHANNEL(rioc
));
4092 rioc
->rdmaout
= rdma
;
4093 rioc
->rdmain
= rdma
->return_path
;
4094 qemu_file_set_hooks(rioc
->file
, &rdma_write_hooks
);
4099 static void rdma_accept_incoming_migration(void *opaque
)
4101 RDMAContext
*rdma
= opaque
;
4104 Error
*local_err
= NULL
;
4106 trace_qemu_rdma_accept_incoming_migration();
4107 ret
= qemu_rdma_accept(rdma
);
4110 fprintf(stderr
, "RDMA ERROR: Migration initialization failed\n");
4114 trace_qemu_rdma_accept_incoming_migration_accepted();
4116 if (rdma
->is_return_path
) {
4120 f
= rdma_new_input(rdma
);
4122 fprintf(stderr
, "RDMA ERROR: could not open RDMA for input\n");
4123 qemu_rdma_cleanup(rdma
);
4127 rdma
->migration_started_on_destination
= 1;
4128 migration_fd_process_incoming(f
, &local_err
);
4130 error_reportf_err(local_err
, "RDMA ERROR:");
4134 void rdma_start_incoming_migration(const char *host_port
, Error
**errp
)
4138 Error
*local_err
= NULL
;
4140 trace_rdma_start_incoming_migration();
4142 /* Avoid ram_block_discard_disable(), cannot change during migration. */
4143 if (ram_block_discard_is_required()) {
4144 error_setg(errp
, "RDMA: cannot disable RAM discard");
4148 rdma
= qemu_rdma_data_init(host_port
, &local_err
);
4153 ret
= qemu_rdma_dest_init(rdma
, &local_err
);
4159 trace_rdma_start_incoming_migration_after_dest_init();
4161 ret
= rdma_listen(rdma
->listen_id
, 5);
4164 ERROR(errp
, "listening on socket!");
4168 trace_rdma_start_incoming_migration_after_rdma_listen();
4170 qemu_set_fd_handler(rdma
->channel
->fd
, rdma_accept_incoming_migration
,
4171 NULL
, (void *)(intptr_t)rdma
);
4175 qemu_rdma_cleanup(rdma
);
4177 error_propagate(errp
, local_err
);
4180 g_free(rdma
->host_port
);
4185 void rdma_start_outgoing_migration(void *opaque
,
4186 const char *host_port
, Error
**errp
)
4188 MigrationState
*s
= opaque
;
4189 RDMAContext
*rdma_return_path
= NULL
;
4193 /* Avoid ram_block_discard_disable(), cannot change during migration. */
4194 if (ram_block_discard_is_required()) {
4195 error_setg(errp
, "RDMA: cannot disable RAM discard");
4199 rdma
= qemu_rdma_data_init(host_port
, errp
);
4204 ret
= qemu_rdma_source_init(rdma
, migrate_rdma_pin_all(), errp
);
4210 trace_rdma_start_outgoing_migration_after_rdma_source_init();
4211 ret
= qemu_rdma_connect(rdma
, errp
, false);
4217 /* RDMA postcopy need a separate queue pair for return path */
4218 if (migrate_postcopy() || migrate_return_path()) {
4219 rdma_return_path
= qemu_rdma_data_init(host_port
, errp
);
4221 if (rdma_return_path
== NULL
) {
4222 goto return_path_err
;
4225 ret
= qemu_rdma_source_init(rdma_return_path
,
4226 migrate_rdma_pin_all(), errp
);
4229 goto return_path_err
;
4232 ret
= qemu_rdma_connect(rdma_return_path
, errp
, true);
4235 goto return_path_err
;
4238 rdma
->return_path
= rdma_return_path
;
4239 rdma_return_path
->return_path
= rdma
;
4240 rdma_return_path
->is_return_path
= true;
4243 trace_rdma_start_outgoing_migration_after_rdma_connect();
4245 s
->to_dst_file
= rdma_new_output(rdma
);
4246 migrate_fd_connect(s
, NULL
);
4249 qemu_rdma_cleanup(rdma
);
4252 g_free(rdma_return_path
);