Archipelago » qemu

/*

 * RDMA protocol and interfaces

 *

 * Copyright IBM, Corp. 2010-2013

 *

 * Authors:

 *  Michael R. Hines <mrhines@us.ibm.com>

 *  Jiuxing Liu <jl@us.ibm.com>

 *

 * This work is licensed under the terms of the GNU GPL, version 2 or

 * later.  See the COPYING file in the top-level directory.

 *

 */

#include "qemu-common.h"

#include "migration/migration.h"

#include "migration/qemu-file.h"

#include "exec/cpu-common.h"

#include "qemu/main-loop.h"

#include "qemu/sockets.h"

#include "qemu/bitmap.h"

#include "block/coroutine.h"

#include <stdio.h>

#include <sys/types.h>

#include <sys/socket.h>

#include <netdb.h>

#include <arpa/inet.h>

#include <string.h>

#include <rdma/rdma_cma.h>

#define DEBUG_RDMA

//#define DEBUG_RDMA_VERBOSE

//#define DEBUG_RDMA_REALLY_VERBOSE

#ifdef DEBUG_RDMA

#define DPRINTF(fmt, ...) \

    do { printf("rdma: " fmt, ## __VA_ARGS__); } while (0)

#else

#define DPRINTF(fmt, ...) \

    do { } while (0)

#endif

#ifdef DEBUG_RDMA_VERBOSE

#define DDPRINTF(fmt, ...) \

    do { printf("rdma: " fmt, ## __VA_ARGS__); } while (0)

#else

#define DDPRINTF(fmt, ...) \

    do { } while (0)

#endif

#ifdef DEBUG_RDMA_REALLY_VERBOSE

#define DDDPRINTF(fmt, ...) \

    do { printf("rdma: " fmt, ## __VA_ARGS__); } while (0)

#else

#define DDDPRINTF(fmt, ...) \

    do { } while (0)

#endif

/*

 * Print and error on both the Monitor and the Log file.

 */

#define ERROR(errp, fmt, ...) \

    do { \

        fprintf(stderr, "RDMA ERROR: " fmt, ## __VA_ARGS__); \

        if (errp && (*(errp) == NULL)) { \

            error_setg(errp, "RDMA ERROR: " fmt, ## __VA_ARGS__); \

        } \

    } while (0)

#define RDMA_RESOLVE_TIMEOUT_MS 10000

/* Do not merge data if larger than this. */

#define RDMA_MERGE_MAX (2 * 1024 * 1024)

#define RDMA_SIGNALED_SEND_MAX (RDMA_MERGE_MAX / 4096)

#define RDMA_REG_CHUNK_SHIFT 20 /* 1 MB */

/*

 * This is only for non-live state being migrated.

 * Instead of RDMA_WRITE messages, we use RDMA_SEND

 * messages for that state, which requires a different

 * delivery design than main memory.

 */

#define RDMA_SEND_INCREMENT 32768

/*

 * Maximum size infiniband SEND message

 */

#define RDMA_CONTROL_MAX_BUFFER (512 * 1024)

#define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096

#define RDMA_CONTROL_VERSION_CURRENT 1

/*

 * Capabilities for negotiation.

 */

#define RDMA_CAPABILITY_PIN_ALL 0x01

/*

 * Add the other flags above to this list of known capabilities

 * as they are introduced.

 */

static uint32_t known_capabilities = RDMA_CAPABILITY_PIN_ALL;

#define CHECK_ERROR_STATE() \

    do { \

        if (rdma->error_state) { \

            if (!rdma->error_reported) { \

                fprintf(stderr, "RDMA is in an error state waiting migration" \

                                " to abort!\n"); \

                rdma->error_reported = 1; \

            } \

            return rdma->error_state; \

        } \

    } while (0);

/*

 * A work request ID is 64-bits and we split up these bits

 * into 3 parts:

 *

 * bits 0-15 : type of control message, 2^16

 * bits 16-29: ram block index, 2^14

 * bits 30-63: ram block chunk number, 2^34

 *

 * The last two bit ranges are only used for RDMA writes,

 * in order to track their completion and potentially

 * also track unregistration status of the message.

 */

#define RDMA_WRID_TYPE_SHIFT  0UL

#define RDMA_WRID_BLOCK_SHIFT 16UL

#define RDMA_WRID_CHUNK_SHIFT 30UL

#define RDMA_WRID_TYPE_MASK \

    ((1UL << RDMA_WRID_BLOCK_SHIFT) - 1UL)

#define RDMA_WRID_BLOCK_MASK \

    (~RDMA_WRID_TYPE_MASK & ((1UL << RDMA_WRID_CHUNK_SHIFT) - 1UL))

#define RDMA_WRID_CHUNK_MASK (~RDMA_WRID_BLOCK_MASK & ~RDMA_WRID_TYPE_MASK)

/*

 * RDMA migration protocol:

 * 1. RDMA Writes (data messages, i.e. RAM)

 * 2. IB Send/Recv (control channel messages)

 */

enum {

    RDMA_WRID_NONE = 0,

    RDMA_WRID_RDMA_WRITE = 1,

    RDMA_WRID_SEND_CONTROL = 2000,

    RDMA_WRID_RECV_CONTROL = 4000,

};

const char *wrid_desc[] = {

    [RDMA_WRID_NONE] = "NONE",

    [RDMA_WRID_RDMA_WRITE] = "WRITE RDMA",

    [RDMA_WRID_SEND_CONTROL] = "CONTROL SEND",

    [RDMA_WRID_RECV_CONTROL] = "CONTROL RECV",

};

/*

 * Work request IDs for IB SEND messages only (not RDMA writes).

 * This is used by the migration protocol to transmit

 * control messages (such as device state and registration commands)

 *

 * We could use more WRs, but we have enough for now.

 */

enum {

    RDMA_WRID_READY = 0,

    RDMA_WRID_DATA,

    RDMA_WRID_CONTROL,

    RDMA_WRID_MAX,

};

/*

 * SEND/RECV IB Control Messages.

 */

enum {

    RDMA_CONTROL_NONE = 0,

    RDMA_CONTROL_ERROR,

    RDMA_CONTROL_READY,               /* ready to receive */

    RDMA_CONTROL_QEMU_FILE,           /* QEMUFile-transmitted bytes */

    RDMA_CONTROL_RAM_BLOCKS_REQUEST,  /* RAMBlock synchronization */

    RDMA_CONTROL_RAM_BLOCKS_RESULT,   /* RAMBlock synchronization */

    RDMA_CONTROL_COMPRESS,            /* page contains repeat values */

    RDMA_CONTROL_REGISTER_REQUEST,    /* dynamic page registration */

    RDMA_CONTROL_REGISTER_RESULT,     /* key to use after registration */

    RDMA_CONTROL_REGISTER_FINISHED,   /* current iteration finished */

    RDMA_CONTROL_UNREGISTER_REQUEST,  /* dynamic UN-registration */

    RDMA_CONTROL_UNREGISTER_FINISHED, /* unpinning finished */

};

const char *control_desc[] = {

    [RDMA_CONTROL_NONE] = "NONE",

    [RDMA_CONTROL_ERROR] = "ERROR",

    [RDMA_CONTROL_READY] = "READY",

    [RDMA_CONTROL_QEMU_FILE] = "QEMU FILE",

    [RDMA_CONTROL_RAM_BLOCKS_REQUEST] = "RAM BLOCKS REQUEST",

    [RDMA_CONTROL_RAM_BLOCKS_RESULT] = "RAM BLOCKS RESULT",

    [RDMA_CONTROL_COMPRESS] = "COMPRESS",

    [RDMA_CONTROL_REGISTER_REQUEST] = "REGISTER REQUEST",

    [RDMA_CONTROL_REGISTER_RESULT] = "REGISTER RESULT",

    [RDMA_CONTROL_REGISTER_FINISHED] = "REGISTER FINISHED",

    [RDMA_CONTROL_UNREGISTER_REQUEST] = "UNREGISTER REQUEST",

    [RDMA_CONTROL_UNREGISTER_FINISHED] = "UNREGISTER FINISHED",

};

/*

 * Memory and MR structures used to represent an IB Send/Recv work request.

 * This is *not* used for RDMA writes, only IB Send/Recv.

 */

typedef struct {

    uint8_t  control[RDMA_CONTROL_MAX_BUFFER]; /* actual buffer to register */

    struct   ibv_mr *control_mr;               /* registration metadata */

    size_t   control_len;                      /* length of the message */

    uint8_t *control_curr;                     /* start of unconsumed bytes */

} RDMAWorkRequestData;

/*

 * Negotiate RDMA capabilities during connection-setup time.

 */

typedef struct {

    uint32_t version;

    uint32_t flags;

} RDMACapabilities;

static void caps_to_network(RDMACapabilities *cap)

{

    cap->version = htonl(cap->version);

    cap->flags = htonl(cap->flags);

}

static void network_to_caps(RDMACapabilities *cap)

{

    cap->version = ntohl(cap->version);

    cap->flags = ntohl(cap->flags);

}

/*

 * Representation of a RAMBlock from an RDMA perspective.

 * This is not transmitted, only local.

 * This and subsequent structures cannot be linked lists

 * because we're using a single IB message to transmit

 * the information. It's small anyway, so a list is overkill.

 */

typedef struct RDMALocalBlock {

    uint8_t  *local_host_addr; /* local virtual address */

    uint64_t remote_host_addr; /* remote virtual address */

    uint64_t offset;

    uint64_t length;

    struct   ibv_mr **pmr;     /* MRs for chunk-level registration */

    struct   ibv_mr *mr;       /* MR for non-chunk-level registration */

    uint32_t *remote_keys;     /* rkeys for chunk-level registration */

    uint32_t remote_rkey;      /* rkeys for non-chunk-level registration */

    int      index;            /* which block are we */

    bool     is_ram_block;

    int      nb_chunks;

    unsigned long *transit_bitmap;

    unsigned long *unregister_bitmap;

} RDMALocalBlock;

/*

 * Also represents a RAMblock, but only on the dest.

 * This gets transmitted by the dest during connection-time

 * to the source VM and then is used to populate the

 * corresponding RDMALocalBlock with

 * the information needed to perform the actual RDMA.

 */

typedef struct QEMU_PACKED RDMARemoteBlock {

    uint64_t remote_host_addr;

    uint64_t offset;

    uint64_t length;

    uint32_t remote_rkey;

    uint32_t padding;

} RDMARemoteBlock;

static uint64_t htonll(uint64_t v)

{

    union { uint32_t lv[2]; uint64_t llv; } u;

    u.lv[0] = htonl(v >> 32);

    u.lv[1] = htonl(v & 0xFFFFFFFFULL);

    return u.llv;

}

static uint64_t ntohll(uint64_t v) {

    union { uint32_t lv[2]; uint64_t llv; } u;

    u.llv = v;

    return ((uint64_t)ntohl(u.lv[0]) << 32) | (uint64_t) ntohl(u.lv[1]);

}

static void remote_block_to_network(RDMARemoteBlock *rb)

{

    rb->remote_host_addr = htonll(rb->remote_host_addr);

    rb->offset = htonll(rb->offset);

    rb->length = htonll(rb->length);

    rb->remote_rkey = htonl(rb->remote_rkey);

}

static void network_to_remote_block(RDMARemoteBlock *rb)

{

    rb->remote_host_addr = ntohll(rb->remote_host_addr);

    rb->offset = ntohll(rb->offset);

    rb->length = ntohll(rb->length);

    rb->remote_rkey = ntohl(rb->remote_rkey);

}

/*

 * Virtual address of the above structures used for transmitting

 * the RAMBlock descriptions at connection-time.

 * This structure is *not* transmitted.

 */

typedef struct RDMALocalBlocks {

    int nb_blocks;

    bool     init;             /* main memory init complete */

    RDMALocalBlock *block;

} RDMALocalBlocks;

/*

 * Main data structure for RDMA state.

 * While there is only one copy of this structure being allocated right now,

 * this is the place where one would start if you wanted to consider

 * having more than one RDMA connection open at the same time.

 */

typedef struct RDMAContext {

    char *host;

    int port;

    RDMAWorkRequestData wr_data[RDMA_WRID_MAX + 1];

    /*

     * This is used by *_exchange_send() to figure out whether or not

     * the initial "READY" message has already been received or not.

     * This is because other functions may potentially poll() and detect

     * the READY message before send() does, in which case we need to

     * know if it completed.

     */

    int control_ready_expected;

    /* number of outstanding writes */

    int nb_sent;

    /* store info about current buffer so that we can

       merge it with future sends */

    uint64_t current_addr;

    uint64_t current_length;

    /* index of ram block the current buffer belongs to */

    int current_index;

    /* index of the chunk in the current ram block */

    int current_chunk;

    bool pin_all;

    /*

     * infiniband-specific variables for opening the device

     * and maintaining connection state and so forth.

     *

     * cm_id also has ibv_context, rdma_event_channel, and ibv_qp in

     * cm_id->verbs, cm_id->channel, and cm_id->qp.

     */

    struct rdma_cm_id *cm_id;               /* connection manager ID */

    struct rdma_cm_id *listen_id;

    struct ibv_context          *verbs;

    struct rdma_event_channel   *channel;

    struct ibv_qp *qp;                      /* queue pair */

    struct ibv_comp_channel *comp_channel;  /* completion channel */

    struct ibv_pd *pd;                      /* protection domain */

    struct ibv_cq *cq;                      /* completion queue */

    /*

     * If a previous write failed (perhaps because of a failed

     * memory registration, then do not attempt any future work

     * and remember the error state.

     */

    int error_state;

    int error_reported;

    /*

     * Description of ram blocks used throughout the code.

     */

    RDMALocalBlocks local_ram_blocks;

    RDMARemoteBlock *block;

    /*

     * Migration on *destination* started.

     * Then use coroutine yield function.

     * Source runs in a thread, so we don't care.

     */

    int migration_started_on_destination;

    int total_registrations;

    int total_writes;

    int unregister_current, unregister_next;

    uint64_t unregistrations[RDMA_SIGNALED_SEND_MAX];

    GHashTable *blockmap;

} RDMAContext;

/*

 * Interface to the rest of the migration call stack.

 */

typedef struct QEMUFileRDMA {

    RDMAContext *rdma;

    size_t len;

    void *file;

} QEMUFileRDMA;

/*

 * Main structure for IB Send/Recv control messages.

 * This gets prepended at the beginning of every Send/Recv.

 */

typedef struct QEMU_PACKED {

    uint32_t len;     /* Total length of data portion */

    uint32_t type;    /* which control command to perform */

    uint32_t repeat;  /* number of commands in data portion of same type */

    uint32_t padding;

} RDMAControlHeader;

static void control_to_network(RDMAControlHeader *control)

{

    control->type = htonl(control->type);

    control->len = htonl(control->len);

    control->repeat = htonl(control->repeat);

}

static void network_to_control(RDMAControlHeader *control)

{

    control->type = ntohl(control->type);

    control->len = ntohl(control->len);

    control->repeat = ntohl(control->repeat);

}

/*

 * Register a single Chunk.

 * Information sent by the source VM to inform the dest

 * to register an single chunk of memory before we can perform

 * the actual RDMA operation.

 */

typedef struct QEMU_PACKED {

    union QEMU_PACKED {

        uint64_t current_addr;  /* offset into the ramblock of the chunk */

        uint64_t chunk;         /* chunk to lookup if unregistering */

    } key;

    uint32_t current_index; /* which ramblock the chunk belongs to */

    uint32_t padding;

    uint64_t chunks;            /* how many sequential chunks to register */

} RDMARegister;

static void register_to_network(RDMARegister *reg)

{

    reg->key.current_addr = htonll(reg->key.current_addr);

    reg->current_index = htonl(reg->current_index);

    reg->chunks = htonll(reg->chunks);

}

static void network_to_register(RDMARegister *reg)

{

    reg->key.current_addr = ntohll(reg->key.current_addr);

    reg->current_index = ntohl(reg->current_index);

    reg->chunks = ntohll(reg->chunks);

}

typedef struct QEMU_PACKED {

    uint32_t value;     /* if zero, we will madvise() */

    uint32_t block_idx; /* which ram block index */

    uint64_t offset;    /* where in the remote ramblock this chunk */

    uint64_t length;    /* length of the chunk */

} RDMACompress;

static void compress_to_network(RDMACompress *comp)

{

    comp->value = htonl(comp->value);

    comp->block_idx = htonl(comp->block_idx);

    comp->offset = htonll(comp->offset);

    comp->length = htonll(comp->length);

}

static void network_to_compress(RDMACompress *comp)

{

    comp->value = ntohl(comp->value);

    comp->block_idx = ntohl(comp->block_idx);

    comp->offset = ntohll(comp->offset);

    comp->length = ntohll(comp->length);

}

/*

 * The result of the dest's memory registration produces an "rkey"

 * which the source VM must reference in order to perform

 * the RDMA operation.

 */

typedef struct QEMU_PACKED {

    uint32_t rkey;

    uint32_t padding;

    uint64_t host_addr;

} RDMARegisterResult;

static void result_to_network(RDMARegisterResult *result)

{

    result->rkey = htonl(result->rkey);

    result->host_addr = htonll(result->host_addr);

};

static void network_to_result(RDMARegisterResult *result)

{

    result->rkey = ntohl(result->rkey);

    result->host_addr = ntohll(result->host_addr);

};

const char *print_wrid(int wrid);

static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,

                                   uint8_t *data, RDMAControlHeader *resp,

                                   int *resp_idx,

                                   int (*callback)(RDMAContext *rdma));

static inline uint64_t ram_chunk_index(uint8_t *start, uint8_t *host)

{

    return ((uintptr_t) host - (uintptr_t) start) >> RDMA_REG_CHUNK_SHIFT;

}

static inline uint8_t *ram_chunk_start(RDMALocalBlock *rdma_ram_block,

                                       uint64_t i)

{

    return (uint8_t *) (((uintptr_t) rdma_ram_block->local_host_addr)

                                    + (i << RDMA_REG_CHUNK_SHIFT));

}

static inline uint8_t *ram_chunk_end(RDMALocalBlock *rdma_ram_block, uint64_t i)

{

    uint8_t *result = ram_chunk_start(rdma_ram_block, i) +

                                         (1UL << RDMA_REG_CHUNK_SHIFT);

    if (result > (rdma_ram_block->local_host_addr + rdma_ram_block->length)) {

        result = rdma_ram_block->local_host_addr + rdma_ram_block->length;

    }

    return result;

}

static int __qemu_rdma_add_block(RDMAContext *rdma, void *host_addr,

                         ram_addr_t block_offset, uint64_t length)

{

    RDMALocalBlocks *local = &rdma->local_ram_blocks;

    RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,

        (void *) block_offset);

    RDMALocalBlock *old = local->block;

    assert(block == NULL);

    local->block = g_malloc0(sizeof(RDMALocalBlock) * (local->nb_blocks + 1));

    if (local->nb_blocks) {

        int x;

        for (x = 0; x < local->nb_blocks; x++) {

            g_hash_table_remove(rdma->blockmap, (void *)old[x].offset);

            g_hash_table_insert(rdma->blockmap, (void *)old[x].offset,

                                                &local->block[x]);

        }

        memcpy(local->block, old, sizeof(RDMALocalBlock) * local->nb_blocks);

        g_free(old);

    }

    block = &local->block[local->nb_blocks];

    block->local_host_addr = host_addr;

    block->offset = block_offset;

    block->length = length;

    block->index = local->nb_blocks;

    block->nb_chunks = ram_chunk_index(host_addr, host_addr + length) + 1UL;

    block->transit_bitmap = bitmap_new(block->nb_chunks);

    bitmap_clear(block->transit_bitmap, 0, block->nb_chunks);

    block->unregister_bitmap = bitmap_new(block->nb_chunks);

    bitmap_clear(block->unregister_bitmap, 0, block->nb_chunks);

    block->remote_keys = g_malloc0(block->nb_chunks * sizeof(uint32_t));

    block->is_ram_block = local->init ? false : true;

    g_hash_table_insert(rdma->blockmap, (void *) block_offset, block);

    DDPRINTF("Added Block: %d, addr: %" PRIu64 ", offset: %" PRIu64

           " length: %" PRIu64 " end: %" PRIu64 " bits %" PRIu64 " chunks %d\n",

            local->nb_blocks, (uint64_t) block->local_host_addr, block->offset,

            block->length, (uint64_t) (block->local_host_addr + block->length),

                BITS_TO_LONGS(block->nb_chunks) *

                    sizeof(unsigned long) * 8, block->nb_chunks);

    local->nb_blocks++;

    return 0;

}

/*

 * Memory regions need to be registered with the device and queue pairs setup

 * in advanced before the migration starts. This tells us where the RAM blocks

 * are so that we can register them individually.

 */

static void qemu_rdma_init_one_block(void *host_addr,

    ram_addr_t block_offset, ram_addr_t length, void *opaque)

{

    __qemu_rdma_add_block(opaque, host_addr, block_offset, length);

}

/*

 * Identify the RAMBlocks and their quantity. They will be references to

 * identify chunk boundaries inside each RAMBlock and also be referenced

 * during dynamic page registration.

 */

static int qemu_rdma_init_ram_blocks(RDMAContext *rdma)

{

    RDMALocalBlocks *local = &rdma->local_ram_blocks;

    assert(rdma->blockmap == NULL);

    rdma->blockmap = g_hash_table_new(g_direct_hash, g_direct_equal);

    memset(local, 0, sizeof *local);

    qemu_ram_foreach_block(qemu_rdma_init_one_block, rdma);

    DPRINTF("Allocated %d local ram block structures\n", local->nb_blocks);

    rdma->block = (RDMARemoteBlock *) g_malloc0(sizeof(RDMARemoteBlock) *

                        rdma->local_ram_blocks.nb_blocks);

    local->init = true;

    return 0;

}

static int __qemu_rdma_delete_block(RDMAContext *rdma, ram_addr_t block_offset)

{

    RDMALocalBlocks *local = &rdma->local_ram_blocks;

    RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,

        (void *) block_offset);

    RDMALocalBlock *old = local->block;

    int x;

    assert(block);

    if (block->pmr) {

        int j;

        for (j = 0; j < block->nb_chunks; j++) {

            if (!block->pmr[j]) {

                continue;

            }

            ibv_dereg_mr(block->pmr[j]);

            rdma->total_registrations--;

        }

        g_free(block->pmr);

        block->pmr = NULL;

    }

    if (block->mr) {

        ibv_dereg_mr(block->mr);

        rdma->total_registrations--;

        block->mr = NULL;

    }

    g_free(block->transit_bitmap);

    block->transit_bitmap = NULL;

    g_free(block->unregister_bitmap);

    block->unregister_bitmap = NULL;

    g_free(block->remote_keys);

    block->remote_keys = NULL;

    for (x = 0; x < local->nb_blocks; x++) {

        g_hash_table_remove(rdma->blockmap, (void *)old[x].offset);

    }

    if (local->nb_blocks > 1) {

        local->block = g_malloc0(sizeof(RDMALocalBlock) *

                                    (local->nb_blocks - 1));

        if (block->index) {

            memcpy(local->block, old, sizeof(RDMALocalBlock) * block->index);

        }

        if (block->index < (local->nb_blocks - 1)) {

            memcpy(local->block + block->index, old + (block->index + 1),

                sizeof(RDMALocalBlock) *

                    (local->nb_blocks - (block->index + 1)));

        }

    } else {

        assert(block == local->block);

        local->block = NULL;

    }

    DDPRINTF("Deleted Block: %d, addr: %" PRIu64 ", offset: %" PRIu64

           " length: %" PRIu64 " end: %" PRIu64 " bits %" PRIu64 " chunks %d\n",

            local->nb_blocks, (uint64_t) block->local_host_addr, block->offset,

            block->length, (uint64_t) (block->local_host_addr + block->length),

                BITS_TO_LONGS(block->nb_chunks) *

                    sizeof(unsigned long) * 8, block->nb_chunks);

    g_free(old);

    local->nb_blocks--;

    if (local->nb_blocks) {

        for (x = 0; x < local->nb_blocks; x++) {

            g_hash_table_insert(rdma->blockmap, (void *)local->block[x].offset,

                                                &local->block[x]);

        }

    }

    return 0;

}

/*

 * Put in the log file which RDMA device was opened and the details

 * associated with that device.

 */

static void qemu_rdma_dump_id(const char *who, struct ibv_context *verbs)

{

    printf("%s RDMA Device opened: kernel name %s "

           "uverbs device name %s, "

           "infiniband_verbs class device path %s,"

           " infiniband class device path %s\n",

                who,

                verbs->device->name,

                verbs->device->dev_name,

                verbs->device->dev_path,

                verbs->device->ibdev_path);

}

/*

 * Put in the log file the RDMA gid addressing information,

 * useful for folks who have trouble understanding the

 * RDMA device hierarchy in the kernel.

 */

static void qemu_rdma_dump_gid(const char *who, struct rdma_cm_id *id)

{

    char sgid[33];

    char dgid[33];

    inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.sgid, sgid, sizeof sgid);

    inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.dgid, dgid, sizeof dgid);

    DPRINTF("%s Source GID: %s, Dest GID: %s\n", who, sgid, dgid);

}

/*

 * Figure out which RDMA device corresponds to the requested IP hostname

 * Also create the initial connection manager identifiers for opening

 * the connection.

 */

static int qemu_rdma_resolve_host(RDMAContext *rdma, Error **errp)

{

    int ret;

    struct addrinfo *res;

    char port_str[16];

    struct rdma_cm_event *cm_event;

    char ip[40] = "unknown";

    if (rdma->host == NULL || !strcmp(rdma->host, "")) {

        ERROR(errp, "RDMA hostname has not been set\n");

        return -1;

    }

    /* create CM channel */

    rdma->channel = rdma_create_event_channel();

    if (!rdma->channel) {

        ERROR(errp, "could not create CM channel\n");

        return -1;

    }

    /* create CM id */

    ret = rdma_create_id(rdma->channel, &rdma->cm_id, NULL, RDMA_PS_TCP);

    if (ret) {

        ERROR(errp, "could not create channel id\n");

        goto err_resolve_create_id;

    }

    snprintf(port_str, 16, "%d", rdma->port);

    port_str[15] = '\0';

    ret = getaddrinfo(rdma->host, port_str, NULL, &res);

    if (ret < 0) {

        ERROR(errp, "could not getaddrinfo address %s\n", rdma->host);

        goto err_resolve_get_addr;

    }

    inet_ntop(AF_INET, &((struct sockaddr_in *) res->ai_addr)->sin_addr,

                                ip, sizeof ip);

    DPRINTF("%s => %s\n", rdma->host, ip);

    /* resolve the first address */

    ret = rdma_resolve_addr(rdma->cm_id, NULL, res->ai_addr,

            RDMA_RESOLVE_TIMEOUT_MS);

    if (ret) {

        ERROR(errp, "could not resolve address %s\n", rdma->host);

        goto err_resolve_get_addr;

    }

    qemu_rdma_dump_gid("source_resolve_addr", rdma->cm_id);

    ret = rdma_get_cm_event(rdma->channel, &cm_event);

    if (ret) {

        ERROR(errp, "could not perform event_addr_resolved\n");

        goto err_resolve_get_addr;

    }

    if (cm_event->event != RDMA_CM_EVENT_ADDR_RESOLVED) {

        ERROR(errp, "result not equal to event_addr_resolved %s\n",

                rdma_event_str(cm_event->event));

        perror("rdma_resolve_addr");

        goto err_resolve_get_addr;

    }

    rdma_ack_cm_event(cm_event);

    /* resolve route */

    ret = rdma_resolve_route(rdma->cm_id, RDMA_RESOLVE_TIMEOUT_MS);

    if (ret) {

        ERROR(errp, "could not resolve rdma route\n");

        goto err_resolve_get_addr;

    }

    ret = rdma_get_cm_event(rdma->channel, &cm_event);

    if (ret) {

        ERROR(errp, "could not perform event_route_resolved\n");

        goto err_resolve_get_addr;

    }

    if (cm_event->event != RDMA_CM_EVENT_ROUTE_RESOLVED) {

        ERROR(errp, "result not equal to event_route_resolved: %s\n",

                        rdma_event_str(cm_event->event));

        rdma_ack_cm_event(cm_event);

        goto err_resolve_get_addr;

    }

    rdma_ack_cm_event(cm_event);

    rdma->verbs = rdma->cm_id->verbs;

    qemu_rdma_dump_id("source_resolve_host", rdma->cm_id->verbs);

    qemu_rdma_dump_gid("source_resolve_host", rdma->cm_id);

    return 0;

err_resolve_get_addr:

    rdma_destroy_id(rdma->cm_id);

    rdma->cm_id = NULL;

err_resolve_create_id:

    rdma_destroy_event_channel(rdma->channel);

    rdma->channel = NULL;

    return -1;

}

/*

 * Create protection domain and completion queues

 */

static int qemu_rdma_alloc_pd_cq(RDMAContext *rdma)

{

    /* allocate pd */

    rdma->pd = ibv_alloc_pd(rdma->verbs);

    if (!rdma->pd) {

        fprintf(stderr, "failed to allocate protection domain\n");

        return -1;

    }

    /* create completion channel */

    rdma->comp_channel = ibv_create_comp_channel(rdma->verbs);

    if (!rdma->comp_channel) {

        fprintf(stderr, "failed to allocate completion channel\n");

        goto err_alloc_pd_cq;

    }

    /*

     * Completion queue can be filled by both read and write work requests,

     * so must reflect the sum of both possible queue sizes.

     */

    rdma->cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3),

            NULL, rdma->comp_channel, 0);

    if (!rdma->cq) {

        fprintf(stderr, "failed to allocate completion queue\n");

        goto err_alloc_pd_cq;

    }

    return 0;

err_alloc_pd_cq:

    if (rdma->pd) {

        ibv_dealloc_pd(rdma->pd);

    }

    if (rdma->comp_channel) {

        ibv_destroy_comp_channel(rdma->comp_channel);

    }

    rdma->pd = NULL;

    rdma->comp_channel = NULL;