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"
21 #include "migration.h"
22 #include "qemu-file.h"
24 #include "qemu-file-channel.h"
25 #include "qemu/error-report.h"
26 #include "qemu/main-loop.h"
27 #include "qemu/module.h"
29 #include "qemu/sockets.h"
30 #include "qemu/bitmap.h"
31 #include "qemu/coroutine.h"
32 #include "exec/memory.h"
33 #include <sys/socket.h>
35 #include <arpa/inet.h>
36 #include <rdma/rdma_cma.h>
38 #include "qom/object.h"
42 * Print and error on both the Monitor and the Log file.
44 #define ERROR(errp, fmt, ...) \
46 fprintf(stderr, "RDMA ERROR: " fmt "\n", ## __VA_ARGS__); \
47 if (errp && (*(errp) == NULL)) { \
48 error_setg(errp, "RDMA ERROR: " fmt, ## __VA_ARGS__); \
52 #define RDMA_RESOLVE_TIMEOUT_MS 10000
54 /* Do not merge data if larger than this. */
55 #define RDMA_MERGE_MAX (2 * 1024 * 1024)
56 #define RDMA_SIGNALED_SEND_MAX (RDMA_MERGE_MAX / 4096)
58 #define RDMA_REG_CHUNK_SHIFT 20 /* 1 MB */
61 * This is only for non-live state being migrated.
62 * Instead of RDMA_WRITE messages, we use RDMA_SEND
63 * messages for that state, which requires a different
64 * delivery design than main memory.
66 #define RDMA_SEND_INCREMENT 32768
69 * Maximum size infiniband SEND message
71 #define RDMA_CONTROL_MAX_BUFFER (512 * 1024)
72 #define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096
74 #define RDMA_CONTROL_VERSION_CURRENT 1
76 * Capabilities for negotiation.
78 #define RDMA_CAPABILITY_PIN_ALL 0x01
81 * Add the other flags above to this list of known capabilities
82 * as they are introduced.
84 static uint32_t known_capabilities
= RDMA_CAPABILITY_PIN_ALL
;
86 #define CHECK_ERROR_STATE() \
88 if (rdma->error_state) { \
89 if (!rdma->error_reported) { \
90 error_report("RDMA is in an error state waiting migration" \
92 rdma->error_reported = 1; \
94 return rdma->error_state; \
99 * A work request ID is 64-bits and we split up these bits
102 * bits 0-15 : type of control message, 2^16
103 * bits 16-29: ram block index, 2^14
104 * bits 30-63: ram block chunk number, 2^34
106 * The last two bit ranges are only used for RDMA writes,
107 * in order to track their completion and potentially
108 * also track unregistration status of the message.
110 #define RDMA_WRID_TYPE_SHIFT 0UL
111 #define RDMA_WRID_BLOCK_SHIFT 16UL
112 #define RDMA_WRID_CHUNK_SHIFT 30UL
114 #define RDMA_WRID_TYPE_MASK \
115 ((1UL << RDMA_WRID_BLOCK_SHIFT) - 1UL)
117 #define RDMA_WRID_BLOCK_MASK \
118 (~RDMA_WRID_TYPE_MASK & ((1UL << RDMA_WRID_CHUNK_SHIFT) - 1UL))
120 #define RDMA_WRID_CHUNK_MASK (~RDMA_WRID_BLOCK_MASK & ~RDMA_WRID_TYPE_MASK)
123 * RDMA migration protocol:
124 * 1. RDMA Writes (data messages, i.e. RAM)
125 * 2. IB Send/Recv (control channel messages)
129 RDMA_WRID_RDMA_WRITE
= 1,
130 RDMA_WRID_SEND_CONTROL
= 2000,
131 RDMA_WRID_RECV_CONTROL
= 4000,
134 static const char *wrid_desc
[] = {
135 [RDMA_WRID_NONE
] = "NONE",
136 [RDMA_WRID_RDMA_WRITE
] = "WRITE RDMA",
137 [RDMA_WRID_SEND_CONTROL
] = "CONTROL SEND",
138 [RDMA_WRID_RECV_CONTROL
] = "CONTROL RECV",
142 * Work request IDs for IB SEND messages only (not RDMA writes).
143 * This is used by the migration protocol to transmit
144 * control messages (such as device state and registration commands)
146 * We could use more WRs, but we have enough for now.
156 * SEND/RECV IB Control Messages.
159 RDMA_CONTROL_NONE
= 0,
161 RDMA_CONTROL_READY
, /* ready to receive */
162 RDMA_CONTROL_QEMU_FILE
, /* QEMUFile-transmitted bytes */
163 RDMA_CONTROL_RAM_BLOCKS_REQUEST
, /* RAMBlock synchronization */
164 RDMA_CONTROL_RAM_BLOCKS_RESULT
, /* RAMBlock synchronization */
165 RDMA_CONTROL_COMPRESS
, /* page contains repeat values */
166 RDMA_CONTROL_REGISTER_REQUEST
, /* dynamic page registration */
167 RDMA_CONTROL_REGISTER_RESULT
, /* key to use after registration */
168 RDMA_CONTROL_REGISTER_FINISHED
, /* current iteration finished */
169 RDMA_CONTROL_UNREGISTER_REQUEST
, /* dynamic UN-registration */
170 RDMA_CONTROL_UNREGISTER_FINISHED
, /* unpinning finished */
175 * Memory and MR structures used to represent an IB Send/Recv work request.
176 * This is *not* used for RDMA writes, only IB Send/Recv.
179 uint8_t control
[RDMA_CONTROL_MAX_BUFFER
]; /* actual buffer to register */
180 struct ibv_mr
*control_mr
; /* registration metadata */
181 size_t control_len
; /* length of the message */
182 uint8_t *control_curr
; /* start of unconsumed bytes */
183 } RDMAWorkRequestData
;
186 * Negotiate RDMA capabilities during connection-setup time.
193 static void caps_to_network(RDMACapabilities
*cap
)
195 cap
->version
= htonl(cap
->version
);
196 cap
->flags
= htonl(cap
->flags
);
199 static void network_to_caps(RDMACapabilities
*cap
)
201 cap
->version
= ntohl(cap
->version
);
202 cap
->flags
= ntohl(cap
->flags
);
206 * Representation of a RAMBlock from an RDMA perspective.
207 * This is not transmitted, only local.
208 * This and subsequent structures cannot be linked lists
209 * because we're using a single IB message to transmit
210 * the information. It's small anyway, so a list is overkill.
212 typedef struct RDMALocalBlock
{
214 uint8_t *local_host_addr
; /* local virtual address */
215 uint64_t remote_host_addr
; /* remote virtual address */
218 struct ibv_mr
**pmr
; /* MRs for chunk-level registration */
219 struct ibv_mr
*mr
; /* MR for non-chunk-level registration */
220 uint32_t *remote_keys
; /* rkeys for chunk-level registration */
221 uint32_t remote_rkey
; /* rkeys for non-chunk-level registration */
222 int index
; /* which block are we */
223 unsigned int src_index
; /* (Only used on dest) */
226 unsigned long *transit_bitmap
;
227 unsigned long *unregister_bitmap
;
231 * Also represents a RAMblock, but only on the dest.
232 * This gets transmitted by the dest during connection-time
233 * to the source VM and then is used to populate the
234 * corresponding RDMALocalBlock with
235 * the information needed to perform the actual RDMA.
237 typedef struct QEMU_PACKED RDMADestBlock
{
238 uint64_t remote_host_addr
;
241 uint32_t remote_rkey
;
245 static const char *control_desc(unsigned int rdma_control
)
247 static const char *strs
[] = {
248 [RDMA_CONTROL_NONE
] = "NONE",
249 [RDMA_CONTROL_ERROR
] = "ERROR",
250 [RDMA_CONTROL_READY
] = "READY",
251 [RDMA_CONTROL_QEMU_FILE
] = "QEMU FILE",
252 [RDMA_CONTROL_RAM_BLOCKS_REQUEST
] = "RAM BLOCKS REQUEST",
253 [RDMA_CONTROL_RAM_BLOCKS_RESULT
] = "RAM BLOCKS RESULT",
254 [RDMA_CONTROL_COMPRESS
] = "COMPRESS",
255 [RDMA_CONTROL_REGISTER_REQUEST
] = "REGISTER REQUEST",
256 [RDMA_CONTROL_REGISTER_RESULT
] = "REGISTER RESULT",
257 [RDMA_CONTROL_REGISTER_FINISHED
] = "REGISTER FINISHED",
258 [RDMA_CONTROL_UNREGISTER_REQUEST
] = "UNREGISTER REQUEST",
259 [RDMA_CONTROL_UNREGISTER_FINISHED
] = "UNREGISTER FINISHED",
262 if (rdma_control
> RDMA_CONTROL_UNREGISTER_FINISHED
) {
263 return "??BAD CONTROL VALUE??";
266 return strs
[rdma_control
];
269 static uint64_t htonll(uint64_t v
)
271 union { uint32_t lv
[2]; uint64_t llv
; } u
;
272 u
.lv
[0] = htonl(v
>> 32);
273 u
.lv
[1] = htonl(v
& 0xFFFFFFFFULL
);
277 static uint64_t ntohll(uint64_t v
)
279 union { uint32_t lv
[2]; uint64_t llv
; } u
;
281 return ((uint64_t)ntohl(u
.lv
[0]) << 32) | (uint64_t) ntohl(u
.lv
[1]);
284 static void dest_block_to_network(RDMADestBlock
*db
)
286 db
->remote_host_addr
= htonll(db
->remote_host_addr
);
287 db
->offset
= htonll(db
->offset
);
288 db
->length
= htonll(db
->length
);
289 db
->remote_rkey
= htonl(db
->remote_rkey
);
292 static void network_to_dest_block(RDMADestBlock
*db
)
294 db
->remote_host_addr
= ntohll(db
->remote_host_addr
);
295 db
->offset
= ntohll(db
->offset
);
296 db
->length
= ntohll(db
->length
);
297 db
->remote_rkey
= ntohl(db
->remote_rkey
);
301 * Virtual address of the above structures used for transmitting
302 * the RAMBlock descriptions at connection-time.
303 * This structure is *not* transmitted.
305 typedef struct RDMALocalBlocks
{
307 bool init
; /* main memory init complete */
308 RDMALocalBlock
*block
;
312 * Main data structure for RDMA state.
313 * While there is only one copy of this structure being allocated right now,
314 * this is the place where one would start if you wanted to consider
315 * having more than one RDMA connection open at the same time.
317 typedef struct RDMAContext
{
322 RDMAWorkRequestData wr_data
[RDMA_WRID_MAX
];
325 * This is used by *_exchange_send() to figure out whether or not
326 * the initial "READY" message has already been received or not.
327 * This is because other functions may potentially poll() and detect
328 * the READY message before send() does, in which case we need to
329 * know if it completed.
331 int control_ready_expected
;
333 /* number of outstanding writes */
336 /* store info about current buffer so that we can
337 merge it with future sends */
338 uint64_t current_addr
;
339 uint64_t current_length
;
340 /* index of ram block the current buffer belongs to */
342 /* index of the chunk in the current ram block */
348 * infiniband-specific variables for opening the device
349 * and maintaining connection state and so forth.
351 * cm_id also has ibv_context, rdma_event_channel, and ibv_qp in
352 * cm_id->verbs, cm_id->channel, and cm_id->qp.
354 struct rdma_cm_id
*cm_id
; /* connection manager ID */
355 struct rdma_cm_id
*listen_id
;
358 struct ibv_context
*verbs
;
359 struct rdma_event_channel
*channel
;
360 struct ibv_qp
*qp
; /* queue pair */
361 struct ibv_comp_channel
*comp_channel
; /* completion channel */
362 struct ibv_pd
*pd
; /* protection domain */
363 struct ibv_cq
*cq
; /* completion queue */
366 * If a previous write failed (perhaps because of a failed
367 * memory registration, then do not attempt any future work
368 * and remember the error state.
375 * Description of ram blocks used throughout the code.
377 RDMALocalBlocks local_ram_blocks
;
378 RDMADestBlock
*dest_blocks
;
380 /* Index of the next RAMBlock received during block registration */
381 unsigned int next_src_index
;
384 * Migration on *destination* started.
385 * Then use coroutine yield function.
386 * Source runs in a thread, so we don't care.
388 int migration_started_on_destination
;
390 int total_registrations
;
393 int unregister_current
, unregister_next
;
394 uint64_t unregistrations
[RDMA_SIGNALED_SEND_MAX
];
396 GHashTable
*blockmap
;
398 /* the RDMAContext for return path */
399 struct RDMAContext
*return_path
;
403 #define TYPE_QIO_CHANNEL_RDMA "qio-channel-rdma"
404 OBJECT_DECLARE_SIMPLE_TYPE(QIOChannelRDMA
, QIO_CHANNEL_RDMA
)
408 struct QIOChannelRDMA
{
411 RDMAContext
*rdmaout
;
413 bool blocking
; /* XXX we don't actually honour this yet */
417 * Main structure for IB Send/Recv control messages.
418 * This gets prepended at the beginning of every Send/Recv.
420 typedef struct QEMU_PACKED
{
421 uint32_t len
; /* Total length of data portion */
422 uint32_t type
; /* which control command to perform */
423 uint32_t repeat
; /* number of commands in data portion of same type */
427 static void control_to_network(RDMAControlHeader
*control
)
429 control
->type
= htonl(control
->type
);
430 control
->len
= htonl(control
->len
);
431 control
->repeat
= htonl(control
->repeat
);
434 static void network_to_control(RDMAControlHeader
*control
)
436 control
->type
= ntohl(control
->type
);
437 control
->len
= ntohl(control
->len
);
438 control
->repeat
= ntohl(control
->repeat
);
442 * Register a single Chunk.
443 * Information sent by the source VM to inform the dest
444 * to register an single chunk of memory before we can perform
445 * the actual RDMA operation.
447 typedef struct QEMU_PACKED
{
449 uint64_t current_addr
; /* offset into the ram_addr_t space */
450 uint64_t chunk
; /* chunk to lookup if unregistering */
452 uint32_t current_index
; /* which ramblock the chunk belongs to */
454 uint64_t chunks
; /* how many sequential chunks to register */
457 static void register_to_network(RDMAContext
*rdma
, RDMARegister
*reg
)
459 RDMALocalBlock
*local_block
;
460 local_block
= &rdma
->local_ram_blocks
.block
[reg
->current_index
];
462 if (local_block
->is_ram_block
) {
464 * current_addr as passed in is an address in the local ram_addr_t
465 * space, we need to translate this for the destination
467 reg
->key
.current_addr
-= local_block
->offset
;
468 reg
->key
.current_addr
+= rdma
->dest_blocks
[reg
->current_index
].offset
;
470 reg
->key
.current_addr
= htonll(reg
->key
.current_addr
);
471 reg
->current_index
= htonl(reg
->current_index
);
472 reg
->chunks
= htonll(reg
->chunks
);
475 static void network_to_register(RDMARegister
*reg
)
477 reg
->key
.current_addr
= ntohll(reg
->key
.current_addr
);
478 reg
->current_index
= ntohl(reg
->current_index
);
479 reg
->chunks
= ntohll(reg
->chunks
);
482 typedef struct QEMU_PACKED
{
483 uint32_t value
; /* if zero, we will madvise() */
484 uint32_t block_idx
; /* which ram block index */
485 uint64_t offset
; /* Address in remote ram_addr_t space */
486 uint64_t length
; /* length of the chunk */
489 static void compress_to_network(RDMAContext
*rdma
, RDMACompress
*comp
)
491 comp
->value
= htonl(comp
->value
);
493 * comp->offset as passed in is an address in the local ram_addr_t
494 * space, we need to translate this for the destination
496 comp
->offset
-= rdma
->local_ram_blocks
.block
[comp
->block_idx
].offset
;
497 comp
->offset
+= rdma
->dest_blocks
[comp
->block_idx
].offset
;
498 comp
->block_idx
= htonl(comp
->block_idx
);
499 comp
->offset
= htonll(comp
->offset
);
500 comp
->length
= htonll(comp
->length
);
503 static void network_to_compress(RDMACompress
*comp
)
505 comp
->value
= ntohl(comp
->value
);
506 comp
->block_idx
= ntohl(comp
->block_idx
);
507 comp
->offset
= ntohll(comp
->offset
);
508 comp
->length
= ntohll(comp
->length
);
512 * The result of the dest's memory registration produces an "rkey"
513 * which the source VM must reference in order to perform
514 * the RDMA operation.
516 typedef struct QEMU_PACKED
{
520 } RDMARegisterResult
;
522 static void result_to_network(RDMARegisterResult
*result
)
524 result
->rkey
= htonl(result
->rkey
);
525 result
->host_addr
= htonll(result
->host_addr
);
528 static void network_to_result(RDMARegisterResult
*result
)
530 result
->rkey
= ntohl(result
->rkey
);
531 result
->host_addr
= ntohll(result
->host_addr
);
534 const char *print_wrid(int wrid
);
535 static int qemu_rdma_exchange_send(RDMAContext
*rdma
, RDMAControlHeader
*head
,
536 uint8_t *data
, RDMAControlHeader
*resp
,
538 int (*callback
)(RDMAContext
*rdma
));
540 static inline uint64_t ram_chunk_index(const uint8_t *start
,
543 return ((uintptr_t) host
- (uintptr_t) start
) >> RDMA_REG_CHUNK_SHIFT
;
546 static inline uint8_t *ram_chunk_start(const RDMALocalBlock
*rdma_ram_block
,
549 return (uint8_t *)(uintptr_t)(rdma_ram_block
->local_host_addr
+
550 (i
<< RDMA_REG_CHUNK_SHIFT
));
553 static inline uint8_t *ram_chunk_end(const RDMALocalBlock
*rdma_ram_block
,
556 uint8_t *result
= ram_chunk_start(rdma_ram_block
, i
) +
557 (1UL << RDMA_REG_CHUNK_SHIFT
);
559 if (result
> (rdma_ram_block
->local_host_addr
+ rdma_ram_block
->length
)) {
560 result
= rdma_ram_block
->local_host_addr
+ rdma_ram_block
->length
;
566 static int rdma_add_block(RDMAContext
*rdma
, const char *block_name
,
568 ram_addr_t block_offset
, uint64_t length
)
570 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
571 RDMALocalBlock
*block
;
572 RDMALocalBlock
*old
= local
->block
;
574 local
->block
= g_new0(RDMALocalBlock
, local
->nb_blocks
+ 1);
576 if (local
->nb_blocks
) {
579 if (rdma
->blockmap
) {
580 for (x
= 0; x
< local
->nb_blocks
; x
++) {
581 g_hash_table_remove(rdma
->blockmap
,
582 (void *)(uintptr_t)old
[x
].offset
);
583 g_hash_table_insert(rdma
->blockmap
,
584 (void *)(uintptr_t)old
[x
].offset
,
588 memcpy(local
->block
, old
, sizeof(RDMALocalBlock
) * local
->nb_blocks
);
592 block
= &local
->block
[local
->nb_blocks
];
594 block
->block_name
= g_strdup(block_name
);
595 block
->local_host_addr
= host_addr
;
596 block
->offset
= block_offset
;
597 block
->length
= length
;
598 block
->index
= local
->nb_blocks
;
599 block
->src_index
= ~0U; /* Filled in by the receipt of the block list */
600 block
->nb_chunks
= ram_chunk_index(host_addr
, host_addr
+ length
) + 1UL;
601 block
->transit_bitmap
= bitmap_new(block
->nb_chunks
);
602 bitmap_clear(block
->transit_bitmap
, 0, block
->nb_chunks
);
603 block
->unregister_bitmap
= bitmap_new(block
->nb_chunks
);
604 bitmap_clear(block
->unregister_bitmap
, 0, block
->nb_chunks
);
605 block
->remote_keys
= g_new0(uint32_t, block
->nb_chunks
);
607 block
->is_ram_block
= local
->init
? false : true;
609 if (rdma
->blockmap
) {
610 g_hash_table_insert(rdma
->blockmap
, (void *)(uintptr_t)block_offset
, block
);
613 trace_rdma_add_block(block_name
, local
->nb_blocks
,
614 (uintptr_t) block
->local_host_addr
,
615 block
->offset
, block
->length
,
616 (uintptr_t) (block
->local_host_addr
+ block
->length
),
617 BITS_TO_LONGS(block
->nb_chunks
) *
618 sizeof(unsigned long) * 8,
627 * Memory regions need to be registered with the device and queue pairs setup
628 * in advanced before the migration starts. This tells us where the RAM blocks
629 * are so that we can register them individually.
631 static int qemu_rdma_init_one_block(RAMBlock
*rb
, void *opaque
)
633 const char *block_name
= qemu_ram_get_idstr(rb
);
634 void *host_addr
= qemu_ram_get_host_addr(rb
);
635 ram_addr_t block_offset
= qemu_ram_get_offset(rb
);
636 ram_addr_t length
= qemu_ram_get_used_length(rb
);
637 return rdma_add_block(opaque
, block_name
, host_addr
, block_offset
, length
);
641 * Identify the RAMBlocks and their quantity. They will be references to
642 * identify chunk boundaries inside each RAMBlock and also be referenced
643 * during dynamic page registration.
645 static int qemu_rdma_init_ram_blocks(RDMAContext
*rdma
)
647 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
650 assert(rdma
->blockmap
== NULL
);
651 memset(local
, 0, sizeof *local
);
652 ret
= foreach_not_ignored_block(qemu_rdma_init_one_block
, rdma
);
656 trace_qemu_rdma_init_ram_blocks(local
->nb_blocks
);
657 rdma
->dest_blocks
= g_new0(RDMADestBlock
,
658 rdma
->local_ram_blocks
.nb_blocks
);
664 * Note: If used outside of cleanup, the caller must ensure that the destination
665 * block structures are also updated
667 static int rdma_delete_block(RDMAContext
*rdma
, RDMALocalBlock
*block
)
669 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
670 RDMALocalBlock
*old
= local
->block
;
673 if (rdma
->blockmap
) {
674 g_hash_table_remove(rdma
->blockmap
, (void *)(uintptr_t)block
->offset
);
679 for (j
= 0; j
< block
->nb_chunks
; j
++) {
680 if (!block
->pmr
[j
]) {
683 ibv_dereg_mr(block
->pmr
[j
]);
684 rdma
->total_registrations
--;
691 ibv_dereg_mr(block
->mr
);
692 rdma
->total_registrations
--;
696 g_free(block
->transit_bitmap
);
697 block
->transit_bitmap
= NULL
;
699 g_free(block
->unregister_bitmap
);
700 block
->unregister_bitmap
= NULL
;
702 g_free(block
->remote_keys
);
703 block
->remote_keys
= NULL
;
705 g_free(block
->block_name
);
706 block
->block_name
= NULL
;
708 if (rdma
->blockmap
) {
709 for (x
= 0; x
< local
->nb_blocks
; x
++) {
710 g_hash_table_remove(rdma
->blockmap
,
711 (void *)(uintptr_t)old
[x
].offset
);
715 if (local
->nb_blocks
> 1) {
717 local
->block
= g_new0(RDMALocalBlock
, local
->nb_blocks
- 1);
720 memcpy(local
->block
, old
, sizeof(RDMALocalBlock
) * block
->index
);
723 if (block
->index
< (local
->nb_blocks
- 1)) {
724 memcpy(local
->block
+ block
->index
, old
+ (block
->index
+ 1),
725 sizeof(RDMALocalBlock
) *
726 (local
->nb_blocks
- (block
->index
+ 1)));
727 for (x
= block
->index
; x
< local
->nb_blocks
- 1; x
++) {
728 local
->block
[x
].index
--;
732 assert(block
== local
->block
);
736 trace_rdma_delete_block(block
, (uintptr_t)block
->local_host_addr
,
737 block
->offset
, block
->length
,
738 (uintptr_t)(block
->local_host_addr
+ block
->length
),
739 BITS_TO_LONGS(block
->nb_chunks
) *
740 sizeof(unsigned long) * 8, block
->nb_chunks
);
746 if (local
->nb_blocks
&& rdma
->blockmap
) {
747 for (x
= 0; x
< local
->nb_blocks
; x
++) {
748 g_hash_table_insert(rdma
->blockmap
,
749 (void *)(uintptr_t)local
->block
[x
].offset
,
758 * Put in the log file which RDMA device was opened and the details
759 * associated with that device.
761 static void qemu_rdma_dump_id(const char *who
, struct ibv_context
*verbs
)
763 struct ibv_port_attr port
;
765 if (ibv_query_port(verbs
, 1, &port
)) {
766 error_report("Failed to query port information");
770 printf("%s RDMA Device opened: kernel name %s "
771 "uverbs device name %s, "
772 "infiniband_verbs class device path %s, "
773 "infiniband class device path %s, "
774 "transport: (%d) %s\n",
777 verbs
->device
->dev_name
,
778 verbs
->device
->dev_path
,
779 verbs
->device
->ibdev_path
,
781 (port
.link_layer
== IBV_LINK_LAYER_INFINIBAND
) ? "Infiniband" :
782 ((port
.link_layer
== IBV_LINK_LAYER_ETHERNET
)
783 ? "Ethernet" : "Unknown"));
787 * Put in the log file the RDMA gid addressing information,
788 * useful for folks who have trouble understanding the
789 * RDMA device hierarchy in the kernel.
791 static void qemu_rdma_dump_gid(const char *who
, struct rdma_cm_id
*id
)
795 inet_ntop(AF_INET6
, &id
->route
.addr
.addr
.ibaddr
.sgid
, sgid
, sizeof sgid
);
796 inet_ntop(AF_INET6
, &id
->route
.addr
.addr
.ibaddr
.dgid
, dgid
, sizeof dgid
);
797 trace_qemu_rdma_dump_gid(who
, sgid
, dgid
);
801 * As of now, IPv6 over RoCE / iWARP is not supported by linux.
802 * We will try the next addrinfo struct, and fail if there are
803 * no other valid addresses to bind against.
805 * If user is listening on '[::]', then we will not have a opened a device
806 * yet and have no way of verifying if the device is RoCE or not.
808 * In this case, the source VM will throw an error for ALL types of
809 * connections (both IPv4 and IPv6) if the destination machine does not have
810 * a regular infiniband network available for use.
812 * The only way to guarantee that an error is thrown for broken kernels is
813 * for the management software to choose a *specific* interface at bind time
814 * and validate what time of hardware it is.
816 * Unfortunately, this puts the user in a fix:
818 * If the source VM connects with an IPv4 address without knowing that the
819 * destination has bound to '[::]' the migration will unconditionally fail
820 * unless the management software is explicitly listening on the IPv4
821 * address while using a RoCE-based device.
823 * If the source VM connects with an IPv6 address, then we're OK because we can
824 * throw an error on the source (and similarly on the destination).
826 * But in mixed environments, this will be broken for a while until it is fixed
829 * We do provide a *tiny* bit of help in this function: We can list all of the
830 * devices in the system and check to see if all the devices are RoCE or
833 * If we detect that we have a *pure* RoCE environment, then we can safely
834 * thrown an error even if the management software has specified '[::]' as the
837 * However, if there is are multiple hetergeneous devices, then we cannot make
838 * this assumption and the user just has to be sure they know what they are
841 * Patches are being reviewed on linux-rdma.
843 static int qemu_rdma_broken_ipv6_kernel(struct ibv_context
*verbs
, Error
**errp
)
845 /* This bug only exists in linux, to our knowledge. */
847 struct ibv_port_attr port_attr
;
850 * Verbs are only NULL if management has bound to '[::]'.
852 * Let's iterate through all the devices and see if there any pure IB
853 * devices (non-ethernet).
855 * If not, then we can safely proceed with the migration.
856 * Otherwise, there are no guarantees until the bug is fixed in linux.
860 struct ibv_device
**dev_list
= ibv_get_device_list(&num_devices
);
861 bool roce_found
= false;
862 bool ib_found
= false;
864 for (x
= 0; x
< num_devices
; x
++) {
865 verbs
= ibv_open_device(dev_list
[x
]);
867 if (errno
== EPERM
) {
874 if (ibv_query_port(verbs
, 1, &port_attr
)) {
875 ibv_close_device(verbs
);
876 ERROR(errp
, "Could not query initial IB port");
880 if (port_attr
.link_layer
== IBV_LINK_LAYER_INFINIBAND
) {
882 } else if (port_attr
.link_layer
== IBV_LINK_LAYER_ETHERNET
) {
886 ibv_close_device(verbs
);
892 fprintf(stderr
, "WARN: migrations may fail:"
893 " IPv6 over RoCE / iWARP in linux"
894 " is broken. But since you appear to have a"
895 " mixed RoCE / IB environment, be sure to only"
896 " migrate over the IB fabric until the kernel "
897 " fixes the bug.\n");
899 ERROR(errp
, "You only have RoCE / iWARP devices in your systems"
900 " and your management software has specified '[::]'"
901 ", but IPv6 over RoCE / iWARP is not supported in Linux.");
910 * If we have a verbs context, that means that some other than '[::]' was
911 * used by the management software for binding. In which case we can
912 * actually warn the user about a potentially broken kernel.
915 /* IB ports start with 1, not 0 */
916 if (ibv_query_port(verbs
, 1, &port_attr
)) {
917 ERROR(errp
, "Could not query initial IB port");
921 if (port_attr
.link_layer
== IBV_LINK_LAYER_ETHERNET
) {
922 ERROR(errp
, "Linux kernel's RoCE / iWARP does not support IPv6 "
923 "(but patches on linux-rdma in progress)");
933 * Figure out which RDMA device corresponds to the requested IP hostname
934 * Also create the initial connection manager identifiers for opening
937 static int qemu_rdma_resolve_host(RDMAContext
*rdma
, Error
**errp
)
940 struct rdma_addrinfo
*res
;
942 struct rdma_cm_event
*cm_event
;
943 char ip
[40] = "unknown";
944 struct rdma_addrinfo
*e
;
946 if (rdma
->host
== NULL
|| !strcmp(rdma
->host
, "")) {
947 ERROR(errp
, "RDMA hostname has not been set");
951 /* create CM channel */
952 rdma
->channel
= rdma_create_event_channel();
953 if (!rdma
->channel
) {
954 ERROR(errp
, "could not create CM channel");
959 ret
= rdma_create_id(rdma
->channel
, &rdma
->cm_id
, NULL
, RDMA_PS_TCP
);
961 ERROR(errp
, "could not create channel id");
962 goto err_resolve_create_id
;
965 snprintf(port_str
, 16, "%d", rdma
->port
);
968 ret
= rdma_getaddrinfo(rdma
->host
, port_str
, NULL
, &res
);
970 ERROR(errp
, "could not rdma_getaddrinfo address %s", rdma
->host
);
971 goto err_resolve_get_addr
;
974 for (e
= res
; e
!= NULL
; e
= e
->ai_next
) {
975 inet_ntop(e
->ai_family
,
976 &((struct sockaddr_in
*) e
->ai_dst_addr
)->sin_addr
, ip
, sizeof ip
);
977 trace_qemu_rdma_resolve_host_trying(rdma
->host
, ip
);
979 ret
= rdma_resolve_addr(rdma
->cm_id
, NULL
, e
->ai_dst_addr
,
980 RDMA_RESOLVE_TIMEOUT_MS
);
982 if (e
->ai_family
== AF_INET6
) {
983 ret
= qemu_rdma_broken_ipv6_kernel(rdma
->cm_id
->verbs
, errp
);
992 rdma_freeaddrinfo(res
);
993 ERROR(errp
, "could not resolve address %s", rdma
->host
);
994 goto err_resolve_get_addr
;
997 rdma_freeaddrinfo(res
);
998 qemu_rdma_dump_gid("source_resolve_addr", rdma
->cm_id
);
1000 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
1002 ERROR(errp
, "could not perform event_addr_resolved");
1003 goto err_resolve_get_addr
;
1006 if (cm_event
->event
!= RDMA_CM_EVENT_ADDR_RESOLVED
) {
1007 ERROR(errp
, "result not equal to event_addr_resolved %s",
1008 rdma_event_str(cm_event
->event
));
1009 perror("rdma_resolve_addr");
1010 rdma_ack_cm_event(cm_event
);
1012 goto err_resolve_get_addr
;
1014 rdma_ack_cm_event(cm_event
);
1017 ret
= rdma_resolve_route(rdma
->cm_id
, RDMA_RESOLVE_TIMEOUT_MS
);
1019 ERROR(errp
, "could not resolve rdma route");
1020 goto err_resolve_get_addr
;
1023 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
1025 ERROR(errp
, "could not perform event_route_resolved");
1026 goto err_resolve_get_addr
;
1028 if (cm_event
->event
!= RDMA_CM_EVENT_ROUTE_RESOLVED
) {
1029 ERROR(errp
, "result not equal to event_route_resolved: %s",
1030 rdma_event_str(cm_event
->event
));
1031 rdma_ack_cm_event(cm_event
);
1033 goto err_resolve_get_addr
;
1035 rdma_ack_cm_event(cm_event
);
1036 rdma
->verbs
= rdma
->cm_id
->verbs
;
1037 qemu_rdma_dump_id("source_resolve_host", rdma
->cm_id
->verbs
);
1038 qemu_rdma_dump_gid("source_resolve_host", rdma
->cm_id
);
1041 err_resolve_get_addr
:
1042 rdma_destroy_id(rdma
->cm_id
);
1044 err_resolve_create_id
:
1045 rdma_destroy_event_channel(rdma
->channel
);
1046 rdma
->channel
= NULL
;
1051 * Create protection domain and completion queues
1053 static int qemu_rdma_alloc_pd_cq(RDMAContext
*rdma
)
1056 rdma
->pd
= ibv_alloc_pd(rdma
->verbs
);
1058 error_report("failed to allocate protection domain");
1062 /* create completion channel */
1063 rdma
->comp_channel
= ibv_create_comp_channel(rdma
->verbs
);
1064 if (!rdma
->comp_channel
) {
1065 error_report("failed to allocate completion channel");
1066 goto err_alloc_pd_cq
;
1070 * Completion queue can be filled by both read and write work requests,
1071 * so must reflect the sum of both possible queue sizes.
1073 rdma
->cq
= ibv_create_cq(rdma
->verbs
, (RDMA_SIGNALED_SEND_MAX
* 3),
1074 NULL
, rdma
->comp_channel
, 0);
1076 error_report("failed to allocate completion queue");
1077 goto err_alloc_pd_cq
;
1084 ibv_dealloc_pd(rdma
->pd
);
1086 if (rdma
->comp_channel
) {
1087 ibv_destroy_comp_channel(rdma
->comp_channel
);
1090 rdma
->comp_channel
= NULL
;
1096 * Create queue pairs.
1098 static int qemu_rdma_alloc_qp(RDMAContext
*rdma
)
1100 struct ibv_qp_init_attr attr
= { 0 };
1103 attr
.cap
.max_send_wr
= RDMA_SIGNALED_SEND_MAX
;
1104 attr
.cap
.max_recv_wr
= 3;
1105 attr
.cap
.max_send_sge
= 1;
1106 attr
.cap
.max_recv_sge
= 1;
1107 attr
.send_cq
= rdma
->cq
;
1108 attr
.recv_cq
= rdma
->cq
;
1109 attr
.qp_type
= IBV_QPT_RC
;
1111 ret
= rdma_create_qp(rdma
->cm_id
, rdma
->pd
, &attr
);
1116 rdma
->qp
= rdma
->cm_id
->qp
;
1120 static int qemu_rdma_reg_whole_ram_blocks(RDMAContext
*rdma
)
1123 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
1125 for (i
= 0; i
< local
->nb_blocks
; i
++) {
1126 local
->block
[i
].mr
=
1127 ibv_reg_mr(rdma
->pd
,
1128 local
->block
[i
].local_host_addr
,
1129 local
->block
[i
].length
,
1130 IBV_ACCESS_LOCAL_WRITE
|
1131 IBV_ACCESS_REMOTE_WRITE
1133 if (!local
->block
[i
].mr
) {
1134 perror("Failed to register local dest ram block!\n");
1137 rdma
->total_registrations
++;
1140 if (i
>= local
->nb_blocks
) {
1144 for (i
--; i
>= 0; i
--) {
1145 ibv_dereg_mr(local
->block
[i
].mr
);
1146 rdma
->total_registrations
--;
1154 * Find the ram block that corresponds to the page requested to be
1155 * transmitted by QEMU.
1157 * Once the block is found, also identify which 'chunk' within that
1158 * block that the page belongs to.
1160 * This search cannot fail or the migration will fail.
1162 static int qemu_rdma_search_ram_block(RDMAContext
*rdma
,
1163 uintptr_t block_offset
,
1166 uint64_t *block_index
,
1167 uint64_t *chunk_index
)
1169 uint64_t current_addr
= block_offset
+ offset
;
1170 RDMALocalBlock
*block
= g_hash_table_lookup(rdma
->blockmap
,
1171 (void *) block_offset
);
1173 assert(current_addr
>= block
->offset
);
1174 assert((current_addr
+ length
) <= (block
->offset
+ block
->length
));
1176 *block_index
= block
->index
;
1177 *chunk_index
= ram_chunk_index(block
->local_host_addr
,
1178 block
->local_host_addr
+ (current_addr
- block
->offset
));
1184 * Register a chunk with IB. If the chunk was already registered
1185 * previously, then skip.
1187 * Also return the keys associated with the registration needed
1188 * to perform the actual RDMA operation.
1190 static int qemu_rdma_register_and_get_keys(RDMAContext
*rdma
,
1191 RDMALocalBlock
*block
, uintptr_t host_addr
,
1192 uint32_t *lkey
, uint32_t *rkey
, int chunk
,
1193 uint8_t *chunk_start
, uint8_t *chunk_end
)
1197 *lkey
= block
->mr
->lkey
;
1200 *rkey
= block
->mr
->rkey
;
1205 /* allocate memory to store chunk MRs */
1207 block
->pmr
= g_new0(struct ibv_mr
*, block
->nb_chunks
);
1211 * If 'rkey', then we're the destination, so grant access to the source.
1213 * If 'lkey', then we're the source VM, so grant access only to ourselves.
1215 if (!block
->pmr
[chunk
]) {
1216 uint64_t len
= chunk_end
- chunk_start
;
1218 trace_qemu_rdma_register_and_get_keys(len
, chunk_start
);
1220 block
->pmr
[chunk
] = ibv_reg_mr(rdma
->pd
,
1222 (rkey
? (IBV_ACCESS_LOCAL_WRITE
|
1223 IBV_ACCESS_REMOTE_WRITE
) : 0));
1225 if (!block
->pmr
[chunk
]) {
1226 perror("Failed to register chunk!");
1227 fprintf(stderr
, "Chunk details: block: %d chunk index %d"
1228 " start %" PRIuPTR
" end %" PRIuPTR
1230 " local %" PRIuPTR
" registrations: %d\n",
1231 block
->index
, chunk
, (uintptr_t)chunk_start
,
1232 (uintptr_t)chunk_end
, host_addr
,
1233 (uintptr_t)block
->local_host_addr
,
1234 rdma
->total_registrations
);
1237 rdma
->total_registrations
++;
1241 *lkey
= block
->pmr
[chunk
]->lkey
;
1244 *rkey
= block
->pmr
[chunk
]->rkey
;
1250 * Register (at connection time) the memory used for control
1253 static int qemu_rdma_reg_control(RDMAContext
*rdma
, int idx
)
1255 rdma
->wr_data
[idx
].control_mr
= ibv_reg_mr(rdma
->pd
,
1256 rdma
->wr_data
[idx
].control
, RDMA_CONTROL_MAX_BUFFER
,
1257 IBV_ACCESS_LOCAL_WRITE
| IBV_ACCESS_REMOTE_WRITE
);
1258 if (rdma
->wr_data
[idx
].control_mr
) {
1259 rdma
->total_registrations
++;
1262 error_report("qemu_rdma_reg_control failed");
1266 const char *print_wrid(int wrid
)
1268 if (wrid
>= RDMA_WRID_RECV_CONTROL
) {
1269 return wrid_desc
[RDMA_WRID_RECV_CONTROL
];
1271 return wrid_desc
[wrid
];
1275 * RDMA requires memory registration (mlock/pinning), but this is not good for
1278 * In preparation for the future where LRU information or workload-specific
1279 * writable writable working set memory access behavior is available to QEMU
1280 * it would be nice to have in place the ability to UN-register/UN-pin
1281 * particular memory regions from the RDMA hardware when it is determine that
1282 * those regions of memory will likely not be accessed again in the near future.
1284 * While we do not yet have such information right now, the following
1285 * compile-time option allows us to perform a non-optimized version of this
1288 * By uncommenting this option, you will cause *all* RDMA transfers to be
1289 * unregistered immediately after the transfer completes on both sides of the
1290 * connection. This has no effect in 'rdma-pin-all' mode, only regular mode.
1292 * This will have a terrible impact on migration performance, so until future
1293 * workload information or LRU information is available, do not attempt to use
1294 * this feature except for basic testing.
1296 /* #define RDMA_UNREGISTRATION_EXAMPLE */
1299 * Perform a non-optimized memory unregistration after every transfer
1300 * for demonstration purposes, only if pin-all is not requested.
1302 * Potential optimizations:
1303 * 1. Start a new thread to run this function continuously
1305 - and for receipt of unregister messages
1307 * 3. Use workload hints.
1309 static int qemu_rdma_unregister_waiting(RDMAContext
*rdma
)
1311 while (rdma
->unregistrations
[rdma
->unregister_current
]) {
1313 uint64_t wr_id
= rdma
->unregistrations
[rdma
->unregister_current
];
1315 (wr_id
& RDMA_WRID_CHUNK_MASK
) >> RDMA_WRID_CHUNK_SHIFT
;
1317 (wr_id
& RDMA_WRID_BLOCK_MASK
) >> RDMA_WRID_BLOCK_SHIFT
;
1318 RDMALocalBlock
*block
=
1319 &(rdma
->local_ram_blocks
.block
[index
]);
1320 RDMARegister reg
= { .current_index
= index
};
1321 RDMAControlHeader resp
= { .type
= RDMA_CONTROL_UNREGISTER_FINISHED
,
1323 RDMAControlHeader head
= { .len
= sizeof(RDMARegister
),
1324 .type
= RDMA_CONTROL_UNREGISTER_REQUEST
,
1328 trace_qemu_rdma_unregister_waiting_proc(chunk
,
1329 rdma
->unregister_current
);
1331 rdma
->unregistrations
[rdma
->unregister_current
] = 0;
1332 rdma
->unregister_current
++;
1334 if (rdma
->unregister_current
== RDMA_SIGNALED_SEND_MAX
) {
1335 rdma
->unregister_current
= 0;
1340 * Unregistration is speculative (because migration is single-threaded
1341 * and we cannot break the protocol's inifinband message ordering).
1342 * Thus, if the memory is currently being used for transmission,
1343 * then abort the attempt to unregister and try again
1344 * later the next time a completion is received for this memory.
1346 clear_bit(chunk
, block
->unregister_bitmap
);
1348 if (test_bit(chunk
, block
->transit_bitmap
)) {
1349 trace_qemu_rdma_unregister_waiting_inflight(chunk
);
1353 trace_qemu_rdma_unregister_waiting_send(chunk
);
1355 ret
= ibv_dereg_mr(block
->pmr
[chunk
]);
1356 block
->pmr
[chunk
] = NULL
;
1357 block
->remote_keys
[chunk
] = 0;
1360 perror("unregistration chunk failed");
1363 rdma
->total_registrations
--;
1365 reg
.key
.chunk
= chunk
;
1366 register_to_network(rdma
, ®
);
1367 ret
= qemu_rdma_exchange_send(rdma
, &head
, (uint8_t *) ®
,
1373 trace_qemu_rdma_unregister_waiting_complete(chunk
);
1379 static uint64_t qemu_rdma_make_wrid(uint64_t wr_id
, uint64_t index
,
1382 uint64_t result
= wr_id
& RDMA_WRID_TYPE_MASK
;
1384 result
|= (index
<< RDMA_WRID_BLOCK_SHIFT
);
1385 result
|= (chunk
<< RDMA_WRID_CHUNK_SHIFT
);
1391 * Set bit for unregistration in the next iteration.
1392 * We cannot transmit right here, but will unpin later.
1394 static void qemu_rdma_signal_unregister(RDMAContext
*rdma
, uint64_t index
,
1395 uint64_t chunk
, uint64_t wr_id
)
1397 if (rdma
->unregistrations
[rdma
->unregister_next
] != 0) {
1398 error_report("rdma migration: queue is full");
1400 RDMALocalBlock
*block
= &(rdma
->local_ram_blocks
.block
[index
]);
1402 if (!test_and_set_bit(chunk
, block
->unregister_bitmap
)) {
1403 trace_qemu_rdma_signal_unregister_append(chunk
,
1404 rdma
->unregister_next
);
1406 rdma
->unregistrations
[rdma
->unregister_next
++] =
1407 qemu_rdma_make_wrid(wr_id
, index
, chunk
);
1409 if (rdma
->unregister_next
== RDMA_SIGNALED_SEND_MAX
) {
1410 rdma
->unregister_next
= 0;
1413 trace_qemu_rdma_signal_unregister_already(chunk
);
1419 * Consult the connection manager to see a work request
1420 * (of any kind) has completed.
1421 * Return the work request ID that completed.
1423 static uint64_t qemu_rdma_poll(RDMAContext
*rdma
, uint64_t *wr_id_out
,
1430 ret
= ibv_poll_cq(rdma
->cq
, 1, &wc
);
1433 *wr_id_out
= RDMA_WRID_NONE
;
1438 error_report("ibv_poll_cq return %d", ret
);
1442 wr_id
= wc
.wr_id
& RDMA_WRID_TYPE_MASK
;
1444 if (wc
.status
!= IBV_WC_SUCCESS
) {
1445 fprintf(stderr
, "ibv_poll_cq wc.status=%d %s!\n",
1446 wc
.status
, ibv_wc_status_str(wc
.status
));
1447 fprintf(stderr
, "ibv_poll_cq wrid=%s!\n", wrid_desc
[wr_id
]);
1452 if (rdma
->control_ready_expected
&&
1453 (wr_id
>= RDMA_WRID_RECV_CONTROL
)) {
1454 trace_qemu_rdma_poll_recv(wrid_desc
[RDMA_WRID_RECV_CONTROL
],
1455 wr_id
- RDMA_WRID_RECV_CONTROL
, wr_id
, rdma
->nb_sent
);
1456 rdma
->control_ready_expected
= 0;
1459 if (wr_id
== RDMA_WRID_RDMA_WRITE
) {
1461 (wc
.wr_id
& RDMA_WRID_CHUNK_MASK
) >> RDMA_WRID_CHUNK_SHIFT
;
1463 (wc
.wr_id
& RDMA_WRID_BLOCK_MASK
) >> RDMA_WRID_BLOCK_SHIFT
;
1464 RDMALocalBlock
*block
= &(rdma
->local_ram_blocks
.block
[index
]);
1466 trace_qemu_rdma_poll_write(print_wrid(wr_id
), wr_id
, rdma
->nb_sent
,
1467 index
, chunk
, block
->local_host_addr
,
1468 (void *)(uintptr_t)block
->remote_host_addr
);
1470 clear_bit(chunk
, block
->transit_bitmap
);
1472 if (rdma
->nb_sent
> 0) {
1476 if (!rdma
->pin_all
) {
1478 * FYI: If one wanted to signal a specific chunk to be unregistered
1479 * using LRU or workload-specific information, this is the function
1480 * you would call to do so. That chunk would then get asynchronously
1481 * unregistered later.
1483 #ifdef RDMA_UNREGISTRATION_EXAMPLE
1484 qemu_rdma_signal_unregister(rdma
, index
, chunk
, wc
.wr_id
);
1488 trace_qemu_rdma_poll_other(print_wrid(wr_id
), wr_id
, rdma
->nb_sent
);
1491 *wr_id_out
= wc
.wr_id
;
1493 *byte_len
= wc
.byte_len
;
1499 /* Wait for activity on the completion channel.
1500 * Returns 0 on success, none-0 on error.
1502 static int qemu_rdma_wait_comp_channel(RDMAContext
*rdma
)
1504 struct rdma_cm_event
*cm_event
;
1508 * Coroutine doesn't start until migration_fd_process_incoming()
1509 * so don't yield unless we know we're running inside of a coroutine.
1511 if (rdma
->migration_started_on_destination
&&
1512 migration_incoming_get_current()->state
== MIGRATION_STATUS_ACTIVE
) {
1513 yield_until_fd_readable(rdma
->comp_channel
->fd
);
1515 /* This is the source side, we're in a separate thread
1516 * or destination prior to migration_fd_process_incoming()
1517 * after postcopy, the destination also in a separate thread.
1518 * we can't yield; so we have to poll the fd.
1519 * But we need to be able to handle 'cancel' or an error
1520 * without hanging forever.
1522 while (!rdma
->error_state
&& !rdma
->received_error
) {
1524 pfds
[0].fd
= rdma
->comp_channel
->fd
;
1525 pfds
[0].events
= G_IO_IN
| G_IO_HUP
| G_IO_ERR
;
1526 pfds
[0].revents
= 0;
1528 pfds
[1].fd
= rdma
->channel
->fd
;
1529 pfds
[1].events
= G_IO_IN
| G_IO_HUP
| G_IO_ERR
;
1530 pfds
[1].revents
= 0;
1532 /* 0.1s timeout, should be fine for a 'cancel' */
1533 switch (qemu_poll_ns(pfds
, 2, 100 * 1000 * 1000)) {
1535 case 1: /* fd active */
1536 if (pfds
[0].revents
) {
1540 if (pfds
[1].revents
) {
1541 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
1543 rdma_ack_cm_event(cm_event
);
1546 error_report("receive cm event while wait comp channel,"
1547 "cm event is %d", cm_event
->event
);
1548 if (cm_event
->event
== RDMA_CM_EVENT_DISCONNECTED
||
1549 cm_event
->event
== RDMA_CM_EVENT_DEVICE_REMOVAL
) {
1555 case 0: /* Timeout, go around again */
1558 default: /* Error of some type -
1559 * I don't trust errno from qemu_poll_ns
1561 error_report("%s: poll failed", __func__
);
1565 if (migrate_get_current()->state
== MIGRATION_STATUS_CANCELLING
) {
1566 /* Bail out and let the cancellation happen */
1572 if (rdma
->received_error
) {
1575 return rdma
->error_state
;
1579 * Block until the next work request has completed.
1581 * First poll to see if a work request has already completed,
1584 * If we encounter completed work requests for IDs other than
1585 * the one we're interested in, then that's generally an error.
1587 * The only exception is actual RDMA Write completions. These
1588 * completions only need to be recorded, but do not actually
1589 * need further processing.
1591 static int qemu_rdma_block_for_wrid(RDMAContext
*rdma
, int wrid_requested
,
1594 int num_cq_events
= 0, ret
= 0;
1597 uint64_t wr_id
= RDMA_WRID_NONE
, wr_id_in
;
1599 if (ibv_req_notify_cq(rdma
->cq
, 0)) {
1603 while (wr_id
!= wrid_requested
) {
1604 ret
= qemu_rdma_poll(rdma
, &wr_id_in
, byte_len
);
1609 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
1611 if (wr_id
== RDMA_WRID_NONE
) {
1614 if (wr_id
!= wrid_requested
) {
1615 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested
),
1616 wrid_requested
, print_wrid(wr_id
), wr_id
);
1620 if (wr_id
== wrid_requested
) {
1625 ret
= qemu_rdma_wait_comp_channel(rdma
);
1627 goto err_block_for_wrid
;
1630 ret
= ibv_get_cq_event(rdma
->comp_channel
, &cq
, &cq_ctx
);
1632 perror("ibv_get_cq_event");
1633 goto err_block_for_wrid
;
1638 ret
= -ibv_req_notify_cq(cq
, 0);
1640 goto err_block_for_wrid
;
1643 while (wr_id
!= wrid_requested
) {
1644 ret
= qemu_rdma_poll(rdma
, &wr_id_in
, byte_len
);
1646 goto err_block_for_wrid
;
1649 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
1651 if (wr_id
== RDMA_WRID_NONE
) {
1654 if (wr_id
!= wrid_requested
) {
1655 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested
),
1656 wrid_requested
, print_wrid(wr_id
), wr_id
);
1660 if (wr_id
== wrid_requested
) {
1661 goto success_block_for_wrid
;
1665 success_block_for_wrid
:
1666 if (num_cq_events
) {
1667 ibv_ack_cq_events(cq
, num_cq_events
);
1672 if (num_cq_events
) {
1673 ibv_ack_cq_events(cq
, num_cq_events
);
1676 rdma
->error_state
= ret
;
1681 * Post a SEND message work request for the control channel
1682 * containing some data and block until the post completes.
1684 static int qemu_rdma_post_send_control(RDMAContext
*rdma
, uint8_t *buf
,
1685 RDMAControlHeader
*head
)
1688 RDMAWorkRequestData
*wr
= &rdma
->wr_data
[RDMA_WRID_CONTROL
];
1689 struct ibv_send_wr
*bad_wr
;
1690 struct ibv_sge sge
= {
1691 .addr
= (uintptr_t)(wr
->control
),
1692 .length
= head
->len
+ sizeof(RDMAControlHeader
),
1693 .lkey
= wr
->control_mr
->lkey
,
1695 struct ibv_send_wr send_wr
= {
1696 .wr_id
= RDMA_WRID_SEND_CONTROL
,
1697 .opcode
= IBV_WR_SEND
,
1698 .send_flags
= IBV_SEND_SIGNALED
,
1703 trace_qemu_rdma_post_send_control(control_desc(head
->type
));
1706 * We don't actually need to do a memcpy() in here if we used
1707 * the "sge" properly, but since we're only sending control messages
1708 * (not RAM in a performance-critical path), then its OK for now.
1710 * The copy makes the RDMAControlHeader simpler to manipulate
1711 * for the time being.
1713 assert(head
->len
<= RDMA_CONTROL_MAX_BUFFER
- sizeof(*head
));
1714 memcpy(wr
->control
, head
, sizeof(RDMAControlHeader
));
1715 control_to_network((void *) wr
->control
);
1718 memcpy(wr
->control
+ sizeof(RDMAControlHeader
), buf
, head
->len
);
1722 ret
= ibv_post_send(rdma
->qp
, &send_wr
, &bad_wr
);
1725 error_report("Failed to use post IB SEND for control");
1729 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_SEND_CONTROL
, NULL
);
1731 error_report("rdma migration: send polling control error");
1738 * Post a RECV work request in anticipation of some future receipt
1739 * of data on the control channel.
1741 static int qemu_rdma_post_recv_control(RDMAContext
*rdma
, int idx
)
1743 struct ibv_recv_wr
*bad_wr
;
1744 struct ibv_sge sge
= {
1745 .addr
= (uintptr_t)(rdma
->wr_data
[idx
].control
),
1746 .length
= RDMA_CONTROL_MAX_BUFFER
,
1747 .lkey
= rdma
->wr_data
[idx
].control_mr
->lkey
,
1750 struct ibv_recv_wr recv_wr
= {
1751 .wr_id
= RDMA_WRID_RECV_CONTROL
+ idx
,
1757 if (ibv_post_recv(rdma
->qp
, &recv_wr
, &bad_wr
)) {
1765 * Block and wait for a RECV control channel message to arrive.
1767 static int qemu_rdma_exchange_get_response(RDMAContext
*rdma
,
1768 RDMAControlHeader
*head
, int expecting
, int idx
)
1771 int ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RECV_CONTROL
+ idx
,
1775 error_report("rdma migration: recv polling control error!");
1779 network_to_control((void *) rdma
->wr_data
[idx
].control
);
1780 memcpy(head
, rdma
->wr_data
[idx
].control
, sizeof(RDMAControlHeader
));
1782 trace_qemu_rdma_exchange_get_response_start(control_desc(expecting
));
1784 if (expecting
== RDMA_CONTROL_NONE
) {
1785 trace_qemu_rdma_exchange_get_response_none(control_desc(head
->type
),
1787 } else if (head
->type
!= expecting
|| head
->type
== RDMA_CONTROL_ERROR
) {
1788 error_report("Was expecting a %s (%d) control message"
1789 ", but got: %s (%d), length: %d",
1790 control_desc(expecting
), expecting
,
1791 control_desc(head
->type
), head
->type
, head
->len
);
1792 if (head
->type
== RDMA_CONTROL_ERROR
) {
1793 rdma
->received_error
= true;
1797 if (head
->len
> RDMA_CONTROL_MAX_BUFFER
- sizeof(*head
)) {
1798 error_report("too long length: %d", head
->len
);
1801 if (sizeof(*head
) + head
->len
!= byte_len
) {
1802 error_report("Malformed length: %d byte_len %d", head
->len
, byte_len
);
1810 * When a RECV work request has completed, the work request's
1811 * buffer is pointed at the header.
1813 * This will advance the pointer to the data portion
1814 * of the control message of the work request's buffer that
1815 * was populated after the work request finished.
1817 static void qemu_rdma_move_header(RDMAContext
*rdma
, int idx
,
1818 RDMAControlHeader
*head
)
1820 rdma
->wr_data
[idx
].control_len
= head
->len
;
1821 rdma
->wr_data
[idx
].control_curr
=
1822 rdma
->wr_data
[idx
].control
+ sizeof(RDMAControlHeader
);
1826 * This is an 'atomic' high-level operation to deliver a single, unified
1827 * control-channel message.
1829 * Additionally, if the user is expecting some kind of reply to this message,
1830 * they can request a 'resp' response message be filled in by posting an
1831 * additional work request on behalf of the user and waiting for an additional
1834 * The extra (optional) response is used during registration to us from having
1835 * to perform an *additional* exchange of message just to provide a response by
1836 * instead piggy-backing on the acknowledgement.
1838 static int qemu_rdma_exchange_send(RDMAContext
*rdma
, RDMAControlHeader
*head
,
1839 uint8_t *data
, RDMAControlHeader
*resp
,
1841 int (*callback
)(RDMAContext
*rdma
))
1846 * Wait until the dest is ready before attempting to deliver the message
1847 * by waiting for a READY message.
1849 if (rdma
->control_ready_expected
) {
1850 RDMAControlHeader resp
;
1851 ret
= qemu_rdma_exchange_get_response(rdma
,
1852 &resp
, RDMA_CONTROL_READY
, RDMA_WRID_READY
);
1859 * If the user is expecting a response, post a WR in anticipation of it.
1862 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_DATA
);
1864 error_report("rdma migration: error posting"
1865 " extra control recv for anticipated result!");
1871 * Post a WR to replace the one we just consumed for the READY message.
1873 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
1875 error_report("rdma migration: error posting first control recv!");
1880 * Deliver the control message that was requested.
1882 ret
= qemu_rdma_post_send_control(rdma
, data
, head
);
1885 error_report("Failed to send control buffer!");
1890 * If we're expecting a response, block and wait for it.
1894 trace_qemu_rdma_exchange_send_issue_callback();
1895 ret
= callback(rdma
);
1901 trace_qemu_rdma_exchange_send_waiting(control_desc(resp
->type
));
1902 ret
= qemu_rdma_exchange_get_response(rdma
, resp
,
1903 resp
->type
, RDMA_WRID_DATA
);
1909 qemu_rdma_move_header(rdma
, RDMA_WRID_DATA
, resp
);
1911 *resp_idx
= RDMA_WRID_DATA
;
1913 trace_qemu_rdma_exchange_send_received(control_desc(resp
->type
));
1916 rdma
->control_ready_expected
= 1;
1922 * This is an 'atomic' high-level operation to receive a single, unified
1923 * control-channel message.
1925 static int qemu_rdma_exchange_recv(RDMAContext
*rdma
, RDMAControlHeader
*head
,
1928 RDMAControlHeader ready
= {
1930 .type
= RDMA_CONTROL_READY
,
1936 * Inform the source that we're ready to receive a message.
1938 ret
= qemu_rdma_post_send_control(rdma
, NULL
, &ready
);
1941 error_report("Failed to send control buffer!");
1946 * Block and wait for the message.
1948 ret
= qemu_rdma_exchange_get_response(rdma
, head
,
1949 expecting
, RDMA_WRID_READY
);
1955 qemu_rdma_move_header(rdma
, RDMA_WRID_READY
, head
);
1958 * Post a new RECV work request to replace the one we just consumed.
1960 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
1962 error_report("rdma migration: error posting second control recv!");
1970 * Write an actual chunk of memory using RDMA.
1972 * If we're using dynamic registration on the dest-side, we have to
1973 * send a registration command first.
1975 static int qemu_rdma_write_one(QEMUFile
*f
, RDMAContext
*rdma
,
1976 int current_index
, uint64_t current_addr
,
1980 struct ibv_send_wr send_wr
= { 0 };
1981 struct ibv_send_wr
*bad_wr
;
1982 int reg_result_idx
, ret
, count
= 0;
1983 uint64_t chunk
, chunks
;
1984 uint8_t *chunk_start
, *chunk_end
;
1985 RDMALocalBlock
*block
= &(rdma
->local_ram_blocks
.block
[current_index
]);
1987 RDMARegisterResult
*reg_result
;
1988 RDMAControlHeader resp
= { .type
= RDMA_CONTROL_REGISTER_RESULT
};
1989 RDMAControlHeader head
= { .len
= sizeof(RDMARegister
),
1990 .type
= RDMA_CONTROL_REGISTER_REQUEST
,
1995 sge
.addr
= (uintptr_t)(block
->local_host_addr
+
1996 (current_addr
- block
->offset
));
1997 sge
.length
= length
;
1999 chunk
= ram_chunk_index(block
->local_host_addr
,
2000 (uint8_t *)(uintptr_t)sge
.addr
);
2001 chunk_start
= ram_chunk_start(block
, chunk
);
2003 if (block
->is_ram_block
) {
2004 chunks
= length
/ (1UL << RDMA_REG_CHUNK_SHIFT
);
2006 if (chunks
&& ((length
% (1UL << RDMA_REG_CHUNK_SHIFT
)) == 0)) {
2010 chunks
= block
->length
/ (1UL << RDMA_REG_CHUNK_SHIFT
);
2012 if (chunks
&& ((block
->length
% (1UL << RDMA_REG_CHUNK_SHIFT
)) == 0)) {
2017 trace_qemu_rdma_write_one_top(chunks
+ 1,
2019 (1UL << RDMA_REG_CHUNK_SHIFT
) / 1024 / 1024);
2021 chunk_end
= ram_chunk_end(block
, chunk
+ chunks
);
2023 if (!rdma
->pin_all
) {
2024 #ifdef RDMA_UNREGISTRATION_EXAMPLE
2025 qemu_rdma_unregister_waiting(rdma
);
2029 while (test_bit(chunk
, block
->transit_bitmap
)) {
2031 trace_qemu_rdma_write_one_block(count
++, current_index
, chunk
,
2032 sge
.addr
, length
, rdma
->nb_sent
, block
->nb_chunks
);
2034 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RDMA_WRITE
, NULL
);
2037 error_report("Failed to Wait for previous write to complete "
2038 "block %d chunk %" PRIu64
2039 " current %" PRIu64
" len %" PRIu64
" %d",
2040 current_index
, chunk
, sge
.addr
, length
, rdma
->nb_sent
);
2045 if (!rdma
->pin_all
|| !block
->is_ram_block
) {
2046 if (!block
->remote_keys
[chunk
]) {
2048 * This chunk has not yet been registered, so first check to see
2049 * if the entire chunk is zero. If so, tell the other size to
2050 * memset() + madvise() the entire chunk without RDMA.
2053 if (buffer_is_zero((void *)(uintptr_t)sge
.addr
, length
)) {
2054 RDMACompress comp
= {
2055 .offset
= current_addr
,
2057 .block_idx
= current_index
,
2061 head
.len
= sizeof(comp
);
2062 head
.type
= RDMA_CONTROL_COMPRESS
;
2064 trace_qemu_rdma_write_one_zero(chunk
, sge
.length
,
2065 current_index
, current_addr
);
2067 compress_to_network(rdma
, &comp
);
2068 ret
= qemu_rdma_exchange_send(rdma
, &head
,
2069 (uint8_t *) &comp
, NULL
, NULL
, NULL
);
2075 acct_update_position(f
, sge
.length
, true);
2081 * Otherwise, tell other side to register.
2083 reg
.current_index
= current_index
;
2084 if (block
->is_ram_block
) {
2085 reg
.key
.current_addr
= current_addr
;
2087 reg
.key
.chunk
= chunk
;
2089 reg
.chunks
= chunks
;
2091 trace_qemu_rdma_write_one_sendreg(chunk
, sge
.length
, current_index
,
2094 register_to_network(rdma
, ®
);
2095 ret
= qemu_rdma_exchange_send(rdma
, &head
, (uint8_t *) ®
,
2096 &resp
, ®_result_idx
, NULL
);
2101 /* try to overlap this single registration with the one we sent. */
2102 if (qemu_rdma_register_and_get_keys(rdma
, block
, sge
.addr
,
2103 &sge
.lkey
, NULL
, chunk
,
2104 chunk_start
, chunk_end
)) {
2105 error_report("cannot get lkey");
2109 reg_result
= (RDMARegisterResult
*)
2110 rdma
->wr_data
[reg_result_idx
].control_curr
;
2112 network_to_result(reg_result
);
2114 trace_qemu_rdma_write_one_recvregres(block
->remote_keys
[chunk
],
2115 reg_result
->rkey
, chunk
);
2117 block
->remote_keys
[chunk
] = reg_result
->rkey
;
2118 block
->remote_host_addr
= reg_result
->host_addr
;
2120 /* already registered before */
2121 if (qemu_rdma_register_and_get_keys(rdma
, block
, sge
.addr
,
2122 &sge
.lkey
, NULL
, chunk
,
2123 chunk_start
, chunk_end
)) {
2124 error_report("cannot get lkey!");
2129 send_wr
.wr
.rdma
.rkey
= block
->remote_keys
[chunk
];
2131 send_wr
.wr
.rdma
.rkey
= block
->remote_rkey
;
2133 if (qemu_rdma_register_and_get_keys(rdma
, block
, sge
.addr
,
2134 &sge
.lkey
, NULL
, chunk
,
2135 chunk_start
, chunk_end
)) {
2136 error_report("cannot get lkey!");
2142 * Encode the ram block index and chunk within this wrid.
2143 * We will use this information at the time of completion
2144 * to figure out which bitmap to check against and then which
2145 * chunk in the bitmap to look for.
2147 send_wr
.wr_id
= qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE
,
2148 current_index
, chunk
);
2150 send_wr
.opcode
= IBV_WR_RDMA_WRITE
;
2151 send_wr
.send_flags
= IBV_SEND_SIGNALED
;
2152 send_wr
.sg_list
= &sge
;
2153 send_wr
.num_sge
= 1;
2154 send_wr
.wr
.rdma
.remote_addr
= block
->remote_host_addr
+
2155 (current_addr
- block
->offset
);
2157 trace_qemu_rdma_write_one_post(chunk
, sge
.addr
, send_wr
.wr
.rdma
.remote_addr
,
2161 * ibv_post_send() does not return negative error numbers,
2162 * per the specification they are positive - no idea why.
2164 ret
= ibv_post_send(rdma
->qp
, &send_wr
, &bad_wr
);
2166 if (ret
== ENOMEM
) {
2167 trace_qemu_rdma_write_one_queue_full();
2168 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RDMA_WRITE
, NULL
);
2170 error_report("rdma migration: failed to make "
2171 "room in full send queue! %d", ret
);
2177 } else if (ret
> 0) {
2178 perror("rdma migration: post rdma write failed");
2182 set_bit(chunk
, block
->transit_bitmap
);
2183 acct_update_position(f
, sge
.length
, false);
2184 rdma
->total_writes
++;
2190 * Push out any unwritten RDMA operations.
2192 * We support sending out multiple chunks at the same time.
2193 * Not all of them need to get signaled in the completion queue.
2195 static int qemu_rdma_write_flush(QEMUFile
*f
, RDMAContext
*rdma
)
2199 if (!rdma
->current_length
) {
2203 ret
= qemu_rdma_write_one(f
, rdma
,
2204 rdma
->current_index
, rdma
->current_addr
, rdma
->current_length
);
2212 trace_qemu_rdma_write_flush(rdma
->nb_sent
);
2215 rdma
->current_length
= 0;
2216 rdma
->current_addr
= 0;
2221 static inline int qemu_rdma_buffer_mergable(RDMAContext
*rdma
,
2222 uint64_t offset
, uint64_t len
)
2224 RDMALocalBlock
*block
;
2228 if (rdma
->current_index
< 0) {
2232 if (rdma
->current_chunk
< 0) {
2236 block
= &(rdma
->local_ram_blocks
.block
[rdma
->current_index
]);
2237 host_addr
= block
->local_host_addr
+ (offset
- block
->offset
);
2238 chunk_end
= ram_chunk_end(block
, rdma
->current_chunk
);
2240 if (rdma
->current_length
== 0) {
2245 * Only merge into chunk sequentially.
2247 if (offset
!= (rdma
->current_addr
+ rdma
->current_length
)) {
2251 if (offset
< block
->offset
) {
2255 if ((offset
+ len
) > (block
->offset
+ block
->length
)) {
2259 if ((host_addr
+ len
) > chunk_end
) {
2267 * We're not actually writing here, but doing three things:
2269 * 1. Identify the chunk the buffer belongs to.
2270 * 2. If the chunk is full or the buffer doesn't belong to the current
2271 * chunk, then start a new chunk and flush() the old chunk.
2272 * 3. To keep the hardware busy, we also group chunks into batches
2273 * and only require that a batch gets acknowledged in the completion
2274 * queue instead of each individual chunk.
2276 static int qemu_rdma_write(QEMUFile
*f
, RDMAContext
*rdma
,
2277 uint64_t block_offset
, uint64_t offset
,
2280 uint64_t current_addr
= block_offset
+ offset
;
2281 uint64_t index
= rdma
->current_index
;
2282 uint64_t chunk
= rdma
->current_chunk
;
2285 /* If we cannot merge it, we flush the current buffer first. */
2286 if (!qemu_rdma_buffer_mergable(rdma
, current_addr
, len
)) {
2287 ret
= qemu_rdma_write_flush(f
, rdma
);
2291 rdma
->current_length
= 0;
2292 rdma
->current_addr
= current_addr
;
2294 ret
= qemu_rdma_search_ram_block(rdma
, block_offset
,
2295 offset
, len
, &index
, &chunk
);
2297 error_report("ram block search failed");
2300 rdma
->current_index
= index
;
2301 rdma
->current_chunk
= chunk
;
2305 rdma
->current_length
+= len
;
2307 /* flush it if buffer is too large */
2308 if (rdma
->current_length
>= RDMA_MERGE_MAX
) {
2309 return qemu_rdma_write_flush(f
, rdma
);
2315 static void qemu_rdma_cleanup(RDMAContext
*rdma
)
2319 if (rdma
->cm_id
&& rdma
->connected
) {
2320 if ((rdma
->error_state
||
2321 migrate_get_current()->state
== MIGRATION_STATUS_CANCELLING
) &&
2322 !rdma
->received_error
) {
2323 RDMAControlHeader head
= { .len
= 0,
2324 .type
= RDMA_CONTROL_ERROR
,
2327 error_report("Early error. Sending error.");
2328 qemu_rdma_post_send_control(rdma
, NULL
, &head
);
2331 rdma_disconnect(rdma
->cm_id
);
2332 trace_qemu_rdma_cleanup_disconnect();
2333 rdma
->connected
= false;
2336 if (rdma
->channel
) {
2337 qemu_set_fd_handler(rdma
->channel
->fd
, NULL
, NULL
, NULL
);
2339 g_free(rdma
->dest_blocks
);
2340 rdma
->dest_blocks
= NULL
;
2342 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
2343 if (rdma
->wr_data
[idx
].control_mr
) {
2344 rdma
->total_registrations
--;
2345 ibv_dereg_mr(rdma
->wr_data
[idx
].control_mr
);
2347 rdma
->wr_data
[idx
].control_mr
= NULL
;
2350 if (rdma
->local_ram_blocks
.block
) {
2351 while (rdma
->local_ram_blocks
.nb_blocks
) {
2352 rdma_delete_block(rdma
, &rdma
->local_ram_blocks
.block
[0]);
2357 rdma_destroy_qp(rdma
->cm_id
);
2361 ibv_destroy_cq(rdma
->cq
);
2364 if (rdma
->comp_channel
) {
2365 ibv_destroy_comp_channel(rdma
->comp_channel
);
2366 rdma
->comp_channel
= NULL
;
2369 ibv_dealloc_pd(rdma
->pd
);
2373 rdma_destroy_id(rdma
->cm_id
);
2377 /* the destination side, listen_id and channel is shared */
2378 if (rdma
->listen_id
) {
2379 if (!rdma
->is_return_path
) {
2380 rdma_destroy_id(rdma
->listen_id
);
2382 rdma
->listen_id
= NULL
;
2384 if (rdma
->channel
) {
2385 if (!rdma
->is_return_path
) {
2386 rdma_destroy_event_channel(rdma
->channel
);
2388 rdma
->channel
= NULL
;
2392 if (rdma
->channel
) {
2393 rdma_destroy_event_channel(rdma
->channel
);
2394 rdma
->channel
= NULL
;
2397 g_free(rdma
->host_port
);
2399 rdma
->host_port
= NULL
;
2403 static int qemu_rdma_source_init(RDMAContext
*rdma
, bool pin_all
, Error
**errp
)
2406 Error
*local_err
= NULL
, **temp
= &local_err
;
2409 * Will be validated against destination's actual capabilities
2410 * after the connect() completes.
2412 rdma
->pin_all
= pin_all
;
2414 ret
= qemu_rdma_resolve_host(rdma
, temp
);
2416 goto err_rdma_source_init
;
2419 ret
= qemu_rdma_alloc_pd_cq(rdma
);
2421 ERROR(temp
, "rdma migration: error allocating pd and cq! Your mlock()"
2422 " limits may be too low. Please check $ ulimit -a # and "
2423 "search for 'ulimit -l' in the output");
2424 goto err_rdma_source_init
;
2427 ret
= qemu_rdma_alloc_qp(rdma
);
2429 ERROR(temp
, "rdma migration: error allocating qp!");
2430 goto err_rdma_source_init
;
2433 ret
= qemu_rdma_init_ram_blocks(rdma
);
2435 ERROR(temp
, "rdma migration: error initializing ram blocks!");
2436 goto err_rdma_source_init
;
2439 /* Build the hash that maps from offset to RAMBlock */
2440 rdma
->blockmap
= g_hash_table_new(g_direct_hash
, g_direct_equal
);
2441 for (idx
= 0; idx
< rdma
->local_ram_blocks
.nb_blocks
; idx
++) {
2442 g_hash_table_insert(rdma
->blockmap
,
2443 (void *)(uintptr_t)rdma
->local_ram_blocks
.block
[idx
].offset
,
2444 &rdma
->local_ram_blocks
.block
[idx
]);
2447 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
2448 ret
= qemu_rdma_reg_control(rdma
, idx
);
2450 ERROR(temp
, "rdma migration: error registering %d control!",
2452 goto err_rdma_source_init
;
2458 err_rdma_source_init
:
2459 error_propagate(errp
, local_err
);
2460 qemu_rdma_cleanup(rdma
);
2464 static int qemu_get_cm_event_timeout(RDMAContext
*rdma
,
2465 struct rdma_cm_event
**cm_event
,
2466 long msec
, Error
**errp
)
2469 struct pollfd poll_fd
= {
2470 .fd
= rdma
->channel
->fd
,
2476 ret
= poll(&poll_fd
, 1, msec
);
2477 } while (ret
< 0 && errno
== EINTR
);
2480 ERROR(errp
, "poll cm event timeout");
2482 } else if (ret
< 0) {
2483 ERROR(errp
, "failed to poll cm event, errno=%i", errno
);
2485 } else if (poll_fd
.revents
& POLLIN
) {
2486 return rdma_get_cm_event(rdma
->channel
, cm_event
);
2488 ERROR(errp
, "no POLLIN event, revent=%x", poll_fd
.revents
);
2493 static int qemu_rdma_connect(RDMAContext
*rdma
, Error
**errp
, bool return_path
)
2495 RDMACapabilities cap
= {
2496 .version
= RDMA_CONTROL_VERSION_CURRENT
,
2499 struct rdma_conn_param conn_param
= { .initiator_depth
= 2,
2501 .private_data
= &cap
,
2502 .private_data_len
= sizeof(cap
),
2504 struct rdma_cm_event
*cm_event
;
2508 * Only negotiate the capability with destination if the user
2509 * on the source first requested the capability.
2511 if (rdma
->pin_all
) {
2512 trace_qemu_rdma_connect_pin_all_requested();
2513 cap
.flags
|= RDMA_CAPABILITY_PIN_ALL
;
2516 caps_to_network(&cap
);
2518 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
2520 ERROR(errp
, "posting second control recv");
2521 goto err_rdma_source_connect
;
2524 ret
= rdma_connect(rdma
->cm_id
, &conn_param
);
2526 perror("rdma_connect");
2527 ERROR(errp
, "connecting to destination!");
2528 goto err_rdma_source_connect
;
2532 ret
= qemu_get_cm_event_timeout(rdma
, &cm_event
, 5000, errp
);
2534 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
2537 perror("rdma_get_cm_event after rdma_connect");
2538 ERROR(errp
, "connecting to destination!");
2539 goto err_rdma_source_connect
;
2542 if (cm_event
->event
!= RDMA_CM_EVENT_ESTABLISHED
) {
2543 perror("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect");
2544 ERROR(errp
, "connecting to destination!");
2545 rdma_ack_cm_event(cm_event
);
2546 goto err_rdma_source_connect
;
2548 rdma
->connected
= true;
2550 memcpy(&cap
, cm_event
->param
.conn
.private_data
, sizeof(cap
));
2551 network_to_caps(&cap
);
2554 * Verify that the *requested* capabilities are supported by the destination
2555 * and disable them otherwise.
2557 if (rdma
->pin_all
&& !(cap
.flags
& RDMA_CAPABILITY_PIN_ALL
)) {
2558 ERROR(errp
, "Server cannot support pinning all memory. "
2559 "Will register memory dynamically.");
2560 rdma
->pin_all
= false;
2563 trace_qemu_rdma_connect_pin_all_outcome(rdma
->pin_all
);
2565 rdma_ack_cm_event(cm_event
);
2567 rdma
->control_ready_expected
= 1;
2571 err_rdma_source_connect
:
2572 qemu_rdma_cleanup(rdma
);
2576 static int qemu_rdma_dest_init(RDMAContext
*rdma
, Error
**errp
)
2579 struct rdma_cm_id
*listen_id
;
2580 char ip
[40] = "unknown";
2581 struct rdma_addrinfo
*res
, *e
;
2584 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
2585 rdma
->wr_data
[idx
].control_len
= 0;
2586 rdma
->wr_data
[idx
].control_curr
= NULL
;
2589 if (!rdma
->host
|| !rdma
->host
[0]) {
2590 ERROR(errp
, "RDMA host is not set!");
2591 rdma
->error_state
= -EINVAL
;
2594 /* create CM channel */
2595 rdma
->channel
= rdma_create_event_channel();
2596 if (!rdma
->channel
) {
2597 ERROR(errp
, "could not create rdma event channel");
2598 rdma
->error_state
= -EINVAL
;
2603 ret
= rdma_create_id(rdma
->channel
, &listen_id
, NULL
, RDMA_PS_TCP
);
2605 ERROR(errp
, "could not create cm_id!");
2606 goto err_dest_init_create_listen_id
;
2609 snprintf(port_str
, 16, "%d", rdma
->port
);
2610 port_str
[15] = '\0';
2612 ret
= rdma_getaddrinfo(rdma
->host
, port_str
, NULL
, &res
);
2614 ERROR(errp
, "could not rdma_getaddrinfo address %s", rdma
->host
);
2615 goto err_dest_init_bind_addr
;
2618 for (e
= res
; e
!= NULL
; e
= e
->ai_next
) {
2619 inet_ntop(e
->ai_family
,
2620 &((struct sockaddr_in
*) e
->ai_dst_addr
)->sin_addr
, ip
, sizeof ip
);
2621 trace_qemu_rdma_dest_init_trying(rdma
->host
, ip
);
2622 ret
= rdma_bind_addr(listen_id
, e
->ai_dst_addr
);
2626 if (e
->ai_family
== AF_INET6
) {
2627 ret
= qemu_rdma_broken_ipv6_kernel(listen_id
->verbs
, errp
);
2635 rdma_freeaddrinfo(res
);
2637 ERROR(errp
, "Error: could not rdma_bind_addr!");
2638 goto err_dest_init_bind_addr
;
2641 rdma
->listen_id
= listen_id
;
2642 qemu_rdma_dump_gid("dest_init", listen_id
);
2645 err_dest_init_bind_addr
:
2646 rdma_destroy_id(listen_id
);
2647 err_dest_init_create_listen_id
:
2648 rdma_destroy_event_channel(rdma
->channel
);
2649 rdma
->channel
= NULL
;
2650 rdma
->error_state
= ret
;
2655 static void qemu_rdma_return_path_dest_init(RDMAContext
*rdma_return_path
,
2660 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
2661 rdma_return_path
->wr_data
[idx
].control_len
= 0;
2662 rdma_return_path
->wr_data
[idx
].control_curr
= NULL
;
2665 /*the CM channel and CM id is shared*/
2666 rdma_return_path
->channel
= rdma
->channel
;
2667 rdma_return_path
->listen_id
= rdma
->listen_id
;
2669 rdma
->return_path
= rdma_return_path
;
2670 rdma_return_path
->return_path
= rdma
;
2671 rdma_return_path
->is_return_path
= true;
2674 static void *qemu_rdma_data_init(const char *host_port
, Error
**errp
)
2676 RDMAContext
*rdma
= NULL
;
2677 InetSocketAddress
*addr
;
2680 rdma
= g_new0(RDMAContext
, 1);
2681 rdma
->current_index
= -1;
2682 rdma
->current_chunk
= -1;
2684 addr
= g_new(InetSocketAddress
, 1);
2685 if (!inet_parse(addr
, host_port
, NULL
)) {
2686 rdma
->port
= atoi(addr
->port
);
2687 rdma
->host
= g_strdup(addr
->host
);
2688 rdma
->host_port
= g_strdup(host_port
);
2690 ERROR(errp
, "bad RDMA migration address '%s'", host_port
);
2695 qapi_free_InetSocketAddress(addr
);
2702 * QEMUFile interface to the control channel.
2703 * SEND messages for control only.
2704 * VM's ram is handled with regular RDMA messages.
2706 static ssize_t
qio_channel_rdma_writev(QIOChannel
*ioc
,
2707 const struct iovec
*iov
,
2713 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
2714 QEMUFile
*f
= rioc
->file
;
2721 RCU_READ_LOCK_GUARD();
2722 rdma
= qatomic_rcu_read(&rioc
->rdmaout
);
2728 CHECK_ERROR_STATE();
2731 * Push out any writes that
2732 * we're queued up for VM's ram.
2734 ret
= qemu_rdma_write_flush(f
, rdma
);
2736 rdma
->error_state
= ret
;
2740 for (i
= 0; i
< niov
; i
++) {
2741 size_t remaining
= iov
[i
].iov_len
;
2742 uint8_t * data
= (void *)iov
[i
].iov_base
;
2744 RDMAControlHeader head
;
2746 len
= MIN(remaining
, RDMA_SEND_INCREMENT
);
2750 head
.type
= RDMA_CONTROL_QEMU_FILE
;
2752 ret
= qemu_rdma_exchange_send(rdma
, &head
, data
, NULL
, NULL
, NULL
);
2755 rdma
->error_state
= ret
;
2767 static size_t qemu_rdma_fill(RDMAContext
*rdma
, uint8_t *buf
,
2768 size_t size
, int idx
)
2772 if (rdma
->wr_data
[idx
].control_len
) {
2773 trace_qemu_rdma_fill(rdma
->wr_data
[idx
].control_len
, size
);
2775 len
= MIN(size
, rdma
->wr_data
[idx
].control_len
);
2776 memcpy(buf
, rdma
->wr_data
[idx
].control_curr
, len
);
2777 rdma
->wr_data
[idx
].control_curr
+= len
;
2778 rdma
->wr_data
[idx
].control_len
-= len
;
2785 * QEMUFile interface to the control channel.
2786 * RDMA links don't use bytestreams, so we have to
2787 * return bytes to QEMUFile opportunistically.
2789 static ssize_t
qio_channel_rdma_readv(QIOChannel
*ioc
,
2790 const struct iovec
*iov
,
2796 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
2798 RDMAControlHeader head
;
2803 RCU_READ_LOCK_GUARD();
2804 rdma
= qatomic_rcu_read(&rioc
->rdmain
);
2810 CHECK_ERROR_STATE();
2812 for (i
= 0; i
< niov
; i
++) {
2813 size_t want
= iov
[i
].iov_len
;
2814 uint8_t *data
= (void *)iov
[i
].iov_base
;
2817 * First, we hold on to the last SEND message we
2818 * were given and dish out the bytes until we run
2821 ret
= qemu_rdma_fill(rdma
, data
, want
, 0);
2824 /* Got what we needed, so go to next iovec */
2829 /* If we got any data so far, then don't wait
2830 * for more, just return what we have */
2836 /* We've got nothing at all, so lets wait for
2839 ret
= qemu_rdma_exchange_recv(rdma
, &head
, RDMA_CONTROL_QEMU_FILE
);
2842 rdma
->error_state
= ret
;
2847 * SEND was received with new bytes, now try again.
2849 ret
= qemu_rdma_fill(rdma
, data
, want
, 0);
2853 /* Still didn't get enough, so lets just return */
2856 return QIO_CHANNEL_ERR_BLOCK
;
2866 * Block until all the outstanding chunks have been delivered by the hardware.
2868 static int qemu_rdma_drain_cq(QEMUFile
*f
, RDMAContext
*rdma
)
2872 if (qemu_rdma_write_flush(f
, rdma
) < 0) {
2876 while (rdma
->nb_sent
) {
2877 ret
= qemu_rdma_block_for_wrid(rdma
, RDMA_WRID_RDMA_WRITE
, NULL
);
2879 error_report("rdma migration: complete polling error!");
2884 qemu_rdma_unregister_waiting(rdma
);
2890 static int qio_channel_rdma_set_blocking(QIOChannel
*ioc
,
2894 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
2895 /* XXX we should make readv/writev actually honour this :-) */
2896 rioc
->blocking
= blocking
;
2901 typedef struct QIOChannelRDMASource QIOChannelRDMASource
;
2902 struct QIOChannelRDMASource
{
2904 QIOChannelRDMA
*rioc
;
2905 GIOCondition condition
;
2909 qio_channel_rdma_source_prepare(GSource
*source
,
2912 QIOChannelRDMASource
*rsource
= (QIOChannelRDMASource
*)source
;
2914 GIOCondition cond
= 0;
2917 RCU_READ_LOCK_GUARD();
2918 if (rsource
->condition
== G_IO_IN
) {
2919 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmain
);
2921 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmaout
);
2925 error_report("RDMAContext is NULL when prepare Gsource");
2929 if (rdma
->wr_data
[0].control_len
) {
2934 return cond
& rsource
->condition
;
2938 qio_channel_rdma_source_check(GSource
*source
)
2940 QIOChannelRDMASource
*rsource
= (QIOChannelRDMASource
*)source
;
2942 GIOCondition cond
= 0;
2944 RCU_READ_LOCK_GUARD();
2945 if (rsource
->condition
== G_IO_IN
) {
2946 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmain
);
2948 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmaout
);
2952 error_report("RDMAContext is NULL when check Gsource");
2956 if (rdma
->wr_data
[0].control_len
) {
2961 return cond
& rsource
->condition
;
2965 qio_channel_rdma_source_dispatch(GSource
*source
,
2966 GSourceFunc callback
,
2969 QIOChannelFunc func
= (QIOChannelFunc
)callback
;
2970 QIOChannelRDMASource
*rsource
= (QIOChannelRDMASource
*)source
;
2972 GIOCondition cond
= 0;
2974 RCU_READ_LOCK_GUARD();
2975 if (rsource
->condition
== G_IO_IN
) {
2976 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmain
);
2978 rdma
= qatomic_rcu_read(&rsource
->rioc
->rdmaout
);
2982 error_report("RDMAContext is NULL when dispatch Gsource");
2986 if (rdma
->wr_data
[0].control_len
) {
2991 return (*func
)(QIO_CHANNEL(rsource
->rioc
),
2992 (cond
& rsource
->condition
),
2997 qio_channel_rdma_source_finalize(GSource
*source
)
2999 QIOChannelRDMASource
*ssource
= (QIOChannelRDMASource
*)source
;
3001 object_unref(OBJECT(ssource
->rioc
));
3004 GSourceFuncs qio_channel_rdma_source_funcs
= {
3005 qio_channel_rdma_source_prepare
,
3006 qio_channel_rdma_source_check
,
3007 qio_channel_rdma_source_dispatch
,
3008 qio_channel_rdma_source_finalize
3011 static GSource
*qio_channel_rdma_create_watch(QIOChannel
*ioc
,
3012 GIOCondition condition
)
3014 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
3015 QIOChannelRDMASource
*ssource
;
3018 source
= g_source_new(&qio_channel_rdma_source_funcs
,
3019 sizeof(QIOChannelRDMASource
));
3020 ssource
= (QIOChannelRDMASource
*)source
;
3022 ssource
->rioc
= rioc
;
3023 object_ref(OBJECT(rioc
));
3025 ssource
->condition
= condition
;
3030 static void qio_channel_rdma_set_aio_fd_handler(QIOChannel
*ioc
,
3033 IOHandler
*io_write
,
3036 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
3038 aio_set_fd_handler(ctx
, rioc
->rdmain
->comp_channel
->fd
,
3039 false, io_read
, io_write
, NULL
, opaque
);
3041 aio_set_fd_handler(ctx
, rioc
->rdmaout
->comp_channel
->fd
,
3042 false, io_read
, io_write
, NULL
, opaque
);
3046 struct rdma_close_rcu
{
3047 struct rcu_head rcu
;
3048 RDMAContext
*rdmain
;
3049 RDMAContext
*rdmaout
;
3052 /* callback from qio_channel_rdma_close via call_rcu */
3053 static void qio_channel_rdma_close_rcu(struct rdma_close_rcu
*rcu
)
3056 qemu_rdma_cleanup(rcu
->rdmain
);
3060 qemu_rdma_cleanup(rcu
->rdmaout
);
3063 g_free(rcu
->rdmain
);
3064 g_free(rcu
->rdmaout
);
3068 static int qio_channel_rdma_close(QIOChannel
*ioc
,
3071 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
3072 RDMAContext
*rdmain
, *rdmaout
;
3073 struct rdma_close_rcu
*rcu
= g_new(struct rdma_close_rcu
, 1);
3075 trace_qemu_rdma_close();
3077 rdmain
= rioc
->rdmain
;
3079 qatomic_rcu_set(&rioc
->rdmain
, NULL
);
3082 rdmaout
= rioc
->rdmaout
;
3084 qatomic_rcu_set(&rioc
->rdmaout
, NULL
);
3087 rcu
->rdmain
= rdmain
;
3088 rcu
->rdmaout
= rdmaout
;
3089 call_rcu(rcu
, qio_channel_rdma_close_rcu
, rcu
);
3095 qio_channel_rdma_shutdown(QIOChannel
*ioc
,
3096 QIOChannelShutdown how
,
3099 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(ioc
);
3100 RDMAContext
*rdmain
, *rdmaout
;
3102 RCU_READ_LOCK_GUARD();
3104 rdmain
= qatomic_rcu_read(&rioc
->rdmain
);
3105 rdmaout
= qatomic_rcu_read(&rioc
->rdmain
);
3108 case QIO_CHANNEL_SHUTDOWN_READ
:
3110 rdmain
->error_state
= -1;
3113 case QIO_CHANNEL_SHUTDOWN_WRITE
:
3115 rdmaout
->error_state
= -1;
3118 case QIO_CHANNEL_SHUTDOWN_BOTH
:
3121 rdmain
->error_state
= -1;
3124 rdmaout
->error_state
= -1;
3135 * This means that 'block_offset' is a full virtual address that does not
3136 * belong to a RAMBlock of the virtual machine and instead
3137 * represents a private malloc'd memory area that the caller wishes to
3141 * Offset is an offset to be added to block_offset and used
3142 * to also lookup the corresponding RAMBlock.
3145 * Initiate an transfer this size.
3148 * A 'hint' or 'advice' that means that we wish to speculatively
3149 * and asynchronously unregister this memory. In this case, there is no
3150 * guarantee that the unregister will actually happen, for example,
3151 * if the memory is being actively transmitted. Additionally, the memory
3152 * may be re-registered at any future time if a write within the same
3153 * chunk was requested again, even if you attempted to unregister it
3156 * @size < 0 : TODO, not yet supported
3157 * Unregister the memory NOW. This means that the caller does not
3158 * expect there to be any future RDMA transfers and we just want to clean
3159 * things up. This is used in case the upper layer owns the memory and
3160 * cannot wait for qemu_fclose() to occur.
3162 * @bytes_sent : User-specificed pointer to indicate how many bytes were
3163 * sent. Usually, this will not be more than a few bytes of
3164 * the protocol because most transfers are sent asynchronously.
3166 static size_t qemu_rdma_save_page(QEMUFile
*f
, void *opaque
,
3167 ram_addr_t block_offset
, ram_addr_t offset
,
3168 size_t size
, uint64_t *bytes_sent
)
3170 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(opaque
);
3174 RCU_READ_LOCK_GUARD();
3175 rdma
= qatomic_rcu_read(&rioc
->rdmaout
);
3181 CHECK_ERROR_STATE();
3183 if (migration_in_postcopy()) {
3184 return RAM_SAVE_CONTROL_NOT_SUPP
;
3191 * Add this page to the current 'chunk'. If the chunk
3192 * is full, or the page doesn't belong to the current chunk,
3193 * an actual RDMA write will occur and a new chunk will be formed.
3195 ret
= qemu_rdma_write(f
, rdma
, block_offset
, offset
, size
);
3197 error_report("rdma migration: write error! %d", ret
);
3202 * We always return 1 bytes because the RDMA
3203 * protocol is completely asynchronous. We do not yet know
3204 * whether an identified chunk is zero or not because we're
3205 * waiting for other pages to potentially be merged with
3206 * the current chunk. So, we have to call qemu_update_position()
3207 * later on when the actual write occurs.
3213 uint64_t index
, chunk
;
3215 /* TODO: Change QEMUFileOps prototype to be signed: size_t => long
3217 ret = qemu_rdma_drain_cq(f, rdma);
3219 fprintf(stderr, "rdma: failed to synchronously drain"
3220 " completion queue before unregistration.\n");
3226 ret
= qemu_rdma_search_ram_block(rdma
, block_offset
,
3227 offset
, size
, &index
, &chunk
);
3230 error_report("ram block search failed");
3234 qemu_rdma_signal_unregister(rdma
, index
, chunk
, 0);
3237 * TODO: Synchronous, guaranteed unregistration (should not occur during
3238 * fast-path). Otherwise, unregisters will process on the next call to
3239 * qemu_rdma_drain_cq()
3241 qemu_rdma_unregister_waiting(rdma);
3247 * Drain the Completion Queue if possible, but do not block,
3250 * If nothing to poll, the end of the iteration will do this
3251 * again to make sure we don't overflow the request queue.
3254 uint64_t wr_id
, wr_id_in
;
3255 int ret
= qemu_rdma_poll(rdma
, &wr_id_in
, NULL
);
3257 error_report("rdma migration: polling error! %d", ret
);
3261 wr_id
= wr_id_in
& RDMA_WRID_TYPE_MASK
;
3263 if (wr_id
== RDMA_WRID_NONE
) {
3268 return RAM_SAVE_CONTROL_DELAYED
;
3270 rdma
->error_state
= ret
;
3274 static void rdma_accept_incoming_migration(void *opaque
);
3276 static void rdma_cm_poll_handler(void *opaque
)
3278 RDMAContext
*rdma
= opaque
;
3280 struct rdma_cm_event
*cm_event
;
3281 MigrationIncomingState
*mis
= migration_incoming_get_current();
3283 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
3285 error_report("get_cm_event failed %d", errno
);
3288 rdma_ack_cm_event(cm_event
);
3290 if (cm_event
->event
== RDMA_CM_EVENT_DISCONNECTED
||
3291 cm_event
->event
== RDMA_CM_EVENT_DEVICE_REMOVAL
) {
3292 if (!rdma
->error_state
&&
3293 migration_incoming_get_current()->state
!=
3294 MIGRATION_STATUS_COMPLETED
) {
3295 error_report("receive cm event, cm event is %d", cm_event
->event
);
3296 rdma
->error_state
= -EPIPE
;
3297 if (rdma
->return_path
) {
3298 rdma
->return_path
->error_state
= -EPIPE
;
3302 if (mis
->migration_incoming_co
) {
3303 qemu_coroutine_enter(mis
->migration_incoming_co
);
3309 static int qemu_rdma_accept(RDMAContext
*rdma
)
3311 RDMACapabilities cap
;
3312 struct rdma_conn_param conn_param
= {
3313 .responder_resources
= 2,
3314 .private_data
= &cap
,
3315 .private_data_len
= sizeof(cap
),
3317 RDMAContext
*rdma_return_path
= NULL
;
3318 struct rdma_cm_event
*cm_event
;
3319 struct ibv_context
*verbs
;
3323 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
3325 goto err_rdma_dest_wait
;
3328 if (cm_event
->event
!= RDMA_CM_EVENT_CONNECT_REQUEST
) {
3329 rdma_ack_cm_event(cm_event
);
3330 goto err_rdma_dest_wait
;
3334 * initialize the RDMAContext for return path for postcopy after first
3335 * connection request reached.
3337 if (migrate_postcopy() && !rdma
->is_return_path
) {
3338 rdma_return_path
= qemu_rdma_data_init(rdma
->host_port
, NULL
);
3339 if (rdma_return_path
== NULL
) {
3340 rdma_ack_cm_event(cm_event
);
3341 goto err_rdma_dest_wait
;
3344 qemu_rdma_return_path_dest_init(rdma_return_path
, rdma
);
3347 memcpy(&cap
, cm_event
->param
.conn
.private_data
, sizeof(cap
));
3349 network_to_caps(&cap
);
3351 if (cap
.version
< 1 || cap
.version
> RDMA_CONTROL_VERSION_CURRENT
) {
3352 error_report("Unknown source RDMA version: %d, bailing...",
3354 rdma_ack_cm_event(cm_event
);
3355 goto err_rdma_dest_wait
;
3359 * Respond with only the capabilities this version of QEMU knows about.
3361 cap
.flags
&= known_capabilities
;
3364 * Enable the ones that we do know about.
3365 * Add other checks here as new ones are introduced.
3367 if (cap
.flags
& RDMA_CAPABILITY_PIN_ALL
) {
3368 rdma
->pin_all
= true;
3371 rdma
->cm_id
= cm_event
->id
;
3372 verbs
= cm_event
->id
->verbs
;
3374 rdma_ack_cm_event(cm_event
);
3376 trace_qemu_rdma_accept_pin_state(rdma
->pin_all
);
3378 caps_to_network(&cap
);
3380 trace_qemu_rdma_accept_pin_verbsc(verbs
);
3383 rdma
->verbs
= verbs
;
3384 } else if (rdma
->verbs
!= verbs
) {
3385 error_report("ibv context not matching %p, %p!", rdma
->verbs
,
3387 goto err_rdma_dest_wait
;
3390 qemu_rdma_dump_id("dest_init", verbs
);
3392 ret
= qemu_rdma_alloc_pd_cq(rdma
);
3394 error_report("rdma migration: error allocating pd and cq!");
3395 goto err_rdma_dest_wait
;
3398 ret
= qemu_rdma_alloc_qp(rdma
);
3400 error_report("rdma migration: error allocating qp!");
3401 goto err_rdma_dest_wait
;
3404 ret
= qemu_rdma_init_ram_blocks(rdma
);
3406 error_report("rdma migration: error initializing ram blocks!");
3407 goto err_rdma_dest_wait
;
3410 for (idx
= 0; idx
< RDMA_WRID_MAX
; idx
++) {
3411 ret
= qemu_rdma_reg_control(rdma
, idx
);
3413 error_report("rdma: error registering %d control", idx
);
3414 goto err_rdma_dest_wait
;
3418 /* Accept the second connection request for return path */
3419 if (migrate_postcopy() && !rdma
->is_return_path
) {
3420 qemu_set_fd_handler(rdma
->channel
->fd
, rdma_accept_incoming_migration
,
3422 (void *)(intptr_t)rdma
->return_path
);
3424 qemu_set_fd_handler(rdma
->channel
->fd
, rdma_cm_poll_handler
,
3428 ret
= rdma_accept(rdma
->cm_id
, &conn_param
);
3430 error_report("rdma_accept returns %d", ret
);
3431 goto err_rdma_dest_wait
;
3434 ret
= rdma_get_cm_event(rdma
->channel
, &cm_event
);
3436 error_report("rdma_accept get_cm_event failed %d", ret
);
3437 goto err_rdma_dest_wait
;
3440 if (cm_event
->event
!= RDMA_CM_EVENT_ESTABLISHED
) {
3441 error_report("rdma_accept not event established");
3442 rdma_ack_cm_event(cm_event
);
3443 goto err_rdma_dest_wait
;
3446 rdma_ack_cm_event(cm_event
);
3447 rdma
->connected
= true;
3449 ret
= qemu_rdma_post_recv_control(rdma
, RDMA_WRID_READY
);
3451 error_report("rdma migration: error posting second control recv");
3452 goto err_rdma_dest_wait
;
3455 qemu_rdma_dump_gid("dest_connect", rdma
->cm_id
);
3460 rdma
->error_state
= ret
;
3461 qemu_rdma_cleanup(rdma
);
3462 g_free(rdma_return_path
);
3466 static int dest_ram_sort_func(const void *a
, const void *b
)
3468 unsigned int a_index
= ((const RDMALocalBlock
*)a
)->src_index
;
3469 unsigned int b_index
= ((const RDMALocalBlock
*)b
)->src_index
;
3471 return (a_index
< b_index
) ? -1 : (a_index
!= b_index
);
3475 * During each iteration of the migration, we listen for instructions
3476 * by the source VM to perform dynamic page registrations before they
3477 * can perform RDMA operations.
3479 * We respond with the 'rkey'.
3481 * Keep doing this until the source tells us to stop.
3483 static int qemu_rdma_registration_handle(QEMUFile
*f
, void *opaque
)
3485 RDMAControlHeader reg_resp
= { .len
= sizeof(RDMARegisterResult
),
3486 .type
= RDMA_CONTROL_REGISTER_RESULT
,
3489 RDMAControlHeader unreg_resp
= { .len
= 0,
3490 .type
= RDMA_CONTROL_UNREGISTER_FINISHED
,
3493 RDMAControlHeader blocks
= { .type
= RDMA_CONTROL_RAM_BLOCKS_RESULT
,
3495 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(opaque
);
3497 RDMALocalBlocks
*local
;
3498 RDMAControlHeader head
;
3499 RDMARegister
*reg
, *registers
;
3501 RDMARegisterResult
*reg_result
;
3502 static RDMARegisterResult results
[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE
];
3503 RDMALocalBlock
*block
;
3510 RCU_READ_LOCK_GUARD();
3511 rdma
= qatomic_rcu_read(&rioc
->rdmain
);
3517 CHECK_ERROR_STATE();
3519 local
= &rdma
->local_ram_blocks
;
3521 trace_qemu_rdma_registration_handle_wait();
3523 ret
= qemu_rdma_exchange_recv(rdma
, &head
, RDMA_CONTROL_NONE
);
3529 if (head
.repeat
> RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE
) {
3530 error_report("rdma: Too many requests in this message (%d)."
3531 "Bailing.", head
.repeat
);
3536 switch (head
.type
) {
3537 case RDMA_CONTROL_COMPRESS
:
3538 comp
= (RDMACompress
*) rdma
->wr_data
[idx
].control_curr
;
3539 network_to_compress(comp
);
3541 trace_qemu_rdma_registration_handle_compress(comp
->length
,
3544 if (comp
->block_idx
>= rdma
->local_ram_blocks
.nb_blocks
) {
3545 error_report("rdma: 'compress' bad block index %u (vs %d)",
3546 (unsigned int)comp
->block_idx
,
3547 rdma
->local_ram_blocks
.nb_blocks
);
3551 block
= &(rdma
->local_ram_blocks
.block
[comp
->block_idx
]);
3553 host_addr
= block
->local_host_addr
+
3554 (comp
->offset
- block
->offset
);
3556 ram_handle_compressed(host_addr
, comp
->value
, comp
->length
);
3559 case RDMA_CONTROL_REGISTER_FINISHED
:
3560 trace_qemu_rdma_registration_handle_finished();
3563 case RDMA_CONTROL_RAM_BLOCKS_REQUEST
:
3564 trace_qemu_rdma_registration_handle_ram_blocks();
3566 /* Sort our local RAM Block list so it's the same as the source,
3567 * we can do this since we've filled in a src_index in the list
3568 * as we received the RAMBlock list earlier.
3570 qsort(rdma
->local_ram_blocks
.block
,
3571 rdma
->local_ram_blocks
.nb_blocks
,
3572 sizeof(RDMALocalBlock
), dest_ram_sort_func
);
3573 for (i
= 0; i
< local
->nb_blocks
; i
++) {
3574 local
->block
[i
].index
= i
;
3577 if (rdma
->pin_all
) {
3578 ret
= qemu_rdma_reg_whole_ram_blocks(rdma
);
3580 error_report("rdma migration: error dest "
3581 "registering ram blocks");
3587 * Dest uses this to prepare to transmit the RAMBlock descriptions
3588 * to the source VM after connection setup.
3589 * Both sides use the "remote" structure to communicate and update
3590 * their "local" descriptions with what was sent.
3592 for (i
= 0; i
< local
->nb_blocks
; i
++) {
3593 rdma
->dest_blocks
[i
].remote_host_addr
=
3594 (uintptr_t)(local
->block
[i
].local_host_addr
);
3596 if (rdma
->pin_all
) {
3597 rdma
->dest_blocks
[i
].remote_rkey
= local
->block
[i
].mr
->rkey
;
3600 rdma
->dest_blocks
[i
].offset
= local
->block
[i
].offset
;
3601 rdma
->dest_blocks
[i
].length
= local
->block
[i
].length
;
3603 dest_block_to_network(&rdma
->dest_blocks
[i
]);
3604 trace_qemu_rdma_registration_handle_ram_blocks_loop(
3605 local
->block
[i
].block_name
,
3606 local
->block
[i
].offset
,
3607 local
->block
[i
].length
,
3608 local
->block
[i
].local_host_addr
,
3609 local
->block
[i
].src_index
);
3612 blocks
.len
= rdma
->local_ram_blocks
.nb_blocks
3613 * sizeof(RDMADestBlock
);
3616 ret
= qemu_rdma_post_send_control(rdma
,
3617 (uint8_t *) rdma
->dest_blocks
, &blocks
);
3620 error_report("rdma migration: error sending remote info");
3625 case RDMA_CONTROL_REGISTER_REQUEST
:
3626 trace_qemu_rdma_registration_handle_register(head
.repeat
);
3628 reg_resp
.repeat
= head
.repeat
;
3629 registers
= (RDMARegister
*) rdma
->wr_data
[idx
].control_curr
;
3631 for (count
= 0; count
< head
.repeat
; count
++) {
3633 uint8_t *chunk_start
, *chunk_end
;
3635 reg
= ®isters
[count
];
3636 network_to_register(reg
);
3638 reg_result
= &results
[count
];
3640 trace_qemu_rdma_registration_handle_register_loop(count
,
3641 reg
->current_index
, reg
->key
.current_addr
, reg
->chunks
);
3643 if (reg
->current_index
>= rdma
->local_ram_blocks
.nb_blocks
) {
3644 error_report("rdma: 'register' bad block index %u (vs %d)",
3645 (unsigned int)reg
->current_index
,
3646 rdma
->local_ram_blocks
.nb_blocks
);
3650 block
= &(rdma
->local_ram_blocks
.block
[reg
->current_index
]);
3651 if (block
->is_ram_block
) {
3652 if (block
->offset
> reg
->key
.current_addr
) {
3653 error_report("rdma: bad register address for block %s"
3654 " offset: %" PRIx64
" current_addr: %" PRIx64
,
3655 block
->block_name
, block
->offset
,
3656 reg
->key
.current_addr
);
3660 host_addr
= (block
->local_host_addr
+
3661 (reg
->key
.current_addr
- block
->offset
));
3662 chunk
= ram_chunk_index(block
->local_host_addr
,
3663 (uint8_t *) host_addr
);
3665 chunk
= reg
->key
.chunk
;
3666 host_addr
= block
->local_host_addr
+
3667 (reg
->key
.chunk
* (1UL << RDMA_REG_CHUNK_SHIFT
));
3668 /* Check for particularly bad chunk value */
3669 if (host_addr
< (void *)block
->local_host_addr
) {
3670 error_report("rdma: bad chunk for block %s"
3672 block
->block_name
, reg
->key
.chunk
);
3677 chunk_start
= ram_chunk_start(block
, chunk
);
3678 chunk_end
= ram_chunk_end(block
, chunk
+ reg
->chunks
);
3679 /* avoid "-Waddress-of-packed-member" warning */
3680 uint32_t tmp_rkey
= 0;
3681 if (qemu_rdma_register_and_get_keys(rdma
, block
,
3682 (uintptr_t)host_addr
, NULL
, &tmp_rkey
,
3683 chunk
, chunk_start
, chunk_end
)) {
3684 error_report("cannot get rkey");
3688 reg_result
->rkey
= tmp_rkey
;
3690 reg_result
->host_addr
= (uintptr_t)block
->local_host_addr
;
3692 trace_qemu_rdma_registration_handle_register_rkey(
3695 result_to_network(reg_result
);
3698 ret
= qemu_rdma_post_send_control(rdma
,
3699 (uint8_t *) results
, ®_resp
);
3702 error_report("Failed to send control buffer");
3706 case RDMA_CONTROL_UNREGISTER_REQUEST
:
3707 trace_qemu_rdma_registration_handle_unregister(head
.repeat
);
3708 unreg_resp
.repeat
= head
.repeat
;
3709 registers
= (RDMARegister
*) rdma
->wr_data
[idx
].control_curr
;
3711 for (count
= 0; count
< head
.repeat
; count
++) {
3712 reg
= ®isters
[count
];
3713 network_to_register(reg
);
3715 trace_qemu_rdma_registration_handle_unregister_loop(count
,
3716 reg
->current_index
, reg
->key
.chunk
);
3718 block
= &(rdma
->local_ram_blocks
.block
[reg
->current_index
]);
3720 ret
= ibv_dereg_mr(block
->pmr
[reg
->key
.chunk
]);
3721 block
->pmr
[reg
->key
.chunk
] = NULL
;
3724 perror("rdma unregistration chunk failed");
3729 rdma
->total_registrations
--;
3731 trace_qemu_rdma_registration_handle_unregister_success(
3735 ret
= qemu_rdma_post_send_control(rdma
, NULL
, &unreg_resp
);
3738 error_report("Failed to send control buffer");
3742 case RDMA_CONTROL_REGISTER_RESULT
:
3743 error_report("Invalid RESULT message at dest.");
3747 error_report("Unknown control message %s", control_desc(head
.type
));
3754 rdma
->error_state
= ret
;
3760 * Called via a ram_control_load_hook during the initial RAM load section which
3761 * lists the RAMBlocks by name. This lets us know the order of the RAMBlocks
3763 * We've already built our local RAMBlock list, but not yet sent the list to
3767 rdma_block_notification_handle(QIOChannelRDMA
*rioc
, const char *name
)
3773 RCU_READ_LOCK_GUARD();
3774 rdma
= qatomic_rcu_read(&rioc
->rdmain
);
3780 /* Find the matching RAMBlock in our local list */
3781 for (curr
= 0; curr
< rdma
->local_ram_blocks
.nb_blocks
; curr
++) {
3782 if (!strcmp(rdma
->local_ram_blocks
.block
[curr
].block_name
, name
)) {
3789 error_report("RAMBlock '%s' not found on destination", name
);
3793 rdma
->local_ram_blocks
.block
[curr
].src_index
= rdma
->next_src_index
;
3794 trace_rdma_block_notification_handle(name
, rdma
->next_src_index
);
3795 rdma
->next_src_index
++;
3800 static int rdma_load_hook(QEMUFile
*f
, void *opaque
, uint64_t flags
, void *data
)
3803 case RAM_CONTROL_BLOCK_REG
:
3804 return rdma_block_notification_handle(opaque
, data
);
3806 case RAM_CONTROL_HOOK
:
3807 return qemu_rdma_registration_handle(f
, opaque
);
3810 /* Shouldn't be called with any other values */
3815 static int qemu_rdma_registration_start(QEMUFile
*f
, void *opaque
,
3816 uint64_t flags
, void *data
)
3818 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(opaque
);
3821 RCU_READ_LOCK_GUARD();
3822 rdma
= qatomic_rcu_read(&rioc
->rdmaout
);
3827 CHECK_ERROR_STATE();
3829 if (migration_in_postcopy()) {
3833 trace_qemu_rdma_registration_start(flags
);
3834 qemu_put_be64(f
, RAM_SAVE_FLAG_HOOK
);
3841 * Inform dest that dynamic registrations are done for now.
3842 * First, flush writes, if any.
3844 static int qemu_rdma_registration_stop(QEMUFile
*f
, void *opaque
,
3845 uint64_t flags
, void *data
)
3847 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(opaque
);
3849 RDMAControlHeader head
= { .len
= 0, .repeat
= 1 };
3852 RCU_READ_LOCK_GUARD();
3853 rdma
= qatomic_rcu_read(&rioc
->rdmaout
);
3858 CHECK_ERROR_STATE();
3860 if (migration_in_postcopy()) {
3865 ret
= qemu_rdma_drain_cq(f
, rdma
);
3871 if (flags
== RAM_CONTROL_SETUP
) {
3872 RDMAControlHeader resp
= {.type
= RDMA_CONTROL_RAM_BLOCKS_RESULT
};
3873 RDMALocalBlocks
*local
= &rdma
->local_ram_blocks
;
3874 int reg_result_idx
, i
, nb_dest_blocks
;
3876 head
.type
= RDMA_CONTROL_RAM_BLOCKS_REQUEST
;
3877 trace_qemu_rdma_registration_stop_ram();
3880 * Make sure that we parallelize the pinning on both sides.
3881 * For very large guests, doing this serially takes a really
3882 * long time, so we have to 'interleave' the pinning locally
3883 * with the control messages by performing the pinning on this
3884 * side before we receive the control response from the other
3885 * side that the pinning has completed.
3887 ret
= qemu_rdma_exchange_send(rdma
, &head
, NULL
, &resp
,
3888 ®_result_idx
, rdma
->pin_all
?
3889 qemu_rdma_reg_whole_ram_blocks
: NULL
);
3891 fprintf(stderr
, "receiving remote info!");
3895 nb_dest_blocks
= resp
.len
/ sizeof(RDMADestBlock
);
3898 * The protocol uses two different sets of rkeys (mutually exclusive):
3899 * 1. One key to represent the virtual address of the entire ram block.
3900 * (dynamic chunk registration disabled - pin everything with one rkey.)
3901 * 2. One to represent individual chunks within a ram block.
3902 * (dynamic chunk registration enabled - pin individual chunks.)
3904 * Once the capability is successfully negotiated, the destination transmits
3905 * the keys to use (or sends them later) including the virtual addresses
3906 * and then propagates the remote ram block descriptions to his local copy.
3909 if (local
->nb_blocks
!= nb_dest_blocks
) {
3910 fprintf(stderr
, "ram blocks mismatch (Number of blocks %d vs %d) "
3911 "Your QEMU command line parameters are probably "
3912 "not identical on both the source and destination.",
3913 local
->nb_blocks
, nb_dest_blocks
);
3914 rdma
->error_state
= -EINVAL
;
3918 qemu_rdma_move_header(rdma
, reg_result_idx
, &resp
);
3919 memcpy(rdma
->dest_blocks
,
3920 rdma
->wr_data
[reg_result_idx
].control_curr
, resp
.len
);
3921 for (i
= 0; i
< nb_dest_blocks
; i
++) {
3922 network_to_dest_block(&rdma
->dest_blocks
[i
]);
3924 /* We require that the blocks are in the same order */
3925 if (rdma
->dest_blocks
[i
].length
!= local
->block
[i
].length
) {
3926 fprintf(stderr
, "Block %s/%d has a different length %" PRIu64
3927 "vs %" PRIu64
, local
->block
[i
].block_name
, i
,
3928 local
->block
[i
].length
,
3929 rdma
->dest_blocks
[i
].length
);
3930 rdma
->error_state
= -EINVAL
;
3933 local
->block
[i
].remote_host_addr
=
3934 rdma
->dest_blocks
[i
].remote_host_addr
;
3935 local
->block
[i
].remote_rkey
= rdma
->dest_blocks
[i
].remote_rkey
;
3939 trace_qemu_rdma_registration_stop(flags
);
3941 head
.type
= RDMA_CONTROL_REGISTER_FINISHED
;
3942 ret
= qemu_rdma_exchange_send(rdma
, &head
, NULL
, NULL
, NULL
, NULL
);
3950 rdma
->error_state
= ret
;
3954 static const QEMUFileHooks rdma_read_hooks
= {
3955 .hook_ram_load
= rdma_load_hook
,
3958 static const QEMUFileHooks rdma_write_hooks
= {
3959 .before_ram_iterate
= qemu_rdma_registration_start
,
3960 .after_ram_iterate
= qemu_rdma_registration_stop
,
3961 .save_page
= qemu_rdma_save_page
,
3965 static void qio_channel_rdma_finalize(Object
*obj
)
3967 QIOChannelRDMA
*rioc
= QIO_CHANNEL_RDMA(obj
);
3969 qemu_rdma_cleanup(rioc
->rdmain
);
3970 g_free(rioc
->rdmain
);
3971 rioc
->rdmain
= NULL
;
3973 if (rioc
->rdmaout
) {
3974 qemu_rdma_cleanup(rioc
->rdmaout
);
3975 g_free(rioc
->rdmaout
);
3976 rioc
->rdmaout
= NULL
;
3980 static void qio_channel_rdma_class_init(ObjectClass
*klass
,
3981 void *class_data G_GNUC_UNUSED
)
3983 QIOChannelClass
*ioc_klass
= QIO_CHANNEL_CLASS(klass
);
3985 ioc_klass
->io_writev
= qio_channel_rdma_writev
;
3986 ioc_klass
->io_readv
= qio_channel_rdma_readv
;
3987 ioc_klass
->io_set_blocking
= qio_channel_rdma_set_blocking
;
3988 ioc_klass
->io_close
= qio_channel_rdma_close
;
3989 ioc_klass
->io_create_watch
= qio_channel_rdma_create_watch
;
3990 ioc_klass
->io_set_aio_fd_handler
= qio_channel_rdma_set_aio_fd_handler
;
3991 ioc_klass
->io_shutdown
= qio_channel_rdma_shutdown
;
3994 static const TypeInfo qio_channel_rdma_info
= {
3995 .parent
= TYPE_QIO_CHANNEL
,
3996 .name
= TYPE_QIO_CHANNEL_RDMA
,
3997 .instance_size
= sizeof(QIOChannelRDMA
),
3998 .instance_finalize
= qio_channel_rdma_finalize
,
3999 .class_init
= qio_channel_rdma_class_init
,
4002 static void qio_channel_rdma_register_types(void)
4004 type_register_static(&qio_channel_rdma_info
);
4007 type_init(qio_channel_rdma_register_types
);
4009 static QEMUFile
*qemu_fopen_rdma(RDMAContext
*rdma
, const char *mode
)
4011 QIOChannelRDMA
*rioc
;
4013 if (qemu_file_mode_is_not_valid(mode
)) {
4017 rioc
= QIO_CHANNEL_RDMA(object_new(TYPE_QIO_CHANNEL_RDMA
));
4019 if (mode
[0] == 'w') {
4020 rioc
->file
= qemu_fopen_channel_output(QIO_CHANNEL(rioc
));
4021 rioc
->rdmaout
= rdma
;
4022 rioc
->rdmain
= rdma
->return_path
;
4023 qemu_file_set_hooks(rioc
->file
, &rdma_write_hooks
);
4025 rioc
->file
= qemu_fopen_channel_input(QIO_CHANNEL(rioc
));
4026 rioc
->rdmain
= rdma
;
4027 rioc
->rdmaout
= rdma
->return_path
;
4028 qemu_file_set_hooks(rioc
->file
, &rdma_read_hooks
);
4034 static void rdma_accept_incoming_migration(void *opaque
)
4036 RDMAContext
*rdma
= opaque
;
4039 Error
*local_err
= NULL
;
4041 trace_qemu_rdma_accept_incoming_migration();
4042 ret
= qemu_rdma_accept(rdma
);
4045 fprintf(stderr
, "RDMA ERROR: Migration initialization failed\n");
4049 trace_qemu_rdma_accept_incoming_migration_accepted();
4051 if (rdma
->is_return_path
) {
4055 f
= qemu_fopen_rdma(rdma
, "rb");
4057 fprintf(stderr
, "RDMA ERROR: could not qemu_fopen_rdma\n");
4058 qemu_rdma_cleanup(rdma
);
4062 rdma
->migration_started_on_destination
= 1;
4063 migration_fd_process_incoming(f
, &local_err
);
4065 error_reportf_err(local_err
, "RDMA ERROR:");
4069 void rdma_start_incoming_migration(const char *host_port
, Error
**errp
)
4072 RDMAContext
*rdma
, *rdma_return_path
= NULL
;
4073 Error
*local_err
= NULL
;
4075 trace_rdma_start_incoming_migration();
4077 /* Avoid ram_block_discard_disable(), cannot change during migration. */
4078 if (ram_block_discard_is_required()) {
4079 error_setg(errp
, "RDMA: cannot disable RAM discard");
4083 rdma
= qemu_rdma_data_init(host_port
, &local_err
);
4088 ret
= qemu_rdma_dest_init(rdma
, &local_err
);
4094 trace_rdma_start_incoming_migration_after_dest_init();
4096 ret
= rdma_listen(rdma
->listen_id
, 5);
4099 ERROR(errp
, "listening on socket!");
4103 trace_rdma_start_incoming_migration_after_rdma_listen();
4105 qemu_set_fd_handler(rdma
->channel
->fd
, rdma_accept_incoming_migration
,
4106 NULL
, (void *)(intptr_t)rdma
);
4110 qemu_rdma_cleanup(rdma
);
4112 error_propagate(errp
, local_err
);
4115 g_free(rdma
->host_port
);
4118 g_free(rdma_return_path
);
4121 void rdma_start_outgoing_migration(void *opaque
,
4122 const char *host_port
, Error
**errp
)
4124 MigrationState
*s
= opaque
;
4125 RDMAContext
*rdma_return_path
= NULL
;
4129 /* Avoid ram_block_discard_disable(), cannot change during migration. */
4130 if (ram_block_discard_is_required()) {
4131 error_setg(errp
, "RDMA: cannot disable RAM discard");
4135 rdma
= qemu_rdma_data_init(host_port
, errp
);
4140 ret
= qemu_rdma_source_init(rdma
,
4141 s
->enabled_capabilities
[MIGRATION_CAPABILITY_RDMA_PIN_ALL
], errp
);
4147 trace_rdma_start_outgoing_migration_after_rdma_source_init();
4148 ret
= qemu_rdma_connect(rdma
, errp
, false);
4154 /* RDMA postcopy need a separate queue pair for return path */
4155 if (migrate_postcopy()) {
4156 rdma_return_path
= qemu_rdma_data_init(host_port
, errp
);
4158 if (rdma_return_path
== NULL
) {
4159 goto return_path_err
;
4162 ret
= qemu_rdma_source_init(rdma_return_path
,
4163 s
->enabled_capabilities
[MIGRATION_CAPABILITY_RDMA_PIN_ALL
], errp
);
4166 goto return_path_err
;
4169 ret
= qemu_rdma_connect(rdma_return_path
, errp
, true);
4172 goto return_path_err
;
4175 rdma
->return_path
= rdma_return_path
;
4176 rdma_return_path
->return_path
= rdma
;
4177 rdma_return_path
->is_return_path
= true;
4180 trace_rdma_start_outgoing_migration_after_rdma_connect();
4182 s
->to_dst_file
= qemu_fopen_rdma(rdma
, "wb");
4183 migrate_fd_connect(s
, NULL
);
4186 qemu_rdma_cleanup(rdma
);
4189 g_free(rdma_return_path
);