text: /*
   * 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/error-report.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>
  #include "trace.h"
  
  /*
   * Print and error on both the Monitor and the Log file.
   */
  #define ERROR(errp, fmt, ...) \
      do { \
          fprintf(stderr, "RDMA ERROR: " fmt "\n", ## __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) { \
                  error_report("RDMA is in an error state waiting migration" \
                                  " to abort!"); \
                  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,
  };
  
  static 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 */
  };
  
  static 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 RDMADestBlock {
      uint64_t remote_host_addr;
      uint64_t offset;
      uint64_t length;
      uint32_t remote_rkey;
      uint32_t padding;
  } RDMADestBlock;
  
  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 dest_block_to_network(RDMADestBlock *db)
  {
      db->remote_host_addr = htonll(db->remote_host_addr);
      db->offset = htonll(db->offset);
      db->length = htonll(db->length);
      db->remote_rkey = htonl(db->remote_rkey);
  }
  
  static void network_to_dest_block(RDMADestBlock *db)
  {
      db->remote_host_addr = ntohll(db->remote_host_addr);
      db->offset = ntohll(db->offset);
      db->length = ntohll(db->length);
      db->remote_rkey = ntohl(db->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];
  
      /*
       * 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;
      bool connected;
  
      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;
      RDMADestBlock  *dest_blocks;
  
      /*
       * 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(const uint8_t *start,
                                         const uint8_t *host)
  {
      return ((uintptr_t) host - (uintptr_t) start) >> RDMA_REG_CHUNK_SHIFT;
  }
  
  static inline uint8_t *ram_chunk_start(const 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(const 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 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 *)(uintptr_t)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 *)(uintptr_t)old[x].offset);
              g_hash_table_insert(rdma->blockmap,
                                  (void *)(uintptr_t)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);
  
      trace_rdma_add_block(local->nb_blocks, (uintptr_t) block->local_host_addr,
                           block->offset, block->length,
                           (uintptr_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 int qemu_rdma_init_one_block(const char *block_name, void *host_addr,
      ram_addr_t block_offset, ram_addr_t length, void *opaque)
  {
      return 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);
      trace_qemu_rdma_init_ram_blocks(local->nb_blocks);
      rdma->dest_blocks = (RDMADestBlock *) g_malloc0(sizeof(RDMADestBlock) *
                          rdma->local_ram_blocks.nb_blocks);
      local->init = true;
      return 0;
  }
  
  static int 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 *)(uintptr_t)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;
      }
  
      trace_rdma_delete_block(local->nb_blocks,
                             (uintptr_t)block->local_host_addr,
                             block->offset, block->length,
                              (uintptr_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 *)(uintptr_t)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)
  {
      struct ibv_port_attr port;
  
      if (ibv_query_port(verbs, 1, &port)) {
          error_report("Failed to query port information");
          return;
      }
  
      printf("%s RDMA Device opened: kernel name %s "
             "uverbs device name %s, "
             "infiniband_verbs class device path %s, "
             "infiniband class device path %s, "
             "transport: (%d) %s\n",
                  who,
                  verbs->device->name,
                  verbs->device->dev_name,
                  verbs->device->dev_path,
                  verbs->device->ibdev_path,
                  port.link_layer,
                  (port.link_layer == IBV_LINK_LAYER_INFINIBAND) ? "Infiniband" :
                   ((port.link_layer == IBV_LINK_LAYER_ETHERNET)
                      ? "Ethernet" : "Unknown"));
  }
  
  /*
   * 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);
      trace_qemu_rdma_dump_gid(who, sgid, dgid);
  }
  
  /*
   * As of now, IPv6 over RoCE / iWARP is not supported by linux.
   * We will try the next addrinfo struct, and fail if there are
   * no other valid addresses to bind against.
   *
   * If user is listening on '[::]', then we will not have a opened a device
   * yet and have no way of verifying if the device is RoCE or not.
   *
   * In this case, the source VM will throw an error for ALL types of
   * connections (both IPv4 and IPv6) if the destination machine does not have
   * a regular infiniband network available for use.
   *
   * The only way to guarantee that an error is thrown for broken kernels is
   * for the management software to choose a *specific* interface at bind time
   * and validate what time of hardware it is.
   *
   * Unfortunately, this puts the user in a fix:
   *
   *  If the source VM connects with an IPv4 address without knowing that the
   *  destination has bound to '[::]' the migration will unconditionally fail
   *  unless the management software is explicitly listening on the the IPv4
   *  address while using a RoCE-based device.
   *
   *  If the source VM connects with an IPv6 address, then we're OK because we can
   *  throw an error on the source (and similarly on the destination).
   *
   *  But in mixed environments, this will be broken for a while until it is fixed
   *  inside linux.
   *
   * We do provide a *tiny* bit of help in this function: We can list all of the
   * devices in the system and check to see if all the devices are RoCE or
   * Infiniband.
   *
   * If we detect that we have a *pure* RoCE environment, then we can safely
   * thrown an error even if the management software has specified '[::]' as the
   * bind address.
   *
   * However, if there is are multiple hetergeneous devices, then we cannot make
   * this assumption and the user just has to be sure they know what they are
   * doing.
   *
   * Patches are being reviewed on linux-rdma.
   */
  static int qemu_rdma_broken_ipv6_kernel(Error **errp, struct ibv_context *verbs)
  {
      struct ibv_port_attr port_attr;
  
      /* This bug only exists in linux, to our knowledge. */
  #ifdef CONFIG_LINUX
  
      /*
       * Verbs are only NULL if management has bound to '[::]'.
       *
       * Let's iterate through all the devices and see if there any pure IB
       * devices (non-ethernet).
       *
       * If not, then we can safely proceed with the migration.
       * Otherwise, there are no guarantees until the bug is fixed in linux.
       */
      if (!verbs) {
          int num_devices, x;
          struct ibv_device ** dev_list = ibv_get_device_list(&num_devices);
          bool roce_found = false;
          bool ib_found = false;
  
          for (x = 0; x < num_devices; x++) {
              verbs = ibv_open_device(dev_list[x]);
              if (!verbs) {
                  if (errno == EPERM) {
                      continue;
                  } else {
                      return -EINVAL;
                  }
              }
  
              if (ibv_query_port(verbs, 1, &port_attr)) {
                  ibv_close_device(verbs);
                  ERROR(errp, "Could not query initial IB port");
                  return -EINVAL;
              }
  
              if (port_attr.link_layer == IBV_LINK_LAYER_INFINIBAND) {
                  ib_found = true;
              } else if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
                  roce_found = true;
              }
  
              ibv_close_device(verbs);
  
          }
  
          if (roce_found) {
              if (ib_found) {
                  fprintf(stderr, "WARN: migrations may fail:"
                                  " IPv6 over RoCE / iWARP in linux"
                                  " is broken. But since you appear to have a"
                                  " mixed RoCE / IB environment, be sure to only"
                                  " migrate over the IB fabric until the kernel "
                                  " fixes the bug.\n");
              } else {
                  ERROR(errp, "You only have RoCE / iWARP devices in your systems"
                              " and your management software has specified '[::]'"
                              ", but IPv6 over RoCE / iWARP is not supported in Linux.");
                  return -ENONET;
              }
          }
  
          return 0;
      }
  
      /*
       * If we have a verbs context, that means that some other than '[::]' was
       * used by the management software for binding. In which case we can
       * actually warn the user about a potentially broken kernel.
       */
  
      /* IB ports start with 1, not 0 */
      if (ibv_query_port(verbs, 1, &port_attr)) {
          ERROR(errp, "Could not query initial IB port");
          return -EINVAL;
      }
  
      if (port_attr.link_layer == IBV_LINK_LAYER_ETHERNET) {
          ERROR(errp, "Linux kernel's RoCE / iWARP does not support IPv6 "
                      "(but patches on linux-rdma in progress)");
          return -ENONET;
      }
  
  #endif
  
      return 0;
  }
  
  /*
   * 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 rdma_addrinfo *res;
      char port_str[16];
      struct rdma_cm_event *cm_event;
      char ip[40] = "unknown";
      struct rdma_addrinfo *e;
  
      if (rdma->host == NULL || !strcmp(rdma->host, "")) {
          ERROR(errp, "RDMA hostname has not been set");
          return -EINVAL;
      }
  
      /* create CM channel */
      rdma->channel = rdma_create_event_channel();
      if (!rdma->channel) {
          ERROR(errp, "could not create CM channel");
          return -EINVAL;
      }
  
      /* 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");
          goto err_resolve_create_id;
      }
  
      snprintf(port_str, 16, "%d", rdma->port);
      port_str[15] = '\0';
  
      ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
      if (ret < 0) {
          ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
          goto err_resolve_get_addr;
      }
  
      for (e = res; e != NULL; e = e->ai_next) {
          inet_ntop(e->ai_family,
              &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
          trace_qemu_rdma_resolve_host_trying(rdma->host, ip);
  
          ret = rdma_resolve_addr(rdma->cm_id, NULL, e->ai_dst_addr,
                  RDMA_RESOLVE_TIMEOUT_MS);
          if (!ret) {
              if (e->ai_family == AF_INET6) {
                  ret = qemu_rdma_broken_ipv6_kernel(errp, rdma->cm_id->verbs);
                  if (ret) {
                      continue;
                  }
              }
              goto route;
          }
      }
  
      ERROR(errp, "could not resolve address %s", rdma->host);
      goto err_resolve_get_addr;
  
  route:
      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");
          goto err_resolve_get_addr;
      }
  
      if (cm_event->event != RDMA_CM_EVENT_ADDR_RESOLVED) {
          ERROR(errp, "result not equal to event_addr_resolved %s",
                  rdma_event_str(cm_event->event));
          perror("rdma_resolve_addr");
          rdma_ack_cm_event(cm_event);
          ret = -EINVAL;
          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");
          goto err_resolve_get_addr;
      }
  
      ret = rdma_get_cm_event(rdma->channel, &cm_event);
      if (ret) {
          ERROR(errp, "could not perform event_route_resolved");
          goto err_resolve_get_addr;
      }
      if (cm_event->event != RDMA_CM_EVENT_ROUTE_RESOLVED) {
          ERROR(errp, "result not equal to event_route_resolved: %s",
                          rdma_event_str(cm_event->event));
          rdma_ack_cm_event(cm_event);
          ret = -EINVAL;
          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 ret;
  }
  
  /*
   * 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) {
          error_report("failed to allocate protection domain");
          return -1;
      }
  
      /* create completion channel */
      rdma->comp_channel = ibv_create_comp_channel(rdma->verbs);
      if (!rdma->comp_channel) {
          error_report("failed to allocate completion channel");
          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) {
          error_report("failed to allocate completion queue");
          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;
      return -1;
  
  }
  
  /*
   * Create queue pairs.
   */
  static int qemu_rdma_alloc_qp(RDMAContext *rdma)
  {
      struct ibv_qp_init_attr attr = { 0 };
      int ret;
  
      attr.cap.max_send_wr = RDMA_SIGNALED_SEND_MAX;
      attr.cap.max_recv_wr = 3;
      attr.cap.max_send_sge = 1;
      attr.cap.max_recv_sge = 1;
      attr.send_cq = rdma->cq;
      attr.recv_cq = rdma->cq;
      attr.qp_type = IBV_QPT_RC;
  
      ret = rdma_create_qp(rdma->cm_id, rdma->pd, &attr);
      if (ret) {
          return -1;
      }
  
      rdma->qp = rdma->cm_id->qp;
      return 0;
  }
  
  static int qemu_rdma_reg_whole_ram_blocks(RDMAContext *rdma)
  {
      int i;
      RDMALocalBlocks *local = &rdma->local_ram_blocks;
  
      for (i = 0; i < local->nb_blocks; i++) {
          local->block[i].mr =
              ibv_reg_mr(rdma->pd,
                      local->block[i].local_host_addr,
                      local->block[i].length,
                      IBV_ACCESS_LOCAL_WRITE |
                      IBV_ACCESS_REMOTE_WRITE
                      );
          if (!local->block[i].mr) {
              perror("Failed to register local dest ram block!\n");
              break;
          }
          rdma->total_registrations++;
      }
  
      if (i >= local->nb_blocks) {
          return 0;
      }
  
      for (i--; i >= 0; i--) {
          ibv_dereg_mr(local->block[i].mr);
          rdma->total_registrations--;
      }
  
      return -1;
  
  }
  
  /*
   * Find the ram block that corresponds to the page requested to be
   * transmitted by QEMU.
   *
   * Once the block is found, also identify which 'chunk' within that
   * block that the page belongs to.
   *
   * This search cannot fail or the migration will fail.
   */
  static int qemu_rdma_search_ram_block(RDMAContext *rdma,
                                        uintptr_t block_offset,
                                        uint64_t offset,
                                        uint64_t length,
                                        uint64_t *block_index,
                                        uint64_t *chunk_index)
  {
      uint64_t current_addr = block_offset + offset;
      RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
                                                  (void *) block_offset);
      assert(block);
      assert(current_addr >= block->offset);
      assert((current_addr + length) <= (block->offset + block->length));
  
      *block_index = block->index;
      *chunk_index = ram_chunk_index(block->local_host_addr,
                  block->local_host_addr + (current_addr - block->offset));
  
      return 0;
  }
  
  /*
   * Register a chunk with IB. If the chunk was already registered
   * previously, then skip.
   *
   * Also return the keys associated with the registration needed
   * to perform the actual RDMA operation.
   */
  static int qemu_rdma_register_and_get_keys(RDMAContext *rdma,
          RDMALocalBlock *block, uintptr_t host_addr,
          uint32_t *lkey, uint32_t *rkey, int chunk,
          uint8_t *chunk_start, uint8_t *chunk_end)
  {
      if (block->mr) {
          if (lkey) {
              *lkey = block->mr->lkey;
          }
          if (rkey) {
              *rkey = block->mr->rkey;
          }
          return 0;
      }
  
      /* allocate memory to store chunk MRs */
      if (!block->pmr) {
          block->pmr = g_malloc0(block->nb_chunks * sizeof(struct ibv_mr *));
      }
  
      /*
       * If 'rkey', then we're the destination, so grant access to the source.
       *
       * If 'lkey', then we're the source VM, so grant access only to ourselves.
       */
      if (!block->pmr[chunk]) {
          uint64_t len = chunk_end - chunk_start;
  
          trace_qemu_rdma_register_and_get_keys(len, chunk_start);
  
          block->pmr[chunk] = ibv_reg_mr(rdma->pd,
                  chunk_start, len,
                  (rkey ? (IBV_ACCESS_LOCAL_WRITE |
                          IBV_ACCESS_REMOTE_WRITE) : 0));
  
          if (!block->pmr[chunk]) {
              perror("Failed to register chunk!");
              fprintf(stderr, "Chunk details: block: %d chunk index %d"
                              " start %" PRIuPTR " end %" PRIuPTR
                              " host %" PRIuPTR
                              " local %" PRIuPTR " registrations: %d\n",
                              block->index, chunk, (uintptr_t)chunk_start,
                              (uintptr_t)chunk_end, host_addr,
                              (uintptr_t)block->local_host_addr,
                              rdma->total_registrations);
              return -1;
          }
          rdma->total_registrations++;
      }
  
      if (lkey) {
          *lkey = block->pmr[chunk]->lkey;
      }
      if (rkey) {
          *rkey = block->pmr[chunk]->rkey;
      }
      return 0;
  }
  
  /*
   * Register (at connection time) the memory used for control
   * channel messages.
   */
  static int qemu_rdma_reg_control(RDMAContext *rdma, int idx)
  {
      rdma->wr_data[idx].control_mr = ibv_reg_mr(rdma->pd,
              rdma->wr_data[idx].control, RDMA_CONTROL_MAX_BUFFER,
              IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
      if (rdma->wr_data[idx].control_mr) {
          rdma->total_registrations++;
          return 0;
      }
      error_report("qemu_rdma_reg_control failed");
      return -1;
  }
  
  const char *print_wrid(int wrid)
  {
      if (wrid >= RDMA_WRID_RECV_CONTROL) {
          return wrid_desc[RDMA_WRID_RECV_CONTROL];
      }
      return wrid_desc[wrid];
  }
  
  /*
   * RDMA requires memory registration (mlock/pinning), but this is not good for
   * overcommitment.
   *
   * In preparation for the future where LRU information or workload-specific
   * writable writable working set memory access behavior is available to QEMU
   * it would be nice to have in place the ability to UN-register/UN-pin
   * particular memory regions from the RDMA hardware when it is determine that
   * those regions of memory will likely not be accessed again in the near future.
   *
   * While we do not yet have such information right now, the following
   * compile-time option allows us to perform a non-optimized version of this
   * behavior.
   *
   * By uncommenting this option, you will cause *all* RDMA transfers to be
   * unregistered immediately after the transfer completes on both sides of the
   * connection. This has no effect in 'rdma-pin-all' mode, only regular mode.
   *
   * This will have a terrible impact on migration performance, so until future
   * workload information or LRU information is available, do not attempt to use
   * this feature except for basic testing.
   */
  //#define RDMA_UNREGISTRATION_EXAMPLE
  
  /*
   * Perform a non-optimized memory unregistration after every transfer
   * for demonsration purposes, only if pin-all is not requested.
   *
   * Potential optimizations:
   * 1. Start a new thread to run this function continuously
          - for bit clearing
          - and for receipt of unregister messages
   * 2. Use an LRU.
   * 3. Use workload hints.
   */
  static int qemu_rdma_unregister_waiting(RDMAContext *rdma)
  {
      while (rdma->unregistrations[rdma->unregister_current]) {
          int ret;
          uint64_t wr_id = rdma->unregistrations[rdma->unregister_current];
          uint64_t chunk =
              (wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
          uint64_t index =
              (wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
          RDMALocalBlock *block =
              &(rdma->local_ram_blocks.block[index]);
          RDMARegister reg = { .current_index = index };
          RDMAControlHeader resp = { .type = RDMA_CONTROL_UNREGISTER_FINISHED,
                                   };
          RDMAControlHeader head = { .len = sizeof(RDMARegister),
                                     .type = RDMA_CONTROL_UNREGISTER_REQUEST,
                                     .repeat = 1,
                                   };
  
          trace_qemu_rdma_unregister_waiting_proc(chunk,
                                                  rdma->unregister_current);
  
          rdma->unregistrations[rdma->unregister_current] = 0;
          rdma->unregister_current++;
  
          if (rdma->unregister_current == RDMA_SIGNALED_SEND_MAX) {
              rdma->unregister_current = 0;
          }
  
  
          /*
           * Unregistration is speculative (because migration is single-threaded
           * and we cannot break the protocol's inifinband message ordering).
           * Thus, if the memory is currently being used for transmission,
           * then abort the attempt to unregister and try again
           * later the next time a completion is received for this memory.
           */
          clear_bit(chunk, block->unregister_bitmap);
  
          if (test_bit(chunk, block->transit_bitmap)) {
              trace_qemu_rdma_unregister_waiting_inflight(chunk);
              continue;
          }
  
          trace_qemu_rdma_unregister_waiting_send(chunk);
  
          ret = ibv_dereg_mr(block->pmr[chunk]);
          block->pmr[chunk] = NULL;
          block->remote_keys[chunk] = 0;
  
          if (ret != 0) {
              perror("unregistration chunk failed");
              return -ret;
          }
          rdma->total_registrations--;
  
          reg.key.chunk = chunk;
          register_to_network(&reg);
          ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) &reg,
                                  &resp, NULL, NULL);
          if (ret < 0) {
              return ret;
          }
  
          trace_qemu_rdma_unregister_waiting_complete(chunk);
      }
  
      return 0;
  }
  
  static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index,
                                           uint64_t chunk)
  {
      uint64_t result = wr_id & RDMA_WRID_TYPE_MASK;
  
      result |= (index << RDMA_WRID_BLOCK_SHIFT);
      result |= (chunk << RDMA_WRID_CHUNK_SHIFT);
  
      return result;
  }
  
  /*
   * Set bit for unregistration in the next iteration.
   * We cannot transmit right here, but will unpin later.
   */
  static void qemu_rdma_signal_unregister(RDMAContext *rdma, uint64_t index,
                                          uint64_t chunk, uint64_t wr_id)
  {
      if (rdma->unregistrations[rdma->unregister_next] != 0) {
          error_report("rdma migration: queue is full");
      } else {
          RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
  
          if (!test_and_set_bit(chunk, block->unregister_bitmap)) {
              trace_qemu_rdma_signal_unregister_append(chunk,
                                                       rdma->unregister_next);
  
              rdma->unregistrations[rdma->unregister_next++] =
                      qemu_rdma_make_wrid(wr_id, index, chunk);
  
              if (rdma->unregister_next == RDMA_SIGNALED_SEND_MAX) {
                  rdma->unregister_next = 0;
              }
          } else {
              trace_qemu_rdma_signal_unregister_already(chunk);
          }
      }
  }
  
  /*
   * Consult the connection manager to see a work request
   * (of any kind) has completed.
   * Return the work request ID that completed.
   */
  static uint64_t qemu_rdma_poll(RDMAContext *rdma, uint64_t *wr_id_out,
                                 uint32_t *byte_len)
  {
      int ret;
      struct ibv_wc wc;
      uint64_t wr_id;
  
      ret = ibv_poll_cq(rdma->cq, 1, &wc);
  
      if (!ret) {
          *wr_id_out = RDMA_WRID_NONE;
          return 0;
      }
  
      if (ret < 0) {
          error_report("ibv_poll_cq return %d", ret);
          return ret;
      }
  
      wr_id = wc.wr_id & RDMA_WRID_TYPE_MASK;
  
      if (wc.status != IBV_WC_SUCCESS) {
          fprintf(stderr, "ibv_poll_cq wc.status=%d %s!\n",
                          wc.status, ibv_wc_status_str(wc.status));
          fprintf(stderr, "ibv_poll_cq wrid=%s!\n", wrid_desc[wr_id]);
  
          return -1;
      }
  
      if (rdma->control_ready_expected &&
          (wr_id >= RDMA_WRID_RECV_CONTROL)) {
          trace_qemu_rdma_poll_recv(wrid_desc[RDMA_WRID_RECV_CONTROL],
                    wr_id - RDMA_WRID_RECV_CONTROL, wr_id, rdma->nb_sent);
          rdma->control_ready_expected = 0;
      }
  
      if (wr_id == RDMA_WRID_RDMA_WRITE) {
          uint64_t chunk =
              (wc.wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
          uint64_t index =
              (wc.wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
          RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
  
          trace_qemu_rdma_poll_write(print_wrid(wr_id), wr_id, rdma->nb_sent,
                                     index, chunk, block->local_host_addr,
                                     (void *)(uintptr_t)block->remote_host_addr);
  
          clear_bit(chunk, block->transit_bitmap);
  
          if (rdma->nb_sent > 0) {
              rdma->nb_sent--;
          }
  
          if (!rdma->pin_all) {
              /*
               * FYI: If one wanted to signal a specific chunk to be unregistered
               * using LRU or workload-specific information, this is the function
               * you would call to do so. That chunk would then get asynchronously
               * unregistered later.
               */
  #ifdef RDMA_UNREGISTRATION_EXAMPLE
              qemu_rdma_signal_unregister(rdma, index, chunk, wc.wr_id);
  #endif
          }
      } else {
          trace_qemu_rdma_poll_other(print_wrid(wr_id), wr_id, rdma->nb_sent);
      }
  
      *wr_id_out = wc.wr_id;
      if (byte_len) {
          *byte_len = wc.byte_len;
      }
  
      return  0;
  }
  
  /*
   * Block until the next work request has completed.
   *
   * First poll to see if a work request has already completed,
   * otherwise block.
   *
   * If we encounter completed work requests for IDs other than
   * the one we're interested in, then that's generally an error.
   *
   * The only exception is actual RDMA Write completions. These
   * completions only need to be recorded, but do not actually
   * need further processing.
   */
  static int qemu_rdma_block_for_wrid(RDMAContext *rdma, int wrid_requested,
                                      uint32_t *byte_len)
  {
      int num_cq_events = 0, ret = 0;
      struct ibv_cq *cq;
      void *cq_ctx;
      uint64_t wr_id = RDMA_WRID_NONE, wr_id_in;
  
      if (ibv_req_notify_cq(rdma->cq, 0)) {
          return -1;
      }
      /* poll cq first */
      while (wr_id != wrid_requested) {
          ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
          if (ret < 0) {
              return ret;
          }
  
          wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
  
          if (wr_id == RDMA_WRID_NONE) {
              break;
          }
          if (wr_id != wrid_requested) {
              trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested),
                         wrid_requested, print_wrid(wr_id), wr_id);
          }
      }
  
      if (wr_id == wrid_requested) {
          return 0;
      }
  
      while (1) {
          /*
           * Coroutine doesn't start until process_incoming_migration()
           * so don't yield unless we know we're running inside of a coroutine.
           */
          if (rdma->migration_started_on_destination) {
              yield_until_fd_readable(rdma->comp_channel->fd);
          }
  
          if (ibv_get_cq_event(rdma->comp_channel, &cq, &cq_ctx)) {
              perror("ibv_get_cq_event");
              goto err_block_for_wrid;
          }
  
          num_cq_events++;
  
          if (ibv_req_notify_cq(cq, 0)) {
              goto err_block_for_wrid;
          }
  
          while (wr_id != wrid_requested) {
              ret = qemu_rdma_poll(rdma, &wr_id_in, byte_len);
              if (ret < 0) {
                  goto err_block_for_wrid;
              }
  
              wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
  
              if (wr_id == RDMA_WRID_NONE) {
                  break;
              }
              if (wr_id != wrid_requested) {
                  trace_qemu_rdma_block_for_wrid_miss(print_wrid(wrid_requested),
                                     wrid_requested, print_wrid(wr_id), wr_id);
              }
          }
  
          if (wr_id == wrid_requested) {
              goto success_block_for_wrid;
          }
      }
  
  success_block_for_wrid:
      if (num_cq_events) {
          ibv_ack_cq_events(cq, num_cq_events);
      }
      return 0;
  
  err_block_for_wrid:
      if (num_cq_events) {
          ibv_ack_cq_events(cq, num_cq_events);
      }
      return ret;
  }
  
  /*
   * Post a SEND message work request for the control channel
   * containing some data and block until the post completes.
   */
  static int qemu_rdma_post_send_control(RDMAContext *rdma, uint8_t *buf,
                                         RDMAControlHeader *head)
  {
      int ret = 0;
      RDMAWorkRequestData *wr = &rdma->wr_data[RDMA_WRID_CONTROL];
      struct ibv_send_wr *bad_wr;
      struct ibv_sge sge = {
                             .addr = (uintptr_t)(wr->control),
                             .length = head->len + sizeof(RDMAControlHeader),
                             .lkey = wr->control_mr->lkey,
                           };
      struct ibv_send_wr send_wr = {
                                     .wr_id = RDMA_WRID_SEND_CONTROL,
                                     .opcode = IBV_WR_SEND,
                                     .send_flags = IBV_SEND_SIGNALED,
                                     .sg_list = &sge,
                                     .num_sge = 1,
                                  };
  
      trace_qemu_rdma_post_send_control(control_desc[head->type]);
  
      /*
       * We don't actually need to do a memcpy() in here if we used
       * the "sge" properly, but since we're only sending control messages
       * (not RAM in a performance-critical path), then its OK for now.
       *
       * The copy makes the RDMAControlHeader simpler to manipulate
       * for the time being.
       */
      assert(head->len <= RDMA_CONTROL_MAX_BUFFER - sizeof(*head));
      memcpy(wr->control, head, sizeof(RDMAControlHeader));
      control_to_network((void *) wr->control);
  
      if (buf) {
          memcpy(wr->control + sizeof(RDMAControlHeader), buf, head->len);
      }
  
  
      ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
  
      if (ret > 0) {
          error_report("Failed to use post IB SEND for control");
          return -ret;
      }
  
      ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_SEND_CONTROL, NULL);
      if (ret < 0) {
          error_report("rdma migration: send polling control error");
      }
  
      return ret;
  }
  
  /*
   * Post a RECV work request in anticipation of some future receipt
   * of data on the control channel.
   */
  static int qemu_rdma_post_recv_control(RDMAContext *rdma, int idx)
  {
      struct ibv_recv_wr *bad_wr;
      struct ibv_sge sge = {
                              .addr = (uintptr_t)(rdma->wr_data[idx].control),
                              .length = RDMA_CONTROL_MAX_BUFFER,
                              .lkey = rdma->wr_data[idx].control_mr->lkey,
                           };
  
      struct ibv_recv_wr recv_wr = {
                                      .wr_id = RDMA_WRID_RECV_CONTROL + idx,
                                      .sg_list = &sge,
                                      .num_sge = 1,
                                   };
  
  
      if (ibv_post_recv(rdma->qp, &recv_wr, &bad_wr)) {
          return -1;
      }
  
      return 0;
  }
  
  /*
   * Block and wait for a RECV control channel message to arrive.
   */
  static int qemu_rdma_exchange_get_response(RDMAContext *rdma,
                  RDMAControlHeader *head, int expecting, int idx)
  {
      uint32_t byte_len;
      int ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RECV_CONTROL + idx,
                                         &byte_len);
  
      if (ret < 0) {
          error_report("rdma migration: recv polling control error!");
          return ret;
      }
  
      network_to_control((void *) rdma->wr_data[idx].control);
      memcpy(head, rdma->wr_data[idx].control, sizeof(RDMAControlHeader));
  
      trace_qemu_rdma_exchange_get_response_start(control_desc[expecting]);
  
      if (expecting == RDMA_CONTROL_NONE) {
          trace_qemu_rdma_exchange_get_response_none(control_desc[head->type],
                                               head->type);
      } else if (head->type != expecting || head->type == RDMA_CONTROL_ERROR) {
          error_report("Was expecting a %s (%d) control message"
                  ", but got: %s (%d), length: %d",
                  control_desc[expecting], expecting,
                  control_desc[head->type], head->type, head->len);
          return -EIO;
      }
      if (head->len > RDMA_CONTROL_MAX_BUFFER - sizeof(*head)) {
          error_report("too long length: %d", head->len);
          return -EINVAL;
      }
      if (sizeof(*head) + head->len != byte_len) {
          error_report("Malformed length: %d byte_len %d", head->len, byte_len);
          return -EINVAL;
      }
  
      return 0;
  }
  
  /*
   * When a RECV work request has completed, the work request's
   * buffer is pointed at the header.
   *
   * This will advance the pointer to the data portion
   * of the control message of the work request's buffer that
   * was populated after the work request finished.
   */
  static void qemu_rdma_move_header(RDMAContext *rdma, int idx,
                                    RDMAControlHeader *head)
  {
      rdma->wr_data[idx].control_len = head->len;
      rdma->wr_data[idx].control_curr =
          rdma->wr_data[idx].control + sizeof(RDMAControlHeader);
  }
  
  /*
   * This is an 'atomic' high-level operation to deliver a single, unified
   * control-channel message.
   *
   * Additionally, if the user is expecting some kind of reply to this message,
   * they can request a 'resp' response message be filled in by posting an
   * additional work request on behalf of the user and waiting for an additional
   * completion.
   *
   * The extra (optional) response is used during registration to us from having
   * to perform an *additional* exchange of message just to provide a response by
   * instead piggy-backing on the acknowledgement.
   */
  static int qemu_rdma_exchange_send(RDMAContext *rdma, RDMAControlHeader *head,
                                     uint8_t *data, RDMAControlHeader *resp,
                                     int *resp_idx,
                                     int (*callback)(RDMAContext *rdma))
  {
      int ret = 0;
  
      /*
       * Wait until the dest is ready before attempting to deliver the message
       * by waiting for a READY message.
       */
      if (rdma->control_ready_expected) {
          RDMAControlHeader resp;
          ret = qemu_rdma_exchange_get_response(rdma,
                                      &resp, RDMA_CONTROL_READY, RDMA_WRID_READY);
          if (ret < 0) {
              return ret;
          }
      }
  
      /*
       * If the user is expecting a response, post a WR in anticipation of it.
       */
      if (resp) {
          ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_DATA);
          if (ret) {
              error_report("rdma migration: error posting"
                      " extra control recv for anticipated result!");
              return ret;
          }
      }
  
      /*
       * Post a WR to replace the one we just consumed for the READY message.
       */
      ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
      if (ret) {
          error_report("rdma migration: error posting first control recv!");
          return ret;
      }
  
      /*
       * Deliver the control message that was requested.
       */
      ret = qemu_rdma_post_send_control(rdma, data, head);
  
      if (ret < 0) {
          error_report("Failed to send control buffer!");
          return ret;
      }
  
      /*
       * If we're expecting a response, block and wait for it.
       */
      if (resp) {
          if (callback) {
              trace_qemu_rdma_exchange_send_issue_callback();
              ret = callback(rdma);
              if (ret < 0) {
                  return ret;
              }
          }
  
          trace_qemu_rdma_exchange_send_waiting(control_desc[resp->type]);
          ret = qemu_rdma_exchange_get_response(rdma, resp,
                                                resp->type, RDMA_WRID_DATA);
  
          if (ret < 0) {
              return ret;
          }
  
          qemu_rdma_move_header(rdma, RDMA_WRID_DATA, resp);
          if (resp_idx) {
              *resp_idx = RDMA_WRID_DATA;
          }
          trace_qemu_rdma_exchange_send_received(control_desc[resp->type]);
      }
  
      rdma->control_ready_expected = 1;
  
      return 0;
  }
  
  /*
   * This is an 'atomic' high-level operation to receive a single, unified
   * control-channel message.
   */
  static int qemu_rdma_exchange_recv(RDMAContext *rdma, RDMAControlHeader *head,
                                  int expecting)
  {
      RDMAControlHeader ready = {
                                  .len = 0,
                                  .type = RDMA_CONTROL_READY,
                                  .repeat = 1,
                                };
      int ret;
  
      /*
       * Inform the source that we're ready to receive a message.
       */
      ret = qemu_rdma_post_send_control(rdma, NULL, &ready);
  
      if (ret < 0) {
          error_report("Failed to send control buffer!");
          return ret;
      }
  
      /*
       * Block and wait for the message.
       */
      ret = qemu_rdma_exchange_get_response(rdma, head,
                                            expecting, RDMA_WRID_READY);
  
      if (ret < 0) {
          return ret;
      }
  
      qemu_rdma_move_header(rdma, RDMA_WRID_READY, head);
  
      /*
       * Post a new RECV work request to replace the one we just consumed.
       */
      ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
      if (ret) {
          error_report("rdma migration: error posting second control recv!");
          return ret;
      }
  
      return 0;
  }
  
  /*
   * Write an actual chunk of memory using RDMA.
   *
   * If we're using dynamic registration on the dest-side, we have to
   * send a registration command first.
   */
  static int qemu_rdma_write_one(QEMUFile *f, RDMAContext *rdma,
                                 int current_index, uint64_t current_addr,
                                 uint64_t length)
  {
      struct ibv_sge sge;
      struct ibv_send_wr send_wr = { 0 };
      struct ibv_send_wr *bad_wr;
      int reg_result_idx, ret, count = 0;
      uint64_t chunk, chunks;
      uint8_t *chunk_start, *chunk_end;
      RDMALocalBlock *block = &(rdma->local_ram_blocks.block[current_index]);
      RDMARegister reg;
      RDMARegisterResult *reg_result;
      RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT };
      RDMAControlHeader head = { .len = sizeof(RDMARegister),
                                 .type = RDMA_CONTROL_REGISTER_REQUEST,
                                 .repeat = 1,
                               };
  
  retry:
      sge.addr = (uintptr_t)(block->local_host_addr +
                              (current_addr - block->offset));
      sge.length = length;
  
      chunk = ram_chunk_index(block->local_host_addr,
                              (uint8_t *)(uintptr_t)sge.addr);
      chunk_start = ram_chunk_start(block, chunk);
  
      if (block->is_ram_block) {
          chunks = length / (1UL << RDMA_REG_CHUNK_SHIFT);
  
          if (chunks && ((length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
              chunks--;
          }
      } else {
          chunks = block->length / (1UL << RDMA_REG_CHUNK_SHIFT);
  
          if (chunks && ((block->length % (1UL << RDMA_REG_CHUNK_SHIFT)) == 0)) {
              chunks--;
          }
      }
  
      trace_qemu_rdma_write_one_top(chunks + 1,
                                    (chunks + 1) *
                                    (1UL << RDMA_REG_CHUNK_SHIFT) / 1024 / 1024);
  
      chunk_end = ram_chunk_end(block, chunk + chunks);
  
      if (!rdma->pin_all) {
  #ifdef RDMA_UNREGISTRATION_EXAMPLE
          qemu_rdma_unregister_waiting(rdma);
  #endif
      }
  
      while (test_bit(chunk, block->transit_bitmap)) {
          (void)count;
          trace_qemu_rdma_write_one_block(count++, current_index, chunk,
                  sge.addr, length, rdma->nb_sent, block->nb_chunks);
  
          ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
  
          if (ret < 0) {
              error_report("Failed to Wait for previous write to complete "
                      "block %d chunk %" PRIu64
                      " current %" PRIu64 " len %" PRIu64 " %d",
                      current_index, chunk, sge.addr, length, rdma->nb_sent);
              return ret;
          }
      }
  
      if (!rdma->pin_all || !block->is_ram_block) {
          if (!block->remote_keys[chunk]) {
              /*
               * This chunk has not yet been registered, so first check to see
               * if the entire chunk is zero. If so, tell the other size to
               * memset() + madvise() the entire chunk without RDMA.
               */
  
              if (can_use_buffer_find_nonzero_offset((void *)(uintptr_t)sge.addr,
                                                     length)
                     && buffer_find_nonzero_offset((void *)(uintptr_t)sge.addr,
                                                      length) == length) {
                  RDMACompress comp = {
                                          .offset = current_addr,
                                          .value = 0,
                                          .block_idx = current_index,
                                          .length = length,
                                      };
  
                  head.len = sizeof(comp);
                  head.type = RDMA_CONTROL_COMPRESS;
  
                  trace_qemu_rdma_write_one_zero(chunk, sge.length,
                                                 current_index, current_addr);
  
                  compress_to_network(&comp);
                  ret = qemu_rdma_exchange_send(rdma, &head,
                                  (uint8_t *) &comp, NULL, NULL, NULL);
  
                  if (ret < 0) {
                      return -EIO;
                  }
  
                  acct_update_position(f, sge.length, true);
  
                  return 1;
              }
  
              /*
               * Otherwise, tell other side to register.
               */
              reg.current_index = current_index;
              if (block->is_ram_block) {
                  reg.key.current_addr = current_addr;
              } else {
                  reg.key.chunk = chunk;
              }
              reg.chunks = chunks;
  
              trace_qemu_rdma_write_one_sendreg(chunk, sge.length, current_index,
                                                current_addr);
  
              register_to_network(&reg);
              ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) &reg,
                                      &resp, &reg_result_idx, NULL);
              if (ret < 0) {
                  return ret;
              }
  
              /* try to overlap this single registration with the one we sent. */
              if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
                                                  &sge.lkey, NULL, chunk,
                                                  chunk_start, chunk_end)) {
                  error_report("cannot get lkey");
                  return -EINVAL;
              }
  
              reg_result = (RDMARegisterResult *)
                      rdma->wr_data[reg_result_idx].control_curr;
  
              network_to_result(reg_result);
  
              trace_qemu_rdma_write_one_recvregres(block->remote_keys[chunk],
                                                   reg_result->rkey, chunk);
  
              block->remote_keys[chunk] = reg_result->rkey;
              block->remote_host_addr = reg_result->host_addr;
          } else {
              /* already registered before */
              if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
                                                  &sge.lkey, NULL, chunk,
                                                  chunk_start, chunk_end)) {
                  error_report("cannot get lkey!");
                  return -EINVAL;
              }
          }
  
          send_wr.wr.rdma.rkey = block->remote_keys[chunk];
      } else {
          send_wr.wr.rdma.rkey = block->remote_rkey;
  
          if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
                                                       &sge.lkey, NULL, chunk,
                                                       chunk_start, chunk_end)) {
              error_report("cannot get lkey!");
              return -EINVAL;
          }
      }
  
      /*
       * Encode the ram block index and chunk within this wrid.
       * We will use this information at the time of completion
       * to figure out which bitmap to check against and then which
       * chunk in the bitmap to look for.
       */
      send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE,
                                          current_index, chunk);
  
      send_wr.opcode = IBV_WR_RDMA_WRITE;
      send_wr.send_flags = IBV_SEND_SIGNALED;
      send_wr.sg_list = &sge;
      send_wr.num_sge = 1;
      send_wr.wr.rdma.remote_addr = block->remote_host_addr +
                                  (current_addr - block->offset);
  
      trace_qemu_rdma_write_one_post(chunk, sge.addr, send_wr.wr.rdma.remote_addr,
                                     sge.length);
  
      /*
       * ibv_post_send() does not return negative error numbers,
       * per the specification they are positive - no idea why.
       */
      ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
  
      if (ret == ENOMEM) {
          trace_qemu_rdma_write_one_queue_full();
          ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
          if (ret < 0) {
              error_report("rdma migration: failed to make "
                           "room in full send queue! %d", ret);
              return ret;
          }
  
          goto retry;
  
      } else if (ret > 0) {
          perror("rdma migration: post rdma write failed");
          return -ret;
      }
  
      set_bit(chunk, block->transit_bitmap);
      acct_update_position(f, sge.length, false);
      rdma->total_writes++;
  
      return 0;
  }
  
  /*
   * Push out any unwritten RDMA operations.
   *
   * We support sending out multiple chunks at the same time.
   * Not all of them need to get signaled in the completion queue.
   */
  static int qemu_rdma_write_flush(QEMUFile *f, RDMAContext *rdma)
  {
      int ret;
  
      if (!rdma->current_length) {
          return 0;
      }
  
      ret = qemu_rdma_write_one(f, rdma,
              rdma->current_index, rdma->current_addr, rdma->current_length);
  
      if (ret < 0) {
          return ret;
      }
  
      if (ret == 0) {
          rdma->nb_sent++;
          trace_qemu_rdma_write_flush(rdma->nb_sent);
      }
  
      rdma->current_length = 0;
      rdma->current_addr = 0;
  
      return 0;
  }
  
  static inline int qemu_rdma_buffer_mergable(RDMAContext *rdma,
                      uint64_t offset, uint64_t len)
  {
      RDMALocalBlock *block;
      uint8_t *host_addr;
      uint8_t *chunk_end;
  
      if (rdma->current_index < 0) {
          return 0;
      }
  
      if (rdma->current_chunk < 0) {
          return 0;
      }
  
      block = &(rdma->local_ram_blocks.block[rdma->current_index]);
      host_addr = block->local_host_addr + (offset - block->offset);
      chunk_end = ram_chunk_end(block, rdma->current_chunk);
  
      if (rdma->current_length == 0) {
          return 0;
      }
  
      /*
       * Only merge into chunk sequentially.
       */
      if (offset != (rdma->current_addr + rdma->current_length)) {
          return 0;
      }
  
      if (offset < block->offset) {
          return 0;
      }
  
      if ((offset + len) > (block->offset + block->length)) {
          return 0;
      }
  
      if ((host_addr + len) > chunk_end) {
          return 0;
      }
  
      return 1;
  }
  
  /*
   * We're not actually writing here, but doing three things:
   *
   * 1. Identify the chunk the buffer belongs to.
   * 2. If the chunk is full or the buffer doesn't belong to the current
   *    chunk, then start a new chunk and flush() the old chunk.
   * 3. To keep the hardware busy, we also group chunks into batches
   *    and only require that a batch gets acknowledged in the completion
   *    qeueue instead of each individual chunk.
   */
  static int qemu_rdma_write(QEMUFile *f, RDMAContext *rdma,
                             uint64_t block_offset, uint64_t offset,
                             uint64_t len)
  {
      uint64_t current_addr = block_offset + offset;
      uint64_t index = rdma->current_index;
      uint64_t chunk = rdma->current_chunk;
      int ret;
  
      /* If we cannot merge it, we flush the current buffer first. */
      if (!qemu_rdma_buffer_mergable(rdma, current_addr, len)) {
          ret = qemu_rdma_write_flush(f, rdma);
          if (ret) {
              return ret;
          }
          rdma->current_length = 0;
          rdma->current_addr = current_addr;
  
          ret = qemu_rdma_search_ram_block(rdma, block_offset,
                                           offset, len, &index, &chunk);
          if (ret) {
              error_report("ram block search failed");
              return ret;
          }
          rdma->current_index = index;
          rdma->current_chunk = chunk;
      }
  
      /* merge it */
      rdma->current_length += len;
  
      /* flush it if buffer is too large */
      if (rdma->current_length >= RDMA_MERGE_MAX) {
          return qemu_rdma_write_flush(f, rdma);
      }
  
      return 0;
  }
  
  static void qemu_rdma_cleanup(RDMAContext *rdma)
  {
      struct rdma_cm_event *cm_event;
      int ret, idx;
  
      if (rdma->cm_id && rdma->connected) {
          if (rdma->error_state) {
              RDMAControlHeader head = { .len = 0,
                                         .type = RDMA_CONTROL_ERROR,
                                         .repeat = 1,
                                       };
              error_report("Early error. Sending error.");
              qemu_rdma_post_send_control(rdma, NULL, &head);
          }
  
          ret = rdma_disconnect(rdma->cm_id);
          if (!ret) {
              trace_qemu_rdma_cleanup_waiting_for_disconnect();
              ret = rdma_get_cm_event(rdma->channel, &cm_event);
              if (!ret) {
                  rdma_ack_cm_event(cm_event);
              }
          }
          trace_qemu_rdma_cleanup_disconnect();
          rdma->connected = false;
      }
  
      g_free(rdma->dest_blocks);
      rdma->dest_blocks = NULL;
  
      for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
          if (rdma->wr_data[idx].control_mr) {
              rdma->total_registrations--;
              ibv_dereg_mr(rdma->wr_data[idx].control_mr);
          }
          rdma->wr_data[idx].control_mr = NULL;
      }
  
      if (rdma->local_ram_blocks.block) {
          while (rdma->local_ram_blocks.nb_blocks) {
              rdma_delete_block(rdma, rdma->local_ram_blocks.block->offset);
          }
      }
  
      if (rdma->qp) {
          rdma_destroy_qp(rdma->cm_id);
          rdma->qp = NULL;
      }
      if (rdma->cq) {
          ibv_destroy_cq(rdma->cq);
          rdma->cq = NULL;
      }
      if (rdma->comp_channel) {
          ibv_destroy_comp_channel(rdma->comp_channel);
          rdma->comp_channel = NULL;
      }
      if (rdma->pd) {
          ibv_dealloc_pd(rdma->pd);
          rdma->pd = NULL;
      }
      if (rdma->cm_id) {
          rdma_destroy_id(rdma->cm_id);
          rdma->cm_id = NULL;
      }
      if (rdma->listen_id) {
          rdma_destroy_id(rdma->listen_id);
          rdma->listen_id = NULL;
      }
      if (rdma->channel) {
          rdma_destroy_event_channel(rdma->channel);
          rdma->channel = NULL;
      }
      g_free(rdma->host);
      rdma->host = NULL;
  }
  
  
  static int qemu_rdma_source_init(RDMAContext *rdma, Error **errp, bool pin_all)
  {
      int ret, idx;
      Error *local_err = NULL, **temp = &local_err;
  
      /*
       * Will be validated against destination's actual capabilities
       * after the connect() completes.
       */
      rdma->pin_all = pin_all;
  
      ret = qemu_rdma_resolve_host(rdma, temp);
      if (ret) {
          goto err_rdma_source_init;
      }
  
      ret = qemu_rdma_alloc_pd_cq(rdma);
      if (ret) {
          ERROR(temp, "rdma migration: error allocating pd and cq! Your mlock()"
                      " limits may be too low. Please check $ ulimit -a # and "
                      "search for 'ulimit -l' in the output");
          goto err_rdma_source_init;
      }
  
      ret = qemu_rdma_alloc_qp(rdma);
      if (ret) {
          ERROR(temp, "rdma migration: error allocating qp!");
          goto err_rdma_source_init;
      }
  
      ret = qemu_rdma_init_ram_blocks(rdma);
      if (ret) {
          ERROR(temp, "rdma migration: error initializing ram blocks!");
          goto err_rdma_source_init;
      }
  
      for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
          ret = qemu_rdma_reg_control(rdma, idx);
          if (ret) {
              ERROR(temp, "rdma migration: error registering %d control!",
                                                              idx);
              goto err_rdma_source_init;
          }
      }
  
      return 0;
  
  err_rdma_source_init:
      error_propagate(errp, local_err);
      qemu_rdma_cleanup(rdma);
      return -1;
  }
  
  static int qemu_rdma_connect(RDMAContext *rdma, Error **errp)
  {
      RDMACapabilities cap = {
                                  .version = RDMA_CONTROL_VERSION_CURRENT,
                                  .flags = 0,
                             };
      struct rdma_conn_param conn_param = { .initiator_depth = 2,
                                            .retry_count = 5,
                                            .private_data = &cap,
                                            .private_data_len = sizeof(cap),
                                          };
      struct rdma_cm_event *cm_event;
      int ret;
  
      /*
       * Only negotiate the capability with destination if the user
       * on the source first requested the capability.
       */
      if (rdma->pin_all) {
          trace_qemu_rdma_connect_pin_all_requested();
          cap.flags |= RDMA_CAPABILITY_PIN_ALL;
      }
  
      caps_to_network(&cap);
  
      ret = rdma_connect(rdma->cm_id, &conn_param);
      if (ret) {
          perror("rdma_connect");
          ERROR(errp, "connecting to destination!");
          goto err_rdma_source_connect;
      }
  
      ret = rdma_get_cm_event(rdma->channel, &cm_event);
      if (ret) {
          perror("rdma_get_cm_event after rdma_connect");
          ERROR(errp, "connecting to destination!");
          rdma_ack_cm_event(cm_event);
          goto err_rdma_source_connect;
      }
  
      if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
          perror("rdma_get_cm_event != EVENT_ESTABLISHED after rdma_connect");
          ERROR(errp, "connecting to destination!");
          rdma_ack_cm_event(cm_event);
          goto err_rdma_source_connect;
      }
      rdma->connected = true;
  
      memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
      network_to_caps(&cap);
  
      /*
       * Verify that the *requested* capabilities are supported by the destination
       * and disable them otherwise.
       */
      if (rdma->pin_all && !(cap.flags & RDMA_CAPABILITY_PIN_ALL)) {
          ERROR(errp, "Server cannot support pinning all memory. "
                          "Will register memory dynamically.");
          rdma->pin_all = false;
      }
  
      trace_qemu_rdma_connect_pin_all_outcome(rdma->pin_all);
  
      rdma_ack_cm_event(cm_event);
  
      ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
      if (ret) {
          ERROR(errp, "posting second control recv!");
          goto err_rdma_source_connect;
      }
  
      rdma->control_ready_expected = 1;
      rdma->nb_sent = 0;
      return 0;
  
  err_rdma_source_connect:
      qemu_rdma_cleanup(rdma);
      return -1;
  }
  
  static int qemu_rdma_dest_init(RDMAContext *rdma, Error **errp)
  {
      int ret, idx;
      struct rdma_cm_id *listen_id;
      char ip[40] = "unknown";
      struct rdma_addrinfo *res, *e;
      char port_str[16];
  
      for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
          rdma->wr_data[idx].control_len = 0;
          rdma->wr_data[idx].control_curr = NULL;
      }
  
      if (!rdma->host || !rdma->host[0]) {
          ERROR(errp, "RDMA host is not set!");
          rdma->error_state = -EINVAL;
          return -1;
      }
      /* create CM channel */
      rdma->channel = rdma_create_event_channel();
      if (!rdma->channel) {
          ERROR(errp, "could not create rdma event channel");
          rdma->error_state = -EINVAL;
          return -1;
      }
  
      /* create CM id */
      ret = rdma_create_id(rdma->channel, &listen_id, NULL, RDMA_PS_TCP);
      if (ret) {
          ERROR(errp, "could not create cm_id!");
          goto err_dest_init_create_listen_id;
      }
  
      snprintf(port_str, 16, "%d", rdma->port);
      port_str[15] = '\0';
  
      ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
      if (ret < 0) {
          ERROR(errp, "could not rdma_getaddrinfo address %s", rdma->host);
          goto err_dest_init_bind_addr;
      }
  
      for (e = res; e != NULL; e = e->ai_next) {
          inet_ntop(e->ai_family,
              &((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
          trace_qemu_rdma_dest_init_trying(rdma->host, ip);
          ret = rdma_bind_addr(listen_id, e->ai_dst_addr);
          if (ret) {
              continue;
          }
          if (e->ai_family == AF_INET6) {
              ret = qemu_rdma_broken_ipv6_kernel(errp, listen_id->verbs);
              if (ret) {
                  continue;
              }
          }
          break;
      }
  
      if (!e) {
          ERROR(errp, "Error: could not rdma_bind_addr!");
          goto err_dest_init_bind_addr;
      }
  
      rdma->listen_id = listen_id;
      qemu_rdma_dump_gid("dest_init", listen_id);
      return 0;
  
  err_dest_init_bind_addr:
      rdma_destroy_id(listen_id);
  err_dest_init_create_listen_id:
      rdma_destroy_event_channel(rdma->channel);
      rdma->channel = NULL;
      rdma->error_state = ret;
      return ret;
  
  }
  
  static void *qemu_rdma_data_init(const char *host_port, Error **errp)
  {
      RDMAContext *rdma = NULL;
      InetSocketAddress *addr;
  
      if (host_port) {
          rdma = g_malloc0(sizeof(RDMAContext));
          rdma->current_index = -1;
          rdma->current_chunk = -1;
  
          addr = inet_parse(host_port, NULL);
          if (addr != NULL) {
              rdma->port = atoi(addr->port);
              rdma->host = g_strdup(addr->host);
          } else {
              ERROR(errp, "bad RDMA migration address '%s'", host_port);
              g_free(rdma);
              rdma = NULL;
          }
  
          qapi_free_InetSocketAddress(addr);
      }
  
      return rdma;
  }
  
  /*
   * QEMUFile interface to the control channel.
   * SEND messages for control only.
   * VM's ram is handled with regular RDMA messages.
   */
  static int qemu_rdma_put_buffer(void *opaque, const uint8_t *buf,
                                  int64_t pos, int size)
  {
      QEMUFileRDMA *r = opaque;
      QEMUFile *f = r->file;
      RDMAContext *rdma = r->rdma;
      size_t remaining = size;
      uint8_t * data = (void *) buf;
      int ret;
  
      CHECK_ERROR_STATE();
  
      /*
       * Push out any writes that
       * we're queued up for VM's ram.
       */
      ret = qemu_rdma_write_flush(f, rdma);
      if (ret < 0) {
          rdma->error_state = ret;
          return ret;
      }
  
      while (remaining) {
          RDMAControlHeader head;
  
          r->len = MIN(remaining, RDMA_SEND_INCREMENT);
          remaining -= r->len;
  
          head.len = r->len;
          head.type = RDMA_CONTROL_QEMU_FILE;
  
          ret = qemu_rdma_exchange_send(rdma, &head, data, NULL, NULL, NULL);
  
          if (ret < 0) {
              rdma->error_state = ret;
              return ret;
          }
  
          data += r->len;
      }
  
      return size;
  }
  
  static size_t qemu_rdma_fill(RDMAContext *rdma, uint8_t *buf,
                               int size, int idx)
  {
      size_t len = 0;
  
      if (rdma->wr_data[idx].control_len) {
          trace_qemu_rdma_fill(rdma->wr_data[idx].control_len, size);
  
          len = MIN(size, rdma->wr_data[idx].control_len);
          memcpy(buf, rdma->wr_data[idx].control_curr, len);
          rdma->wr_data[idx].control_curr += len;
          rdma->wr_data[idx].control_len -= len;
      }
  
      return len;
  }
  
  /*
   * QEMUFile interface to the control channel.
   * RDMA links don't use bytestreams, so we have to
   * return bytes to QEMUFile opportunistically.
   */
  static int qemu_rdma_get_buffer(void *opaque, uint8_t *buf,
                                  int64_t pos, int size)
  {
      QEMUFileRDMA *r = opaque;
      RDMAContext *rdma = r->rdma;
      RDMAControlHeader head;
      int ret = 0;
  
      CHECK_ERROR_STATE();
  
      /*
       * First, we hold on to the last SEND message we
       * were given and dish out the bytes until we run
       * out of bytes.
       */
      r->len = qemu_rdma_fill(r->rdma, buf, size, 0);
      if (r->len) {
          return r->len;
      }
  
      /*
       * Once we run out, we block and wait for another
       * SEND message to arrive.
       */
      ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_QEMU_FILE);
  
      if (ret < 0) {
          rdma->error_state = ret;
          return ret;
      }
  
      /*
       * SEND was received with new bytes, now try again.
       */
      return qemu_rdma_fill(r->rdma, buf, size, 0);
  }
  
  /*
   * Block until all the outstanding chunks have been delivered by the hardware.
   */
  static int qemu_rdma_drain_cq(QEMUFile *f, RDMAContext *rdma)
  {
      int ret;
  
      if (qemu_rdma_write_flush(f, rdma) < 0) {
          return -EIO;
      }
  
      while (rdma->nb_sent) {
          ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
          if (ret < 0) {
              error_report("rdma migration: complete polling error!");
              return -EIO;
          }
      }
  
      qemu_rdma_unregister_waiting(rdma);
  
      return 0;
  }
  
  static int qemu_rdma_close(void *opaque)
  {
      trace_qemu_rdma_close();
      QEMUFileRDMA *r = opaque;
      if (r->rdma) {
          qemu_rdma_cleanup(r->rdma);
          g_free(r->rdma);
      }
      g_free(r);
      return 0;
  }
  
  /*
   * Parameters:
   *    @offset == 0 :
   *        This means that 'block_offset' is a full virtual address that does not
   *        belong to a RAMBlock of the virtual machine and instead
   *        represents a private malloc'd memory area that the caller wishes to
   *        transfer.
   *
   *    @offset != 0 :
   *        Offset is an offset to be added to block_offset and used
   *        to also lookup the corresponding RAMBlock.
   *
   *    @size > 0 :
   *        Initiate an transfer this size.
   *
   *    @size == 0 :
   *        A 'hint' or 'advice' that means that we wish to speculatively
   *        and asynchronously unregister this memory. In this case, there is no
   *        guarantee that the unregister will actually happen, for example,
   *        if the memory is being actively transmitted. Additionally, the memory
   *        may be re-registered at any future time if a write within the same
   *        chunk was requested again, even if you attempted to unregister it
   *        here.
   *
   *    @size < 0 : TODO, not yet supported
   *        Unregister the memory NOW. This means that the caller does not
   *        expect there to be any future RDMA transfers and we just want to clean
   *        things up. This is used in case the upper layer owns the memory and
   *        cannot wait for qemu_fclose() to occur.
   *
   *    @bytes_sent : User-specificed pointer to indicate how many bytes were
   *                  sent. Usually, this will not be more than a few bytes of
   *                  the protocol because most transfers are sent asynchronously.
   */
  static size_t qemu_rdma_save_page(QEMUFile *f, void *opaque,
                                    ram_addr_t block_offset, ram_addr_t offset,
                                    size_t size, uint64_t *bytes_sent)
  {
      QEMUFileRDMA *rfile = opaque;
      RDMAContext *rdma = rfile->rdma;
      int ret;
  
      CHECK_ERROR_STATE();
  
      qemu_fflush(f);
  
      if (size > 0) {
          /*
           * Add this page to the current 'chunk'. If the chunk
           * is full, or the page doen't belong to the current chunk,
           * an actual RDMA write will occur and a new chunk will be formed.
           */
          ret = qemu_rdma_write(f, rdma, block_offset, offset, size);
          if (ret < 0) {
              error_report("rdma migration: write error! %d", ret);
              goto err;
          }
  
          /*
           * We always return 1 bytes because the RDMA
           * protocol is completely asynchronous. We do not yet know
           * whether an  identified chunk is zero or not because we're
           * waiting for other pages to potentially be merged with
           * the current chunk. So, we have to call qemu_update_position()
           * later on when the actual write occurs.
           */
          if (bytes_sent) {
              *bytes_sent = 1;
          }
      } else {
          uint64_t index, chunk;
  
          /* TODO: Change QEMUFileOps prototype to be signed: size_t => long
          if (size < 0) {
              ret = qemu_rdma_drain_cq(f, rdma);
              if (ret < 0) {
                  fprintf(stderr, "rdma: failed to synchronously drain"
                                  " completion queue before unregistration.\n");
                  goto err;
              }
          }
          */
  
          ret = qemu_rdma_search_ram_block(rdma, block_offset,
                                           offset, size, &index, &chunk);
  
          if (ret) {
              error_report("ram block search failed");
              goto err;
          }
  
          qemu_rdma_signal_unregister(rdma, index, chunk, 0);
  
          /*
           * TODO: Synchronous, guaranteed unregistration (should not occur during
           * fast-path). Otherwise, unregisters will process on the next call to
           * qemu_rdma_drain_cq()
          if (size < 0) {
              qemu_rdma_unregister_waiting(rdma);
          }
          */
      }
  
      /*
       * Drain the Completion Queue if possible, but do not block,
       * just poll.
       *
       * If nothing to poll, the end of the iteration will do this
       * again to make sure we don't overflow the request queue.
       */
      while (1) {
          uint64_t wr_id, wr_id_in;
          int ret = qemu_rdma_poll(rdma, &wr_id_in, NULL);
          if (ret < 0) {
              error_report("rdma migration: polling error! %d", ret);
              goto err;
          }
  
          wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
  
          if (wr_id == RDMA_WRID_NONE) {
              break;
          }
      }
  
      return RAM_SAVE_CONTROL_DELAYED;
  err:
      rdma->error_state = ret;
      return ret;
  }
  
  static int qemu_rdma_accept(RDMAContext *rdma)
  {
      RDMACapabilities cap;
      struct rdma_conn_param conn_param = {
                                              .responder_resources = 2,
                                              .private_data = &cap,
                                              .private_data_len = sizeof(cap),
                                           };
      struct rdma_cm_event *cm_event;
      struct ibv_context *verbs;
      int ret = -EINVAL;
      int idx;
  
      ret = rdma_get_cm_event(rdma->channel, &cm_event);
      if (ret) {
          goto err_rdma_dest_wait;
      }
  
      if (cm_event->event != RDMA_CM_EVENT_CONNECT_REQUEST) {
          rdma_ack_cm_event(cm_event);
          goto err_rdma_dest_wait;
      }
  
      memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
  
      network_to_caps(&cap);
  
      if (cap.version < 1 || cap.version > RDMA_CONTROL_VERSION_CURRENT) {
              error_report("Unknown source RDMA version: %d, bailing...",
                              cap.version);
              rdma_ack_cm_event(cm_event);
              goto err_rdma_dest_wait;
      }
  
      /*
       * Respond with only the capabilities this version of QEMU knows about.
       */
      cap.flags &= known_capabilities;
  
      /*
       * Enable the ones that we do know about.
       * Add other checks here as new ones are introduced.
       */
      if (cap.flags & RDMA_CAPABILITY_PIN_ALL) {
          rdma->pin_all = true;
      }
  
      rdma->cm_id = cm_event->id;
      verbs = cm_event->id->verbs;
  
      rdma_ack_cm_event(cm_event);
  
      trace_qemu_rdma_accept_pin_state(rdma->pin_all);
  
      caps_to_network(&cap);
  
      trace_qemu_rdma_accept_pin_verbsc(verbs);
  
      if (!rdma->verbs) {
          rdma->verbs = verbs;
      } else if (rdma->verbs != verbs) {
              error_report("ibv context not matching %p, %p!", rdma->verbs,
                           verbs);
              goto err_rdma_dest_wait;
      }
  
      qemu_rdma_dump_id("dest_init", verbs);
  
      ret = qemu_rdma_alloc_pd_cq(rdma);
      if (ret) {
          error_report("rdma migration: error allocating pd and cq!");
          goto err_rdma_dest_wait;
      }
  
      ret = qemu_rdma_alloc_qp(rdma);
      if (ret) {
          error_report("rdma migration: error allocating qp!");
          goto err_rdma_dest_wait;
      }
  
      ret = qemu_rdma_init_ram_blocks(rdma);
      if (ret) {
          error_report("rdma migration: error initializing ram blocks!");
          goto err_rdma_dest_wait;
      }
  
      for (idx = 0; idx < RDMA_WRID_MAX; idx++) {
          ret = qemu_rdma_reg_control(rdma, idx);
          if (ret) {
              error_report("rdma: error registering %d control", idx);
              goto err_rdma_dest_wait;
          }
      }
  
      qemu_set_fd_handler(rdma->channel->fd, NULL, NULL, NULL);
  
      ret = rdma_accept(rdma->cm_id, &conn_param);
      if (ret) {
          error_report("rdma_accept returns %d", ret);
          goto err_rdma_dest_wait;
      }
  
      ret = rdma_get_cm_event(rdma->channel, &cm_event);
      if (ret) {
          error_report("rdma_accept get_cm_event failed %d", ret);
          goto err_rdma_dest_wait;
      }
  
      if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
          error_report("rdma_accept not event established");
          rdma_ack_cm_event(cm_event);
          goto err_rdma_dest_wait;
      }
  
      rdma_ack_cm_event(cm_event);
      rdma->connected = true;
  
      ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY);
      if (ret) {
          error_report("rdma migration: error posting second control recv");
          goto err_rdma_dest_wait;
      }
  
      qemu_rdma_dump_gid("dest_connect", rdma->cm_id);
  
      return 0;
  
  err_rdma_dest_wait:
      rdma->error_state = ret;
      qemu_rdma_cleanup(rdma);
      return ret;
  }
  
  /*
   * During each iteration of the migration, we listen for instructions
   * by the source VM to perform dynamic page registrations before they
   * can perform RDMA operations.
   *
   * We respond with the 'rkey'.
   *
   * Keep doing this until the source tells us to stop.
   */
  static int qemu_rdma_registration_handle(QEMUFile *f, void *opaque,
                                           uint64_t flags)
  {
      RDMAControlHeader reg_resp = { .len = sizeof(RDMARegisterResult),
                                 .type = RDMA_CONTROL_REGISTER_RESULT,
                                 .repeat = 0,
                               };
      RDMAControlHeader unreg_resp = { .len = 0,
                                 .type = RDMA_CONTROL_UNREGISTER_FINISHED,
                                 .repeat = 0,
                               };
      RDMAControlHeader blocks = { .type = RDMA_CONTROL_RAM_BLOCKS_RESULT,
                                   .repeat = 1 };
      QEMUFileRDMA *rfile = opaque;
      RDMAContext *rdma = rfile->rdma;
      RDMALocalBlocks *local = &rdma->local_ram_blocks;
      RDMAControlHeader head;
      RDMARegister *reg, *registers;
      RDMACompress *comp;
      RDMARegisterResult *reg_result;
      static RDMARegisterResult results[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE];
      RDMALocalBlock *block;
      void *host_addr;
      int ret = 0;
      int idx = 0;
      int count = 0;
      int i = 0;
  
      CHECK_ERROR_STATE();
  
      do {
          trace_qemu_rdma_registration_handle_wait(flags);
  
          ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_NONE);
  
          if (ret < 0) {
              break;
          }
  
          if (head.repeat > RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE) {
              error_report("rdma: Too many requests in this message (%d)."
                              "Bailing.", head.repeat);
              ret = -EIO;
              break;
          }
  
          switch (head.type) {
          case RDMA_CONTROL_COMPRESS:
              comp = (RDMACompress *) rdma->wr_data[idx].control_curr;
              network_to_compress(comp);
  
              trace_qemu_rdma_registration_handle_compress(comp->length,
                                                           comp->block_idx,
                                                           comp->offset);
              block = &(rdma->local_ram_blocks.block[comp->block_idx]);
  
              host_addr = block->local_host_addr +
                              (comp->offset - block->offset);
  
              ram_handle_compressed(host_addr, comp->value, comp->length);
              break;
  
          case RDMA_CONTROL_REGISTER_FINISHED:
              trace_qemu_rdma_registration_handle_finished();
              goto out;
  
          case RDMA_CONTROL_RAM_BLOCKS_REQUEST:
              trace_qemu_rdma_registration_handle_ram_blocks();
  
              if (rdma->pin_all) {
                  ret = qemu_rdma_reg_whole_ram_blocks(rdma);
                  if (ret) {
                      error_report("rdma migration: error dest "
                                      "registering ram blocks");
                      goto out;
                  }
              }
  
              /*
               * Dest uses this to prepare to transmit the RAMBlock descriptions
               * to the source VM after connection setup.
               * Both sides use the "remote" structure to communicate and update
               * their "local" descriptions with what was sent.
               */
              for (i = 0; i < local->nb_blocks; i++) {
                  rdma->dest_blocks[i].remote_host_addr =
                      (uintptr_t)(local->block[i].local_host_addr);
  
                  if (rdma->pin_all) {
                      rdma->dest_blocks[i].remote_rkey = local->block[i].mr->rkey;
                  }
  
                  rdma->dest_blocks[i].offset = local->block[i].offset;
                  rdma->dest_blocks[i].length = local->block[i].length;
  
                  dest_block_to_network(&rdma->dest_blocks[i]);
              }
  
              blocks.len = rdma->local_ram_blocks.nb_blocks
                                                  * sizeof(RDMADestBlock);
  
  
              ret = qemu_rdma_post_send_control(rdma,
                                          (uint8_t *) rdma->dest_blocks, &blocks);
  
              if (ret < 0) {
                  error_report("rdma migration: error sending remote info");
                  goto out;
              }
  
              break;
          case RDMA_CONTROL_REGISTER_REQUEST:
              trace_qemu_rdma_registration_handle_register(head.repeat);
  
              reg_resp.repeat = head.repeat;
              registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
  
              for (count = 0; count < head.repeat; count++) {
                  uint64_t chunk;
                  uint8_t *chunk_start, *chunk_end;
  
                  reg = &registers[count];
                  network_to_register(reg);
  
                  reg_result = &results[count];
  
                  trace_qemu_rdma_registration_handle_register_loop(count,
                           reg->current_index, reg->key.current_addr, reg->chunks);
  
                  block = &(rdma->local_ram_blocks.block[reg->current_index]);
                  if (block->is_ram_block) {
                      host_addr = (block->local_host_addr +
                                  (reg->key.current_addr - block->offset));
                      chunk = ram_chunk_index(block->local_host_addr,
                                              (uint8_t *) host_addr);
                  } else {
                      chunk = reg->key.chunk;
                      host_addr = block->local_host_addr +
                          (reg->key.chunk * (1UL << RDMA_REG_CHUNK_SHIFT));
                  }
                  chunk_start = ram_chunk_start(block, chunk);
                  chunk_end = ram_chunk_end(block, chunk + reg->chunks);
                  if (qemu_rdma_register_and_get_keys(rdma, block,
                              (uintptr_t)host_addr, NULL, &reg_result->rkey,
                              chunk, chunk_start, chunk_end)) {
                      error_report("cannot get rkey");
                      ret = -EINVAL;
                      goto out;
                  }
  
                  reg_result->host_addr = (uintptr_t)block->local_host_addr;
  
                  trace_qemu_rdma_registration_handle_register_rkey(
                                                             reg_result->rkey);
  
                  result_to_network(reg_result);
              }
  
              ret = qemu_rdma_post_send_control(rdma,
                              (uint8_t *) results, &reg_resp);
  
              if (ret < 0) {
                  error_report("Failed to send control buffer");
                  goto out;
              }
              break;
          case RDMA_CONTROL_UNREGISTER_REQUEST:
              trace_qemu_rdma_registration_handle_unregister(head.repeat);
              unreg_resp.repeat = head.repeat;
              registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
  
              for (count = 0; count < head.repeat; count++) {
                  reg = &registers[count];
                  network_to_register(reg);
  
                  trace_qemu_rdma_registration_handle_unregister_loop(count,
                             reg->current_index, reg->key.chunk);
  
                  block = &(rdma->local_ram_blocks.block[reg->current_index]);
  
                  ret = ibv_dereg_mr(block->pmr[reg->key.chunk]);
                  block->pmr[reg->key.chunk] = NULL;
  
                  if (ret != 0) {
                      perror("rdma unregistration chunk failed");
                      ret = -ret;
                      goto out;
                  }
  
                  rdma->total_registrations--;
  
                  trace_qemu_rdma_registration_handle_unregister_success(
                                                         reg->key.chunk);
              }
  
              ret = qemu_rdma_post_send_control(rdma, NULL, &unreg_resp);
  
              if (ret < 0) {
                  error_report("Failed to send control buffer");
                  goto out;
              }
              break;
          case RDMA_CONTROL_REGISTER_RESULT:
              error_report("Invalid RESULT message at dest.");
              ret = -EIO;
              goto out;
          default:
              error_report("Unknown control message %s", control_desc[head.type]);
              ret = -EIO;
              goto out;
          }
      } while (1);
  out:
      if (ret < 0) {
          rdma->error_state = ret;
      }
      return ret;
  }
  
  static int qemu_rdma_registration_start(QEMUFile *f, void *opaque,
                                          uint64_t flags)
  {
      QEMUFileRDMA *rfile = opaque;
      RDMAContext *rdma = rfile->rdma;
  
      CHECK_ERROR_STATE();
  
      trace_qemu_rdma_registration_start(flags);
      qemu_put_be64(f, RAM_SAVE_FLAG_HOOK);
      qemu_fflush(f);
  
      return 0;
  }
  
  /*
   * Inform dest that dynamic registrations are done for now.
   * First, flush writes, if any.
   */
  static int qemu_rdma_registration_stop(QEMUFile *f, void *opaque,
                                         uint64_t flags)
  {
      Error *local_err = NULL, **errp = &local_err;
      QEMUFileRDMA *rfile = opaque;
      RDMAContext *rdma = rfile->rdma;
      RDMAControlHeader head = { .len = 0, .repeat = 1 };
      int ret = 0;
  
      CHECK_ERROR_STATE();
  
      qemu_fflush(f);
      ret = qemu_rdma_drain_cq(f, rdma);
  
      if (ret < 0) {
          goto err;
      }
  
      if (flags == RAM_CONTROL_SETUP) {
          RDMAControlHeader resp = {.type = RDMA_CONTROL_RAM_BLOCKS_RESULT };
          RDMALocalBlocks *local = &rdma->local_ram_blocks;
          int reg_result_idx, i, j, nb_dest_blocks;
  
          head.type = RDMA_CONTROL_RAM_BLOCKS_REQUEST;
          trace_qemu_rdma_registration_stop_ram();
  
          /*
           * Make sure that we parallelize the pinning on both sides.
           * For very large guests, doing this serially takes a really
           * long time, so we have to 'interleave' the pinning locally
           * with the control messages by performing the pinning on this
           * side before we receive the control response from the other
           * side that the pinning has completed.
           */
          ret = qemu_rdma_exchange_send(rdma, &head, NULL, &resp,
                      &reg_result_idx, rdma->pin_all ?
                      qemu_rdma_reg_whole_ram_blocks : NULL);
          if (ret < 0) {
              ERROR(errp, "receiving remote info!");
              return ret;
          }
  
          nb_dest_blocks = resp.len / sizeof(RDMADestBlock);
  
          /*
           * The protocol uses two different sets of rkeys (mutually exclusive):
           * 1. One key to represent the virtual address of the entire ram block.
           *    (dynamic chunk registration disabled - pin everything with one rkey.)
           * 2. One to represent individual chunks within a ram block.
           *    (dynamic chunk registration enabled - pin individual chunks.)
           *
           * Once the capability is successfully negotiated, the destination transmits
           * the keys to use (or sends them later) including the virtual addresses
           * and then propagates the remote ram block descriptions to his local copy.
           */
  
          if (local->nb_blocks != nb_dest_blocks) {
              ERROR(errp, "ram blocks mismatch #1! "
                          "Your QEMU command line parameters are probably "
                          "not identical on both the source and destination.");
              return -EINVAL;
          }
  
          qemu_rdma_move_header(rdma, reg_result_idx, &resp);
          memcpy(rdma->dest_blocks,
              rdma->wr_data[reg_result_idx].control_curr, resp.len);
          for (i = 0; i < nb_dest_blocks; i++) {
              network_to_dest_block(&rdma->dest_blocks[i]);
  
              /* search local ram blocks */
              for (j = 0; j < local->nb_blocks; j++) {
                  if (rdma->dest_blocks[i].offset != local->block[j].offset) {
                      continue;
                  }
  
                  if (rdma->dest_blocks[i].length != local->block[j].length) {
                      ERROR(errp, "ram blocks mismatch #2! "
                          "Your QEMU command line parameters are probably "
                          "not identical on both the source and destination.");
                      return -EINVAL;
                  }
                  local->block[j].remote_host_addr =
                          rdma->dest_blocks[i].remote_host_addr;
                  local->block[j].remote_rkey = rdma->dest_blocks[i].remote_rkey;
                  break;
              }
  
              if (j >= local->nb_blocks) {
                  ERROR(errp, "ram blocks mismatch #3! "
                          "Your QEMU command line parameters are probably "
                          "not identical on both the source and destination.");
                  return -EINVAL;
              }
          }
      }
  
      trace_qemu_rdma_registration_stop(flags);
  
      head.type = RDMA_CONTROL_REGISTER_FINISHED;
      ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL);
  
      if (ret < 0) {
          goto err;
      }
  
      return 0;
  err:
      rdma->error_state = ret;
      return ret;
  }
  
  static int qemu_rdma_get_fd(void *opaque)
  {
      QEMUFileRDMA *rfile = opaque;
      RDMAContext *rdma = rfile->rdma;
  
      return rdma->comp_channel->fd;
  }
  
  static const QEMUFileOps rdma_read_ops = {
      .get_buffer    = qemu_rdma_get_buffer,
      .get_fd        = qemu_rdma_get_fd,
      .close         = qemu_rdma_close,
      .hook_ram_load = qemu_rdma_registration_handle,
  };
  
  static const QEMUFileOps rdma_write_ops = {
      .put_buffer         = qemu_rdma_put_buffer,
      .close              = qemu_rdma_close,
      .before_ram_iterate = qemu_rdma_registration_start,
      .after_ram_iterate  = qemu_rdma_registration_stop,
      .save_page          = qemu_rdma_save_page,
  };
  
  static void *qemu_fopen_rdma(RDMAContext *rdma, const char *mode)
  {
      QEMUFileRDMA *r = g_malloc0(sizeof(QEMUFileRDMA));
  
      if (qemu_file_mode_is_not_valid(mode)) {
          return NULL;
      }
  
      r->rdma = rdma;
  
      if (mode[0] == 'w') {
          r->file = qemu_fopen_ops(r, &rdma_write_ops);
      } else {
          r->file = qemu_fopen_ops(r, &rdma_read_ops);
      }
  
      return r->file;
  }
  
  static void rdma_accept_incoming_migration(void *opaque)
  {
      RDMAContext *rdma = opaque;
      int ret;
      QEMUFile *f;
      Error *local_err = NULL, **errp = &local_err;
  
      trace_qemu_dma_accept_incoming_migration();
      ret = qemu_rdma_accept(rdma);
  
      if (ret) {
          ERROR(errp, "RDMA Migration initialization failed!");
          return;
      }
  
      trace_qemu_dma_accept_incoming_migration_accepted();
  
      f = qemu_fopen_rdma(rdma, "rb");
      if (f == NULL) {
          ERROR(errp, "could not qemu_fopen_rdma!");
          qemu_rdma_cleanup(rdma);
          return;
      }
  
      rdma->migration_started_on_destination = 1;
      process_incoming_migration(f);
  }
  
  void rdma_start_incoming_migration(const char *host_port, Error **errp)
  {
      int ret;
      RDMAContext *rdma;
      Error *local_err = NULL;
  
      trace_rdma_start_incoming_migration();
      rdma = qemu_rdma_data_init(host_port, &local_err);
  
      if (rdma == NULL) {
          goto err;
      }
  
      ret = qemu_rdma_dest_init(rdma, &local_err);
  
      if (ret) {
          goto err;
      }
  
      trace_rdma_start_incoming_migration_after_dest_init();
  
      ret = rdma_listen(rdma->listen_id, 5);
  
      if (ret) {
          ERROR(errp, "listening on socket!");
          goto err;
      }
  
      trace_rdma_start_incoming_migration_after_rdma_listen();
  
      qemu_set_fd_handler(rdma->channel->fd, rdma_accept_incoming_migration,
                          NULL, (void *)(intptr_t)rdma);
      return;
  err:
      error_propagate(errp, local_err);
      g_free(rdma);
  }
  
  void rdma_start_outgoing_migration(void *opaque,
                              const char *host_port, Error **errp)
  {
      MigrationState *s = opaque;
      Error *local_err = NULL, **temp = &local_err;
      RDMAContext *rdma = qemu_rdma_data_init(host_port, &local_err);
      int ret = 0;
  
      if (rdma == NULL) {
          ERROR(temp, "Failed to initialize RDMA data structures! %d", ret);
          goto err;
      }
  
      ret = qemu_rdma_source_init(rdma, &local_err,
          s->enabled_capabilities[MIGRATION_CAPABILITY_RDMA_PIN_ALL]);
  
      if (ret) {
          goto err;
      }
  
      trace_rdma_start_outgoing_migration_after_rdma_source_init();
      ret = qemu_rdma_connect(rdma, &local_err);
  
      if (ret) {
          goto err;
      }
  
      trace_rdma_start_outgoing_migration_after_rdma_connect();
  
      s->file = qemu_fopen_rdma(rdma, "wb");
      migrate_fd_connect(s);
      return;
  err:
      error_propagate(errp, local_err);
      g_free(rdma);
      migrate_fd_error(s);
  }
  