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