1 /* 2 * RDMA protocol and interfaces 3 * 4 * Copyright IBM, Corp. 2010-2013 5 * 6 * Authors: 7 * Michael R. Hines <mrhines@us.ibm.com> 8 * Jiuxing Liu <jl@us.ibm.com> 9 * 10 * This work is licensed under the terms of the GNU GPL, version 2 or 11 * later. See the COPYING file in the top-level directory. 12 * 13 */ 14 #include "qemu-common.h" 15 #include "migration/migration.h" 16 #include "migration/qemu-file.h" 17 #include "exec/cpu-common.h" 18 #include "qemu/main-loop.h" 19 #include "qemu/sockets.h" 20 #include "qemu/bitmap.h" 21 #include "block/coroutine.h" 22 #include <stdio.h> 23 #include <sys/types.h> 24 #include <sys/socket.h> 25 #include <netdb.h> 26 #include <arpa/inet.h> 27 #include <string.h> 28 #include <rdma/rdma_cma.h> 29 #include "trace.h" 30 31 /* 32 * Print and error on both the Monitor and the Log file. 33 */ 34 #define ERROR(errp, fmt, ...) \ 35 do { \ 36 fprintf(stderr, "RDMA ERROR: " fmt "\n", ## __VA_ARGS__); \ 37 if (errp && (*(errp) == NULL)) { \ 38 error_setg(errp, "RDMA ERROR: " fmt, ## __VA_ARGS__); \ 39 } \ 40 } while (0) 41 42 #define RDMA_RESOLVE_TIMEOUT_MS 10000 43 44 /* Do not merge data if larger than this. */ 45 #define RDMA_MERGE_MAX (2 * 1024 * 1024) 46 #define RDMA_SIGNALED_SEND_MAX (RDMA_MERGE_MAX / 4096) 47 48 #define RDMA_REG_CHUNK_SHIFT 20 /* 1 MB */ 49 50 /* 51 * This is only for non-live state being migrated. 52 * Instead of RDMA_WRITE messages, we use RDMA_SEND 53 * messages for that state, which requires a different 54 * delivery design than main memory. 55 */ 56 #define RDMA_SEND_INCREMENT 32768 57 58 /* 59 * Maximum size infiniband SEND message 60 */ 61 #define RDMA_CONTROL_MAX_BUFFER (512 * 1024) 62 #define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096 63 64 #define RDMA_CONTROL_VERSION_CURRENT 1 65 /* 66 * Capabilities for negotiation. 67 */ 68 #define RDMA_CAPABILITY_PIN_ALL 0x01 69 70 /* 71 * Add the other flags above to this list of known capabilities 72 * as they are introduced. 73 */ 74 static uint32_t known_capabilities = RDMA_CAPABILITY_PIN_ALL; 75 76 #define CHECK_ERROR_STATE() \ 77 do { \ 78 if (rdma->error_state) { \ 79 if (!rdma->error_reported) { \ 80 error_report("RDMA is in an error state waiting migration" \ 81 " to abort!"); \ 82 rdma->error_reported = 1; \ 83 } \ 84 return rdma->error_state; \ 85 } \ 86 } while (0); 87 88 /* 89 * A work request ID is 64-bits and we split up these bits 90 * into 3 parts: 91 * 92 * bits 0-15 : type of control message, 2^16 93 * bits 16-29: ram block index, 2^14 94 * bits 30-63: ram block chunk number, 2^34 95 * 96 * The last two bit ranges are only used for RDMA writes, 97 * in order to track their completion and potentially 98 * also track unregistration status of the message. 99 */ 100 #define RDMA_WRID_TYPE_SHIFT 0UL 101 #define RDMA_WRID_BLOCK_SHIFT 16UL 102 #define RDMA_WRID_CHUNK_SHIFT 30UL 103 104 #define RDMA_WRID_TYPE_MASK \ 105 ((1UL << RDMA_WRID_BLOCK_SHIFT) - 1UL) 106 107 #define RDMA_WRID_BLOCK_MASK \ 108 (~RDMA_WRID_TYPE_MASK & ((1UL << RDMA_WRID_CHUNK_SHIFT) - 1UL)) 109 110 #define RDMA_WRID_CHUNK_MASK (~RDMA_WRID_BLOCK_MASK & ~RDMA_WRID_TYPE_MASK) 111 112 /* 113 * RDMA migration protocol: 114 * 1. RDMA Writes (data messages, i.e. RAM) 115 * 2. IB Send/Recv (control channel messages) 116 */ 117 enum { 118 RDMA_WRID_NONE = 0, 119 RDMA_WRID_RDMA_WRITE = 1, 120 RDMA_WRID_SEND_CONTROL = 2000, 121 RDMA_WRID_RECV_CONTROL = 4000, 122 }; 123 124 const char *wrid_desc[] = { 125 [RDMA_WRID_NONE] = "NONE", 126 [RDMA_WRID_RDMA_WRITE] = "WRITE RDMA", 127 [RDMA_WRID_SEND_CONTROL] = "CONTROL SEND", 128 [RDMA_WRID_RECV_CONTROL] = "CONTROL RECV", 129 }; 130 131 /* 132 * Work request IDs for IB SEND messages only (not RDMA writes). 133 * This is used by the migration protocol to transmit 134 * control messages (such as device state and registration commands) 135 * 136 * We could use more WRs, but we have enough for now. 137 */ 138 enum { 139 RDMA_WRID_READY = 0, 140 RDMA_WRID_DATA, 141 RDMA_WRID_CONTROL, 142 RDMA_WRID_MAX, 143 }; 144 145 /* 146 * SEND/RECV IB Control Messages. 147 */ 148 enum { 149 RDMA_CONTROL_NONE = 0, 150 RDMA_CONTROL_ERROR, 151 RDMA_CONTROL_READY, /* ready to receive */ 152 RDMA_CONTROL_QEMU_FILE, /* QEMUFile-transmitted bytes */ 153 RDMA_CONTROL_RAM_BLOCKS_REQUEST, /* RAMBlock synchronization */ 154 RDMA_CONTROL_RAM_BLOCKS_RESULT, /* RAMBlock synchronization */ 155 RDMA_CONTROL_COMPRESS, /* page contains repeat values */ 156 RDMA_CONTROL_REGISTER_REQUEST, /* dynamic page registration */ 157 RDMA_CONTROL_REGISTER_RESULT, /* key to use after registration */ 158 RDMA_CONTROL_REGISTER_FINISHED, /* current iteration finished */ 159 RDMA_CONTROL_UNREGISTER_REQUEST, /* dynamic UN-registration */ 160 RDMA_CONTROL_UNREGISTER_FINISHED, /* unpinning finished */ 161 }; 162 163 const char *control_desc[] = { 164 [RDMA_CONTROL_NONE] = "NONE", 165 [RDMA_CONTROL_ERROR] = "ERROR", 166 [RDMA_CONTROL_READY] = "READY", 167 [RDMA_CONTROL_QEMU_FILE] = "QEMU FILE", 168 [RDMA_CONTROL_RAM_BLOCKS_REQUEST] = "RAM BLOCKS REQUEST", 169 [RDMA_CONTROL_RAM_BLOCKS_RESULT] = "RAM BLOCKS RESULT", 170 [RDMA_CONTROL_COMPRESS] = "COMPRESS", 171 [RDMA_CONTROL_REGISTER_REQUEST] = "REGISTER REQUEST", 172 [RDMA_CONTROL_REGISTER_RESULT] = "REGISTER RESULT", 173 [RDMA_CONTROL_REGISTER_FINISHED] = "REGISTER FINISHED", 174 [RDMA_CONTROL_UNREGISTER_REQUEST] = "UNREGISTER REQUEST", 175 [RDMA_CONTROL_UNREGISTER_FINISHED] = "UNREGISTER FINISHED", 176 }; 177 178 /* 179 * Memory and MR structures used to represent an IB Send/Recv work request. 180 * This is *not* used for RDMA writes, only IB Send/Recv. 181 */ 182 typedef struct { 183 uint8_t control[RDMA_CONTROL_MAX_BUFFER]; /* actual buffer to register */ 184 struct ibv_mr *control_mr; /* registration metadata */ 185 size_t control_len; /* length of the message */ 186 uint8_t *control_curr; /* start of unconsumed bytes */ 187 } RDMAWorkRequestData; 188 189 /* 190 * Negotiate RDMA capabilities during connection-setup time. 191 */ 192 typedef struct { 193 uint32_t version; 194 uint32_t flags; 195 } RDMACapabilities; 196 197 static void caps_to_network(RDMACapabilities *cap) 198 { 199 cap->version = htonl(cap->version); 200 cap->flags = htonl(cap->flags); 201 } 202 203 static void network_to_caps(RDMACapabilities *cap) 204 { 205 cap->version = ntohl(cap->version); 206 cap->flags = ntohl(cap->flags); 207 } 208 209 /* 210 * Representation of a RAMBlock from an RDMA perspective. 211 * This is not transmitted, only local. 212 * This and subsequent structures cannot be linked lists 213 * because we're using a single IB message to transmit 214 * the information. It's small anyway, so a list is overkill. 215 */ 216 typedef struct RDMALocalBlock { 217 uint8_t *local_host_addr; /* local virtual address */ 218 uint64_t remote_host_addr; /* remote virtual address */ 219 uint64_t offset; 220 uint64_t length; 221 struct ibv_mr **pmr; /* MRs for chunk-level registration */ 222 struct ibv_mr *mr; /* MR for non-chunk-level registration */ 223 uint32_t *remote_keys; /* rkeys for chunk-level registration */ 224 uint32_t remote_rkey; /* rkeys for non-chunk-level registration */ 225 int index; /* which block are we */ 226 bool is_ram_block; 227 int nb_chunks; 228 unsigned long *transit_bitmap; 229 unsigned long *unregister_bitmap; 230 } RDMALocalBlock; 231 232 /* 233 * Also represents a RAMblock, but only on the dest. 234 * This gets transmitted by the dest during connection-time 235 * to the source VM and then is used to populate the 236 * corresponding RDMALocalBlock with 237 * the information needed to perform the actual RDMA. 238 */ 239 typedef struct QEMU_PACKED RDMARemoteBlock { 240 uint64_t remote_host_addr; 241 uint64_t offset; 242 uint64_t length; 243 uint32_t remote_rkey; 244 uint32_t padding; 245 } RDMARemoteBlock; 246 247 static uint64_t htonll(uint64_t v) 248 { 249 union { uint32_t lv[2]; uint64_t llv; } u; 250 u.lv[0] = htonl(v >> 32); 251 u.lv[1] = htonl(v & 0xFFFFFFFFULL); 252 return u.llv; 253 } 254 255 static uint64_t ntohll(uint64_t v) { 256 union { uint32_t lv[2]; uint64_t llv; } u; 257 u.llv = v; 258 return ((uint64_t)ntohl(u.lv[0]) << 32) | (uint64_t) ntohl(u.lv[1]); 259 } 260 261 static void remote_block_to_network(RDMARemoteBlock *rb) 262 { 263 rb->remote_host_addr = htonll(rb->remote_host_addr); 264 rb->offset = htonll(rb->offset); 265 rb->length = htonll(rb->length); 266 rb->remote_rkey = htonl(rb->remote_rkey); 267 } 268 269 static void network_to_remote_block(RDMARemoteBlock *rb) 270 { 271 rb->remote_host_addr = ntohll(rb->remote_host_addr); 272 rb->offset = ntohll(rb->offset); 273 rb->length = ntohll(rb->length); 274 rb->remote_rkey = ntohl(rb->remote_rkey); 275 } 276 277 /* 278 * Virtual address of the above structures used for transmitting 279 * the RAMBlock descriptions at connection-time. 280 * This structure is *not* transmitted. 281 */ 282 typedef struct RDMALocalBlocks { 283 int nb_blocks; 284 bool init; /* main memory init complete */ 285 RDMALocalBlock *block; 286 } RDMALocalBlocks; 287 288 /* 289 * Main data structure for RDMA state. 290 * While there is only one copy of this structure being allocated right now, 291 * this is the place where one would start if you wanted to consider 292 * having more than one RDMA connection open at the same time. 293 */ 294 typedef struct RDMAContext { 295 char *host; 296 int port; 297 298 RDMAWorkRequestData wr_data[RDMA_WRID_MAX]; 299 300 /* 301 * This is used by *_exchange_send() to figure out whether or not 302 * the initial "READY" message has already been received or not. 303 * This is because other functions may potentially poll() and detect 304 * the READY message before send() does, in which case we need to 305 * know if it completed. 306 */ 307 int control_ready_expected; 308 309 /* number of outstanding writes */ 310 int nb_sent; 311 312 /* store info about current buffer so that we can 313 merge it with future sends */ 314 uint64_t current_addr; 315 uint64_t current_length; 316 /* index of ram block the current buffer belongs to */ 317 int current_index; 318 /* index of the chunk in the current ram block */ 319 int current_chunk; 320 321 bool pin_all; 322 323 /* 324 * infiniband-specific variables for opening the device 325 * and maintaining connection state and so forth. 326 * 327 * cm_id also has ibv_context, rdma_event_channel, and ibv_qp in 328 * cm_id->verbs, cm_id->channel, and cm_id->qp. 329 */ 330 struct rdma_cm_id *cm_id; /* connection manager ID */ 331 struct rdma_cm_id *listen_id; 332 bool connected; 333 334 struct ibv_context *verbs; 335 struct rdma_event_channel *channel; 336 struct ibv_qp *qp; /* queue pair */ 337 struct ibv_comp_channel *comp_channel; /* completion channel */ 338 struct ibv_pd *pd; /* protection domain */ 339 struct ibv_cq *cq; /* completion queue */ 340 341 /* 342 * If a previous write failed (perhaps because of a failed 343 * memory registration, then do not attempt any future work 344 * and remember the error state. 345 */ 346 int error_state; 347 int error_reported; 348 349 /* 350 * Description of ram blocks used throughout the code. 351 */ 352 RDMALocalBlocks local_ram_blocks; 353 RDMARemoteBlock *block; 354 355 /* 356 * Migration on *destination* started. 357 * Then use coroutine yield function. 358 * Source runs in a thread, so we don't care. 359 */ 360 int migration_started_on_destination; 361 362 int total_registrations; 363 int total_writes; 364 365 int unregister_current, unregister_next; 366 uint64_t unregistrations[RDMA_SIGNALED_SEND_MAX]; 367 368 GHashTable *blockmap; 369 } RDMAContext; 370 371 /* 372 * Interface to the rest of the migration call stack. 373 */ 374 typedef struct QEMUFileRDMA { 375 RDMAContext *rdma; 376 size_t len; 377 void *file; 378 } QEMUFileRDMA; 379 380 /* 381 * Main structure for IB Send/Recv control messages. 382 * This gets prepended at the beginning of every Send/Recv. 383 */ 384 typedef struct QEMU_PACKED { 385 uint32_t len; /* Total length of data portion */ 386 uint32_t type; /* which control command to perform */ 387 uint32_t repeat; /* number of commands in data portion of same type */ 388 uint32_t padding; 389 } RDMAControlHeader; 390 391 static void control_to_network(RDMAControlHeader *control) 392 { 393 control->type = htonl(control->type); 394 control->len = htonl(control->len); 395 control->repeat = htonl(control->repeat); 396 } 397 398 static void network_to_control(RDMAControlHeader *control) 399 { 400 control->type = ntohl(control->type); 401 control->len = ntohl(control->len); 402 control->repeat = ntohl(control->repeat); 403 } 404 405 /* 406 * Register a single Chunk. 407 * Information sent by the source VM to inform the dest 408 * to register an single chunk of memory before we can perform 409 * the actual RDMA operation. 410 */ 411 typedef struct QEMU_PACKED { 412 union QEMU_PACKED { 413 uint64_t current_addr; /* offset into the ramblock of the chunk */ 414 uint64_t chunk; /* chunk to lookup if unregistering */ 415 } key; 416 uint32_t current_index; /* which ramblock the chunk belongs to */ 417 uint32_t padding; 418 uint64_t chunks; /* how many sequential chunks to register */ 419 } RDMARegister; 420 421 static void register_to_network(RDMARegister *reg) 422 { 423 reg->key.current_addr = htonll(reg->key.current_addr); 424 reg->current_index = htonl(reg->current_index); 425 reg->chunks = htonll(reg->chunks); 426 } 427 428 static void network_to_register(RDMARegister *reg) 429 { 430 reg->key.current_addr = ntohll(reg->key.current_addr); 431 reg->current_index = ntohl(reg->current_index); 432 reg->chunks = ntohll(reg->chunks); 433 } 434 435 typedef struct QEMU_PACKED { 436 uint32_t value; /* if zero, we will madvise() */ 437 uint32_t block_idx; /* which ram block index */ 438 uint64_t offset; /* where in the remote ramblock this chunk */ 439 uint64_t length; /* length of the chunk */ 440 } RDMACompress; 441 442 static void compress_to_network(RDMACompress *comp) 443 { 444 comp->value = htonl(comp->value); 445 comp->block_idx = htonl(comp->block_idx); 446 comp->offset = htonll(comp->offset); 447 comp->length = htonll(comp->length); 448 } 449 450 static void network_to_compress(RDMACompress *comp) 451 { 452 comp->value = ntohl(comp->value); 453 comp->block_idx = ntohl(comp->block_idx); 454 comp->offset = ntohll(comp->offset); 455 comp->length = ntohll(comp->length); 456 } 457 458 /* 459 * The result of the dest's memory registration produces an "rkey" 460 * which the source VM must reference in order to perform 461 * the RDMA operation. 462 */ 463 typedef struct QEMU_PACKED { 464 uint32_t rkey; 465 uint32_t padding; 466 uint64_t host_addr; 467 } RDMARegisterResult; 468 469 static void result_to_network(RDMARegisterResult *result) 470 { 471 result->rkey = htonl(result->rkey); 472 result->host_addr = htonll(result->host_addr); 473 }; 474 475 static void network_to_result(RDMARegisterResult *result) 476 { 477 result->rkey = ntohl(result->rkey); 478 result->host_addr = ntohll(result->host_addr); 479 }; 480 481 const char *print_wrid(int wrid); 482 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head, 483 uint8_t *data, RDMAControlHeader *resp, 484 int *resp_idx, 485 int (*callback)(RDMAContext *rdma)); 486 487 static inline uint64_t ram_chunk_index(const uint8_t *start, 488 const uint8_t *host) 489 { 490 return ((uintptr_t) host - (uintptr_t) start) >> RDMA_REG_CHUNK_SHIFT; 491 } 492 493 static inline uint8_t *ram_chunk_start(const RDMALocalBlock *rdma_ram_block, 494 uint64_t i) 495 { 496 return (uint8_t *) (((uintptr_t) rdma_ram_block->local_host_addr) 497 + (i << RDMA_REG_CHUNK_SHIFT)); 498 } 499 500 static inline uint8_t *ram_chunk_end(const RDMALocalBlock *rdma_ram_block, 501 uint64_t i) 502 { 503 uint8_t *result = ram_chunk_start(rdma_ram_block, i) + 504 (1UL << RDMA_REG_CHUNK_SHIFT); 505 506 if (result > (rdma_ram_block->local_host_addr + rdma_ram_block->length)) { 507 result = rdma_ram_block->local_host_addr + rdma_ram_block->length; 508 } 509 510 return result; 511 } 512 513 static int __qemu_rdma_add_block(RDMAContext *rdma, void *host_addr, 514 ram_addr_t block_offset, uint64_t length) 515 { 516 RDMALocalBlocks *local = &rdma->local_ram_blocks; 517 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap, 518 (void *) block_offset); 519 RDMALocalBlock *old = local->block; 520 521 assert(block == NULL); 522 523 local->block = g_malloc0(sizeof(RDMALocalBlock) * (local->nb_blocks + 1)); 524 525 if (local->nb_blocks) { 526 int x; 527 528 for (x = 0; x < local->nb_blocks; x++) { 529 g_hash_table_remove(rdma->blockmap, (void *)old[x].offset); 530 g_hash_table_insert(rdma->blockmap, (void *)old[x].offset, 531 &local->block[x]); 532 } 533 memcpy(local->block, old, sizeof(RDMALocalBlock) * local->nb_blocks); 534 g_free(old); 535 } 536 537 block = &local->block[local->nb_blocks]; 538 539 block->local_host_addr = host_addr; 540 block->offset = block_offset; 541 block->length = length; 542 block->index = local->nb_blocks; 543 block->nb_chunks = ram_chunk_index(host_addr, host_addr + length) + 1UL; 544 block->transit_bitmap = bitmap_new(block->nb_chunks); 545 bitmap_clear(block->transit_bitmap, 0, block->nb_chunks); 546 block->unregister_bitmap = bitmap_new(block->nb_chunks); 547 bitmap_clear(block->unregister_bitmap, 0, block->nb_chunks); 548 block->remote_keys = g_malloc0(block->nb_chunks * sizeof(uint32_t)); 549 550 block->is_ram_block = local->init ? false : true; 551 552 g_hash_table_insert(rdma->blockmap, (void *) block_offset, block); 553 554 trace___qemu_rdma_add_block(local->nb_blocks, 555 (uint64_t) block->local_host_addr, block->offset, 556 block->length, 557 (uint64_t) (block->local_host_addr + block->length), 558 BITS_TO_LONGS(block->nb_chunks) * 559 sizeof(unsigned long) * 8, 560 block->nb_chunks); 561 562 local->nb_blocks++; 563 564 return 0; 565 } 566 567 /* 568 * Memory regions need to be registered with the device and queue pairs setup 569 * in advanced before the migration starts. This tells us where the RAM blocks 570 * are so that we can register them individually. 571 */ 572 static void qemu_rdma_init_one_block(void *host_addr, 573 ram_addr_t block_offset, ram_addr_t length, void *opaque) 574 { 575 __qemu_rdma_add_block(opaque, host_addr, block_offset, length); 576 } 577 578 /* 579 * Identify the RAMBlocks and their quantity. They will be references to 580 * identify chunk boundaries inside each RAMBlock and also be referenced 581 * during dynamic page registration. 582 */ 583 static int qemu_rdma_init_ram_blocks(RDMAContext *rdma) 584 { 585 RDMALocalBlocks *local = &rdma->local_ram_blocks; 586 587 assert(rdma->blockmap == NULL); 588 rdma->blockmap = g_hash_table_new(g_direct_hash, g_direct_equal); 589 memset(local, 0, sizeof *local); 590 qemu_ram_foreach_block(qemu_rdma_init_one_block, rdma); 591 trace_qemu_rdma_init_ram_blocks(local->nb_blocks); 592 rdma->block = (RDMARemoteBlock *) g_malloc0(sizeof(RDMARemoteBlock) * 593 rdma->local_ram_blocks.nb_blocks); 594 local->init = true; 595 return 0; 596 } 597 598 static int __qemu_rdma_delete_block(RDMAContext *rdma, ram_addr_t block_offset) 599 { 600 RDMALocalBlocks *local = &rdma->local_ram_blocks; 601 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap, 602 (void *) block_offset); 603 RDMALocalBlock *old = local->block; 604 int x; 605 606 assert(block); 607 608 if (block->pmr) { 609 int j; 610 611 for (j = 0; j < block->nb_chunks; j++) { 612 if (!block->pmr[j]) { 613 continue; 614 } 615 ibv_dereg_mr(block->pmr[j]); 616 rdma->total_registrations--; 617 } 618 g_free(block->pmr); 619 block->pmr = NULL; 620 } 621 622 if (block->mr) { 623 ibv_dereg_mr(block->mr); 624 rdma->total_registrations--; 625 block->mr = NULL; 626 } 627 628 g_free(block->transit_bitmap); 629 block->transit_bitmap = NULL; 630 631 g_free(block->unregister_bitmap); 632 block->unregister_bitmap = NULL; 633 634 g_free(block->remote_keys); 635 block->remote_keys = NULL; 636 637 for (x = 0; x < local->nb_blocks; x++) { 638 g_hash_table_remove(rdma->blockmap, (void *)old[x].offset); 639 } 640 641 if (local->nb_blocks > 1) { 642 643 local->block = g_malloc0(sizeof(RDMALocalBlock) * 644 (local->nb_blocks - 1)); 645 646 if (block->index) { 647 memcpy(local->block, old, sizeof(RDMALocalBlock) * block->index); 648 } 649 650 if (block->index < (local->nb_blocks - 1)) { 651 memcpy(local->block + block->index, old + (block->index + 1), 652 sizeof(RDMALocalBlock) * 653 (local->nb_blocks - (block->index + 1))); 654 } 655 } else { 656 assert(block == local->block); 657 local->block = NULL; 658 } 659 660 trace___qemu_rdma_delete_block(local->nb_blocks, 661 (uint64_t)block->local_host_addr, 662 block->offset, block->length, 663 (uint64_t)(block->local_host_addr + block->length), 664 BITS_TO_LONGS(block->nb_chunks) * 665 sizeof(unsigned long) * 8, block->nb_chunks); 666 667 g_free(old); 668 669 local->nb_blocks--; 670 671 if (local->nb_blocks) { 672 for (x = 0; x < local->nb_blocks; x++) { 673 g_hash_table_insert(rdma->blockmap, (void *)local->block[x].offset, 674 &local->block[x]); 675 } 676 } 677 678 return 0; 679 } 680 681 /* 682 * Put in the log file which RDMA device was opened and the details 683 * associated with that device. 684 */ 685 static void qemu_rdma_dump_id(const char *who, struct ibv_context *verbs) 686 { 687 struct ibv_port_attr port; 688 689 if (ibv_query_port(verbs, 1, &port)) { 690 error_report("Failed to query port information"); 691 return; 692 } 693 694 printf("%s RDMA Device opened: kernel name %s " 695 "uverbs device name %s, " 696 "infiniband_verbs class device path %s, " 697 "infiniband class device path %s, " 698 "transport: (%d) %s\n", 699 who, 700 verbs->device->name, 701 verbs->device->dev_name, 702 verbs->device->dev_path, 703 verbs->device->ibdev_path, 704 port.link_layer, 705 (port.link_layer == IBV_LINK_LAYER_INFINIBAND) ? "Infiniband" : 706 ((port.link_layer == IBV_LINK_LAYER_ETHERNET) 707 ? "Ethernet" : "Unknown")); 708 } 709 710 /* 711 * Put in the log file the RDMA gid addressing information, 712 * useful for folks who have trouble understanding the 713 * RDMA device hierarchy in the kernel. 714 */ 715 static void qemu_rdma_dump_gid(const char *who, struct rdma_cm_id *id) 716 { 717 char sgid[33]; 718 char dgid[33]; 719 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.sgid, sgid, sizeof sgid); 720 inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.dgid, dgid, sizeof dgid); 721 trace_qemu_rdma_dump_gid(who, sgid, dgid); 722 } 723 724 /* 725 * As of now, IPv6 over RoCE / iWARP is not supported by linux. 726 * We will try the next addrinfo struct, and fail if there are 727 * no other valid addresses to bind against. 728 * 729 * If user is listening on '[::]', then we will not have a opened a device 730 * yet and have no way of verifying if the device is RoCE or not. 731 * 732 * In this case, the source VM will throw an error for ALL types of 733 * connections (both IPv4 and IPv6) if the destination machine does not have 734 * a regular infiniband network available for use. 735 * 736 * The only way to guarantee that an error is thrown for broken kernels is 737 * for the management software to choose a *specific* interface at bind time 738 * and validate what time of hardware it is. 739 * 740 * Unfortunately, this puts the user in a fix: 741 * 742 * If the source VM connects with an IPv4 address without knowing that the 743 * destination has bound to '[::]' the migration will unconditionally fail 744 * unless the management software is explicitly listening on the the IPv4 745 * address while using a RoCE-based device. 746 * 747 * If the source VM connects with an IPv6 address, then we're OK because we can 748 * throw an error on the source (and similarly on the destination). 749 * 750 * But in mixed environments, this will be broken for a while until it is fixed 751 * inside linux. 752 * 753 * We do provide a *tiny* bit of help in this function: We can list all of the 754 * devices in the system and check to see if all the devices are RoCE or 755 * Infiniband. 756 * 757 * If we detect that we have a *pure* RoCE environment, then we can safely 758 * thrown an error even if the management software has specified '[::]' as the 759 * bind address. 760 * 761 * However, if there is are multiple hetergeneous devices, then we cannot make 762 * this assumption and the user just has to be sure they know what they are 763 * doing. 764 * 765 * Patches are being reviewed on linux-rdma. 766 */ 767 static int qemu_rdma_broken_ipv6_kernel(Error **errp, struct ibv_context *verbs) 768 { 769 struct ibv_port_attr port_attr; 770 771 /* This bug only exists in linux, to our knowledge. */ 772 #ifdef CONFIG_LINUX 773 774 /* 775 * Verbs are only NULL if management has bound to '[::]'. 776 * 777 * Let's iterate through all the devices and see if there any pure IB 778 * devices (non-ethernet). 779 * 780 * If not, then we can safely proceed with the migration. 781 * Otherwise, there are no guarantees until the bug is fixed in linux. 782 */ 783 if (!verbs) { 784 int num_devices, x; 785 struct ibv_device ** dev_list = ibv_get_device_list(&num_devices); 786 bool roce_found = false; 787 bool ib_found = false; 788 789 for (x = 0; x < num_devices; x++) { 790 verbs = ibv_open_device(dev_list[x]); 791 792 if (ibv_query_port(verbs, 1, &port_attr)) { 793 ibv_close_device(verbs); 794 ERROR(errp, "Could not query initial IB port"); 795 return -EINVAL; 796 } 797 798 if (port_attr.link_layer == IBV_LINK_LAYER_INFINIBAND) { 799 ib_found = true; 800 } else if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) { 801 roce_found = true; 802 } 803 804 ibv_close_device(verbs); 805 806 } 807 808 if (roce_found) { 809 if (ib_found) { 810 fprintf(stderr, "WARN: migrations may fail:" 811 " IPv6 over RoCE / iWARP in linux" 812 " is broken. But since you appear to have a" 813 " mixed RoCE / IB environment, be sure to only" 814 " migrate over the IB fabric until the kernel " 815 " fixes the bug.\n"); 816 } else { 817 ERROR(errp, "You only have RoCE / iWARP devices in your systems" 818 " and your management software has specified '[::]'" 819 ", but IPv6 over RoCE / iWARP is not supported in Linux."); 820 return -ENONET; 821 } 822 } 823 824 return 0; 825 } 826 827 /* 828 * If we have a verbs context, that means that some other than '[::]' was 829 * used by the management software for binding. In which case we can actually 830 * warn the user about a potential broken kernel; 831 */ 832 833 /* IB ports start with 1, not 0 */ 834 if (ibv_query_port(verbs, 1, &port_attr)) { 835 ERROR(errp, "Could not query initial IB port"); 836 return -EINVAL; 837 } 838 839 if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) { 840 ERROR(errp, "Linux kernel's RoCE / iWARP does not support IPv6 " 841 "(but patches on linux-rdma in progress)"); 842 return -ENONET; 843 } 844 845 #endif 846 847 return 0; 848 } 849 850 /* 851 * Figure out which RDMA device corresponds to the requested IP hostname 852 * Also create the initial connection manager identifiers for opening 853 * the connection. 854 */ 855 static int qemu_rdma_resolve_host(RDMAContext *rdma, Error **errp) 856 { 857 int ret; 858 struct rdma_addrinfo *res; 859 char port_str[16]; 860 struct rdma_cm_event *cm_event; 861 char ip[40] = "unknown"; 862 struct rdma_addrinfo *e; 863 864 if (rdma->host == NULL || !strcmp(rdma->host, "")) { 865 ERROR(errp, "RDMA hostname has not been set"); 866 return -EINVAL; 867 } 868 869 /* create CM channel */ 870 rdma->channel = rdma_create_event_channel(); 871 if (!rdma->channel) { 872 ERROR(errp, "could not create CM channel"); 873 return -EINVAL; 874 } 875 876 /* create CM id */ 877 ret = rdma_create_id(rdma->channel, &rdma->cm_id, NULL, RDMA_PS_TCP); 878 if (ret) { 879 ERROR(errp, "could not create channel id"); 880 goto err_resolve_create_id; 881 } 882 883 snprintf(port_str, 16, "%d", rdma->port); 884 port_str[15] = '\0'; 885 886 ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res); 887 if (ret < 0) { 888 ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host); 889 goto err_resolve_get_addr; 890 } 891 892 for (e = res; e != NULL; e = e->ai_next) { 893 inet_ntop(e->ai_family, 894 &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip); 895 trace_qemu_rdma_resolve_host_trying(rdma->host, ip); 896 897 ret = rdma_resolve_addr(rdma->cm_id, NULL, e->ai_dst_addr, 898 RDMA_RESOLVE_TIMEOUT_MS); 899 if (!ret) { 900 if (e->ai_family == AF_INET6) { 901 ret = qemu_rdma_broken_ipv6_kernel(errp, rdma->cm_id->verbs); 902 if (ret) { 903 continue; 904 } 905 } 906 goto route; 907 } 908 } 909 910 ERROR(errp, "could not resolve address %s", rdma->host); 911 goto err_resolve_get_addr; 912 913 route: 914 qemu_rdma_dump_gid("source_resolve_addr", rdma->cm_id); 915 916 ret = rdma_get_cm_event(rdma->channel, &cm_event); 917 if (ret) { 918 ERROR(errp, "could not perform event_addr_resolved"); 919 goto err_resolve_get_addr; 920 } 921 922 if (cm_event->event != RDMA_CM_EVENT_ADDR_RESOLVED) { 923 ERROR(errp, "result not equal to event_addr_resolved %s", 924 rdma_event_str(cm_event->event)); 925 perror("rdma_resolve_addr"); 926 rdma_ack_cm_event(cm_event); 927 ret = -EINVAL; 928 goto err_resolve_get_addr; 929 } 930 rdma_ack_cm_event(cm_event); 931 932 /* resolve route */ 933 ret = rdma_resolve_route(rdma->cm_id, RDMA_RESOLVE_TIMEOUT_MS); 934 if (ret) { 935 ERROR(errp, "could not resolve rdma route"); 936 goto err_resolve_get_addr; 937 } 938 939 ret = rdma_get_cm_event(rdma->channel, &cm_event); 940 if (ret) { 941 ERROR(errp, "could not perform event_route_resolved"); 942 goto err_resolve_get_addr; 943 } 944 if (cm_event->event != RDMA_CM_EVENT_ROUTE_RESOLVED) { 945 ERROR(errp, "result not equal to event_route_resolved: %s", 946 rdma_event_str(cm_event->event)); 947 rdma_ack_cm_event(cm_event); 948 ret = -EINVAL; 949 goto err_resolve_get_addr; 950 } 951 rdma_ack_cm_event(cm_event); 952 rdma->verbs = rdma->cm_id->verbs; 953 qemu_rdma_dump_id("source_resolve_host", rdma->cm_id->verbs); 954 qemu_rdma_dump_gid("source_resolve_host", rdma->cm_id); 955 return 0; 956 957 err_resolve_get_addr: 958 rdma_destroy_id(rdma->cm_id); 959 rdma->cm_id = NULL; 960 err_resolve_create_id: 961 rdma_destroy_event_channel(rdma->channel); 962 rdma->channel = NULL; 963 return ret; 964 } 965 966 /* 967 * Create protection domain and completion queues 968 */ 969 static int qemu_rdma_alloc_pd_cq(RDMAContext *rdma) 970 { 971 /* allocate pd */ 972 rdma->pd = ibv_alloc_pd(rdma->verbs); 973 if (!rdma->pd) { 974 error_report("failed to allocate protection domain"); 975 return -1; 976 } 977 978 /* create completion channel */ 979 rdma->comp_channel = ibv_create_comp_channel(rdma->verbs); 980 if (!rdma->comp_channel) { 981 error_report("failed to allocate completion channel"); 982 goto err_alloc_pd_cq; 983 } 984 985 /* 986 * Completion queue can be filled by both read and write work requests, 987 * so must reflect the sum of both possible queue sizes. 988 */ 989 rdma->cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3), 990 NULL, rdma->comp_channel, 0); 991 if (!rdma->cq) { 992 error_report("failed to allocate completion queue"); 993 goto err_alloc_pd_cq; 994 } 995 996 return 0; 997 998 err_alloc_pd_cq: 999 if (rdma->pd) { 1000 ibv_dealloc_pd(rdma->pd); 1001 } 1002 if (rdma->comp_channel) { 1003 ibv_destroy_comp_channel(rdma->comp_channel); 1004 } 1005 rdma->pd = NULL; 1006 rdma->comp_channel = NULL; 1007 return -1; 1008 1009 } 1010 1011 /* 1012 * Create queue pairs. 1013 */ 1014 static int qemu_rdma_alloc_qp(RDMAContext *rdma) 1015 { 1016 struct ibv_qp_init_attr attr = { 0 }; 1017 int ret; 1018 1019 attr.cap.max_send_wr = RDMA_SIGNALED_SEND_MAX; 1020 attr.cap.max_recv_wr = 3; 1021 attr.cap.max_send_sge = 1; 1022 attr.cap.max_recv_sge = 1; 1023 attr.send_cq = rdma->cq; 1024 attr.recv_cq = rdma->cq; 1025 attr.qp_type = IBV_QPT_RC; 1026 1027 ret = rdma_create_qp(rdma->cm_id, rdma->pd, &attr); 1028 if (ret) { 1029 return -1; 1030 } 1031 1032 rdma->qp = rdma->cm_id->qp; 1033 return 0; 1034 } 1035 1036 static int qemu_rdma_reg_whole_ram_blocks(RDMAContext *rdma) 1037 { 1038 int i; 1039 RDMALocalBlocks *local = &rdma->local_ram_blocks; 1040 1041 for (i = 0; i < local->nb_blocks; i++) { 1042 local->block[i].mr = 1043 ibv_reg_mr(rdma->pd, 1044 local->block[i].local_host_addr, 1045 local->block[i].length, 1046 IBV_ACCESS_LOCAL_WRITE | 1047 IBV_ACCESS_REMOTE_WRITE 1048 ); 1049 if (!local->block[i].mr) { 1050 perror("Failed to register local dest ram block!\n"); 1051 break; 1052 } 1053 rdma->total_registrations++; 1054 } 1055 1056 if (i >= local->nb_blocks) { 1057 return 0; 1058 } 1059 1060 for (i--; i >= 0; i--) { 1061 ibv_dereg_mr(local->block[i].mr); 1062 rdma->total_registrations--; 1063 } 1064 1065 return -1; 1066 1067 } 1068 1069 /* 1070 * Find the ram block that corresponds to the page requested to be 1071 * transmitted by QEMU. 1072 * 1073 * Once the block is found, also identify which 'chunk' within that 1074 * block that the page belongs to. 1075 * 1076 * This search cannot fail or the migration will fail. 1077 */ 1078 static int qemu_rdma_search_ram_block(RDMAContext *rdma, 1079 uint64_t block_offset, 1080 uint64_t offset, 1081 uint64_t length, 1082 uint64_t *block_index, 1083 uint64_t *chunk_index) 1084 { 1085 uint64_t current_addr = block_offset + offset; 1086 RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap, 1087 (void *) block_offset); 1088 assert(block); 1089 assert(current_addr >= block->offset); 1090 assert((current_addr + length) <= (block->offset + block->length)); 1091 1092 *block_index = block->index; 1093 *chunk_index = ram_chunk_index(block->local_host_addr, 1094 block->local_host_addr + (current_addr - block->offset)); 1095 1096 return 0; 1097 } 1098 1099 /* 1100 * Register a chunk with IB. If the chunk was already registered 1101 * previously, then skip. 1102 * 1103 * Also return the keys associated with the registration needed 1104 * to perform the actual RDMA operation. 1105 */ 1106 static int qemu_rdma_register_and_get_keys(RDMAContext *rdma, 1107 RDMALocalBlock *block, uint8_t *host_addr, 1108 uint32_t *lkey, uint32_t *rkey, int chunk, 1109 uint8_t *chunk_start, uint8_t *chunk_end) 1110 { 1111 if (block->mr) { 1112 if (lkey) { 1113 *lkey = block->mr->lkey; 1114 } 1115 if (rkey) { 1116 *rkey = block->mr->rkey; 1117 } 1118 return 0; 1119 } 1120 1121 /* allocate memory to store chunk MRs */ 1122 if (!block->pmr) { 1123 block->pmr = g_malloc0(block->nb_chunks * sizeof(struct ibv_mr *)); 1124 if (!block->pmr) { 1125 return -1; 1126 } 1127 } 1128 1129 /* 1130 * If 'rkey', then we're the destination, so grant access to the source. 1131 * 1132 * If 'lkey', then we're the source VM, so grant access only to ourselves. 1133 */ 1134 if (!block->pmr[chunk]) { 1135 uint64_t len = chunk_end - chunk_start; 1136 1137 trace_qemu_rdma_register_and_get_keys(len, chunk_start); 1138 1139 block->pmr[chunk] = ibv_reg_mr(rdma->pd, 1140 chunk_start, len, 1141 (rkey ? (IBV_ACCESS_LOCAL_WRITE | 1142 IBV_ACCESS_REMOTE_WRITE) : 0)); 1143 1144 if (!block->pmr[chunk]) { 1145 perror("Failed to register chunk!"); 1146 fprintf(stderr, "Chunk details: block: %d chunk index %d" 1147 " start %" PRIu64 " end %" PRIu64 " host %" PRIu64 1148 " local %" PRIu64 " registrations: %d\n", 1149 block->index, chunk, (uint64_t) chunk_start, 1150 (uint64_t) chunk_end, (uint64_t) host_addr, 1151 (uint64_t) block->local_host_addr, 1152 rdma->total_registrations); 1153 return -1; 1154 } 1155 rdma->total_registrations++; 1156 } 1157 1158 if (lkey) { 1159 *lkey = block->pmr[chunk]->lkey; 1160 } 1161 if (rkey) { 1162 *rkey = block->pmr[chunk]->rkey; 1163 } 1164 return 0; 1165 } 1166 1167 /* 1168 * Register (at connection time) the memory used for control 1169 * channel messages. 1170 */ 1171 static int qemu_rdma_reg_control(RDMAContext *rdma, int idx) 1172 { 1173 rdma->wr_data[idx].control_mr = ibv_reg_mr(rdma->pd, 1174 rdma->wr_data[idx].control, RDMA_CONTROL_MAX_BUFFER, 1175 IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE); 1176 if (rdma->wr_data[idx].control_mr) { 1177 rdma->total_registrations++; 1178 return 0; 1179 } 1180 error_report("qemu_rdma_reg_control failed"); 1181 return -1; 1182 } 1183 1184 const char *print_wrid(int wrid) 1185 { 1186 if (wrid >= RDMA_WRID_RECV_CONTROL) { 1187 return wrid_desc[RDMA_WRID_RECV_CONTROL]; 1188 } 1189 return wrid_desc[wrid]; 1190 } 1191 1192 /* 1193 * RDMA requires memory registration (mlock/pinning), but this is not good for 1194 * overcommitment. 1195 * 1196 * In preparation for the future where LRU information or workload-specific 1197 * writable writable working set memory access behavior is available to QEMU 1198 * it would be nice to have in place the ability to UN-register/UN-pin 1199 * particular memory regions from the RDMA hardware when it is determine that 1200 * those regions of memory will likely not be accessed again in the near future. 1201 * 1202 * While we do not yet have such information right now, the following 1203 * compile-time option allows us to perform a non-optimized version of this 1204 * behavior. 1205 * 1206 * By uncommenting this option, you will cause *all* RDMA transfers to be 1207 * unregistered immediately after the transfer completes on both sides of the 1208 * connection. This has no effect in 'rdma-pin-all' mode, only regular mode. 1209 * 1210 * This will have a terrible impact on migration performance, so until future 1211 * workload information or LRU information is available, do not attempt to use 1212 * this feature except for basic testing. 1213 */ 1214 //#define RDMA_UNREGISTRATION_EXAMPLE 1215 1216 /* 1217 * Perform a non-optimized memory unregistration after every transfer 1218 * for demonsration purposes, only if pin-all is not requested. 1219 * 1220 * Potential optimizations: 1221 * 1. Start a new thread to run this function continuously 1222 - for bit clearing 1223 - and for receipt of unregister messages 1224 * 2. Use an LRU. 1225 * 3. Use workload hints. 1226 */ 1227 static int qemu_rdma_unregister_waiting(RDMAContext *rdma) 1228 { 1229 while (rdma->unregistrations[rdma->unregister_current]) { 1230 int ret; 1231 uint64_t wr_id = rdma->unregistrations[rdma->unregister_current]; 1232 uint64_t chunk = 1233 (wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT; 1234 uint64_t index = 1235 (wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT; 1236 RDMALocalBlock *block = 1237 &(rdma->local_ram_blocks.block[index]); 1238 RDMARegister reg = { .current_index = index }; 1239 RDMAControlHeader resp = { .type = RDMA_CONTROL_UNREGISTER_FINISHED, 1240 }; 1241 RDMAControlHeader head = { .len = sizeof(RDMARegister), 1242 .type = RDMA_CONTROL_UNREGISTER_REQUEST, 1243 .repeat = 1, 1244 }; 1245 1246 trace_qemu_rdma_unregister_waiting_proc(chunk, 1247 rdma->unregister_current); 1248 1249 rdma->unregistrations[rdma->unregister_current] = 0; 1250 rdma->unregister_current++; 1251 1252 if (rdma->unregister_current == RDMA_SIGNALED_SEND_MAX) { 1253 rdma->unregister_current = 0; 1254 } 1255 1256 1257 /* 1258 * Unregistration is speculative (because migration is single-threaded 1259 * and we cannot break the protocol's inifinband message ordering). 1260 * Thus, if the memory is currently being used for transmission, 1261 * then abort the attempt to unregister and try again 1262 * later the next time a completion is received for this memory. 1263 */ 1264 clear_bit(chunk, block->unregister_bitmap); 1265 1266 if (test_bit(chunk, block->transit_bitmap)) { 1267 trace_qemu_rdma_unregister_waiting_inflight(chunk); 1268 continue; 1269 } 1270 1271 trace_qemu_rdma_unregister_waiting_send(chunk); 1272 1273 ret = ibv_dereg_mr(block->pmr[chunk]); 1274 block->pmr[chunk] = NULL; 1275 block->remote_keys[chunk] = 0; 1276 1277 if (ret != 0) { 1278 perror("unregistration chunk failed"); 1279 return -ret; 1280 } 1281 rdma->total_registrations--; 1282 1283 reg.key.chunk = chunk; 1284 register_to_network(®); 1285 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) ®, 1286 &resp, NULL, NULL); 1287 if (ret < 0) { 1288 return ret; 1289 } 1290 1291 trace_qemu_rdma_unregister_waiting_complete(chunk); 1292 } 1293 1294 return 0; 1295 } 1296 1297 static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index, 1298 uint64_t chunk) 1299 { 1300 uint64_t result = wr_id & RDMA_WRID_TYPE_MASK; 1301 1302 result |= (index << RDMA_WRID_BLOCK_SHIFT); 1303 result |= (chunk << RDMA_WRID_CHUNK_SHIFT); 1304 1305 return result; 1306 } 1307 1308 /* 1309 * Set bit for unregistration in the next iteration. 1310 * We cannot transmit right here, but will unpin later. 1311 */ 1312 static void qemu_rdma_signal_unregister(RDMAContext *rdma, uint64_t index, 1313 uint64_t chunk, uint64_t wr_id) 1314 { 1315 if (rdma->unregistrations[rdma->unregister_next] != 0) { 1316 error_report("rdma migration: queue is full"); 1317 } else { 1318 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]); 1319 1320 if (!test_and_set_bit(chunk, block->unregister_bitmap)) { 1321 trace_qemu_rdma_signal_unregister_append(chunk, 1322 rdma->unregister_next); 1323 1324 rdma->unregistrations[rdma->unregister_next++] = 1325 qemu_rdma_make_wrid(wr_id, index, chunk); 1326 1327 if (rdma->unregister_next == RDMA_SIGNALED_SEND_MAX) { 1328 rdma->unregister_next = 0; 1329 } 1330 } else { 1331 trace_qemu_rdma_signal_unregister_already(chunk); 1332 } 1333 } 1334 } 1335 1336 /* 1337 * Consult the connection manager to see a work request 1338 * (of any kind) has completed. 1339 * Return the work request ID that completed. 1340 */ 1341 static uint64_t qemu_rdma_poll(RDMAContext *rdma, uint64_t *wr_id_out, 1342 uint32_t *byte_len) 1343 { 1344 int ret; 1345 struct ibv_wc wc; 1346 uint64_t wr_id; 1347 1348 ret = ibv_poll_cq(rdma->cq, 1, &wc); 1349 1350 if (!ret) { 1351 *wr_id_out = RDMA_WRID_NONE; 1352 return 0; 1353 } 1354 1355 if (ret < 0) { 1356 error_report("ibv_poll_cq return %d", ret); 1357 return ret; 1358 } 1359 1360 wr_id = wc.wr_id & RDMA_WRID_TYPE_MASK; 1361 1362 if (wc.status != IBV_WC_SUCCESS) { 1363 fprintf(stderr, "ibv_poll_cq wc.status=%d %s!\n", 1364 wc.status, ibv_wc_status_str(wc.status)); 1365 fprintf(stderr, "ibv_poll_cq wrid=%s!\n", wrid_desc[wr_id]); 1366 1367 return -1; 1368 } 1369 1370 if (rdma->control_ready_expected && 1371 (wr_id >= RDMA_WRID_RECV_CONTROL)) { 1372 trace_qemu_rdma_poll_recv(wrid_desc[RDMA_WRID_RECV_CONTROL], 1373 wr_id - RDMA_WRID_RECV_CONTROL, wr_id, rdma->nb_sent); 1374 rdma->control_ready_expected = 0; 1375 } 1376 1377 if (wr_id == RDMA_WRID_RDMA_WRITE) { 1378 uint64_t chunk = 1379 (wc.wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT; 1380 uint64_t index = 1381 (wc.wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT; 1382 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]); 1383 1384 trace_qemu_rdma_poll_write(print_wrid(wr_id), wr_id, rdma->nb_sent, 1385 index, chunk, 1386 block->local_host_addr, (void *)block->remote_host_addr); 1387 1388 clear_bit(chunk, block->transit_bitmap); 1389 1390 if (rdma->nb_sent > 0) { 1391 rdma->nb_sent--; 1392 } 1393 1394 if (!rdma->pin_all) { 1395 /* 1396 * FYI: If one wanted to signal a specific chunk to be unregistered 1397 * using LRU or workload-specific information, this is the function 1398 * you would call to do so. That chunk would then get asynchronously 1399 * unregistered later. 1400 */ 1401 #ifdef RDMA_UNREGISTRATION_EXAMPLE 1402 qemu_rdma_signal_unregister(rdma, index, chunk, wc.wr_id); 1403 #endif 1404 } 1405 } else { 1406 trace_qemu_rdma_poll_other(print_wrid(wr_id), wr_id, rdma->nb_sent); 1407 } 1408 1409 *wr_id_out = wc.wr_id; 1410 if (byte_len) { 1411 *byte_len = wc.byte_len; 1412 } 1413 1414 return 0; 1415 } 1416 1417 /* 1418 * Block until the next work request has completed. 1419 * 1420 * First poll to see if a work request has already completed, 1421 * otherwise block. 1422 * 1423 * If we encounter completed work requests for IDs other than 1424 * the one we're interested in, then that's generally an error. 1425 * 1426 * The only exception is actual RDMA Write completions. These 1427 * completions only need to be recorded, but do not actually 1428 * need further processing. 1429 */ 1430 static int qemu_rdma_block_for_wrid(RDMAContext *rdma, int wrid_requested, 1431 uint32_t *byte_len) 1432 { 1433 int num_cq_events = 0, ret = 0; 1434 struct ibv_cq *cq; 1435 void *cq_ctx; 1436 uint64_t wr_id = RDMA_WRID_NONE, wr_id_in; 1437 1438 if (ibv_req_notify_cq(rdma->cq, 0)) { 1439 return -1; 1440 } 1441 /* poll cq first */ 1442 while (wr_id != wrid_requested) { 1443 ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len); 1444 if (ret < 0) { 1445 return ret; 1446 } 1447 1448 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK; 1449 1450 if (wr_id == RDMA_WRID_NONE) { 1451 break; 1452 } 1453 if (wr_id != wrid_requested) { 1454 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested), 1455 wrid_requested, print_wrid(wr_id), wr_id); 1456 } 1457 } 1458 1459 if (wr_id == wrid_requested) { 1460 return 0; 1461 } 1462 1463 while (1) { 1464 /* 1465 * Coroutine doesn't start until process_incoming_migration() 1466 * so don't yield unless we know we're running inside of a coroutine. 1467 */ 1468 if (rdma->migration_started_on_destination) { 1469 yield_until_fd_readable(rdma->comp_channel->fd); 1470 } 1471 1472 if (ibv_get_cq_event(rdma->comp_channel, &cq, &cq_ctx)) { 1473 perror("ibv_get_cq_event"); 1474 goto err_block_for_wrid; 1475 } 1476 1477 num_cq_events++; 1478 1479 if (ibv_req_notify_cq(cq, 0)) { 1480 goto err_block_for_wrid; 1481 } 1482 1483 while (wr_id != wrid_requested) { 1484 ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len); 1485 if (ret < 0) { 1486 goto err_block_for_wrid; 1487 } 1488 1489 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK; 1490 1491 if (wr_id == RDMA_WRID_NONE) { 1492 break; 1493 } 1494 if (wr_id != wrid_requested) { 1495 trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested), 1496 wrid_requested, print_wrid(wr_id), wr_id); 1497 } 1498 } 1499 1500 if (wr_id == wrid_requested) { 1501 goto success_block_for_wrid; 1502 } 1503 } 1504 1505 success_block_for_wrid: 1506 if (num_cq_events) { 1507 ibv_ack_cq_events(cq, num_cq_events); 1508 } 1509 return 0; 1510 1511 err_block_for_wrid: 1512 if (num_cq_events) { 1513 ibv_ack_cq_events(cq, num_cq_events); 1514 } 1515 return ret; 1516 } 1517 1518 /* 1519 * Post a SEND message work request for the control channel 1520 * containing some data and block until the post completes. 1521 */ 1522 static int qemu_rdma_post_send_control(RDMAContext *rdma, uint8_t *buf, 1523 RDMAControlHeader *head) 1524 { 1525 int ret = 0; 1526 RDMAWorkRequestData *wr = &rdma->wr_data[RDMA_WRID_CONTROL]; 1527 struct ibv_send_wr *bad_wr; 1528 struct ibv_sge sge = { 1529 .addr = (uint64_t)(wr->control), 1530 .length = head->len + sizeof(RDMAControlHeader), 1531 .lkey = wr->control_mr->lkey, 1532 }; 1533 struct ibv_send_wr send_wr = { 1534 .wr_id = RDMA_WRID_SEND_CONTROL, 1535 .opcode = IBV_WR_SEND, 1536 .send_flags = IBV_SEND_SIGNALED, 1537 .sg_list = &sge, 1538 .num_sge = 1, 1539 }; 1540 1541 trace_qemu_rdma_post_send_control(control_desc[head->type]); 1542 1543 /* 1544 * We don't actually need to do a memcpy() in here if we used 1545 * the "sge" properly, but since we're only sending control messages 1546 * (not RAM in a performance-critical path), then its OK for now. 1547 * 1548 * The copy makes the RDMAControlHeader simpler to manipulate 1549 * for the time being. 1550 */ 1551 assert(head->len <= RDMA_CONTROL_MAX_BUFFER - sizeof(*head)); 1552 memcpy(wr->control, head, sizeof(RDMAControlHeader)); 1553 control_to_network((void *) wr->control); 1554 1555 if (buf) { 1556 memcpy(wr->control + sizeof(RDMAControlHeader), buf, head->len); 1557 } 1558 1559 1560 ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr); 1561 1562 if (ret > 0) { 1563 error_report("Failed to use post IB SEND for control"); 1564 return -ret; 1565 } 1566 1567 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_SEND_CONTROL, NULL); 1568 if (ret < 0) { 1569 error_report("rdma migration: send polling control error"); 1570 } 1571 1572 return ret; 1573 } 1574 1575 /* 1576 * Post a RECV work request in anticipation of some future receipt 1577 * of data on the control channel. 1578 */ 1579 static int qemu_rdma_post_recv_control(RDMAContext *rdma, int idx) 1580 { 1581 struct ibv_recv_wr *bad_wr; 1582 struct ibv_sge sge = { 1583 .addr = (uint64_t)(rdma->wr_data[idx].control), 1584 .length = RDMA_CONTROL_MAX_BUFFER, 1585 .lkey = rdma->wr_data[idx].control_mr->lkey, 1586 }; 1587 1588 struct ibv_recv_wr recv_wr = { 1589 .wr_id = RDMA_WRID_RECV_CONTROL + idx, 1590 .sg_list = &sge, 1591 .num_sge = 1, 1592 }; 1593 1594 1595 if (ibv_post_recv(rdma->qp, &recv_wr, &bad_wr)) { 1596 return -1; 1597 } 1598 1599 return 0; 1600 } 1601 1602 /* 1603 * Block and wait for a RECV control channel message to arrive. 1604 */ 1605 static int qemu_rdma_exchange_get_response(RDMAContext *rdma, 1606 RDMAControlHeader *head, int expecting, int idx) 1607 { 1608 uint32_t byte_len; 1609 int ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RECV_CONTROL + idx, 1610 &byte_len); 1611 1612 if (ret < 0) { 1613 error_report("rdma migration: recv polling control error!"); 1614 return ret; 1615 } 1616 1617 network_to_control((void *) rdma->wr_data[idx].control); 1618 memcpy(head, rdma->wr_data[idx].control, sizeof(RDMAControlHeader)); 1619 1620 trace_qemu_rdma_exchange_get_response_start(control_desc[expecting]); 1621 1622 if (expecting == RDMA_CONTROL_NONE) { 1623 trace_qemu_rdma_exchange_get_response_none(control_desc[head->type], 1624 head->type); 1625 } else if (head->type != expecting || head->type == RDMA_CONTROL_ERROR) { 1626 error_report("Was expecting a %s (%d) control message" 1627 ", but got: %s (%d), length: %d", 1628 control_desc[expecting], expecting, 1629 control_desc[head->type], head->type, head->len); 1630 return -EIO; 1631 } 1632 if (head->len > RDMA_CONTROL_MAX_BUFFER - sizeof(*head)) { 1633 error_report("too long length: %d\n", head->len); 1634 return -EINVAL; 1635 } 1636 if (sizeof(*head) + head->len != byte_len) { 1637 error_report("Malformed length: %d byte_len %d", head->len, byte_len); 1638 return -EINVAL; 1639 } 1640 1641 return 0; 1642 } 1643 1644 /* 1645 * When a RECV work request has completed, the work request's 1646 * buffer is pointed at the header. 1647 * 1648 * This will advance the pointer to the data portion 1649 * of the control message of the work request's buffer that 1650 * was populated after the work request finished. 1651 */ 1652 static void qemu_rdma_move_header(RDMAContext *rdma, int idx, 1653 RDMAControlHeader *head) 1654 { 1655 rdma->wr_data[idx].control_len = head->len; 1656 rdma->wr_data[idx].control_curr = 1657 rdma->wr_data[idx].control + sizeof(RDMAControlHeader); 1658 } 1659 1660 /* 1661 * This is an 'atomic' high-level operation to deliver a single, unified 1662 * control-channel message. 1663 * 1664 * Additionally, if the user is expecting some kind of reply to this message, 1665 * they can request a 'resp' response message be filled in by posting an 1666 * additional work request on behalf of the user and waiting for an additional 1667 * completion. 1668 * 1669 * The extra (optional) response is used during registration to us from having 1670 * to perform an *additional* exchange of message just to provide a response by 1671 * instead piggy-backing on the acknowledgement. 1672 */ 1673 static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head, 1674 uint8_t *data, RDMAControlHeader *resp, 1675 int *resp_idx, 1676 int (*callback)(RDMAContext *rdma)) 1677 { 1678 int ret = 0; 1679 1680 /* 1681 * Wait until the dest is ready before attempting to deliver the message 1682 * by waiting for a READY message. 1683 */ 1684 if (rdma->control_ready_expected) { 1685 RDMAControlHeader resp; 1686 ret = qemu_rdma_exchange_get_response(rdma, 1687 &resp, RDMA_CONTROL_READY, RDMA_WRID_READY); 1688 if (ret < 0) { 1689 return ret; 1690 } 1691 } 1692 1693 /* 1694 * If the user is expecting a response, post a WR in anticipation of it. 1695 */ 1696 if (resp) { 1697 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_DATA); 1698 if (ret) { 1699 error_report("rdma migration: error posting" 1700 " extra control recv for anticipated result!"); 1701 return ret; 1702 } 1703 } 1704 1705 /* 1706 * Post a WR to replace the one we just consumed for the READY message. 1707 */ 1708 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY); 1709 if (ret) { 1710 error_report("rdma migration: error posting first control recv!"); 1711 return ret; 1712 } 1713 1714 /* 1715 * Deliver the control message that was requested. 1716 */ 1717 ret = qemu_rdma_post_send_control(rdma, data, head); 1718 1719 if (ret < 0) { 1720 error_report("Failed to send control buffer!"); 1721 return ret; 1722 } 1723 1724 /* 1725 * If we're expecting a response, block and wait for it. 1726 */ 1727 if (resp) { 1728 if (callback) { 1729 trace_qemu_rdma_exchange_send_issue_callback(); 1730 ret = callback(rdma); 1731 if (ret < 0) { 1732 return ret; 1733 } 1734 } 1735 1736 trace_qemu_rdma_exchange_send_waiting(control_desc[resp->type]); 1737 ret = qemu_rdma_exchange_get_response(rdma, resp, 1738 resp->type, RDMA_WRID_DATA); 1739 1740 if (ret < 0) { 1741 return ret; 1742 } 1743 1744 qemu_rdma_move_header(rdma, RDMA_WRID_DATA, resp); 1745 if (resp_idx) { 1746 *resp_idx = RDMA_WRID_DATA; 1747 } 1748 trace_qemu_rdma_exchange_send_received(control_desc[resp->type]); 1749 } 1750 1751 rdma->control_ready_expected = 1; 1752 1753 return 0; 1754 } 1755 1756 /* 1757 * This is an 'atomic' high-level operation to receive a single, unified 1758 * control-channel message. 1759 */ 1760 static int qemu_rdma_exchange_recv(RDMAContext *rdma, RDMAControlHeader *head, 1761 int expecting) 1762 { 1763 RDMAControlHeader ready = { 1764 .len = 0, 1765 .type = RDMA_CONTROL_READY, 1766 .repeat = 1, 1767 }; 1768 int ret; 1769 1770 /* 1771 * Inform the source that we're ready to receive a message. 1772 */ 1773 ret = qemu_rdma_post_send_control(rdma, NULL, &ready); 1774 1775 if (ret < 0) { 1776 error_report("Failed to send control buffer!"); 1777 return ret; 1778 } 1779 1780 /* 1781 * Block and wait for the message. 1782 */ 1783 ret = qemu_rdma_exchange_get_response(rdma, head, 1784 expecting, RDMA_WRID_READY); 1785 1786 if (ret < 0) { 1787 return ret; 1788 } 1789 1790 qemu_rdma_move_header(rdma, RDMA_WRID_READY, head); 1791 1792 /* 1793 * Post a new RECV work request to replace the one we just consumed. 1794 */ 1795 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY); 1796 if (ret) { 1797 error_report("rdma migration: error posting second control recv!"); 1798 return ret; 1799 } 1800 1801 return 0; 1802 } 1803 1804 /* 1805 * Write an actual chunk of memory using RDMA. 1806 * 1807 * If we're using dynamic registration on the dest-side, we have to 1808 * send a registration command first. 1809 */ 1810 static int qemu_rdma_write_one(QEMUFile *f, RDMAContext *rdma, 1811 int current_index, uint64_t current_addr, 1812 uint64_t length) 1813 { 1814 struct ibv_sge sge; 1815 struct ibv_send_wr send_wr = { 0 }; 1816 struct ibv_send_wr *bad_wr; 1817 int reg_result_idx, ret, count = 0; 1818 uint64_t chunk, chunks; 1819 uint8_t *chunk_start, *chunk_end; 1820 RDMALocalBlock *block = &(rdma->local_ram_blocks.block[current_index]); 1821 RDMARegister reg; 1822 RDMARegisterResult *reg_result; 1823 RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT }; 1824 RDMAControlHeader head = { .len = sizeof(RDMARegister), 1825 .type = RDMA_CONTROL_REGISTER_REQUEST, 1826 .repeat = 1, 1827 }; 1828 1829 retry: 1830 sge.addr = (uint64_t)(block->local_host_addr + 1831 (current_addr - block->offset)); 1832 sge.length = length; 1833 1834 chunk = ram_chunk_index(block->local_host_addr, (uint8_t *) sge.addr); 1835 chunk_start = ram_chunk_start(block, chunk); 1836 1837 if (block->is_ram_block) { 1838 chunks = length / (1UL << RDMA_REG_CHUNK_SHIFT); 1839 1840 if (chunks && ((length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) { 1841 chunks--; 1842 } 1843 } else { 1844 chunks = block->length / (1UL << RDMA_REG_CHUNK_SHIFT); 1845 1846 if (chunks && ((block->length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) { 1847 chunks--; 1848 } 1849 } 1850 1851 trace_qemu_rdma_write_one_top(chunks + 1, 1852 (chunks + 1) * 1853 (1UL << RDMA_REG_CHUNK_SHIFT) / 1024 / 1024); 1854 1855 chunk_end = ram_chunk_end(block, chunk + chunks); 1856 1857 if (!rdma->pin_all) { 1858 #ifdef RDMA_UNREGISTRATION_EXAMPLE 1859 qemu_rdma_unregister_waiting(rdma); 1860 #endif 1861 } 1862 1863 while (test_bit(chunk, block->transit_bitmap)) { 1864 (void)count; 1865 trace_qemu_rdma_write_one_block(count++, current_index, chunk, 1866 sge.addr, length, rdma->nb_sent, block->nb_chunks); 1867 1868 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL); 1869 1870 if (ret < 0) { 1871 error_report("Failed to Wait for previous write to complete " 1872 "block %d chunk %" PRIu64 1873 " current %" PRIu64 " len %" PRIu64 " %d", 1874 current_index, chunk, sge.addr, length, rdma->nb_sent); 1875 return ret; 1876 } 1877 } 1878 1879 if (!rdma->pin_all || !block->is_ram_block) { 1880 if (!block->remote_keys[chunk]) { 1881 /* 1882 * This chunk has not yet been registered, so first check to see 1883 * if the entire chunk is zero. If so, tell the other size to 1884 * memset() + madvise() the entire chunk without RDMA. 1885 */ 1886 1887 if (can_use_buffer_find_nonzero_offset((void *)sge.addr, length) 1888 && buffer_find_nonzero_offset((void *)sge.addr, 1889 length) == length) { 1890 RDMACompress comp = { 1891 .offset = current_addr, 1892 .value = 0, 1893 .block_idx = current_index, 1894 .length = length, 1895 }; 1896 1897 head.len = sizeof(comp); 1898 head.type = RDMA_CONTROL_COMPRESS; 1899 1900 trace_qemu_rdma_write_one_zero(chunk, sge.length, 1901 current_index, current_addr); 1902 1903 compress_to_network(&comp); 1904 ret = qemu_rdma_exchange_send(rdma, &head, 1905 (uint8_t *) &comp, NULL, NULL, NULL); 1906 1907 if (ret < 0) { 1908 return -EIO; 1909 } 1910 1911 acct_update_position(f, sge.length, true); 1912 1913 return 1; 1914 } 1915 1916 /* 1917 * Otherwise, tell other side to register. 1918 */ 1919 reg.current_index = current_index; 1920 if (block->is_ram_block) { 1921 reg.key.current_addr = current_addr; 1922 } else { 1923 reg.key.chunk = chunk; 1924 } 1925 reg.chunks = chunks; 1926 1927 trace_qemu_rdma_write_one_sendreg(chunk, sge.length, current_index, 1928 current_addr); 1929 1930 register_to_network(®); 1931 ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) ®, 1932 &resp, ®_result_idx, NULL); 1933 if (ret < 0) { 1934 return ret; 1935 } 1936 1937 /* try to overlap this single registration with the one we sent. */ 1938 if (qemu_rdma_register_and_get_keys(rdma, block, 1939 (uint8_t *) sge.addr, 1940 &sge.lkey, NULL, chunk, 1941 chunk_start, chunk_end)) { 1942 error_report("cannot get lkey"); 1943 return -EINVAL; 1944 } 1945 1946 reg_result = (RDMARegisterResult *) 1947 rdma->wr_data[reg_result_idx].control_curr; 1948 1949 network_to_result(reg_result); 1950 1951 trace_qemu_rdma_write_one_recvregres(block->remote_keys[chunk], 1952 reg_result->rkey, chunk); 1953 1954 block->remote_keys[chunk] = reg_result->rkey; 1955 block->remote_host_addr = reg_result->host_addr; 1956 } else { 1957 /* already registered before */ 1958 if (qemu_rdma_register_and_get_keys(rdma, block, 1959 (uint8_t *)sge.addr, 1960 &sge.lkey, NULL, chunk, 1961 chunk_start, chunk_end)) { 1962 error_report("cannot get lkey!"); 1963 return -EINVAL; 1964 } 1965 } 1966 1967 send_wr.wr.rdma.rkey = block->remote_keys[chunk]; 1968 } else { 1969 send_wr.wr.rdma.rkey = block->remote_rkey; 1970 1971 if (qemu_rdma_register_and_get_keys(rdma, block, (uint8_t *)sge.addr, 1972 &sge.lkey, NULL, chunk, 1973 chunk_start, chunk_end)) { 1974 error_report("cannot get lkey!"); 1975 return -EINVAL; 1976 } 1977 } 1978 1979 /* 1980 * Encode the ram block index and chunk within this wrid. 1981 * We will use this information at the time of completion 1982 * to figure out which bitmap to check against and then which 1983 * chunk in the bitmap to look for. 1984 */ 1985 send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE, 1986 current_index, chunk); 1987 1988 send_wr.opcode = IBV_WR_RDMA_WRITE; 1989 send_wr.send_flags = IBV_SEND_SIGNALED; 1990 send_wr.sg_list = &sge; 1991 send_wr.num_sge = 1; 1992 send_wr.wr.rdma.remote_addr = block->remote_host_addr + 1993 (current_addr - block->offset); 1994 1995 trace_qemu_rdma_write_one_post(chunk, sge.addr, send_wr.wr.rdma.remote_addr, 1996 sge.length); 1997 1998 /* 1999 * ibv_post_send() does not return negative error numbers, 2000 * per the specification they are positive - no idea why. 2001 */ 2002 ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr); 2003 2004 if (ret == ENOMEM) { 2005 trace_qemu_rdma_write_one_queue_full(); 2006 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL); 2007 if (ret < 0) { 2008 error_report("rdma migration: failed to make " 2009 "room in full send queue! %d", ret); 2010 return ret; 2011 } 2012 2013 goto retry; 2014 2015 } else if (ret > 0) { 2016 perror("rdma migration: post rdma write failed"); 2017 return -ret; 2018 } 2019 2020 set_bit(chunk, block->transit_bitmap); 2021 acct_update_position(f, sge.length, false); 2022 rdma->total_writes++; 2023 2024 return 0; 2025 } 2026 2027 /* 2028 * Push out any unwritten RDMA operations. 2029 * 2030 * We support sending out multiple chunks at the same time. 2031 * Not all of them need to get signaled in the completion queue. 2032 */ 2033 static int qemu_rdma_write_flush(QEMUFile *f, RDMAContext *rdma) 2034 { 2035 int ret; 2036 2037 if (!rdma->current_length) { 2038 return 0; 2039 } 2040 2041 ret = qemu_rdma_write_one(f, rdma, 2042 rdma->current_index, rdma->current_addr, rdma->current_length); 2043 2044 if (ret < 0) { 2045 return ret; 2046 } 2047 2048 if (ret == 0) { 2049 rdma->nb_sent++; 2050 trace_qemu_rdma_write_flush(rdma->nb_sent); 2051 } 2052 2053 rdma->current_length = 0; 2054 rdma->current_addr = 0; 2055 2056 return 0; 2057 } 2058 2059 static inline int qemu_rdma_buffer_mergable(RDMAContext *rdma, 2060 uint64_t offset, uint64_t len) 2061 { 2062 RDMALocalBlock *block; 2063 uint8_t *host_addr; 2064 uint8_t *chunk_end; 2065 2066 if (rdma->current_index < 0) { 2067 return 0; 2068 } 2069 2070 if (rdma->current_chunk < 0) { 2071 return 0; 2072 } 2073 2074 block = &(rdma->local_ram_blocks.block[rdma->current_index]); 2075 host_addr = block->local_host_addr + (offset - block->offset); 2076 chunk_end = ram_chunk_end(block, rdma->current_chunk); 2077 2078 if (rdma->current_length == 0) { 2079 return 0; 2080 } 2081 2082 /* 2083 * Only merge into chunk sequentially. 2084 */ 2085 if (offset != (rdma->current_addr + rdma->current_length)) { 2086 return 0; 2087 } 2088 2089 if (offset < block->offset) { 2090 return 0; 2091 } 2092 2093 if ((offset + len) > (block->offset + block->length)) { 2094 return 0; 2095 } 2096 2097 if ((host_addr + len) > chunk_end) { 2098 return 0; 2099 } 2100 2101 return 1; 2102 } 2103 2104 /* 2105 * We're not actually writing here, but doing three things: 2106 * 2107 * 1. Identify the chunk the buffer belongs to. 2108 * 2. If the chunk is full or the buffer doesn't belong to the current 2109 * chunk, then start a new chunk and flush() the old chunk. 2110 * 3. To keep the hardware busy, we also group chunks into batches 2111 * and only require that a batch gets acknowledged in the completion 2112 * qeueue instead of each individual chunk. 2113 */ 2114 static int qemu_rdma_write(QEMUFile *f, RDMAContext *rdma, 2115 uint64_t block_offset, uint64_t offset, 2116 uint64_t len) 2117 { 2118 uint64_t current_addr = block_offset + offset; 2119 uint64_t index = rdma->current_index; 2120 uint64_t chunk = rdma->current_chunk; 2121 int ret; 2122 2123 /* If we cannot merge it, we flush the current buffer first. */ 2124 if (!qemu_rdma_buffer_mergable(rdma, current_addr, len)) { 2125 ret = qemu_rdma_write_flush(f, rdma); 2126 if (ret) { 2127 return ret; 2128 } 2129 rdma->current_length = 0; 2130 rdma->current_addr = current_addr; 2131 2132 ret = qemu_rdma_search_ram_block(rdma, block_offset, 2133 offset, len, &index, &chunk); 2134 if (ret) { 2135 error_report("ram block search failed"); 2136 return ret; 2137 } 2138 rdma->current_index = index; 2139 rdma->current_chunk = chunk; 2140 } 2141 2142 /* merge it */ 2143 rdma->current_length += len; 2144 2145 /* flush it if buffer is too large */ 2146 if (rdma->current_length >= RDMA_MERGE_MAX) { 2147 return qemu_rdma_write_flush(f, rdma); 2148 } 2149 2150 return 0; 2151 } 2152 2153 static void qemu_rdma_cleanup(RDMAContext *rdma) 2154 { 2155 struct rdma_cm_event *cm_event; 2156 int ret, idx; 2157 2158 if (rdma->cm_id && rdma->connected) { 2159 if (rdma->error_state) { 2160 RDMAControlHeader head = { .len = 0, 2161 .type = RDMA_CONTROL_ERROR, 2162 .repeat = 1, 2163 }; 2164 error_report("Early error. Sending error."); 2165 qemu_rdma_post_send_control(rdma, NULL, &head); 2166 } 2167 2168 ret = rdma_disconnect(rdma->cm_id); 2169 if (!ret) { 2170 trace_qemu_rdma_cleanup_waiting_for_disconnect(); 2171 ret = rdma_get_cm_event(rdma->channel, &cm_event); 2172 if (!ret) { 2173 rdma_ack_cm_event(cm_event); 2174 } 2175 } 2176 trace_qemu_rdma_cleanup_disconnect(); 2177 rdma->connected = false; 2178 } 2179 2180 g_free(rdma->block); 2181 rdma->block = NULL; 2182 2183 for (idx = 0; idx < RDMA_WRID_MAX; idx++) { 2184 if (rdma->wr_data[idx].control_mr) { 2185 rdma->total_registrations--; 2186 ibv_dereg_mr(rdma->wr_data[idx].control_mr); 2187 } 2188 rdma->wr_data[idx].control_mr = NULL; 2189 } 2190 2191 if (rdma->local_ram_blocks.block) { 2192 while (rdma->local_ram_blocks.nb_blocks) { 2193 __qemu_rdma_delete_block(rdma, 2194 rdma->local_ram_blocks.block->offset); 2195 } 2196 } 2197 2198 if (rdma->cq) { 2199 ibv_destroy_cq(rdma->cq); 2200 rdma->cq = NULL; 2201 } 2202 if (rdma->comp_channel) { 2203 ibv_destroy_comp_channel(rdma->comp_channel); 2204 rdma->comp_channel = NULL; 2205 } 2206 if (rdma->pd) { 2207 ibv_dealloc_pd(rdma->pd); 2208 rdma->pd = NULL; 2209 } 2210 if (rdma->listen_id) { 2211 rdma_destroy_id(rdma->listen_id); 2212 rdma->listen_id = NULL; 2213 } 2214 if (rdma->cm_id) { 2215 if (rdma->qp) { 2216 rdma_destroy_qp(rdma->cm_id); 2217 rdma->qp = NULL; 2218 } 2219 rdma_destroy_id(rdma->cm_id); 2220 rdma->cm_id = NULL; 2221 } 2222 if (rdma->channel) { 2223 rdma_destroy_event_channel(rdma->channel); 2224 rdma->channel = NULL; 2225 } 2226 g_free(rdma->host); 2227 rdma->host = NULL; 2228 } 2229 2230 2231 static int qemu_rdma_source_init(RDMAContext *rdma, Error **errp, bool pin_all) 2232 { 2233 int ret, idx; 2234 Error *local_err = NULL, **temp = &local_err; 2235 2236 /* 2237 * Will be validated against destination's actual capabilities 2238 * after the connect() completes. 2239 */ 2240 rdma->pin_all = pin_all; 2241 2242 ret = qemu_rdma_resolve_host(rdma, temp); 2243 if (ret) { 2244 goto err_rdma_source_init; 2245 } 2246 2247 ret = qemu_rdma_alloc_pd_cq(rdma); 2248 if (ret) { 2249 ERROR(temp, "rdma migration: error allocating pd and cq! Your mlock()" 2250 " limits may be too low. Please check $ ulimit -a # and " 2251 "search for 'ulimit -l' in the output"); 2252 goto err_rdma_source_init; 2253 } 2254 2255 ret = qemu_rdma_alloc_qp(rdma); 2256 if (ret) { 2257 ERROR(temp, "rdma migration: error allocating qp!"); 2258 goto err_rdma_source_init; 2259 } 2260 2261 ret = qemu_rdma_init_ram_blocks(rdma); 2262 if (ret) { 2263 ERROR(temp, "rdma migration: error initializing ram blocks!"); 2264 goto err_rdma_source_init; 2265 } 2266 2267 for (idx = 0; idx < RDMA_WRID_MAX; idx++) { 2268 ret = qemu_rdma_reg_control(rdma, idx); 2269 if (ret) { 2270 ERROR(temp, "rdma migration: error registering %d control!", 2271 idx); 2272 goto err_rdma_source_init; 2273 } 2274 } 2275 2276 return 0; 2277 2278 err_rdma_source_init: 2279 error_propagate(errp, local_err); 2280 qemu_rdma_cleanup(rdma); 2281 return -1; 2282 } 2283 2284 static int qemu_rdma_connect(RDMAContext *rdma, Error **errp) 2285 { 2286 RDMACapabilities cap = { 2287 .version = RDMA_CONTROL_VERSION_CURRENT, 2288 .flags = 0, 2289 }; 2290 struct rdma_conn_param conn_param = { .initiator_depth = 2, 2291 .retry_count = 5, 2292 .private_data = &cap, 2293 .private_data_len = sizeof(cap), 2294 }; 2295 struct rdma_cm_event *cm_event; 2296 int ret; 2297 2298 /* 2299 * Only negotiate the capability with destination if the user 2300 * on the source first requested the capability. 2301 */ 2302 if (rdma->pin_all) { 2303 trace_qemu_rdma_connect_pin_all_requested(); 2304 cap.flags |= RDMA_CAPABILITY_PIN_ALL; 2305 } 2306 2307 caps_to_network(&cap); 2308 2309 ret = rdma_connect(rdma->cm_id, &conn_param); 2310 if (ret) { 2311 perror("rdma_connect"); 2312 ERROR(errp, "connecting to destination!"); 2313 rdma_destroy_id(rdma->cm_id); 2314 rdma->cm_id = NULL; 2315 goto err_rdma_source_connect; 2316 } 2317 2318 ret = rdma_get_cm_event(rdma->channel, &cm_event); 2319 if (ret) { 2320 perror("rdma_get_cm_event after rdma_connect"); 2321 ERROR(errp, "connecting to destination!"); 2322 rdma_ack_cm_event(cm_event); 2323 rdma_destroy_id(rdma->cm_id); 2324 rdma->cm_id = NULL; 2325 goto err_rdma_source_connect; 2326 } 2327 2328 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) { 2329 perror("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect"); 2330 ERROR(errp, "connecting to destination!"); 2331 rdma_ack_cm_event(cm_event); 2332 rdma_destroy_id(rdma->cm_id); 2333 rdma->cm_id = NULL; 2334 goto err_rdma_source_connect; 2335 } 2336 rdma->connected = true; 2337 2338 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap)); 2339 network_to_caps(&cap); 2340 2341 /* 2342 * Verify that the *requested* capabilities are supported by the destination 2343 * and disable them otherwise. 2344 */ 2345 if (rdma->pin_all && !(cap.flags & RDMA_CAPABILITY_PIN_ALL)) { 2346 ERROR(errp, "Server cannot support pinning all memory. " 2347 "Will register memory dynamically."); 2348 rdma->pin_all = false; 2349 } 2350 2351 trace_qemu_rdma_connect_pin_all_outcome(rdma->pin_all); 2352 2353 rdma_ack_cm_event(cm_event); 2354 2355 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY); 2356 if (ret) { 2357 ERROR(errp, "posting second control recv!"); 2358 goto err_rdma_source_connect; 2359 } 2360 2361 rdma->control_ready_expected = 1; 2362 rdma->nb_sent = 0; 2363 return 0; 2364 2365 err_rdma_source_connect: 2366 qemu_rdma_cleanup(rdma); 2367 return -1; 2368 } 2369 2370 static int qemu_rdma_dest_init(RDMAContext *rdma, Error **errp) 2371 { 2372 int ret = -EINVAL, idx; 2373 struct rdma_cm_id *listen_id; 2374 char ip[40] = "unknown"; 2375 struct rdma_addrinfo *res; 2376 char port_str[16]; 2377 2378 for (idx = 0; idx < RDMA_WRID_MAX; idx++) { 2379 rdma->wr_data[idx].control_len = 0; 2380 rdma->wr_data[idx].control_curr = NULL; 2381 } 2382 2383 if (rdma->host == NULL) { 2384 ERROR(errp, "RDMA host is not set!"); 2385 rdma->error_state = -EINVAL; 2386 return -1; 2387 } 2388 /* create CM channel */ 2389 rdma->channel = rdma_create_event_channel(); 2390 if (!rdma->channel) { 2391 ERROR(errp, "could not create rdma event channel"); 2392 rdma->error_state = -EINVAL; 2393 return -1; 2394 } 2395 2396 /* create CM id */ 2397 ret = rdma_create_id(rdma->channel, &listen_id, NULL, RDMA_PS_TCP); 2398 if (ret) { 2399 ERROR(errp, "could not create cm_id!"); 2400 goto err_dest_init_create_listen_id; 2401 } 2402 2403 snprintf(port_str, 16, "%d", rdma->port); 2404 port_str[15] = '\0'; 2405 2406 if (rdma->host && strcmp("", rdma->host)) { 2407 struct rdma_addrinfo *e; 2408 2409 ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res); 2410 if (ret < 0) { 2411 ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host); 2412 goto err_dest_init_bind_addr; 2413 } 2414 2415 for (e = res; e != NULL; e = e->ai_next) { 2416 inet_ntop(e->ai_family, 2417 &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip); 2418 trace_qemu_rdma_dest_init_trying(rdma->host, ip); 2419 ret = rdma_bind_addr(listen_id, e->ai_dst_addr); 2420 if (!ret) { 2421 if (e->ai_family == AF_INET6) { 2422 ret = qemu_rdma_broken_ipv6_kernel(errp, listen_id->verbs); 2423 if (ret) { 2424 continue; 2425 } 2426 } 2427 2428 goto listen; 2429 } 2430 } 2431 2432 ERROR(errp, "Error: could not rdma_bind_addr!"); 2433 goto err_dest_init_bind_addr; 2434 } else { 2435 ERROR(errp, "migration host and port not specified!"); 2436 ret = -EINVAL; 2437 goto err_dest_init_bind_addr; 2438 } 2439 listen: 2440 2441 rdma->listen_id = listen_id; 2442 qemu_rdma_dump_gid("dest_init", listen_id); 2443 return 0; 2444 2445 err_dest_init_bind_addr: 2446 rdma_destroy_id(listen_id); 2447 err_dest_init_create_listen_id: 2448 rdma_destroy_event_channel(rdma->channel); 2449 rdma->channel = NULL; 2450 rdma->error_state = ret; 2451 return ret; 2452 2453 } 2454 2455 static void *qemu_rdma_data_init(const char *host_port, Error **errp) 2456 { 2457 RDMAContext *rdma = NULL; 2458 InetSocketAddress *addr; 2459 2460 if (host_port) { 2461 rdma = g_malloc0(sizeof(RDMAContext)); 2462 memset(rdma, 0, sizeof(RDMAContext)); 2463 rdma->current_index = -1; 2464 rdma->current_chunk = -1; 2465 2466 addr = inet_parse(host_port, NULL); 2467 if (addr != NULL) { 2468 rdma->port = atoi(addr->port); 2469 rdma->host = g_strdup(addr->host); 2470 } else { 2471 ERROR(errp, "bad RDMA migration address '%s'", host_port); 2472 g_free(rdma); 2473 rdma = NULL; 2474 } 2475 2476 qapi_free_InetSocketAddress(addr); 2477 } 2478 2479 return rdma; 2480 } 2481 2482 /* 2483 * QEMUFile interface to the control channel. 2484 * SEND messages for control only. 2485 * VM's ram is handled with regular RDMA messages. 2486 */ 2487 static int qemu_rdma_put_buffer(void *opaque, const uint8_t *buf, 2488 int64_t pos, int size) 2489 { 2490 QEMUFileRDMA *r = opaque; 2491 QEMUFile *f = r->file; 2492 RDMAContext *rdma = r->rdma; 2493 size_t remaining = size; 2494 uint8_t * data = (void *) buf; 2495 int ret; 2496 2497 CHECK_ERROR_STATE(); 2498 2499 /* 2500 * Push out any writes that 2501 * we're queued up for VM's ram. 2502 */ 2503 ret = qemu_rdma_write_flush(f, rdma); 2504 if (ret < 0) { 2505 rdma->error_state = ret; 2506 return ret; 2507 } 2508 2509 while (remaining) { 2510 RDMAControlHeader head; 2511 2512 r->len = MIN(remaining, RDMA_SEND_INCREMENT); 2513 remaining -= r->len; 2514 2515 head.len = r->len; 2516 head.type = RDMA_CONTROL_QEMU_FILE; 2517 2518 ret = qemu_rdma_exchange_send(rdma, &head, data, NULL, NULL, NULL); 2519 2520 if (ret < 0) { 2521 rdma->error_state = ret; 2522 return ret; 2523 } 2524 2525 data += r->len; 2526 } 2527 2528 return size; 2529 } 2530 2531 static size_t qemu_rdma_fill(RDMAContext *rdma, uint8_t *buf, 2532 int size, int idx) 2533 { 2534 size_t len = 0; 2535 2536 if (rdma->wr_data[idx].control_len) { 2537 trace_qemu_rdma_fill(rdma->wr_data[idx].control_len, size); 2538 2539 len = MIN(size, rdma->wr_data[idx].control_len); 2540 memcpy(buf, rdma->wr_data[idx].control_curr, len); 2541 rdma->wr_data[idx].control_curr += len; 2542 rdma->wr_data[idx].control_len -= len; 2543 } 2544 2545 return len; 2546 } 2547 2548 /* 2549 * QEMUFile interface to the control channel. 2550 * RDMA links don't use bytestreams, so we have to 2551 * return bytes to QEMUFile opportunistically. 2552 */ 2553 static int qemu_rdma_get_buffer(void *opaque, uint8_t *buf, 2554 int64_t pos, int size) 2555 { 2556 QEMUFileRDMA *r = opaque; 2557 RDMAContext *rdma = r->rdma; 2558 RDMAControlHeader head; 2559 int ret = 0; 2560 2561 CHECK_ERROR_STATE(); 2562 2563 /* 2564 * First, we hold on to the last SEND message we 2565 * were given and dish out the bytes until we run 2566 * out of bytes. 2567 */ 2568 r->len = qemu_rdma_fill(r->rdma, buf, size, 0); 2569 if (r->len) { 2570 return r->len; 2571 } 2572 2573 /* 2574 * Once we run out, we block and wait for another 2575 * SEND message to arrive. 2576 */ 2577 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_QEMU_FILE); 2578 2579 if (ret < 0) { 2580 rdma->error_state = ret; 2581 return ret; 2582 } 2583 2584 /* 2585 * SEND was received with new bytes, now try again. 2586 */ 2587 return qemu_rdma_fill(r->rdma, buf, size, 0); 2588 } 2589 2590 /* 2591 * Block until all the outstanding chunks have been delivered by the hardware. 2592 */ 2593 static int qemu_rdma_drain_cq(QEMUFile *f, RDMAContext *rdma) 2594 { 2595 int ret; 2596 2597 if (qemu_rdma_write_flush(f, rdma) < 0) { 2598 return -EIO; 2599 } 2600 2601 while (rdma->nb_sent) { 2602 ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL); 2603 if (ret < 0) { 2604 error_report("rdma migration: complete polling error!"); 2605 return -EIO; 2606 } 2607 } 2608 2609 qemu_rdma_unregister_waiting(rdma); 2610 2611 return 0; 2612 } 2613 2614 static int qemu_rdma_close(void *opaque) 2615 { 2616 trace_qemu_rdma_close(); 2617 QEMUFileRDMA *r = opaque; 2618 if (r->rdma) { 2619 qemu_rdma_cleanup(r->rdma); 2620 g_free(r->rdma); 2621 } 2622 g_free(r); 2623 return 0; 2624 } 2625 2626 /* 2627 * Parameters: 2628 * @offset == 0 : 2629 * This means that 'block_offset' is a full virtual address that does not 2630 * belong to a RAMBlock of the virtual machine and instead 2631 * represents a private malloc'd memory area that the caller wishes to 2632 * transfer. 2633 * 2634 * @offset != 0 : 2635 * Offset is an offset to be added to block_offset and used 2636 * to also lookup the corresponding RAMBlock. 2637 * 2638 * @size > 0 : 2639 * Initiate an transfer this size. 2640 * 2641 * @size == 0 : 2642 * A 'hint' or 'advice' that means that we wish to speculatively 2643 * and asynchronously unregister this memory. In this case, there is no 2644 * guarantee that the unregister will actually happen, for example, 2645 * if the memory is being actively transmitted. Additionally, the memory 2646 * may be re-registered at any future time if a write within the same 2647 * chunk was requested again, even if you attempted to unregister it 2648 * here. 2649 * 2650 * @size < 0 : TODO, not yet supported 2651 * Unregister the memory NOW. This means that the caller does not 2652 * expect there to be any future RDMA transfers and we just want to clean 2653 * things up. This is used in case the upper layer owns the memory and 2654 * cannot wait for qemu_fclose() to occur. 2655 * 2656 * @bytes_sent : User-specificed pointer to indicate how many bytes were 2657 * sent. Usually, this will not be more than a few bytes of 2658 * the protocol because most transfers are sent asynchronously. 2659 */ 2660 static size_t qemu_rdma_save_page(QEMUFile *f, void *opaque, 2661 ram_addr_t block_offset, ram_addr_t offset, 2662 size_t size, int *bytes_sent) 2663 { 2664 QEMUFileRDMA *rfile = opaque; 2665 RDMAContext *rdma = rfile->rdma; 2666 int ret; 2667 2668 CHECK_ERROR_STATE(); 2669 2670 qemu_fflush(f); 2671 2672 if (size > 0) { 2673 /* 2674 * Add this page to the current 'chunk'. If the chunk 2675 * is full, or the page doen't belong to the current chunk, 2676 * an actual RDMA write will occur and a new chunk will be formed. 2677 */ 2678 ret = qemu_rdma_write(f, rdma, block_offset, offset, size); 2679 if (ret < 0) { 2680 error_report("rdma migration: write error! %d", ret); 2681 goto err; 2682 } 2683 2684 /* 2685 * We always return 1 bytes because the RDMA 2686 * protocol is completely asynchronous. We do not yet know 2687 * whether an identified chunk is zero or not because we're 2688 * waiting for other pages to potentially be merged with 2689 * the current chunk. So, we have to call qemu_update_position() 2690 * later on when the actual write occurs. 2691 */ 2692 if (bytes_sent) { 2693 *bytes_sent = 1; 2694 } 2695 } else { 2696 uint64_t index, chunk; 2697 2698 /* TODO: Change QEMUFileOps prototype to be signed: size_t => long 2699 if (size < 0) { 2700 ret = qemu_rdma_drain_cq(f, rdma); 2701 if (ret < 0) { 2702 fprintf(stderr, "rdma: failed to synchronously drain" 2703 " completion queue before unregistration.\n"); 2704 goto err; 2705 } 2706 } 2707 */ 2708 2709 ret = qemu_rdma_search_ram_block(rdma, block_offset, 2710 offset, size, &index, &chunk); 2711 2712 if (ret) { 2713 error_report("ram block search failed"); 2714 goto err; 2715 } 2716 2717 qemu_rdma_signal_unregister(rdma, index, chunk, 0); 2718 2719 /* 2720 * TODO: Synchronous, guaranteed unregistration (should not occur during 2721 * fast-path). Otherwise, unregisters will process on the next call to 2722 * qemu_rdma_drain_cq() 2723 if (size < 0) { 2724 qemu_rdma_unregister_waiting(rdma); 2725 } 2726 */ 2727 } 2728 2729 /* 2730 * Drain the Completion Queue if possible, but do not block, 2731 * just poll. 2732 * 2733 * If nothing to poll, the end of the iteration will do this 2734 * again to make sure we don't overflow the request queue. 2735 */ 2736 while (1) { 2737 uint64_t wr_id, wr_id_in; 2738 int ret = qemu_rdma_poll(rdma, &wr_id_in, NULL); 2739 if (ret < 0) { 2740 error_report("rdma migration: polling error! %d", ret); 2741 goto err; 2742 } 2743 2744 wr_id = wr_id_in & RDMA_WRID_TYPE_MASK; 2745 2746 if (wr_id == RDMA_WRID_NONE) { 2747 break; 2748 } 2749 } 2750 2751 return RAM_SAVE_CONTROL_DELAYED; 2752 err: 2753 rdma->error_state = ret; 2754 return ret; 2755 } 2756 2757 static int qemu_rdma_accept(RDMAContext *rdma) 2758 { 2759 RDMACapabilities cap; 2760 struct rdma_conn_param conn_param = { 2761 .responder_resources = 2, 2762 .private_data = &cap, 2763 .private_data_len = sizeof(cap), 2764 }; 2765 struct rdma_cm_event *cm_event; 2766 struct ibv_context *verbs; 2767 int ret = -EINVAL; 2768 int idx; 2769 2770 ret = rdma_get_cm_event(rdma->channel, &cm_event); 2771 if (ret) { 2772 goto err_rdma_dest_wait; 2773 } 2774 2775 if (cm_event->event != RDMA_CM_EVENT_CONNECT_REQUEST) { 2776 rdma_ack_cm_event(cm_event); 2777 goto err_rdma_dest_wait; 2778 } 2779 2780 memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap)); 2781 2782 network_to_caps(&cap); 2783 2784 if (cap.version < 1 || cap.version > RDMA_CONTROL_VERSION_CURRENT) { 2785 error_report("Unknown source RDMA version: %d, bailing...", 2786 cap.version); 2787 rdma_ack_cm_event(cm_event); 2788 goto err_rdma_dest_wait; 2789 } 2790 2791 /* 2792 * Respond with only the capabilities this version of QEMU knows about. 2793 */ 2794 cap.flags &= known_capabilities; 2795 2796 /* 2797 * Enable the ones that we do know about. 2798 * Add other checks here as new ones are introduced. 2799 */ 2800 if (cap.flags & RDMA_CAPABILITY_PIN_ALL) { 2801 rdma->pin_all = true; 2802 } 2803 2804 rdma->cm_id = cm_event->id; 2805 verbs = cm_event->id->verbs; 2806 2807 rdma_ack_cm_event(cm_event); 2808 2809 trace_qemu_rdma_accept_pin_state(rdma->pin_all); 2810 2811 caps_to_network(&cap); 2812 2813 trace_qemu_rdma_accept_pin_verbsc(verbs); 2814 2815 if (!rdma->verbs) { 2816 rdma->verbs = verbs; 2817 } else if (rdma->verbs != verbs) { 2818 error_report("ibv context not matching %p, %p!", rdma->verbs, 2819 verbs); 2820 goto err_rdma_dest_wait; 2821 } 2822 2823 qemu_rdma_dump_id("dest_init", verbs); 2824 2825 ret = qemu_rdma_alloc_pd_cq(rdma); 2826 if (ret) { 2827 error_report("rdma migration: error allocating pd and cq!"); 2828 goto err_rdma_dest_wait; 2829 } 2830 2831 ret = qemu_rdma_alloc_qp(rdma); 2832 if (ret) { 2833 error_report("rdma migration: error allocating qp!"); 2834 goto err_rdma_dest_wait; 2835 } 2836 2837 ret = qemu_rdma_init_ram_blocks(rdma); 2838 if (ret) { 2839 error_report("rdma migration: error initializing ram blocks!"); 2840 goto err_rdma_dest_wait; 2841 } 2842 2843 for (idx = 0; idx < RDMA_WRID_MAX; idx++) { 2844 ret = qemu_rdma_reg_control(rdma, idx); 2845 if (ret) { 2846 error_report("rdma: error registering %d control", idx); 2847 goto err_rdma_dest_wait; 2848 } 2849 } 2850 2851 qemu_set_fd_handler2(rdma->channel->fd, NULL, NULL, NULL, NULL); 2852 2853 ret = rdma_accept(rdma->cm_id, &conn_param); 2854 if (ret) { 2855 error_report("rdma_accept returns %d", ret); 2856 goto err_rdma_dest_wait; 2857 } 2858 2859 ret = rdma_get_cm_event(rdma->channel, &cm_event); 2860 if (ret) { 2861 error_report("rdma_accept get_cm_event failed %d", ret); 2862 goto err_rdma_dest_wait; 2863 } 2864 2865 if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) { 2866 error_report("rdma_accept not event established"); 2867 rdma_ack_cm_event(cm_event); 2868 goto err_rdma_dest_wait; 2869 } 2870 2871 rdma_ack_cm_event(cm_event); 2872 rdma->connected = true; 2873 2874 ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY); 2875 if (ret) { 2876 error_report("rdma migration: error posting second control recv"); 2877 goto err_rdma_dest_wait; 2878 } 2879 2880 qemu_rdma_dump_gid("dest_connect", rdma->cm_id); 2881 2882 return 0; 2883 2884 err_rdma_dest_wait: 2885 rdma->error_state = ret; 2886 qemu_rdma_cleanup(rdma); 2887 return ret; 2888 } 2889 2890 /* 2891 * During each iteration of the migration, we listen for instructions 2892 * by the source VM to perform dynamic page registrations before they 2893 * can perform RDMA operations. 2894 * 2895 * We respond with the 'rkey'. 2896 * 2897 * Keep doing this until the source tells us to stop. 2898 */ 2899 static int qemu_rdma_registration_handle(QEMUFile *f, void *opaque, 2900 uint64_t flags) 2901 { 2902 RDMAControlHeader reg_resp = { .len = sizeof(RDMARegisterResult), 2903 .type = RDMA_CONTROL_REGISTER_RESULT, 2904 .repeat = 0, 2905 }; 2906 RDMAControlHeader unreg_resp = { .len = 0, 2907 .type = RDMA_CONTROL_UNREGISTER_FINISHED, 2908 .repeat = 0, 2909 }; 2910 RDMAControlHeader blocks = { .type = RDMA_CONTROL_RAM_BLOCKS_RESULT, 2911 .repeat = 1 }; 2912 QEMUFileRDMA *rfile = opaque; 2913 RDMAContext *rdma = rfile->rdma; 2914 RDMALocalBlocks *local = &rdma->local_ram_blocks; 2915 RDMAControlHeader head; 2916 RDMARegister *reg, *registers; 2917 RDMACompress *comp; 2918 RDMARegisterResult *reg_result; 2919 static RDMARegisterResult results[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE]; 2920 RDMALocalBlock *block; 2921 void *host_addr; 2922 int ret = 0; 2923 int idx = 0; 2924 int count = 0; 2925 int i = 0; 2926 2927 CHECK_ERROR_STATE(); 2928 2929 do { 2930 trace_qemu_rdma_registration_handle_wait(flags); 2931 2932 ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_NONE); 2933 2934 if (ret < 0) { 2935 break; 2936 } 2937 2938 if (head.repeat > RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE) { 2939 error_report("rdma: Too many requests in this message (%d)." 2940 "Bailing.", head.repeat); 2941 ret = -EIO; 2942 break; 2943 } 2944 2945 switch (head.type) { 2946 case RDMA_CONTROL_COMPRESS: 2947 comp = (RDMACompress *) rdma->wr_data[idx].control_curr; 2948 network_to_compress(comp); 2949 2950 trace_qemu_rdma_registration_handle_compress(comp->length, 2951 comp->block_idx, 2952 comp->offset); 2953 block = &(rdma->local_ram_blocks.block[comp->block_idx]); 2954 2955 host_addr = block->local_host_addr + 2956 (comp->offset - block->offset); 2957 2958 ram_handle_compressed(host_addr, comp->value, comp->length); 2959 break; 2960 2961 case RDMA_CONTROL_REGISTER_FINISHED: 2962 trace_qemu_rdma_registration_handle_finished(); 2963 goto out; 2964 2965 case RDMA_CONTROL_RAM_BLOCKS_REQUEST: 2966 trace_qemu_rdma_registration_handle_ram_blocks(); 2967 2968 if (rdma->pin_all) { 2969 ret = qemu_rdma_reg_whole_ram_blocks(rdma); 2970 if (ret) { 2971 error_report("rdma migration: error dest " 2972 "registering ram blocks"); 2973 goto out; 2974 } 2975 } 2976 2977 /* 2978 * Dest uses this to prepare to transmit the RAMBlock descriptions 2979 * to the source VM after connection setup. 2980 * Both sides use the "remote" structure to communicate and update 2981 * their "local" descriptions with what was sent. 2982 */ 2983 for (i = 0; i < local->nb_blocks; i++) { 2984 rdma->block[i].remote_host_addr = 2985 (uint64_t)(local->block[i].local_host_addr); 2986 2987 if (rdma->pin_all) { 2988 rdma->block[i].remote_rkey = local->block[i].mr->rkey; 2989 } 2990 2991 rdma->block[i].offset = local->block[i].offset; 2992 rdma->block[i].length = local->block[i].length; 2993 2994 remote_block_to_network(&rdma->block[i]); 2995 } 2996 2997 blocks.len = rdma->local_ram_blocks.nb_blocks 2998 * sizeof(RDMARemoteBlock); 2999 3000 3001 ret = qemu_rdma_post_send_control(rdma, 3002 (uint8_t *) rdma->block, &blocks); 3003 3004 if (ret < 0) { 3005 error_report("rdma migration: error sending remote info"); 3006 goto out; 3007 } 3008 3009 break; 3010 case RDMA_CONTROL_REGISTER_REQUEST: 3011 trace_qemu_rdma_registration_handle_register(head.repeat); 3012 3013 reg_resp.repeat = head.repeat; 3014 registers = (RDMARegister *) rdma->wr_data[idx].control_curr; 3015 3016 for (count = 0; count < head.repeat; count++) { 3017 uint64_t chunk; 3018 uint8_t *chunk_start, *chunk_end; 3019 3020 reg = ®isters[count]; 3021 network_to_register(reg); 3022 3023 reg_result = &results[count]; 3024 3025 trace_qemu_rdma_registration_handle_register_loop(count, 3026 reg->current_index, reg->key.current_addr, reg->chunks); 3027 3028 block = &(rdma->local_ram_blocks.block[reg->current_index]); 3029 if (block->is_ram_block) { 3030 host_addr = (block->local_host_addr + 3031 (reg->key.current_addr - block->offset)); 3032 chunk = ram_chunk_index(block->local_host_addr, 3033 (uint8_t *) host_addr); 3034 } else { 3035 chunk = reg->key.chunk; 3036 host_addr = block->local_host_addr + 3037 (reg->key.chunk * (1UL << RDMA_REG_CHUNK_SHIFT)); 3038 } 3039 chunk_start = ram_chunk_start(block, chunk); 3040 chunk_end = ram_chunk_end(block, chunk + reg->chunks); 3041 if (qemu_rdma_register_and_get_keys(rdma, block, 3042 (uint8_t *)host_addr, NULL, ®_result->rkey, 3043 chunk, chunk_start, chunk_end)) { 3044 error_report("cannot get rkey"); 3045 ret = -EINVAL; 3046 goto out; 3047 } 3048 3049 reg_result->host_addr = (uint64_t) block->local_host_addr; 3050 3051 trace_qemu_rdma_registration_handle_register_rkey( 3052 reg_result->rkey); 3053 3054 result_to_network(reg_result); 3055 } 3056 3057 ret = qemu_rdma_post_send_control(rdma, 3058 (uint8_t *) results, ®_resp); 3059 3060 if (ret < 0) { 3061 error_report("Failed to send control buffer"); 3062 goto out; 3063 } 3064 break; 3065 case RDMA_CONTROL_UNREGISTER_REQUEST: 3066 trace_qemu_rdma_registration_handle_unregister(head.repeat); 3067 unreg_resp.repeat = head.repeat; 3068 registers = (RDMARegister *) rdma->wr_data[idx].control_curr; 3069 3070 for (count = 0; count < head.repeat; count++) { 3071 reg = ®isters[count]; 3072 network_to_register(reg); 3073 3074 trace_qemu_rdma_registration_handle_unregister_loop(count, 3075 reg->current_index, reg->key.chunk); 3076 3077 block = &(rdma->local_ram_blocks.block[reg->current_index]); 3078 3079 ret = ibv_dereg_mr(block->pmr[reg->key.chunk]); 3080 block->pmr[reg->key.chunk] = NULL; 3081 3082 if (ret != 0) { 3083 perror("rdma unregistration chunk failed"); 3084 ret = -ret; 3085 goto out; 3086 } 3087 3088 rdma->total_registrations--; 3089 3090 trace_qemu_rdma_registration_handle_unregister_success( 3091 reg->key.chunk); 3092 } 3093 3094 ret = qemu_rdma_post_send_control(rdma, NULL, &unreg_resp); 3095 3096 if (ret < 0) { 3097 error_report("Failed to send control buffer"); 3098 goto out; 3099 } 3100 break; 3101 case RDMA_CONTROL_REGISTER_RESULT: 3102 error_report("Invalid RESULT message at dest."); 3103 ret = -EIO; 3104 goto out; 3105 default: 3106 error_report("Unknown control message %s", control_desc[head.type]); 3107 ret = -EIO; 3108 goto out; 3109 } 3110 } while (1); 3111 out: 3112 if (ret < 0) { 3113 rdma->error_state = ret; 3114 } 3115 return ret; 3116 } 3117 3118 static int qemu_rdma_registration_start(QEMUFile *f, void *opaque, 3119 uint64_t flags) 3120 { 3121 QEMUFileRDMA *rfile = opaque; 3122 RDMAContext *rdma = rfile->rdma; 3123 3124 CHECK_ERROR_STATE(); 3125 3126 trace_qemu_rdma_registration_start(flags); 3127 qemu_put_be64(f, RAM_SAVE_FLAG_HOOK); 3128 qemu_fflush(f); 3129 3130 return 0; 3131 } 3132 3133 /* 3134 * Inform dest that dynamic registrations are done for now. 3135 * First, flush writes, if any. 3136 */ 3137 static int qemu_rdma_registration_stop(QEMUFile *f, void *opaque, 3138 uint64_t flags) 3139 { 3140 Error *local_err = NULL, **errp = &local_err; 3141 QEMUFileRDMA *rfile = opaque; 3142 RDMAContext *rdma = rfile->rdma; 3143 RDMAControlHeader head = { .len = 0, .repeat = 1 }; 3144 int ret = 0; 3145 3146 CHECK_ERROR_STATE(); 3147 3148 qemu_fflush(f); 3149 ret = qemu_rdma_drain_cq(f, rdma); 3150 3151 if (ret < 0) { 3152 goto err; 3153 } 3154 3155 if (flags == RAM_CONTROL_SETUP) { 3156 RDMAControlHeader resp = {.type = RDMA_CONTROL_RAM_BLOCKS_RESULT }; 3157 RDMALocalBlocks *local = &rdma->local_ram_blocks; 3158 int reg_result_idx, i, j, nb_remote_blocks; 3159 3160 head.type = RDMA_CONTROL_RAM_BLOCKS_REQUEST; 3161 trace_qemu_rdma_registration_stop_ram(); 3162 3163 /* 3164 * Make sure that we parallelize the pinning on both sides. 3165 * For very large guests, doing this serially takes a really 3166 * long time, so we have to 'interleave' the pinning locally 3167 * with the control messages by performing the pinning on this 3168 * side before we receive the control response from the other 3169 * side that the pinning has completed. 3170 */ 3171 ret = qemu_rdma_exchange_send(rdma, &head, NULL, &resp, 3172 ®_result_idx, rdma->pin_all ? 3173 qemu_rdma_reg_whole_ram_blocks : NULL); 3174 if (ret < 0) { 3175 ERROR(errp, "receiving remote info!"); 3176 return ret; 3177 } 3178 3179 nb_remote_blocks = resp.len / sizeof(RDMARemoteBlock); 3180 3181 /* 3182 * The protocol uses two different sets of rkeys (mutually exclusive): 3183 * 1. One key to represent the virtual address of the entire ram block. 3184 * (dynamic chunk registration disabled - pin everything with one rkey.) 3185 * 2. One to represent individual chunks within a ram block. 3186 * (dynamic chunk registration enabled - pin individual chunks.) 3187 * 3188 * Once the capability is successfully negotiated, the destination transmits 3189 * the keys to use (or sends them later) including the virtual addresses 3190 * and then propagates the remote ram block descriptions to his local copy. 3191 */ 3192 3193 if (local->nb_blocks != nb_remote_blocks) { 3194 ERROR(errp, "ram blocks mismatch #1! " 3195 "Your QEMU command line parameters are probably " 3196 "not identical on both the source and destination."); 3197 return -EINVAL; 3198 } 3199 3200 qemu_rdma_move_header(rdma, reg_result_idx, &resp); 3201 memcpy(rdma->block, 3202 rdma->wr_data[reg_result_idx].control_curr, resp.len); 3203 for (i = 0; i < nb_remote_blocks; i++) { 3204 network_to_remote_block(&rdma->block[i]); 3205 3206 /* search local ram blocks */ 3207 for (j = 0; j < local->nb_blocks; j++) { 3208 if (rdma->block[i].offset != local->block[j].offset) { 3209 continue; 3210 } 3211 3212 if (rdma->block[i].length != local->block[j].length) { 3213 ERROR(errp, "ram blocks mismatch #2! " 3214 "Your QEMU command line parameters are probably " 3215 "not identical on both the source and destination."); 3216 return -EINVAL; 3217 } 3218 local->block[j].remote_host_addr = 3219 rdma->block[i].remote_host_addr; 3220 local->block[j].remote_rkey = rdma->block[i].remote_rkey; 3221 break; 3222 } 3223 3224 if (j >= local->nb_blocks) { 3225 ERROR(errp, "ram blocks mismatch #3! " 3226 "Your QEMU command line parameters are probably " 3227 "not identical on both the source and destination."); 3228 return -EINVAL; 3229 } 3230 } 3231 } 3232 3233 trace_qemu_rdma_registration_stop(flags); 3234 3235 head.type = RDMA_CONTROL_REGISTER_FINISHED; 3236 ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL); 3237 3238 if (ret < 0) { 3239 goto err; 3240 } 3241 3242 return 0; 3243 err: 3244 rdma->error_state = ret; 3245 return ret; 3246 } 3247 3248 static int qemu_rdma_get_fd(void *opaque) 3249 { 3250 QEMUFileRDMA *rfile = opaque; 3251 RDMAContext *rdma = rfile->rdma; 3252 3253 return rdma->comp_channel->fd; 3254 } 3255 3256 const QEMUFileOps rdma_read_ops = { 3257 .get_buffer = qemu_rdma_get_buffer, 3258 .get_fd = qemu_rdma_get_fd, 3259 .close = qemu_rdma_close, 3260 .hook_ram_load = qemu_rdma_registration_handle, 3261 }; 3262 3263 const QEMUFileOps rdma_write_ops = { 3264 .put_buffer = qemu_rdma_put_buffer, 3265 .close = qemu_rdma_close, 3266 .before_ram_iterate = qemu_rdma_registration_start, 3267 .after_ram_iterate = qemu_rdma_registration_stop, 3268 .save_page = qemu_rdma_save_page, 3269 }; 3270 3271 static void *qemu_fopen_rdma(RDMAContext *rdma, const char *mode) 3272 { 3273 QEMUFileRDMA *r = g_malloc0(sizeof(QEMUFileRDMA)); 3274 3275 if (qemu_file_mode_is_not_valid(mode)) { 3276 return NULL; 3277 } 3278 3279 r->rdma = rdma; 3280 3281 if (mode[0] == 'w') { 3282 r->file = qemu_fopen_ops(r, &rdma_write_ops); 3283 } else { 3284 r->file = qemu_fopen_ops(r, &rdma_read_ops); 3285 } 3286 3287 return r->file; 3288 } 3289 3290 static void rdma_accept_incoming_migration(void *opaque) 3291 { 3292 RDMAContext *rdma = opaque; 3293 int ret; 3294 QEMUFile *f; 3295 Error *local_err = NULL, **errp = &local_err; 3296 3297 trace_qemu_dma_accept_incoming_migration(); 3298 ret = qemu_rdma_accept(rdma); 3299 3300 if (ret) { 3301 ERROR(errp, "RDMA Migration initialization failed!"); 3302 return; 3303 } 3304 3305 trace_qemu_dma_accept_incoming_migration_accepted(); 3306 3307 f = qemu_fopen_rdma(rdma, "rb"); 3308 if (f == NULL) { 3309 ERROR(errp, "could not qemu_fopen_rdma!"); 3310 qemu_rdma_cleanup(rdma); 3311 return; 3312 } 3313 3314 rdma->migration_started_on_destination = 1; 3315 process_incoming_migration(f); 3316 } 3317 3318 void rdma_start_incoming_migration(const char *host_port, Error **errp) 3319 { 3320 int ret; 3321 RDMAContext *rdma; 3322 Error *local_err = NULL; 3323 3324 trace_rdma_start_incoming_migration(); 3325 rdma = qemu_rdma_data_init(host_port, &local_err); 3326 3327 if (rdma == NULL) { 3328 goto err; 3329 } 3330 3331 ret = qemu_rdma_dest_init(rdma, &local_err); 3332 3333 if (ret) { 3334 goto err; 3335 } 3336 3337 trace_rdma_start_incoming_migration_after_dest_init(); 3338 3339 ret = rdma_listen(rdma->listen_id, 5); 3340 3341 if (ret) { 3342 ERROR(errp, "listening on socket!"); 3343 goto err; 3344 } 3345 3346 trace_rdma_start_incoming_migration_after_rdma_listen(); 3347 3348 qemu_set_fd_handler2(rdma->channel->fd, NULL, 3349 rdma_accept_incoming_migration, NULL, 3350 (void *)(intptr_t) rdma); 3351 return; 3352 err: 3353 error_propagate(errp, local_err); 3354 g_free(rdma); 3355 } 3356 3357 void rdma_start_outgoing_migration(void *opaque, 3358 const char *host_port, Error **errp) 3359 { 3360 MigrationState *s = opaque; 3361 Error *local_err = NULL, **temp = &local_err; 3362 RDMAContext *rdma = qemu_rdma_data_init(host_port, &local_err); 3363 int ret = 0; 3364 3365 if (rdma == NULL) { 3366 ERROR(temp, "Failed to initialize RDMA data structures! %d", ret); 3367 goto err; 3368 } 3369 3370 ret = qemu_rdma_source_init(rdma, &local_err, 3371 s->enabled_capabilities[MIGRATION_CAPABILITY_RDMA_PIN_ALL]); 3372 3373 if (ret) { 3374 goto err; 3375 } 3376 3377 trace_rdma_start_outgoing_migration_after_rdma_source_init(); 3378 ret = qemu_rdma_connect(rdma, &local_err); 3379 3380 if (ret) { 3381 goto err; 3382 } 3383 3384 trace_rdma_start_outgoing_migration_after_rdma_connect(); 3385 3386 s->file = qemu_fopen_rdma(rdma, "wb"); 3387 migrate_fd_connect(s); 3388 return; 3389 err: 3390 error_propagate(errp, local_err); 3391 g_free(rdma); 3392 migrate_fd_error(s); 3393 } 3394