<repo_name>wgtdkp/nvds<file_sep>/test/test_boost_asio.cc
#include <bits/stdc++.h>
#include <boost/asio.hpp>

using namespace boost::asio;

int main() {
  io_service io_service;
  ip::tcp::resolver resolver(io_service);
  ip::tcp::resolver::query query("www.wgtdkp.com", "http");
  auto iter = resolver.resolve(query);
  decltype(iter) end;
  while (iter != end) {
    ip::tcp::endpoint ep = *iter++;
    std::cout << ep << std::endl;
  }
  return 0;
}
<file_sep>/src/infiniband.h
#ifndef _NVDS_INFINIBAND_H_
#define _NVDS_INFINIBAND_H_

#include "common.h"
#include "spinlock.h"
#include "transport.h"

#include <cstring>
#include <mutex>
#include <infiniband/verbs.h>
#include <netinet/ip.h>

namespace nvds {

struct Request;
struct Response;

// Derived from RAMCloud Infiniband.h
class Infiniband {
 public:
  static const uint16_t kPort = 1;
  static const uint32_t kMaxSendSge = 12;
  static const uint32_t kMaxRecvSge = 12;
  static const uint32_t kMaxInlineData = 128;

  Infiniband(uint32_t registered_mem_size=1024);
  ~Infiniband();

  // This class includes informations for constructing a connection
	struct QueuePairInfo {
    uint16_t lid;   // Local id
		uint32_t qpn;   // Queue pair number
		uint32_t psn;   // Packet sequence number
		uint32_t rkey;  // Memory region key
    uint64_t vaddr; // Virtual address of memory regions
    // DEBUG
    void Print() const;
	};

  struct QueuePair {
    static const uint32_t kDefaultPsn = 33;
    Infiniband&   ib;
    ibv_qp_type   type;
    ibv_pd*       pd;
    ibv_srq*      srq;
    ibv_qp*       qp;
    ibv_cq*       scq;
    ibv_cq*       rcq;
    uint32_t      psn;

    QueuePair(Infiniband& ib, ibv_qp_type type, 
              uint32_t max_send, uint32_t max_recv);
    ~QueuePair();
    DISALLOW_COPY_AND_ASSIGN(QueuePair);
    uint32_t GetLocalQPNum() const { return qp->qp_num; }
    uint32_t GetPeerQPNum() const;
    int GetState() const;
    void Activate();
    void SetStateRTR(const QueuePairInfo& peer_info);
    void SetStateRTS(const QueuePairInfo& peer_info);
  };

  struct Address {
		int32_t ib_port;
		uint16_t lid;
		uint32_t qpn;

		std::string ToString() const;
    bool operator==(const Address& other) const {
      return ib_port == other.ib_port &&
             lid == other.lid &&
             qpn == other.qpn;
    }
    bool operator!=(const Address& other) const {
      return !(*this == other);
    }
	};

	struct Buffer {
    Buffer*   next;
		char* 		buf;				
		uint32_t 	size;		    // Buffer size
		uint32_t 	msg_len;    // message length
		ibv_mr*		mr;         // IBV memory region
		Address   peer_addr;  // Peer address of this request
		bool 			is_recv;    // Is this a recv buffer

		Buffer(char* b, uint32_t size, ibv_mr* mr, bool is_recv=false)
				: next(nullptr), buf(b), size(size), mr(mr), is_recv(is_recv) {}
		Buffer() : next(nullptr), buf(nullptr), size(0), msg_len(0),
               mr(nullptr), is_recv(false) {}
		DISALLOW_COPY_AND_ASSIGN(Buffer);
    Request* MakeRequest() {
      return reinterpret_cast<Request*>(buf + kIBUDPadding);
    }
    Response* MakeResponse() {
      // TODO(wgtdkp): are you sure ?
      return reinterpret_cast<Response*>(buf + kIBUDPadding);
    }
	};

  // A registered buffer pool
	class RegisteredBuffers {
	 public:
	  RegisteredBuffers(ibv_pd* pd, uint32_t buf_size,
                      uint32_t buf_num, bool is_recv=false);
		~RegisteredBuffers();
		DISALLOW_COPY_AND_ASSIGN(RegisteredBuffers);
    // Used as buffer pool
    Buffer* Alloc() {
      std::lock_guard<Spinlock> _(spinlock_);
      auto ret = root_;
      if (root_ != nullptr) {
        root_ = root_->next;
      }
      return ret;
    }
    void Free(Buffer* b) {
      std::lock_guard<Spinlock> _(spinlock_);
      b->next = root_;
      root_ = b;
    }
    const Buffer* root() const { return root_; }
	 private:
	 	uint32_t buf_size_;
		uint32_t buf_num_;
		void* ptr_;
		Buffer* bufs_;
    Buffer* root_;
    Spinlock spinlock_;
	};

  uint16_t GetLid(uint16_t port);
  Buffer* TryReceive(QueuePair* qp) { return PollCQ(qp->rcq, 1); }
  Buffer* TrySend(QueuePair* qp) { return PollCQ(qp->scq, 1); }
  Buffer* PollCQ(ibv_cq* cq, int num_entries);
  Buffer* Receive(QueuePair* qp);
	void PostReceive(QueuePair* qp, Buffer* b);
  void PostSend(QueuePair* qp, Buffer* b, uint32_t len,
								const Address* peer_addr);
  void PostSendAndWait(QueuePair* qp, Buffer* b, uint32_t len,
											 const Address* peer_addr);
  ibv_ah* GetAddrHandler(const Address& addr);

  void PostWrite(QueuePair& qp, QueuePairInfo& peer_info);
  void PostWriteAndWait(QueuePair& qp, QueuePairInfo& peer_info);


  ibv_context* ctx() { return ctx_; }
  ibv_pd* pd() { return pd_; }
  ibv_mr* mr() { return mr_; }
  char* raw_mem() { return raw_mem_; }
  uint32_t raw_mem_size() { return raw_mem_size_; }

 private:
  ibv_context* ctx_;
  ibv_pd* pd_;
  ibv_mr* mr_;
  char* raw_mem_;
  uint32_t raw_mem_size_;

  std::array<ibv_ah*, UINT16_MAX + 1> addr_handlers_;
};

} // namespace nvds

#endif // _NVDS_INFINIBAND_H_
<file_sep>/src/tablet.cc
#include "tablet.h"

#include "index.h"
#include "request.h"
#include "status.h"

#define OFFSETOF_NVMOBJECT(obj, member) \
    offsetof(NVMObject, member) + obj

namespace nvds {

Tablet::Tablet(const IndexManager& index_manager,
               NVMPtr<NVMTablet> nvm_tablet, bool is_backup)
    : index_manager_(index_manager),
      nvm_tablet_(nvm_tablet), allocator_(&nvm_tablet->data) {
  info_.is_backup = is_backup;
  
  // Memory region
  // FIXME(wgtdkp): how to simulate latency of RDMA read/write to NVM?
  mr_ = ibv_reg_mr(ib_.pd(), nvm_tablet_.ptr(), kNVMTabletSize,
                   IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
  assert(mr_ != nullptr);

  for (size_t i = 0; i < kNumReplicas; ++i) {
    qps_[i] = new Infiniband::QueuePair(ib_, IBV_QPT_RC,
        kMaxIBQueueDepth, kMaxIBQueueDepth);
    //qps_[i]->Plumb();
    info_.qpis[i] = {
      ib_.GetLid(Infiniband::kPort),
      qps_[i]->GetLocalQPNum(),
      Infiniband::QueuePair::kDefaultPsn,
      mr_->rkey,
      reinterpret_cast<uint64_t>(nvm_tablet_.ptr())
    };
  }
}

Tablet::~Tablet() {
  for (ssize_t i = kNumReplicas - 1; i >= 0; --i) {
    delete qps_[i];
  }
  int err = ibv_dereg_mr(mr_);
  assert(err == 0);
}

Status Tablet::Put(const Request* r, ModificationList& modifications) {
  allocator_.set_modifications(&modifications);

  assert(r->type == Request::Type::PUT);
  auto idx = r->key_hash % kHashTableSize;
  uint32_t slot = offsetof(NVMTablet, hash_table) + sizeof(uint32_t) * idx;
  auto head = allocator_.Read<uint32_t>(slot);
  auto q = slot, p = head;
  while (p) {
    if (r->key_len != allocator_.Read<uint16_t>(OFFSETOF_NVMOBJECT(p, key_len)) ||
        allocator_.Memcmp(OFFSETOF_NVMOBJECT(p, data), r->Key(), r->key_len)) {
      q = p;
      p = allocator_.Read<uint32_t>(OFFSETOF_NVMOBJECT(p, next));
      continue;
    }
    // There is already the same key, overwrite it.
    if (r->val_len < allocator_.Read<uint16_t>(OFFSETOF_NVMOBJECT(p, val_len))) {
      // The new value is shorter than the older, store data at its original place.
      allocator_.Write(OFFSETOF_NVMOBJECT(p, val_len), r->val_len);
      allocator_.Memcpy(OFFSETOF_NVMOBJECT(p, data) + r->key_len, r->Val(), r->val_len);
    } else {
      auto next = allocator_.Read<uint32_t>(OFFSETOF_NVMOBJECT(p, next));
      allocator_.Free(p);
      auto size = sizeof(NVMObject) + r->key_len + r->val_len;
      p = allocator_.Alloc(size);
      // TODO(wgtdkp): handle the situation: `no space`.
      assert(p != 0);
      allocator_.Write<NVMObject>(p, {next, r->key_len, r->val_len, r->key_hash});
      allocator_.Memcpy(OFFSETOF_NVMOBJECT(p, data), r->data, r->key_len + r->val_len);
      allocator_.Write(OFFSETOF_NVMOBJECT(q, next), p);
    }
    return Status::OK;
  }

  auto size = sizeof(NVMObject) + r->key_len + r->val_len;
  p = allocator_.Alloc(size);
  // TODO(wgtdkp): handle the situation: `no space`.
  assert(p != 0);
  // TODO(wgtdkp): use single `memcpy`
  allocator_.Write<NVMObject>(p, {head, r->key_len, r->val_len, r->key_hash});
  allocator_.Memcpy(OFFSETOF_NVMOBJECT(p, data), r->data, r->key_len + r->val_len);
  // Insert the new item to head of the bucket list
  allocator_.Write(slot, p);
  return Status::OK;
}

Status Tablet::Add(const Request* r, ModificationList& modifications) {
  allocator_.set_modifications(&modifications);

  assert(r->type == Request::Type::ADD);
  auto idx = r->key_hash % kHashTableSize;
  uint32_t slot = offsetof(NVMTablet, hash_table) + sizeof(uint32_t) * idx;
  auto head = allocator_.Read<uint32_t>(slot);
  auto p = head;
  while (p) {
    if (r->key_len != allocator_.Read<uint16_t>(OFFSETOF_NVMOBJECT(p, key_len)) ||
        allocator_.Memcmp(OFFSETOF_NVMOBJECT(p, data), r->Key(), r->key_len)) {
      p = allocator_.Read<uint32_t>(OFFSETOF_NVMOBJECT(p, next));
      continue;
    }
    // There is already the same key, return Status::ERROR.    
    return Status::ERROR;
  }

  auto size = sizeof(NVMObject) + r->key_len + r->val_len;
  p = allocator_.Alloc(size);
  // TODO(wgtdkp): handle the situation: `no space`.
  assert(p != 0);
  // TODO(wgtdkp): use single `memcpy`
  allocator_.Write<NVMObject>(p, {head, r->key_len, r->val_len, r->key_hash});
  allocator_.Memcpy(OFFSETOF_NVMOBJECT(p, data), r->data, r->key_len + r->val_len);
  // Insert the new item to head of the bucket list
  allocator_.Write(slot, p);
  return Status::OK;
}

Status Tablet::Get(Response* resp, const Request* r,
                   ModificationList& modifications) {
  allocator_.set_modifications(&modifications);

  assert(r->type == Request::Type::GET);
  auto idx = r->key_hash % kHashTableSize;
  auto p = offsetof(NVMTablet, hash_table) + sizeof(uint32_t) * idx;
  p = allocator_.Read<uint32_t>(p);
  while (p) {
    // TODO(wgtdkp): use a different hash to speedup searching
    if (r->key_len == allocator_.Read<uint16_t>(OFFSETOF_NVMOBJECT(p, key_len)) &&
        allocator_.Memcmp(OFFSETOF_NVMOBJECT(p, data), r->Key(), r->key_len) == 0) {
      auto len = allocator_.Read<uint16_t>(OFFSETOF_NVMOBJECT(p, val_len));
      allocator_.Memcpy(resp->val, OFFSETOF_NVMOBJECT(p, data) + r->key_len, len);
      resp->val_len = len;
      return Status::OK;
    }
    p = allocator_.Read<uint32_t>(OFFSETOF_NVMOBJECT(p, next));
  }
  return Status::ERROR;
}

Status Tablet::Del(const Request* r, ModificationList& modifications) {
  allocator_.set_modifications(&modifications);

  assert(r->type == Request::Type::DEL);
  auto idx = r->key_hash % kHashTableSize;
  auto q = offsetof(NVMTablet, hash_table) + sizeof(uint32_t) * idx;
  auto p = allocator_.Read<uint32_t>(q);
  while (p) {
    if (r->key_len == allocator_.Read<uint16_t>(OFFSETOF_NVMOBJECT(p, key_len)) &&
        allocator_.Memcmp(OFFSETOF_NVMOBJECT(p, data), r->Key(), r->key_len) == 0) {
      auto next = allocator_.Read<uint32_t>(OFFSETOF_NVMOBJECT(p, next));
      allocator_.Write(OFFSETOF_NVMOBJECT(q, next), next);
      allocator_.Free(p);
      return Status::OK;
    }
  }
  return Status::ERROR;
}

void Tablet::SettingupQPConnect(TabletId id, const IndexManager& index_manager) {
  info_ = index_manager.GetTablet(id);
  if (info_.is_backup) {
    auto master_info = index_manager.GetTablet(info_.master);
    assert(!master_info.is_backup);
    for (uint32_t i = 0; i < kNumReplicas; ++i) {
      if (master_info.backups[i] == id) {
        qps_[0]->SetStateRTR(master_info.qpis[i]);
        break;
      }
      assert(i < kNumReplicas - 1);
    }
  } else {
    for (uint32_t i = 0; i < kNumReplicas; ++i) {
      auto backup_id = info_.backups[i];
      auto backup_info = index_manager.GetTablet(backup_id);
      assert(backup_info.is_backup);
      qps_[i]->SetStateRTS(backup_info.qpis[0]);
    }
  }
}

int Tablet::Sync(ModificationList& modifications) {
  if (modifications.size() == 0) {
    return 0;
  }
  MergeModifications(modifications);

  for (size_t k = 0; k < info_.backups.size(); ++k) {
    auto backup = index_manager_.GetTablet(info_.backups[k]);
    assert(backup.is_backup);
    size_t i = 0;
    for (const auto& m : modifications) {
      sges[i] = {m.src, m.len, mr_->lkey};
      
      wrs[i].wr.rdma.remote_addr = backup.qpis[0].vaddr + m.des;
      wrs[i].wr.rdma.rkey        = backup.qpis[0].rkey;
      // TODO(wgtdkp): use unique id
      wrs[i].wr_id               = 1;
      wrs[i].sg_list             = &sges[i];
      wrs[i].num_sge             = 1;
      wrs[i].opcode              = IBV_WR_RDMA_WRITE;
      // TODO(wgtdkp): do we really need to signal each send?
      wrs[i].send_flags          = IBV_SEND_INLINE;
      wrs[i].next                = &wrs[i+1];
      ++i;
      // DEBUG
      break;
    }
    wrs[i-1].next = nullptr;
    wrs[i-1].send_flags |= IBV_SEND_SIGNALED;

    struct ibv_send_wr* bad_wr;
    int err = ibv_post_send(qps_[k]->qp, &wrs[0], &bad_wr);
    if (err != 0) {
      throw TransportException(HERE, "ibv_post_send failed", err);
    }
  }

  for (auto& qp : qps_) {
    ibv_wc wc;
    int r;
    while ((r = ibv_poll_cq(qp->scq, 1, &wc)) != 1) {}
    if (wc.status != IBV_WC_SUCCESS) {
      throw TransportException(HERE, wc.status);
    } else {
      // std::clog << "success" << std::endl << std::flush;
    }
  }
  return 0;
}

/*
// DEBUG
static void PrintModifications(const ModificationList& modifications) {
  for (const auto& m : modifications) {
    m.Print();
  }
  std::cout << std::endl << std::flush;
}
*/

void Tablet::MergeModifications(ModificationList& modifications) {
  auto n = modifications.size();
  assert(n > 0);

  //PrintModifications(modifications);

  // Baby, don't worry, n <= 12.
  // This simple algorithm should be faster than `union-find`;
  for (size_t i = 0; i < n; ++i) {
    for (size_t j = i + 1; j < n; ++j) {
      if (modifications[j] < modifications[i])
        std::swap(modifications[j], modifications[i]);
    }
  }

  //PrintModifications(modifications);

  static const size_t gap = 16;
  size_t i = 0;
  uint64_t src = modifications[0].src;
  uint32_t begin = modifications[0].des; 
  uint32_t end = modifications[0].des + modifications[0].len;
  for (const auto& m : modifications) {
    if (m.des > end + gap) {
      modifications[i++] = {begin, src, end - begin};
      begin = m.des;
      end = m.des + m.len;
      src = m.src;
    } else {
      end = std::max(end, m.des + m.len);
    }
  }
  modifications[i++] = {begin, src, end - begin};
  modifications.resize(i);

  //PrintModifications(modifications);
}

} // namespace nvds
<file_sep>/script/startup.sh
#!/bin/bash

../src/build/coordinator > coordinator.txt 2> err_coordinator.txt < /dev/null &

# Waiting for coordinator
sleep 0.1s

first_line=$(head -n 1 coordinator.txt)
first_line_arr=($first_line)

# Compatible with the server's bash
#coord_addr=${first_line_arr[-1]}
coord_addr=${first_line_arr[2]}

while read line; do
    server=($line)
    echo "starting server [${server[0]}:${server[1]}]"
    dir="/home/wgtdkp/nvds/"
    bin="./src/build/server"
    out="./script/server_${server[1]}.txt"
    err="./script/err_server_${server[1]}.txt"
    in="/dev/null"
    ssh root@${server[0]} "cd ${dir}; nohup ${bin} ${server[1]} ${coord_addr} > ${out} 2> ${err} < ${in} &" < /dev/null
done < servers.txt
<file_sep>/test/test.c
extern int test_main();

int main() {
    return test_main();
}
<file_sep>/doc/command.md
1. **_ifconfig -a_** to find the nic device; **_ethtool device_** to show nic information; The machines are 56Gb/s infiniband;
<file_sep>/src/message.h
#ifndef _NVDS_MESSAGE_H_
#define _NVDS_MESSAGE_H_

#include "common.h"
#include "infiniband.h"
#include "json.hpp"

namespace nvds {

class IndexManager;

/*
 The message body is in json format, it looks like:
 {
   "name-0": value,
   "name-1": value
 }
 
 0. REQ_JOIN :
 {
   "size": int,
   "ib_addr": {
     "ib_port": int,
     "lid": int,
     "pqn": int,
   },
   "tablets_vaddr": [int],
   "tablets_rkey": [int]
 }
 1. RES_JOIN :
 {
   "id": int,
   "index_manager": IndexManager
 }
 */

/*
struct Backup {
  ServerId server_id;
  TabletId tablet_id;
  bool operator==(const Backup& other) const {
    return server_id == other.server_id &&
           tablet_id == other.tablet_id;
  }
  bool operator!=(const Backup& other) const {
    return !(*this == other);
  }
};
using Master = Backup;
*/

struct TabletInfo {
  TabletId id;
  ServerId server_id;
  bool is_backup;
  std::array<Infiniband::QueuePairInfo, kNumReplicas> qpis;
  union {
    // If `is_backup` == true,
    // `master` is the master tablet of this backup tablet
    TabletId master;
    // Else, `backups` is the backups of this makster tablet
    std::array<TabletId, kNumReplicas> backups;
  };
  bool operator==(const TabletInfo& other) const {
    return id == other.id &&
           server_id == other.server_id &&
           is_backup == other.is_backup;
  }
  bool operator!=(const TabletInfo& other) const {
    return !(*this == other);
  }
  // DEBUG
  void Print() const;
};

/* Json representation
{
  "id": int,
  "active": bool,
  "addr": string,
  "ib_addr": {
    "ib_port": int,
    "lid": int,
    "qpn": int,
  }
  "tablets": array[TabletInfo],
}
 */
struct ServerInfo {
  ServerId id;
  bool active;
  std::string addr; // Ip address
  Infiniband::Address ib_addr; // Infiniband address
  std::array<TabletId, kNumTabletAndBackupsPerServer> tablets;
};

class Message {
 public:
  enum class SenderType : uint8_t {
    COORDINATOR,
    SERVER,
    CLIENT,
  };

  enum class Type : uint8_t {
    REQ_JOIN,         // erver/client ---> coordinator
    REQ_LEAVE,        // server/client ---> coordinator
    RES_JOIN,         // coordinator   ---> server
    RES_CLUSTER_INFO, // coordinator   ---> client
    QP_INFO_EXCH,     // server/client ---> server
    ACK_REJECT,       // acknowledgement: reject
    ACK_ERROR,        // aknowledgement: error
    ACK_OK,           // acknowledgement: ok
  };
  
  PACKED(struct Header {
    SenderType sender_type;
    Type type;
    uint32_t body_len;
  });

  Message() {}
  Message(Header header, std::string body)
      : header_(std::move(header)), body_(std::move(body)) {
    header_.body_len = body_.size();
  }
  //explicit Message(const char* raw=nullptr);
  ~Message() {}

  static const size_t kHeaderSize = sizeof(Header);

  const Header& header() const { return header_; }
  Header& header() { return header_; }
  const std::string& body() const { return body_; }
  std::string& body() { return body_; }
  
  SenderType sender_type() const { return header_.sender_type; }
  Type type() const { return header_.type; }
  uint32_t body_len() const { return header_.body_len; }

 private:
  Header header_;
  std::string body_;
};

void to_json(nlohmann::json& j, const Infiniband::Address& ia);
void from_json(const nlohmann::json& j, Infiniband::Address& ia);
void to_json(nlohmann::json& j, const Infiniband::QueuePairInfo& qpi);
void from_json(const nlohmann::json& j, Infiniband::QueuePairInfo& qpi);
void to_json(nlohmann::json& j, const TabletInfo& ti);
void from_json(const nlohmann::json& j, TabletInfo& ti);
void to_json(nlohmann::json& j, const ServerInfo& si);
void from_json(const nlohmann::json& j, ServerInfo& si);
void to_json(nlohmann::json& j, const IndexManager& im);
void from_json(const nlohmann::json& j, IndexManager& im);

} // namespace nvds

#endif // _NVDS_MESSAGE_
<file_sep>/script/connect_machine.sh
#!/bin/bash
# How to connect to lab's servers;
sudo route add 172.16.31.10 gw 192.168.1.1
<file_sep>/src/hash.h
#ifndef _NVDS_HASH_H_
#define _NVDS_HASH_H_

#include "MurmurHash2.h"

#include <string>

namespace nvds {

using KeyHash = uint64_t;
static const KeyHash kMaxKeyHash = static_cast<KeyHash>(-1);

struct KeyHashRange {
  KeyHash begin;
  KeyHash end;
};

static inline KeyHash Hash(const std::string& key) {
  return MurmurHash64B(key.c_str(), key.size(), 103);
}

static inline KeyHash Hash(const char* key, size_t key_len) {
  return MurmurHash64B(key, key_len, 103);
}

} // namespace nvds

#endif // _NVDS_HASH_H_
<file_sep>/src/experiments/memc.cc
// clang++ -std=c++11 -Wall -o memc -g memc.cc -lmemcached -lrt

#include <iostream>
#include <random>
#include <string>
#include <vector>

#include <cassert>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>

#include <libmemcached/memcached.h>

using namespace std;

static vector<string> GenRandomStrings(size_t len, size_t n) {
  vector<string> ans;
  default_random_engine g;
  uniform_int_distribution<int> dist('a', 'z');
  for (size_t i = 0; i < n; ++i) {
    ans.emplace_back(len, 0);
    for (size_t j = 0; j < len; ++j) {
      ans[i][j] = dist(g);
    }
  }
  return ans;
}

static inline int64_t TimeDiff(struct timespec* lhs, struct timespec* rhs) {
  return (lhs->tv_sec - rhs->tv_sec) * 1000000000 + 
         (lhs->tv_nsec - rhs->tv_nsec);
}

void Test(const char* server, uint32_t len) {
  auto cfg = "--SERVER=" + string(server);
  memcached_st* memc = memcached(cfg.c_str(), cfg.size());
  assert(memc != nullptr);

  static const size_t n = 10000;

  auto keys = GenRandomStrings(16, n);
  auto vals = GenRandomStrings(len, n);
  
  struct timespec begin, end;
  assert(clock_gettime(CLOCK_REALTIME, &begin) == 0);
  for (size_t i = 0; i < keys.size(); ++i) {
    auto& key = keys[i];
    auto& val = vals[i];
    memcached_return_t rc = memcached_set(memc, key.c_str(), key.size(), val.c_str(), val.size(), 0, 0);
    assert(rc == MEMCACHED_SUCCESS);
  }
  assert(clock_gettime(CLOCK_REALTIME, &end) == 0);
  double t_set = TimeDiff(&end, &begin) * 1.0 / n / 1000;

  assert(clock_gettime(CLOCK_REALTIME, &begin) == 0);
  for (size_t i = 0; i < keys.size(); ++i) {
    auto& key = keys[i];
    size_t val_len;
    uint32_t flags;
    memcached_return_t rc;
    memcached_get(memc, key.c_str(), key.size(), &val_len, &flags, &rc);
    assert(rc == MEMCACHED_SUCCESS || rc == MEMCACHED_NOTFOUND);
  }
  assert(clock_gettime(CLOCK_REALTIME, &end) == 0);
  double t_get = TimeDiff(&end, &begin) * 1.0 / n / 1000;

  memcached_free(memc);

  printf("%u, %f, %f\n", len, t_set, t_get);
}

int main(int argc, const char* argv[]) {
  assert(argc == 3);

  const char* server = argv[1];
  uint32_t len = atoi(argv[2]);

  //for (uint32_t len = 16; len <= 1024; len <<= 1) {
    Test(server, len);
  //}
  return 0;
}
<file_sep>/src/tablet.h
#ifndef _NVDS_TABLET_H_
#define _NVDS_TABLET_H_

#include "allocator.h"
#include "common.h"
#include "hash.h"
#include "message.h"
#include "modification.h"
#include "response.h"

namespace nvds {

struct Request;
class IndexManager;

// A reasonable prime number
static const uint32_t kHashTableSize = 1000003;

struct NVMObject {
  uint32_t next;
  uint16_t key_len;
  uint16_t val_len;
  KeyHash key_hash;    
  char data[0];
};

struct NVMTablet {
  char data[Allocator::kSize];
  std::array<uint32_t, kHashTableSize> hash_table;
  NVMTablet() {
    hash_table.fill(0);
  }
};
static const uint32_t kNVMTabletSize = sizeof(NVMTablet);

class Tablet {
 public:
  //Tablet(const TabletInfo& info, NVMPtr<NVMTablet> nvm_tablet);
  Tablet(const IndexManager& index_manager,
         NVMPtr<NVMTablet> nvm_tablet,
         bool is_backup=false);
  ~Tablet();
  DISALLOW_COPY_AND_ASSIGN(Tablet);

  const TabletInfo& info() const { return info_; }
  Status Get(Response* resp, const Request* r,
                       ModificationList& modifications);
  Status Del(const Request* r, ModificationList& modifications);
  Status Put(const Request* r, ModificationList& modifications);
  // Return: Status::ERROR, if there is already the same key; else, Status::OK;
  Status Add(const Request* r, ModificationList& modifications);
  void SettingupQPConnect(TabletId id, const IndexManager& index_manager);
  int Sync(ModificationList& modifications);

 private:
  static void MergeModifications(ModificationList& modifications);

  const IndexManager& index_manager_;  
  TabletInfo info_;
  NVMPtr<NVMTablet> nvm_tablet_;
  Allocator allocator_;

  // Infiniband
  static const uint32_t kMaxIBQueueDepth = 128;
  Infiniband ib_;
  //Infiniband::Address ib_addr_;
  ibv_mr* mr_;
  std::array<Infiniband::QueuePair*, kNumReplicas> qps_;
  // `kNumReplica` queue pairs share this `rcq_` and `scq_`

  static const uint32_t kNumScatters = 16;
  std::array<struct ibv_sge, kNumScatters> sges;
  std::array<struct ibv_send_wr, kNumScatters> wrs;
};

} // namespace nvds

#endif // _NVDS_TABLET_H_
<file_sep>/src/session.cc
#include "session.h"

namespace nvds {

void Session::Start() {
  AsyncRecvMessage(std::make_shared<Message>());
}

void Session::AsyncRecvMessage(std::shared_ptr<Message> msg) {
  // Avoid out-of-range of this session
  auto self = shared_from_this();
  auto body_handler = [this, self, msg](
      const boost::system::error_code& err,
      size_t bytes_transferred) {
    if (!err) {
      recv_msg_handler_(self, msg);
    } else {
      NVDS_ERR(err.message().c_str());
    }
  };
  auto header_handler = [this, self, msg, body_handler](
      const boost::system::error_code& err,
      size_t bytes_transferred) {
    if (!err) {
      assert(bytes_transferred >= Message::kHeaderSize);

      // Read message body
      size_t len = Message::kHeaderSize + msg->body_len() - bytes_transferred;
      msg->body().resize(len);
      boost::asio::async_read(conn_sock_,
          boost::asio::buffer(&msg->body()[0], msg->body_len()),
          body_handler);
    } else {
      NVDS_ERR(err.message().c_str());
    }
  };
  
  // Read message header
  boost::asio::async_read(conn_sock_,
      boost::asio::buffer(&msg->header(), Message::kHeaderSize),
      header_handler);
}

void Session::AsyncSendMessage(std::shared_ptr<Message> msg) {
  // Avoid out-of-range of this session
  auto self = shared_from_this();
  auto body_handler = [this, self, msg](
      const boost::system::error_code& err,
      size_t bytes_transferred) {
    if (!err) {
      send_msg_handler_(self, msg);
    } else {
      NVDS_ERR(err.message().c_str());
    }
  };
  auto header_handler = [this, self, msg, body_handler](
      const boost::system::error_code& err,
      size_t bytes_transferred) {
    if (!err) {
      assert(bytes_transferred >= Message::kHeaderSize);

      // Send message body
      boost::asio::async_write(conn_sock_,
          boost::asio::buffer(msg->body()), body_handler);
    } else {
      NVDS_ERR(err.message().c_str());
    }
  };
  
  // Read message header
  boost::asio::async_write(conn_sock_,
      boost::asio::buffer(&msg->header(), Message::kHeaderSize),
      header_handler);
}

void Session::SendMessage(const Message& msg) {
  boost::asio::write(conn_sock_,
      boost::asio::buffer(&msg.header(), Message::kHeaderSize));
  boost::asio::write(conn_sock_, boost::asio::buffer(msg.body()));
}

Message Session::RecvMessage() {
  Message msg;
  boost::asio::read(conn_sock_,
      boost::asio::buffer(&msg.header(), Message::kHeaderSize));
  msg.body().resize(msg.body_len());
  boost::asio::read(conn_sock_,
      boost::asio::buffer(&msg.body()[0], msg.body_len()));
  return msg;
}

} // namespace nvds
<file_sep>/doc/note.md
# My Notes

## Plan
1. 先做2个节点之间的通信与query，一开始可先不做复制；测试单一节点的QPS；
2. 考虑持久化的设计；
3. 将其拓展到分布式结构；
4. 再去考虑故障检测，故障恢复等问题；

## RDMA
1. [libvma](https://github.com/Mellanox/libvma): Enable standard socket API over Ethernet or Infiniband  from user-space with full network stack bypass.

2. rdma_cma.h: Used to establish communication over RDMA transports.

## RDMA 理解
1. Channel Adapter: 其实就是指NIC
2. RDMA存在两种典型的semantics: memory semantic 和 channel semantic；memory semantic就是由发起方指定读写的地址，完全不需要host的cpu参与， 这称为one-sided；这对应于RDMA的READ、WRITE；channel semantic对应于RDMA的SEND、RECV

## 测试

### RDMA 延迟
相同机架上两台机器之间的RDMA写延迟为1.21us，读延迟为1.20us；读写数据长度为64Byte；
<file_sep>/src/test_stl_map.cc
#include "hash.h"
#include "measurement.h"

#include <gtest/gtest.h>
#include <bits/stdc++.h>

using namespace std;
using namespace nvds;

static vector<string> GenRandomStrings(size_t len, size_t n) {
  vector<string> ans;
  std::default_random_engine g;
  std::uniform_int_distribution<int> dist('a', 'z');
  for (size_t i = 0; i < n; ++i) {
    ans.emplace_back(len, 0);
    for (size_t j = 0; j < len; ++j) {
      ans[i][j] = dist(g);
    }
  }
  return ans;
}

TEST (AllocatorTest, Alloc) {
  const size_t n = 1000 * 1000;
  auto vals = GenRandomStrings(40, n);
  vector<uint64_t> keys(n, 0);
  transform(vals.begin(), vals.end(), keys.begin(), [](const string& x) -> uint64_t {
      return Hash(x.substr(0, 16));
    });
  
  Measurement m;
  unordered_map<uint64_t, string> um;
  um.reserve(n);
  m.begin();
  for (size_t i = 0; i < n; ++i) {
    um[keys[i]] = vals[i];
  }
  m.end();
  m.Print();
}

int main(int argc, char* argv[]) {
  testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}
<file_sep>/src/index.cc
#include "index.h"

namespace nvds {

const ServerInfo& IndexManager::AddServer(const std::string& addr,
                                          nlohmann::json& msg_body) {
  auto id = AllocServerId();
  Infiniband::Address ib_addr = msg_body["ib_addr"];
  servers_[id] = {id, true, addr, ib_addr};
  std::vector<Infiniband::QueuePairInfo> qpis = msg_body["tablet_qpis"];
  
  auto i = id;
  for (uint32_t j = 0; j < kNumReplicas + 1; ++j) {
    for (uint32_t k = 0; k < kNumTabletsPerServer; ++k) {
      auto tablet_id = CalcTabletId(i, j, k);
      auto tablet_idx = CalcTabletId(0, j, k);
      tablets_[tablet_id].id = tablet_id;
      tablets_[tablet_id].server_id = i;
      tablets_[tablet_id].is_backup = j > 0;

      copy(qpis.begin() + tablet_idx * kNumReplicas,
           qpis.begin() + (tablet_idx + 1) * kNumReplicas,
           tablets_[tablet_id].qpis.begin());

      if (!tablets_[tablet_id].is_backup) {
        auto idx = CalcTabletId(0, i, k);
        key_tablet_map_[idx] = tablet_id;
      } else {
        auto backup_id = tablet_id;
        auto master_id = CalcTabletId((kNumServers + i - j) % kNumServers, 0, k);
        // It is tricky here
        assert(backup_id != 0);
        tablets_[master_id].backups[j-1] = backup_id;
        tablets_[backup_id].master = master_id;
      }
      servers_[i].tablets[tablet_idx] = tablet_id;
    }
  }

  return servers_[id];
}

void IndexManager::PrintTablets() const {
  for (auto& tablet : tablets_) {
    tablet.Print();
    std::cout << std::endl;
  }
}


} // namespace nvds
<file_sep>/src/allocator.h
/*
 * This allocator is a simplified version of `Doug Lea's Malloc`.
 * Reference: http://g.oswego.edu/dl/html/malloc.html
 */

#ifndef _NVDS_ALLOCATOR_H_
#define _NVDS_ALLOCATOR_H_

#include "common.h"
#include "modification.h"

#include <memory.h>

namespace nvds {

class Allocator {
 public:
  static const uint32_t kMaxBlockSize = 1024 + 128;
  static const uint32_t kSize = 64 * 1024 * 1024;

  Allocator(void* base) : Allocator(reinterpret_cast<uintptr_t>(base)) {
    Format();
  }
  Allocator(uintptr_t base)
      : base_(base), cnt_writes_(0), modifications_(nullptr) {
    flm_ = OffsetToPtr<FreeListManager>(0);
  }
  ~Allocator() {}
  DISALLOW_COPY_AND_ASSIGN(Allocator);

  // Return offset to the object
  uint32_t Alloc(uint32_t size) {
    auto blk_size = RoundupBlockSize(size + sizeof(uint32_t));
    assert(blk_size % 16 == 0);
    // The client side should refuse too big kv item.  
    assert(blk_size <= kMaxBlockSize);

    auto blk = AllocBlock(blk_size);
    return blk == 0 ? 0 : blk + sizeof(uint32_t);
  }
  void Free(uint32_t ptr) {
    // TODO(wgtdkp): checking if ptr is actually in this Allocator zone.
    assert(ptr > sizeof(uint32_t) && ptr <= kSize);
    FreeBlock(ptr - sizeof(uint32_t));
  }
    template<typename T>
  T* OffsetToPtr(uint32_t offset) const {
    return reinterpret_cast<T*>(base_ + offset);
  }
  template<typename T>
  void Write(uint32_t offset, const T& val) {
    assert(offset != 0);
    auto ptr = OffsetToPtr<T>(offset);
    *ptr = val;
    modifications_->emplace_back(offset,
        base_ + offset, static_cast<uint32_t>(sizeof(T)));
    ++cnt_writes_;
  }
  template<typename T>
  T Read(uint32_t offset) {
    assert(offset);
    return *OffsetToPtr<T>(offset);
  }
  // TODO(wgtdkp): add latency
  int Memcmp(uint32_t lhs, uint32_t rhs, uint32_t len) {
    return memcmp(OffsetToPtr<char>(lhs), OffsetToPtr<char>(rhs), len);
  }
  int Memcmp(uint32_t lhs, const char* rhs, uint32_t len) {
    return memcmp(OffsetToPtr<char>(lhs), rhs, len);
  }
  int Memcmp(const char* lhs, uint32_t rhs, uint32_t len) {
    return memcmp(lhs, OffsetToPtr<char>(rhs), len);
  }
  void Memcpy(uint32_t des, uint32_t src, uint32_t len) {
    memcpy(OffsetToPtr<char>(des), OffsetToPtr<char>(src), len);
    modifications_->emplace_back(des, base_ + des, len);
  }
  void Memcpy(uint32_t des, const char* src, uint32_t len) {
    memcpy(OffsetToPtr<char>(des), src, len);
    modifications_->emplace_back(des, base_ + des, len);
  }
  void Memcpy(char* des, uint32_t src, uint32_t len) {
    memcpy(des, OffsetToPtr<char>(src), len);
    // Copy to user, no NVM modification
  }
  uintptr_t base() const { return base_; }
  uint64_t cnt_writes() const { return cnt_writes_; }
  void set_modifications(ModificationList* modifications) {
    modifications_ = modifications;
  }

 private:
  static const uint32_t kNumFreeLists = kMaxBlockSize / 16 + 1;
  void Format();

  PACKED(
  struct BlockHeader {
    // Denoting if previous block is free.
    uint32_t free: 1;
    // Block size, include block header.
    uint32_t size: 31;
    // Pointer to previous block in the same free list.
    // Relative to base_.
    uint32_t prev;
    // Pointer to next block in the same free list.
    // Relative to base_.
    uint32_t next;

    static const uint32_t kFreeMask = static_cast<uint32_t>(1) << 31;
    BlockHeader() = delete;
  });

  PACKED(
  struct BlockFooter {
    // Denoting if current block is free.
    uint32_t free: 1;
    // Block size, include block header.
    uint32_t size: 31;

    BlockFooter() = delete;
  });

  PACKED(
  struct FreeListManager {
    // The list of these free blocks with same size.
    // The last list is all free block greater than kMaxBlockSize.
    // So these blocks are not necessary the same size,
    // and not sorted(by addr or size).
    uint32_t free_lists[kNumFreeLists];

    FreeListManager() = delete;
  });

  uint32_t GetBlockSizeByFreeList(uint32_t free_list) {
    free_list /= sizeof(uint32_t);
    if (free_list < kNumFreeLists - 1) {
      return (free_list + 1) * 16;
    }
    assert(free_list == kNumFreeLists - 1);
    auto head = Read<uint32_t>(free_list * sizeof(uint32_t));
    return head == 0 ? 0 : ReadTheSizeTag(head);
  }
  uint32_t GetFreeListByBlockOffset(uint32_t blk) {
    auto blk_size = OffsetToPtr<const BlockHeader>(blk)->size;
    return GetFreeListByBlockSize(blk_size);
  }
  uint32_t GetFreeListByBlockSize(uint32_t blk_size) {
    assert(blk_size >= 16);
    if (blk_size <= kMaxBlockSize) {
      return (blk_size / 16 - 1) * sizeof(uint32_t);
    }
    return GetLastFreeList();
  }
  static uint32_t RoundupBlockSize(uint32_t blk_size) {
    if (blk_size <= kMaxBlockSize) {
      return (blk_size + 16 - 1) / 16 * 16;
    }
    assert(false);
    return 0;
  }
  static uint32_t GetLastFreeList() {
    return (kNumFreeLists - 1) * sizeof(uint32_t);
  }
  void SetTheFreeTag(uint32_t blk, uint32_t blk_size) {
    auto next_blk = blk + blk_size;
    Write(next_blk, Read<uint32_t>(next_blk) | BlockHeader::kFreeMask);
  }
  void ResetTheFreeTag(uint32_t blk, uint32_t blk_size) {
    auto next_blk = blk + blk_size;
    Write(next_blk, Read<uint32_t>(next_blk) & ~BlockHeader::kFreeMask);
  }
  uint32_t ReadTheFreeTag(uint32_t blk, uint32_t blk_size) {
    return Read<uint32_t>(blk + blk_size) & BlockHeader::kFreeMask;
  }
  void SetThePrevFreeTag(uint32_t blk) {
    Write(blk, BlockHeader::kFreeMask | Read<uint32_t>(blk));
  }
  void ResetThePrevFreeTag(uint32_t blk) {
    Write(blk, ~BlockHeader::kFreeMask & Read<uint32_t>(blk));
  }
  uint32_t ReadThePrevFreeTag(uint32_t blk) {
    return Read<uint32_t>(blk) & BlockHeader::kFreeMask;
  }
  uint32_t ReadTheSizeTag(uint32_t blk) {
    return Read<uint32_t>(blk) & ~BlockHeader::kFreeMask;
  }

  uint32_t AllocBlock(uint32_t size);
  void FreeBlock(uint32_t blk);
  
  // Remove the block from the free list, and it keeps free.
  void RemoveBlock(uint32_t blk, uint32_t blk_size);
  uint32_t SplitBlock(uint32_t blk, uint32_t blk_size, uint32_t needed_size);

 private:
  uintptr_t base_;
  FreeListManager* flm_;
  uint64_t cnt_writes_;
  ModificationList* modifications_;
};

} // namespace nvds

#endif // _NVDS_ALLOCATOR_H_
<file_sep>/src/basic_server.cc
#include "basic_server.h"

#include "session.h"

using boost::asio::ip::tcp;

namespace nvds {

BasicServer::BasicServer(uint16_t port)
    : tcp_acceptor_(tcp_service_,
                    tcp::endpoint(tcp::v4(), port)),
      conn_sock_(tcp_service_) {
}

void BasicServer::Accept(MessageHandler recv_msg_handler,
                         MessageHandler send_msg_handler) {
  tcp_acceptor_.async_accept(conn_sock_,
      [this, recv_msg_handler, send_msg_handler](boost::system::error_code err) {
        if (!err) {
          std::make_shared<Session>(std::move(conn_sock_),
              recv_msg_handler, send_msg_handler)->Start();
        } else {
          NVDS_ERR(err.message().c_str());
        }
        Accept(recv_msg_handler, send_msg_handler);
      });
}

} // namespace nvds
<file_sep>/src/common.h
#ifndef _NVDS_COMMON_H_
#define _NVDS_COMMON_H_

#include "nvm.h"

#include <unistd.h>

#include <cassert>
#include <cinttypes>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <array>
#include <iostream>
#include <string>
#include <vector>

#define PACKED(s) s __attribute__((packed))

#ifndef DISALLOW_COPY_AND_ASSIGN
#define DISALLOW_COPY_AND_ASSIGN(TypeName)        \
  TypeName(const TypeName&) = delete;             \
  TypeName& operator=(const TypeName&) = delete;
#endif

#define NVDS_ERR(format, args...) {         \
  std::cerr << "error: "                    \
            << nvds::Format(format, ##args) \
            << std::endl;                   \
}

#define NVDS_LOG(format, args...) {                       \
  std::cout << nvds::Format(format, ##args) << std::endl; \
}

namespace nvds {

using ServerId = uint32_t;
using TabletId = uint32_t;

/*
 * Cluster configuration
 */
// Configurable
static const uint32_t kNumReplicas = 1;
static const uint32_t kNumServers = 2;
static const uint32_t kNumTabletsPerServer = 1;

static const uint16_t kCoordPort = 9090;
static const uint32_t kMaxItemSize = 1024;
static const uint32_t kNumTabletAndBackupsPerServer = kNumTabletsPerServer * (1 + kNumReplicas);
static const uint32_t kNumTablets = kNumTabletsPerServer * kNumServers;
static const uint32_t kNumTabletAndBackups = kNumTabletAndBackupsPerServer * kNumServers;
static_assert(kNumTablets % kNumServers == 0,
              "`kNumTablets` cannot be divisible by `kNumServers`");

/*
 * Infiniband configuration
 */
static const uint32_t kSendBufSize = 1024 + 128;
static const uint32_t kRecvBufSize = 1024 * 2 + 128;
static const uint32_t kIBUDPadding = 40;

std::string Format(const char* format, ...);
std::string Demangle(const char* name);

static inline void memset_volatile(volatile char* ptr, char v, size_t size) {
  auto end = ptr + size;
  while (ptr != end) {
    *ptr++ = v;
  }
}

} // namespace nvds

#endif // _NVDS_COMMON_H_
<file_sep>/test/test_nvds.c
#include "test.h"
#include "nvds.c"

int test_main() {
    return 0;
}
<file_sep>/src/response.h
#ifndef _NVDS_RESPONSE_H_
#define _NVDS_RESPONSE_H_

#include "common.h"
#include "infiniband.h"
#include "request.h"
#include "status.h"

namespace nvds {

struct Response {
  using Type = Request::Type;
  Type type;
  Status status;
  uint16_t val_len;
  char val[0];

  static Response* New(Infiniband::Buffer* b, Type type, Status status) {
    return new (b->buf) Response(type, status);
  }
  static void Del(const Response* r) {
    r->~Response();
  }
  uint32_t Len() const {
    return sizeof(Response) + (type == Type::GET ? val_len : 0);
  }
  void Print() const {
    std::cout << "type: " << (type == Type::GET ? "GET" : type == Type::PUT ? "PUT" : "DEL") << std::endl;
    std::cout << "status: " << (status == Status::OK ? "OK" : status == Status::ERROR ? "ERROR" : "NO_MEM") << std::endl;
    std::cout << "val_len: " << val_len << std::endl;
    std::cout << "val: ";
    for (size_t i = 0; i < val_len; ++i)
      std::cout << val[i];
    std::cout << std::endl;
  }
 private:
  Response(Request::Type t, Status s)
      : type(t), status(s), val_len(0) {
  }
};

} // namespace nvds

#endif // _NVDS_RESPONSE_H_
<file_sep>/src/client.cc
#include "client.h"

#include "request.h"
#include "response.h"

namespace nvds {

using json = nlohmann::json;

Client::Client(const std::string& coord_addr)
    : session_(Connect(coord_addr)),
      send_bufs_(ib_.pd(), kSendBufSize, kMaxIBQueueDepth, false),
      recv_bufs_(ib_.pd(), kRecvBufSize, kMaxIBQueueDepth, true) {
  // Infiniband
  qp_ = new Infiniband::QueuePair(ib_, IBV_QPT_UD,
      kMaxIBQueueDepth, kMaxIBQueueDepth);
  qp_->Activate();
  Join();
}

Client::~Client() {
  Close();
  delete qp_;  
}

void Client::Join() {
  Message msg {
    Message::Header {Message::SenderType::CLIENT, Message::Type::REQ_JOIN, 0},
    json().dump()
  };
  session_.SendMessage(msg);
  msg = session_.RecvMessage();
  assert(msg.sender_type() == Message::SenderType::COORDINATOR);
  assert(msg.type() == Message::Type::RES_JOIN);

  auto j_body = json::parse(msg.body());
  index_manager_ = j_body["index_manager"];
}

tcp::socket Client::Connect(const std::string& coord_addr) {
  tcp::resolver resolver {tcp_service_};
  tcp::socket conn_sock {tcp_service_};
  tcp::resolver::query query {coord_addr, std::to_string(kCoordPort)};
  auto ep = resolver.resolve(query);
  boost::asio::connect(conn_sock, ep);
  return conn_sock;
}

Client::Buffer* Client::RequestAndWait(const char* key, size_t key_len,
    const char* val, size_t val_len, Request::Type type) {
  assert(key_len + val_len <= kMaxItemSize);
  // 0. compute key hash
  auto hash = Hash(key, key_len);
  
  // 1. get tablet and server info
  //auto& tablet = index_manager_.GetTablet(hash);
  auto& server = index_manager_.GetServer(hash);
  // 2. post ib send and recv
  auto sb = send_bufs_.Alloc();
  assert(sb != nullptr);
  auto r = Request::New(sb, type, key, key_len, val, val_len, hash);
  auto rb = recv_bufs_.Alloc();
  assert(rb != nullptr);
  ib_.PostReceive(qp_, rb);
  ib_.PostSendAndWait(qp_, sb, r->Len(), &server.ib_addr);
  Request::Del(r);
  send_bufs_.Free(sb);
  assert(rb == ib_.Receive(qp_));
  return rb;
  //std::string ans(resp->val, resp->val_len);
  //recv_bufs_.Free(rb);
  //return ans;
}

} // namespace nvds
<file_sep>/src/exception.h
#ifndef _NVDS_EXCEPTION_H_
#define _NVDS_EXCEPTION_H_

#include "common.h"
#include "source_location.h"

#include <cstring>
#include <exception>
#include <typeinfo>

namespace nvds {

std::string Demangle(const char* name);

class Exception : public std::exception {
 public:
  explicit Exception (const SourceLocation& loc)
      : msg_(""), err_no_(0), loc_(loc) {}
  Exception(const SourceLocation& loc, const std::string& msg)
      : msg_(msg), err_no_(0), loc_(loc) {}
  Exception(const SourceLocation& loc, int err_no)
      : msg_(strerror(err_no)), err_no_(err_no), loc_(loc) {}
  Exception(const SourceLocation& loc, const std::string& msg, int err_no)
      : msg_(msg + ": " + strerror(err_no)), err_no_(err_no), loc_(loc) {}
  virtual ~Exception() {}
  
  std::string ToString() const {
    auto ret = Demangle(typeid(*this).name()) + ": " +
               msg_ + ", thrown at " + loc_.ToString();
    if (err_no_ != 0) {
        ret += " [errno: " + std::to_string(err_no_) + "]";
    }
    return ret;
  }
  const std::string& msg() const { return msg_; }
  int err_no() const { return err_no_; }
  const SourceLocation& loc() const { return loc_; }

 private:
  std::string msg_;
  int err_no_;
  SourceLocation loc_;
};

} // namespace nvds

#endif // _NVDS_EXCEPTION_H_
<file_sep>/test/test_consistent_hash.c
#include "test.h"
#include "consistent_hash.c"

#include <assert.h>
#include <limits.h>
#include <memory.h>
#include <stdlib.h>
#include <time.h>

clock_t begin;
clock_t delete_clk = 0;

static void __rbtree_print(rbtree_node_t* root) {
  if (root == rbtree_nil)
    return;
  __rbtree_print(root->left);
  printf("%d ", root->key);
  __rbtree_print(root->right);
}

__attribute__((unused)) static void rbtree_print(rbtree_node_t* root) {
  __rbtree_print(root);
  printf("\n");
}

static void statistic(cluster_t* c, int request_num) {
  cluster_node_t** nodes = malloc(sizeof(cluster_node_t*) * c->node_num);
  memset(nodes, 0, sizeof(cluster_node_t*) * c->node_num);
  int* requests = malloc(sizeof(int) * c->node_num);
  memset(requests, 0, sizeof(int) * c->node_num);
  int cnt = 0;

  srand(time(NULL));
  for (int i = 0; i < request_num; ++i) {
    hash_t key = rand() % INT_MAX;
    cluster_node_t* node = cluster_find(c, key);
    assert(node);
    
    int j;
    for (j = 0; j < cnt; ++j) {
      if (nodes[j] == node)
        break;
    }
    ++requests[j];
    if (j == cnt) {
      nodes[j] = node;
      ++cnt;
    }
  }

  printf("total requests: %d\n", request_num);
  printf("total nodes: %d\n", c->node_num);
  printf("average expection: %f%%\n", 100.0 / c->node_num);
  for (int i = 0; i < cnt; ++i) {
    printf("%s: %d: %f%%\n", nodes[i]->name,
        requests[i], 100.0 * requests[i] / request_num);
  }
  printf("\n");
  fflush(stdout);
  free(nodes);
  free(requests);
}

static void test_cluster_remove(cluster_t* c, const char* name) {
  clock_t begin = clock();
  cluster_remove(c, name);
  delete_clk += clock() - begin;
}

static void test(void) {
  cluster_t c;
  cluster_init(&c);
  const char* node_names[] = {
    "192.168.1.87",
    "192.168.1.88",
    "192.168.1.89",
    "192.168.1.90"
  };

  int node_num = sizeof(node_names) / sizeof(const char*);
  for (int i = 0; i < node_num; ++i) {
    printf("add %s\n", node_names[i]);
    fflush(stdout);
    cluster_add(&c, node_names[i]);
  }

  const int request_num = 500000;

  statistic(&c, request_num);

  printf("add %s\n", "192.168.1.91");
  fflush(stdout);
  cluster_add(&c, "192.168.1.91");
  statistic(&c, request_num);
  
  printf("add %s\n", "192.168.1.92");
  fflush(stdout);
  cluster_add(&c, "192.168.1.92");
  printf("add %s\n", "192.168.1.93");
  fflush(stdout);
  cluster_add(&c, "192.168.1.93");
  statistic(&c, request_num);
  
  printf("remove %s\n", "192.168.1.87");
  fflush(stdout);
  test_cluster_remove(&c, "192.168.1.87");
  statistic(&c, request_num);

  //rbtree_print(c.root);
  printf("remove %s\n", "192.168.1.88");
  fflush(stdout);
  test_cluster_remove(&c, "192.168.1.88");
  
  printf("remove %s\n", "192.168.1.89");
  fflush(stdout);
  test_cluster_remove(&c, "192.168.1.89");
  
  printf("remove %s\n", "192.168.1.90");
  fflush(stdout);
  test_cluster_remove(&c, "192.168.1.90");
  
  printf("remove %s\n", "192.168.1.91");
  fflush(stdout);
  test_cluster_remove(&c, "192.168.1.91");
  statistic(&c, request_num);

  cluster_destroy(&c);
}

int test_main(void) {
  begin = clock();
  test();
  printf("total time: %fs\n", 1.0 * (clock() - begin) / CLOCKS_PER_SEC);
  printf("delete time: %fs\n", 1.0 * delete_clk / CLOCKS_PER_SEC);
  return 0;
}
<file_sep>/src/client_main.cc
#include "client.h"

#include <chrono>
#include <iostream>
#include <memory>
#include <random>
#include <thread>

using namespace nvds;

using VecStr = std::vector<std::string>;

static Client* client;

static void SigInt(int signo) {
  std::cout << std::endl << "num_send: " << client->num_send() << std::endl;
  exit(0);
}

static void Usage() {
  std::cout << "usage: " << std::endl;
  std::cout << "client  <coordinator_addr> <item number per thread> <thread numer>" << std::endl;
}

static VecStr GenRandomStrings(size_t len, size_t n) {
  VecStr ans;
  std::default_random_engine g;
  std::uniform_int_distribution<int> dist('a', 'z');
  for (size_t i = 0; i < n; ++i) {
    ans.emplace_back(len, 0);
    for (size_t j = 0; j < len; ++j) {
      ans[i][j] = dist(g);
    }
  }
  return ans;
}

static void Work(double* qps, const std::string& coord_addr, size_t n) {
  using namespace std::chrono;
  try {
    Client c {coord_addr};
    client = &c;
    auto keys = GenRandomStrings(16, n);
    auto vals = VecStr(n, std::string(16, 'a'));
    auto begin = high_resolution_clock::now();
    for (size_t i = 0; i < keys.size(); ++i) {
      c.Put(keys[i], vals[i]);
    }
    auto end = high_resolution_clock::now();
    double t = duration_cast<duration<double>>(end - begin).count();
    *qps = keys.size() / t;
  } catch (boost::system::system_error& e) {
    NVDS_ERR(e.what());
  } catch (TransportException& e) {
    NVDS_ERR(e.msg().c_str());
    NVDS_LOG("initialize infiniband devices failed");
  }
}

static int bench_main(int argc, const char* argv[]) {
  if (argc < 4) {
    Usage();
    return -1;
  }

  const std::string coord_addr = argv[1];
  const size_t num_items = std::stoi(argv[2]);
  const size_t num_threads = std::stoi(argv[3]);
  
  std::vector<double> qpss(num_threads);
  std::vector<std::thread> workers;
  for (size_t i = 0; i < num_threads; ++i) {
    workers.emplace_back(std::bind(Work, &qpss[i], coord_addr, num_items));
  }
  for (size_t i = 0; i < num_threads; ++i) {
    workers[i].join();
  }

  for (auto& qps : qpss) {
    NVDS_LOG("QPS: %.2f", qps);
  }
  NVDS_LOG("total QPS: %.2f", std::accumulate(qpss.begin(), qpss.end(), 0.0));
  return 0;
}

/*
static int function_test_main(int argc, const char* argv[]) {
  assert(argc >= 2);
  std::string coord_addr {argv[1]};
  try {
    Client c {coord_addr};
    client = &c;
    std::string author = "DDST-SJTU";
    std::string version = "0.1.0";
    c.Put("author", "DDST-SJTU");
    c.Put("version", "0.1.0");
    std::cout << "author: "  << c.Get("author")  << std::endl;
    std::cout << "version: " << c.Get("version") << std::endl;

    c.Put("author", "wgtdkp");
    c.Put("version", "0.2.0");
    std::cout << "author: "  << c.Get("author")  << std::endl;
    std::cout << "version: " << c.Get("version") << std::endl;

    c.Del("author");
    c.Del("version");
    std::cout << "author: "  << c.Get("author")  << std::endl;
    std::cout << "version: " << c.Get("version") << std::endl;
  } catch (boost::system::system_error& e) {
    NVDS_ERR(e.what());
  } catch (TransportException& e) {
    NVDS_ERR(e.msg().c_str());
    NVDS_LOG("initialize infiniband devices failed");
  }
  return 0;
}
*/

int main(int argc, const char* argv[]) {
  signal(SIGINT, SigInt);
  //return function_test_main(argc, argv);
  return bench_main(argc, argv);
}
<file_sep>/src/coordinator.h
#ifndef _NVDS_COORDINATOR_H_
#define _NVDS_COORDINATOR_H_

#include "basic_server.h"
#include "index.h"

#include <boost/asio.hpp>

namespace nvds {

class Session;

/*
 * Coordinator do not serve any frequent reuqests from client,
 * like the get/put operation. It accepts various messages,
 * controls server recovery/joining/removing and failure
 * detection. It maintains the metadata of the whole cluster.
 *
 * At any given time, there is at most one active coordinator.
 * But several standby coordinators are preparing to take in charge
 * in case of the primary coordinator is down.
 */
class Coordinator : public BasicServer {
 public:
  Coordinator();
  ~Coordinator() {}
  DISALLOW_COPY_AND_ASSIGN(Coordinator);

  uint64_t total_storage() const { return total_storage_; }

  IndexManager& index_manager() { return index_manager_; }
  const IndexManager& index_manager() const { return index_manager_; }

  void Run() override;

 private:
  void HandleRecvMessage(std::shared_ptr<Session> session,
                         std::shared_ptr<Message> msg);
  void HandleSendMessage(std::shared_ptr<Session> session,
                         std::shared_ptr<Message> msg);
  void HandleMessageFromServer(std::shared_ptr<Session> session,
                               std::shared_ptr<Message> msg);
  void HandleServerRequestJoin(std::shared_ptr<Session> session,
                               std::shared_ptr<Message> req);
  void ResponseAllJoins();
  void HandleMessageFromClient(std::shared_ptr<Session> session,
                               std::shared_ptr<Message> msg);
  void HandleClientRequestJoin(std::shared_ptr<Session> session,
                               std::shared_ptr<Message> msg);

 private:
  uint32_t num_servers_ = 0;
  uint64_t total_storage_ = 0;

  IndexManager index_manager_;
  std::vector<std::shared_ptr<Session>> sessions_;
};

} // namespace nvds

#endif // NVDS_COORDINATOR_H_
<file_sep>/cs/Makefile
CC=gcc -std=c99
CFLAGS=-Wall -g -D_GNU_SOURCE
LDFLAGS= -libverbs -lrdmacm

all: common

common: common.o
	${CC} -o $@ $^ ${LDFLAGS}

ud: ud.o
	${CC} -o $@ $^ ${LDFLAGS}

ud.o: ud.c

common.o: common.c common.h

clean:
	rm -f common common.o
<file_sep>/src/common.cc
#include "common.h"

#include <boost/asio.hpp>
#include <cstdarg>
#include <cxxabi.h>

namespace nvds {

static std::string VFormat(const char* format, va_list ap);

/// A safe version of sprintf.
std::string Format(const char* format, ...) {
  va_list ap;
  va_start(ap, format);
  const auto& s = VFormat(format, ap);
  va_end(ap);
  return s;
}

/// A safe version of vprintf.
static std::string VFormat(const char* format, va_list ap) {
  std::string s;

  // We're not really sure how big of a buffer will be necessary.
  // Try 1K, if not the return value will tell us how much is necessary.
  int bufSize = 1024;
  while (true) {
    char buf[bufSize];
    // vsnprintf trashes the va_list, so copy it first
    va_list aq;
    __va_copy(aq, ap);
    int r = vsnprintf(buf, bufSize, format, aq);
    assert(r >= 0); // old glibc versions returned -1
    if (r < bufSize) {
      s = buf;
      break;
    }
    bufSize = r + 1;
  }
  return s;
}

std::string Demangle(const char* name) {
  int status;
  char* res = abi::__cxa_demangle(name, NULL, NULL, &status);
  if (status != 0) {
    return name;
  }
  std::string ret(res);
  free(res);
  return ret;
}

} // namespace nvds
<file_sep>/src/Makefile
CXX = g++ -std=c++14 -O3
LD = g++
CXXINC = -I./ -I../json/src/
#CXXFLAGS = -g -fsanitize=address -Wall -D_XOPEN_SOURCE -D_GNU_SOURCE $(CXXINC)
CXXFLAGS = -g -Wall -D_XOPEN_SOURCE -D_GNU_SOURCE $(CXXINC)
LDFLAGS = 
#LDLIBS = -lasan -libverbs -lboost_system -lboost_thread -lrt -lpthread -lssl -lcrypto
LDLIBS = -libverbs -lboost_system -lboost_thread -lrt -lpthread -lssl -lcrypto


SRCS = allocator.cc         \
	   basic_server.cc		\
       client.cc			\
	   common.cc			\
	   coordinator.cc		\
	   index.cc				\
	   infiniband.cc		\
	   message.cc			\
	   server.cc			\
	   session.cc			\
	   tablet.cc			\
	   MurmurHash2.cc

OBJS_DIR = build/
OBJS = $(addprefix $(OBJS_DIR), $(SRCS:.cc=.o))

GTEST_DIR = ../googletest/googletest
GTEST_FLAGS = -isystem ${GTEST_DIR}/include -pthread
$(OBJS_DIR)libgtest.a:
	$(CXX) $(GTEST_FLAGS) -I${GTEST_DIR} -c ${GTEST_DIR}/src/gtest-all.cc -o $(OBJS_DIR)gtest-all.o
	ar -rv $@ $(OBJS_DIR)gtest-all.o

default: all

all:
	@mkdir -p $(OBJS_DIR)
	@make coordinator server client library

coordinator: $(OBJS) $(OBJS_DIR)coordinator_main.o
	$(LD) $(LDFLAGS) -o $(OBJS_DIR)$@ $^ $(LDLIBS)

server: $(OBJS) $(OBJS_DIR)server_main.o
	$(LD) $(LDFLAGS) -o $(OBJS_DIR)$@ $^ $(LDLIBS)

client: $(OBJS) $(OBJS_DIR)client_main.o
	$(LD) $(LDFLAGS) -o $(OBJS_DIR)$@ $^ $(LDLIBS)

library: $(OBJS)
	ar rcs $(OBJS_DIR)libnvds.a $(OBJS)

$(OBJS_DIR)%.o: %.cc
	$(CXX) $(CXXFLAGS) -o $@ -c $^
	
$(OBJS_DIR)coordinator_main.o: coordinator_main.cc
	$(CXX) $(CXXFLAGS) -o $@ -c $^

$(OBJS_DIR)server_main.o : server_main.cc
	$(CXX) $(CXXFLAGS) -o $@ -c $^

$(OBJS_DIR)client_main.o : client_main.cc
	$(CXX) $(CXXFLAGS) -o $@ -c $^

$(OBJS_DIR)test_infiniband.o: test_infiniband.cc infiniband.h
	$(CXX) $(CXXFLAGS) $(GTEST_FLAGS) -o $@ -c test_infiniband.cc
test_infiniband: $(OBJS_DIR)test_infiniband.o $(OBJS_DIR)infiniband.o $(OBJS_DIR)common.o $(OBJS_DIR)libgtest.a
	$(LD) $(LDFLAGS) $(GTEST_FLAGS) -o $(OBJS_DIR)$@ $^ $(LDLIBS)

$(OBJS_DIR)test_allocator.o: test_allocator.cc allocator.h
	$(CXX) $(CXXFLAGS) $(GTEST_FLAGS) -o $@ -c test_allocator.cc
test_allocator: $(OBJS_DIR)test_allocator.o $(OBJS_DIR)allocator.o $(OBJS_DIR)libgtest.a
	$(LD) $(LDFLAGS) $(GTEST_FLAGS) -o $(OBJS_DIR)$@ $^ $(LDLIBS)

$(OBJS_DIR)test_stl_map.o: test_stl_map.cc hash.h
	$(CXX) $(CXXFLAGS) $(GTEST_FLAGS) -o $@ -c test_stl_map.cc
test_stl_map: $(OBJS_DIR)test_stl_map.o $(OBJS_DIR)MurmurHash2.o $(OBJS_DIR)libgtest.a
	$(LD) $(LDFLAGS) $(GTEST_FLAGS) -o $(OBJS_DIR)$@ $^ $(LDLIBS)

$(OBJS_DIR)test_message.o: test_message.cc message.h
	$(CXX) $(CXXFLAGS) $(GTEST_FLAGS) -o $@ -c test_message.cc
test_message: $(OBJS_DIR)test_message.o $(OBJS_DIR)message.o $(OBJS_DIR)libgtest.a
	$(LD) $(LDFLAGS) $(GTEST_FLAGS) -o $(OBJS_DIR)$@ $^ $(LDLIBS)

install:
	@sudo mkdir -p /usr/include/nvds/
	@sudo cp ./*.h /usr/include/nvds/
	@sudo cp ../json/src/json.hpp /usr/include/nvds/
	@sudo cp ./build/libnvds.a /usr/lib/

.PHONY: clean

clean-objs:
	@rm -f $(OBJS_DIR)*.o

clean:
	@rm -rf $(OBJS_DIR)
<file_sep>/src/transport.h
#ifndef _NVDS_TRANSPORT_H_
#define _NVDS_TRANSPORT_H_

#include "exception.h"

namespace nvds {
class Message;

struct TransportException : public Exception {
  explicit TransportException(const SourceLocation& loc)
      : Exception(loc) {}
  TransportException(const SourceLocation& loc, const std::string& msg)
      : Exception(loc, msg) {}
  TransportException(const SourceLocation& loc, int err_no)
      : Exception(loc, err_no) {}
  TransportException(const SourceLocation& loc,
                     const std::string& msg, int err_no)
      : Exception(loc, msg, err_no) {}
};

class Transport {
 public:
  static const uint32_t kMaxRPCLen = (1 << 23) + 200;

  class ServerRPC {

  };
};

} // namespace nvds

#endif // _NVDS_TRANSPORT_H_
<file_sep>/src/request.h
#ifndef _NVDS_REQUEST_H_
#define _NVDS_REQUEST_H_

#include "common.h"
#include "infiniband.h"

namespace nvds {

struct Request {
  enum class Type : uint8_t {
    PUT, ADD, GET, DEL
  };
  Type type;
  uint16_t key_len;
  uint16_t val_len;
  KeyHash key_hash;
  // TODO(wgtdkp):
  //uint64_t id;
  // Key data followed by value data
  char data[0];
 
  static Request* New(Infiniband::Buffer* b, Type type,
                      const char* key, size_t key_len,
                      const char* val, size_t val_len, KeyHash key_hash) {
    return new (b->buf) Request(type, key, key_len, val, val_len, key_hash);
  }
  static void Del(const Request* r) {
    // Explicitly call destructor(only when pairing with placement new)
    r->~Request();
  }
  size_t Len() const { return sizeof(Request) + key_len + val_len; }
  char* Key() { return data; }
  const char* Key() const { return data; }
  char* Val() { return data + key_len; }
  const char* Val() const { return data + key_len; }
  void Print() const {
    std::cout << "type: " << (type == Type::PUT ? "PUT" : type == Type::GET ? "GET" : "DEL") << std::endl;
    std::cout << "key_len: " << key_len << std::endl;
    std::cout << "val_len: " << val_len << std::endl;
    std::cout << "key_hash: " << key_hash << std::endl;
    std::cout << "key: ";
    for (size_t i = 0; i < key_len; ++i)
      std::cout << data[i];
    std::cout << std::endl;
    std::cout << "val: ";
    for (size_t i = key_len; i < key_len + val_len; ++i)
      std::cout << data[i];
    std::cout << std::endl;
  }

 private:
  Request(Type type, const char* key, size_t key_len,
      const char* val, size_t val_len, KeyHash key_hash)
      : type(type), key_len(key_len), val_len(val_len), key_hash(key_hash) {
    memcpy(data, key, key_len);
    if (val != nullptr && val_len > 0) {
      memcpy(data + key_len, val, val_len);
    }
  }
};

} // namespace nvds

#endif // _NVDS_SERVER_H_
<file_sep>/src/message.cc
#include "message.h"

#include "index.h"

namespace nvds {

void TabletInfo::Print() const {
  std::cout << "id: " << id << std::endl;
  std::cout << "server id: " << server_id << std::endl;
  std::cout << "is backup: " << is_backup << std::endl;
  if (is_backup) {
    std::cout << "master id: " << master << std::endl;
    qpis[0].Print();
  } else {
    std::cout << "backups id: ";
    for (size_t i = 0; i < backups.size(); ++i) {
      std::cout << backups[i] << (i + 1 == backups.size() ? "\n" : ", ");
    }
    for (auto& qpi : qpis) {
      qpi.Print();
    }
  }
}

void to_json(nlohmann::json& j, const Infiniband::Address& ia) {
  j = {
    {"ib_port", ia.ib_port},
    {"lid", ia.lid},
    {"qpn", ia.qpn}
  };
}

void from_json(const nlohmann::json& j, Infiniband::Address& ia) {
  ia = {
    j["ib_port"],
    j["lid"],
    j["qpn"],
  };
}

void to_json(nlohmann::json& j, const Infiniband::QueuePairInfo& qpi) {
  j["lid"] = qpi.lid;
  j["qpn"] = qpi.qpn;
  j["psn"] = qpi.psn;
  j["rkey"] = qpi.rkey;
  j["vaddr"] = qpi.vaddr;
}

void from_json(const nlohmann::json& j, Infiniband::QueuePairInfo& qpi) {
  qpi.lid = j["lid"];
  qpi.qpn = j["qpn"];
  qpi.psn = j["psn"];
  qpi.rkey = j["rkey"];
  qpi.vaddr = j["vaddr"];
}

void to_json(nlohmann::json& j, const TabletInfo& ti) {
  j["id"] = ti.id;
  j["server_id"] = ti.server_id;
  j["is_backup"] = ti.is_backup;
  j["qpis"] = ti.qpis;
  if (ti.is_backup) {
    j["master"] = ti.master;
  } else {
    j["backups"] = ti.backups;
  }
}

void from_json(const nlohmann::json& j, TabletInfo& ti) {
  ti.id = j["id"];
  ti.server_id = j["server_id"];
  ti.is_backup = j["is_backup"];
  ti.qpis = j["qpis"];
  if (ti.is_backup) {
    ti.master = j["master"];
  } else {
    ti.backups = j["backups"];
  }
}

void to_json(nlohmann::json& j, const ServerInfo& si) {
  j = {
    {"id", si.id},
    {"active", si.active},
    {"addr", si.addr},
    {"ib_addr", si.ib_addr},
    {"tablets", si.tablets}
  };
}

void from_json(const nlohmann::json& j, ServerInfo& si) {
  si = {
    j["id"],
    j["active"],
    j["addr"],
    j["ib_addr"],
    j["tablets"]
  };
}

void to_json(nlohmann::json& j, const IndexManager& im) {
  j = {
    {"key_tablet_map", im.key_tablet_map_},
    {"tablets", im.tablets_},
    {"servers", im.servers_}
  };
}

void from_json(const nlohmann::json& j, IndexManager& im) {
  im.key_tablet_map_ = j["key_tablet_map"];
  im.tablets_ = j["tablets"];
  im.servers_ = j["servers"];
}

} // namespace nvds
<file_sep>/src/test_message.cc
#include "json.hpp"
#include "message.h"
#include <gtest/gtest.h>

using namespace std;
using namespace nvds;
using namespace nlohmann;

TEST (MessageTest, Backup) {
  string r = R"([123, 456])";
  Backup b = json::parse(r);
  EXPECT_EQ(b.server_id, 123);
  EXPECT_EQ(b.tablet_id, 456);

  json j = Backup {111, 222};
  EXPECT_EQ(j[0], 111);
  EXPECT_EQ(j[1], 222);
}

TEST (MessageTest, TabletInfo) {
  string r = R"({
    "id": 3,
    "is_backup": true,
    "master": [123, 456]
  })";
  TabletInfo t = json::parse(r);
  EXPECT_EQ(t.id, 3);
  EXPECT_EQ(t.is_backup, true);
  EXPECT_EQ(t.master.server_id, 123);
  EXPECT_EQ(t.master.tablet_id, 456);
  
  t = TabletInfo { 3, true};
  t.master = {123, 456};
  json j = t;
  EXPECT_EQ(j["id"], 3);
  EXPECT_EQ(j["is_backup"], true);
  EXPECT_EQ(j["master"][0], 123);
  EXPECT_EQ(j["master"][1], 456);
  
  
  r = R"({
    "id": 3,
    "is_backup": false,
    "backups": [[123, 456], [789, 101112]]
  })";
  t = json::parse(r);
  EXPECT_EQ(t.id, 3);
  EXPECT_EQ(t.is_backup, false);
  EXPECT_EQ(t.backups[0].server_id, 123);
  EXPECT_EQ(t.backups[0].tablet_id, 456);
  EXPECT_EQ(t.backups[1].server_id, 789);
  EXPECT_EQ(t.backups[1].tablet_id, 101112);

  //j = TabletInfo { 3, false};
  t = {3, false};
  t.backups[0] = {123, 456};
  t.backups[1] = {789, 101112};
  j = t;
  EXPECT_EQ(j["id"], 3);
  EXPECT_EQ(j["is_backup"], false);
  EXPECT_EQ(j["backups"][0][0], 123);
  EXPECT_EQ(j["backups"][0][1], 456);
  EXPECT_EQ(j["backups"][1][0], 789);
  EXPECT_EQ(j["backups"][1][1], 101112);
}

TEST (MessageTest, InfinibandAddress) {
  string r = R"({
    "ib_port": 1,
    "lid": 2,
    "qpn": 3
  })";
  Infiniband::Address addr = json::parse(r);
  EXPECT_EQ(addr.ib_port, 1);
  EXPECT_EQ(addr.lid, 2);
  EXPECT_EQ(addr.qpn, 3);

  json j = Infiniband::Address {1, 2, 3};
  EXPECT_EQ(j["ib_port"], 1);
  EXPECT_EQ(j["lid"], 2);
  EXPECT_EQ(j["qpn"], 3);
}

TEST (MessageTest, ServerInfo) {
  string r = R"({
    "id": 1,
    "active": true,
    "addr": "192.168.1.12",
    "ib_addr": {
      "ib_port": 11,
      "lid": 12,
      "qpn": 13
    },
    "tablets": [
      {
        "id": 0,
        "is_backup": true,
        "master": [11, 13]
      }
    ]
  })";
  ServerInfo s = json::parse(r);
  EXPECT_EQ(s.id, 1);
  EXPECT_EQ(s.active, true);
  EXPECT_EQ(s.addr, "192.168.1.12");
  EXPECT_EQ(s.ib_addr, Infiniband::Address({11, 12, 13}));
  EXPECT_EQ(s.tablets[0].id, 0);
  EXPECT_EQ(s.tablets[0].is_backup, true);
  EXPECT_EQ(s.tablets[0].master, Backup({11, 13}));

  json j = s;
  EXPECT_EQ(j["id"], 1);
  EXPECT_EQ(j["active"], true);
  EXPECT_EQ(j["addr"], "192.168.1.12");
  EXPECT_EQ(j["ib_addr"], Infiniband::Address({11, 12, 13}));
  EXPECT_EQ(j["tablets"][0]["id"], 0);
  EXPECT_EQ(j["tablets"][0]["is_backup"], true);
  EXPECT_EQ(j["tablets"][0]["master"][0], 11);
  EXPECT_EQ(j["tablets"][0]["master"][1], 13);
}

int main(int argc, char* argv[]) {
  testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
  return 0;
}
<file_sep>/test/test_print.c
#include <stdio.h>
#include <inttypes.h>
int main (void) {
    uint32_t a=1234;
    uint16_t b=5678;
    printf("%" PRIu32 "\n",a);
    printf("%" PRIu16 "\n",b);
    return 0;
}
<file_sep>/src/index.h
/*
 * Mapping items to servers
 */

#ifndef _NVDS_INDEX_H_
#define _NVDS_INDEX_H_

#include "common.h"
#include "hash.h"
#include "infiniband.h"
#include "json.hpp"
#include "message.h"

#include <map>
#include <unordered_map>

namespace nvds {

class IndexManager {
  friend void to_json(nlohmann::json& j, const IndexManager& im);
  friend void from_json(const nlohmann::json& j, IndexManager& im);

 public:
  IndexManager() {}
  ~IndexManager() {}

  const ServerInfo& AddServer(const std::string& addr,
                              nlohmann::json& msg_body);
  const ServerInfo& GetServer(KeyHash key_hash) const {
    auto id = GetServerId(key_hash);
    const auto& ans = GetServer(id);
    assert(ans.id == id);
    return ans;
  }
  const ServerInfo& GetServer(ServerId id) const {
    return servers_[id];
  }
  ServerId GetServerId(KeyHash key_hash) const {       
    return GetTablet(key_hash).server_id;
  }
  // Get the id of the tablet that this key hash locates in.
  TabletId GetTabletId(KeyHash key_hash) const {
    uint32_t idx = key_hash / ((1.0 + kMaxKeyHash) / key_tablet_map_.size());
    return key_tablet_map_[idx];
  }
  const TabletInfo& GetTablet(KeyHash key_hash) const {
    return tablets_[GetTabletId(key_hash)];
  }
  //TabletInfo& GetTablet(KeyHash key_hash) {
  //  return tablets_[GetTabletId(key_hash)];
  //}
  const TabletInfo& GetTablet(TabletId id) const {
    return tablets_[id];
  }

  // DEBUG
  void PrintTablets() const;
  
 private:
  static ServerId AllocServerId() {
    static ServerId id = 0;
    return id++;
  }

  // `r_idx`: replication index; `s_idx`: server index; `t_idx`: tablet index;
  static uint32_t CalcTabletId(uint32_t s_idx, uint32_t r_idx, uint32_t t_idx) {
    return s_idx * kNumTabletAndBackupsPerServer + r_idx * kNumTabletsPerServer + t_idx;
  }

  std::array<TabletId, kNumTablets> key_tablet_map_;
  std::array<TabletInfo, kNumTabletAndBackups> tablets_;
  std::array<ServerInfo, kNumServers> servers_;
};

} // namespace nvds

#endif // _NVDS_INDEX_H_
<file_sep>/src/source_location.h
#ifndef _NVDS_SOURCE_LOCATION_H_
#define _NVDS_SOURCE_LOCATION_H_

#include <cstdint>
#include <string>

namespace nvds {

#define HERE  (SourceLocation {__FILE__, __func__, __LINE__})

struct SourceLocation {
  std::string file;
  std::string func;
  uint32_t line;

  std::string ToString() const {
    return file + ":" + func + ":" + std::to_string(line);
  }
};

} // namespace nvds

#endif // _NVDS_SOURCE_LOCATION_H_<file_sep>/src/spinlock.h
#ifndef _NVDS_SPINLOCK_H_
#define _NVDS_SPINLOCK_H_

#include "common.h"

#include <atomic>

namespace nvds {

class Spinlock {
 public:
  Spinlock() = default;
  DISALLOW_COPY_AND_ASSIGN(Spinlock);
  void lock() {
    while (flag_.test_and_set(std::memory_order_acquire)) {}
  }
  void unlock() {
    flag_.clear(std::memory_order_release);
  }

 private:
  std::atomic_flag flag_ = ATOMIC_FLAG_INIT;
};

} // namespace nvds

#endif // _NVDS_SPINLOCK_H_
<file_sep>/cs/udpp.c
/*
 * Copyright (c) 2005 Topspin Communications.  All rights reserved.
 * Copyright (c) 2008 Intel Corporation.  All rights reserved.
 *
 * This software is available to you under the OpenIB.org BSD license
 * below:
 *
 *     Redistribution and use in source and binary forms, with or
 *     without modification, are permitted provided that the following
 *     conditions are met:
 *
 *      - Redistributions of source code must retain the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer.
 *
 *      - Redistributions in binary form must reproduce the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer in the documentation and/or other materials
 *        provided with the distribution.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AWV
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */


#include <assert.h>
#include <ctype.h>
#include <errno.h>
#include <malloc.h>
#include <memory.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include <arpa/inet.h>
#include <infiniband/verbs.h>
#include <sys/socket.h>
#include <sys/time.h>
// Interprocess communication access
#include <sys/ipc.h>
// Shared memory facility
#include <sys/shm.h>

enum {
	PINGPONG_RECV_WRID = 1,
	PINGPONG_SEND_WRID = 2,
};

struct pingpong_context {
	struct ibv_context		*context;
	struct ibv_comp_channel *channel;
	struct ibv_pd			*pd;
	struct ibv_mr			*mr;
	struct ibv_cq			*cq;
	struct ibv_qp			*qp;
	struct ibv_ah			*ah;
	void					*buf;
	int						size;
	int						rx_depth;
	int						pending;
};

struct pingpong_dest {
	int lid;
	int qpn;
	int psn;
};

static int pp_connect_ctx(struct pingpong_context *ctx, int port, int my_psn,
			  int sl, struct pingpong_dest *dest)
{
	struct ibv_ah_attr ah_attr;
	struct ibv_qp_attr attr;

	memset(&ah_attr, 0, sizeof ah_attr);
	ah_attr.is_global     = 0;
	ah_attr.dlid          = (uint16_t) dest->lid;
	ah_attr.sl            = (uint8_t) sl;
	ah_attr.src_path_bits = 0;
	ah_attr.port_num      = (uint8_t) port;

	attr.qp_state = IBV_QPS_RTR;

	if (ibv_modify_qp(ctx->qp, &attr, IBV_QP_STATE)) {
		fprintf(stderr, "Failed to modify QP to RTR\n");
		return 1;
	}

	attr.qp_state = IBV_QPS_RTS;
	attr.sq_psn	  = my_psn;

	if (ibv_modify_qp(ctx->qp, &attr,
			  IBV_QP_STATE | IBV_QP_SQ_PSN)) {
		fprintf(stderr, "Failed to modify QP to RTS\n");
		return 1;
	}

	ctx->ah = ibv_create_ah(ctx->pd, &ah_attr);
	if (!ctx->ah) {
		fprintf(stderr, "Failed to create AH\n");
		return 1;
	}

	return 0;
}

static struct pingpong_dest *pp_client_exch_dest(const char *servername, int port,
						 const struct pingpong_dest *my_dest)
{
	struct addrinfo *res, *t;
	struct addrinfo hints;
	char service[6];
	char msg[sizeof "0000:000000:000000"];
	int n;
	SOCKET sockfd = INVALID_SOCKET;
	struct pingpong_dest *rem_dest = NULL;

	memset(&hints, 0, sizeof hints);
	hints.ai_family   = AF_UNSPEC;
	hints.ai_socktype = SOCK_STREAM;
	sprintf(service, "%d\0", port);

	n = getaddrinfo(servername, service, &hints, &res);

	if (n != 0) {
		fprintf(stderr, "%s for %s:%d\n", gai_strerror(n), servername, port);
		return NULL;
	}

	for (t = res; t; t = t->ai_next) {
		sockfd = socket(t->ai_family, t->ai_socktype, t->ai_protocol);
		if (sockfd != INVALID_SOCKET) {
			if (!connect(sockfd, t->ai_addr, t->ai_addrlen))
				break;
			closesocket(sockfd);
			sockfd = INVALID_SOCKET;
		}
	}

	freeaddrinfo(res);

	if (sockfd == INVALID_SOCKET) {
		fprintf(stderr, "Couldn't connect to %s:%d\n", servername, port);
		return NULL;
	}

	sprintf(msg, "%04x:%06x:%06x", my_dest->lid, my_dest->qpn, my_dest->psn);
	if (send(sockfd, msg, sizeof msg, 0) != sizeof msg) {
		fprintf(stderr, "Couldn't send local address\n");
		goto out;
	}

	if (recv(sockfd, msg, sizeof msg, 0) != sizeof msg) {
		perror("client recv");
		fprintf(stderr, "Couldn't recv remote address\n");
		goto out;
	}

	send(sockfd, "done", sizeof "done", 0);

	rem_dest = malloc(sizeof *rem_dest);
	if (!rem_dest)
		goto out;

	sscanf(msg, "%x:%x:%x", &rem_dest->lid, &rem_dest->qpn, &rem_dest->psn);

out:
	closesocket(sockfd);
	return rem_dest;
}

static struct pingpong_dest *pp_server_exch_dest(struct pingpong_context *ctx,
						 int ib_port, int port, int sl,
						 const struct pingpong_dest *my_dest)
{
	struct addrinfo *res, *t;
	struct addrinfo hints;
	char service[6];
	char msg[sizeof "0000:000000:000000"];
	int n;
	SOCKET sockfd = INVALID_SOCKET, connfd;
	struct pingpong_dest *rem_dest = NULL;

	memset(&hints, 0, sizeof hints);
	hints.ai_flags    = AI_PASSIVE;
	hints.ai_family   = AF_UNSPEC;
	hints.ai_socktype = SOCK_STREAM;
	sprintf(service, "%d\0", port);

	n = getaddrinfo(NULL, service, &hints, &res);

	if (n != 0) {
		fprintf(stderr, "%s for port %d\n", gai_strerror(n), port);
		return NULL;
	}

	for (t = res; t; t = t->ai_next) {
		sockfd = socket(t->ai_family, t->ai_socktype, t->ai_protocol);
		if (sockfd != INVALID_SOCKET) {
			n = 0;
			setsockopt(sockfd, IPPROTO_IPV6, IPV6_V6ONLY, (char *) &n, sizeof n);
			n = 1;
			setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (char *) &n, sizeof n);

			if (!bind(sockfd, t->ai_addr, t->ai_addrlen))
				break;
			closesocket(sockfd);
			sockfd = INVALID_SOCKET;
		}
	}

	freeaddrinfo(res);

	if (sockfd == INVALID_SOCKET) {
		fprintf(stderr, "Couldn't listen to port %d\n", port);
		return NULL;
	}

	listen(sockfd, 1);
	connfd = accept(sockfd, NULL, 0);
	closesocket(sockfd);
	if (connfd == INVALID_SOCKET) {
		fprintf(stderr, "accept() failed\n");
		return NULL;
	}

	n = recv(connfd, msg, sizeof msg, 0);
	if (n != sizeof msg) {
		perror("server recv");
		fprintf(stderr, "%d/%d: Couldn't recv remote address\n", n, (int) sizeof msg);
		goto out;
	}

	rem_dest = malloc(sizeof *rem_dest);
	if (!rem_dest)
		goto out;

	sscanf(msg, "%x:%x:%x", &rem_dest->lid, &rem_dest->qpn, &rem_dest->psn);

	if (pp_connect_ctx(ctx, ib_port, my_dest->psn, sl, rem_dest)) {
		fprintf(stderr, "Couldn't connect to remote QP\n");
		free(rem_dest);
		rem_dest = NULL;
		goto out;
	}

	sprintf(msg, "%04x:%06x:%06x", my_dest->lid, my_dest->qpn, my_dest->psn);
	if (send(connfd, msg, sizeof msg, 0) != sizeof msg) {
		fprintf(stderr, "Couldn't send local address\n");
		free(rem_dest);
		rem_dest = NULL;
		goto out;
	}

	recv(connfd, msg, sizeof msg, 0);

out:
	closesocket(connfd);
	return rem_dest;
}

static struct pingpong_context *pp_init_ctx(struct ibv_device *ib_dev, int size,
					    int rx_depth, int port,
					    int use_event)
{
	struct pingpong_context *ctx;

	ctx = malloc(sizeof *ctx);
	if (!ctx)
		return NULL;

	ctx->size     = size;
	ctx->rx_depth = rx_depth;

	ctx->buf = malloc(size + 40);
	if (!ctx->buf) {
		fprintf(stderr, "Couldn't allocate work buf.\n");
		return NULL;
	}

	memset(ctx->buf, 0, size + 40);

	ctx->context = ibv_open_device(ib_dev);
	if (!ctx->context) {
		fprintf(stderr, "Couldn't get context for %s\n",
			ibv_get_device_name(ib_dev));
		return NULL;
	}

	if (use_event) {
		ctx->channel = ibv_create_comp_channel(ctx->context);
		if (!ctx->channel) {
			fprintf(stderr, "Couldn't create completion channel\n");
			return NULL;
		}
	} else
		ctx->channel = NULL;

	ctx->pd = ibv_alloc_pd(ctx->context);
	if (!ctx->pd) {
		fprintf(stderr, "Couldn't allocate PD\n");
		return NULL;
	}

	ctx->mr = ibv_reg_mr(ctx->pd, ctx->buf, size + 40, IBV_ACCESS_LOCAL_WRITE);
	if (!ctx->mr) {
		fprintf(stderr, "Couldn't register MR\n");
		return NULL;
	}

	ctx->cq = ibv_create_cq(ctx->context, rx_depth + 1, NULL,
				ctx->channel, 0);
	if (!ctx->cq) {
		fprintf(stderr, "Couldn't create CQ\n");
		return NULL;
	}

	{
		struct ibv_qp_init_attr attr;

		memset(&attr, 0, sizeof attr);
		attr.send_cq = ctx->cq;
		attr.recv_cq = ctx->cq;
		attr.cap.max_send_wr  = 1;
		attr.cap.max_recv_wr  = rx_depth;
		attr.cap.max_send_sge = 1;
		attr.cap.max_recv_sge = 1;
		attr.qp_type = IBV_QPT_UD,

		ctx->qp = ibv_create_qp(ctx->pd, &attr);
		if (!ctx->qp)  {
			fprintf(stderr, "Couldn't create QP\n");
			return NULL;
		}
	}

	{
		struct ibv_qp_attr attr;
		
		memset(&attr, 0, sizeof attr);
		attr.qp_state        = IBV_QPS_INIT;
		attr.pkey_index      = 0;
		attr.port_num        = (uint8_t) port;
		attr.qkey            = 0x11111111;

		if (ibv_modify_qp(ctx->qp, &attr,
				  IBV_QP_STATE              |
				  IBV_QP_PKEY_INDEX         |
				  IBV_QP_PORT               |
				  IBV_QP_QKEY)) {
			fprintf(stderr, "Failed to modify QP to INIT\n");
			return NULL;
		}
	}

	return ctx;
}

int pp_close_ctx(struct pingpong_context *ctx)
{
	if (ibv_destroy_qp(ctx->qp)) {
		fprintf(stderr, "Couldn't destroy QP\n");
		return 1;
	}

	if (ibv_destroy_cq(ctx->cq)) {
		fprintf(stderr, "Couldn't destroy CQ\n");
		return 1;
	}

	if (ibv_dereg_mr(ctx->mr)) {
		fprintf(stderr, "Couldn't deregister MR\n");
		return 1;
	}

	if (ibv_destroy_ah(ctx->ah)) {
		fprintf(stderr, "Couldn't destroy AH\n");
		return 1;
	}

	if (ibv_dealloc_pd(ctx->pd)) {
		fprintf(stderr, "Couldn't deallocate PD\n");
		return 1;
	}

	if (ctx->channel) {
		if (ibv_destroy_comp_channel(ctx->channel)) {
			fprintf(stderr, "Couldn't destroy completion channel\n");
			return 1;
		}
	}

	if (ibv_close_device(ctx->context)) {
		fprintf(stderr, "Couldn't release context\n");
		return 1;
	}

	free(ctx->buf);
	free(ctx);

	return 0;
}

static int pp_post_recv(struct pingpong_context *ctx, int n)
{
	struct ibv_sge list;
	struct ibv_recv_wr wr;
	struct ibv_recv_wr *bad_wr;
	int i;

	list.addr	= (uintptr_t) ctx->buf;
	list.length = ctx->size + 40;
	list.lkey	= ctx->mr->lkey;
	wr.next		= NULL;
	wr.wr_id	= PINGPONG_RECV_WRID;
	wr.sg_list  = &list;
	wr.num_sge  = 1;

	for (i = 0; i < n; ++i)
		if (ibv_post_recv(ctx->qp, &wr, &bad_wr))
			break;

	return i;
}

static int pp_post_send(struct pingpong_context *ctx, uint32_t qpn)
{
	struct ibv_sge list;
	struct ibv_send_wr wr;
	struct ibv_send_wr *bad_wr;

	list.addr	  = (uintptr_t) ctx->buf + 40;
	list.length   = ctx->size;
	list.lkey	  = ctx->mr->lkey;
	wr.next		  = NULL;
	wr.wr_id	  = PINGPONG_SEND_WRID;
	wr.sg_list    = &list;
	wr.num_sge    = 1;
	wr.opcode     = IBV_WR_SEND;
	wr.send_flags = IBV_SEND_SIGNALED;
	wr.wr.ud.ah			 = ctx->ah;
	wr.wr.ud.remote_qpn  = qpn;
	wr.wr.ud.remote_qkey = 0x11111111;

	return ibv_post_send(ctx->qp, &wr, &bad_wr);
}

static void usage(const char *argv0)
{
	printf("Usage:\n");
	printf("  %s            start a server and wait for connection\n", argv0);
	printf("  %s -h <host>  connect to server at <host>\n", argv0);
	printf("\n");
	printf("Options:\n");
	printf("  -p <port>     listen on/connect to port <port> (default 18515)\n");
	printf("  -d <dev>      use IB device <dev> (default first device found)\n");
	printf("  -i <port>     use port <port> of IB device (default 1)\n");
	printf("  -s <size>     size of message to exchange (default 2048)\n");
	printf("  -r <dep>      number of receives to post at a time (default 500)\n");
	printf("  -n <iters>    number of exchanges (default 1000)\n");
	printf("  -e            sleep on CQ events (default poll)\n");
}

int __cdecl main(int argc, char *argv[])
{
	struct ibv_device		**dev_list;
	struct ibv_device		*ib_dev;
	struct pingpong_context *ctx;
	struct pingpong_dest    my_dest;
	struct pingpong_dest    *rem_dest;
	LARGE_INTEGER			start, end, freq;
	char                    *ib_devname = NULL;
	char                    *servername = NULL;
	int                     port = 18515;
	int                     ib_port = 1;
	int                     size = 2048;
	int                     rx_depth = 500;
	int                     iters = 1000;
	int                     use_event = 0;
	int                     routs;
	int                     rcnt, scnt;
	int                     num_cq_events = 0;
	int                     sl = 0;
	WORD					version;
	WSADATA					data;
	int						err;

	srand((unsigned int) time(NULL));
	version = MAKEWORD(2, 2);
	err = WSAStartup(version, &data);
	if (err)
		return -1;

	while (1) {
		int c;

		c = getopt(argc, argv, "h:p:d:i:s:r:n:l:e");
		if (c == -1)
			break;

		switch (c) {
		case 'p':
			port = strtol(optarg, NULL, 0);
			if (port < 0 || port > 65535) {
				usage(argv[0]);
				return 1;
			}
			break;

		case 'd':
			ib_devname = _strdup(optarg);
			break;

		case 'i':
			ib_port = strtol(optarg, NULL, 0);
			if (ib_port < 0) {
				usage(argv[0]);
				return 1;
			}
			break;

		case 's':
			size = strtol(optarg, NULL, 0);
			break;

		case 'r':
			rx_depth = strtol(optarg, NULL, 0);
			break;

		case 'n':
			iters = strtol(optarg, NULL, 0);
			break;

		case 'l':
			sl = strtol(optarg, NULL, 0);
			break;

		case 'e':
			++use_event;
			break;

		case 'h':
			servername = _strdup(optarg);
			break;

		default:
			usage(argv[0]);
			return 1;
		}
	}

	dev_list = ibv_get_device_list(NULL);
	if (!dev_list) {
		fprintf(stderr, "No IB devices found\n");
		return 1;
	}

	if (!ib_devname) {
		ib_dev = *dev_list;
		if (!ib_dev) {
			fprintf(stderr, "No IB devices found\n");
			return 1;
		}
	} else {
		int i;
		for (i = 0; dev_list[i]; ++i)
			if (!strcmp(ibv_get_device_name(dev_list[i]), ib_devname))
				break;
		ib_dev = dev_list[i];
		if (!ib_dev) {
			fprintf(stderr, "IB device %s not found\n", ib_devname);
			return 1;
		}
	}

	ctx = pp_init_ctx(ib_dev, size, rx_depth, ib_port, use_event);
	if (!ctx)
		return 1;

	routs = pp_post_recv(ctx, ctx->rx_depth);
	if (routs < ctx->rx_depth) {
		fprintf(stderr, "Couldn't post receive (%d)\n", routs);
		return 1;
	}

	if (use_event)
		if (ibv_req_notify_cq(ctx->cq, 0)) {
			fprintf(stderr, "Couldn't request CQ notification\n");
			return 1;
		}

	my_dest.lid = pp_get_local_lid(ctx->context, ib_port);
	my_dest.qpn = ctx->qp->qp_num;
	my_dest.psn = rand() & 0xffffff;
	if (!my_dest.lid) {
		fprintf(stderr, "Couldn't get local LID\n");
		return 1;
	}

	printf("  local address:  LID 0x%04x, QPN 0x%06x, PSN 0x%06x\n",
	       my_dest.lid, my_dest.qpn, my_dest.psn);

	if (servername)
		rem_dest = pp_client_exch_dest(servername, port, &my_dest);
	else
		rem_dest = pp_server_exch_dest(ctx, ib_port, port, sl, &my_dest);

	if (!rem_dest)
		return 1;

	printf("  remote address: LID 0x%04x, QPN 0x%06x, PSN 0x%06x\n",
	       rem_dest->lid, rem_dest->qpn, rem_dest->psn);

	if (servername)
		if (pp_connect_ctx(ctx, ib_port, my_dest.psn, sl, rem_dest))
			return 1;

	ctx->pending = PINGPONG_RECV_WRID;

	if (servername) {
		if (pp_post_send(ctx, rem_dest->qpn)) {
			fprintf(stderr, "Couldn't post send\n");
			return 1;
		}
		ctx->pending |= PINGPONG_SEND_WRID;
	}

	if (!QueryPerformanceCounter(&start)) {
		perror("QueryPerformanceCounter");
		return 1;
	}

	rcnt = scnt = 0;
	while (rcnt < iters || scnt < iters) {
		if (use_event) {
			struct ibv_cq *ev_cq;
			void          *ev_ctx;

			if (ibv_get_cq_event(ctx->channel, &ev_cq, &ev_ctx)) {
				fprintf(stderr, "Failed to get cq_event\n");
				return 1;
			}

			++num_cq_events;

			if (ev_cq != ctx->cq) {
				fprintf(stderr, "CQ event for unknown CQ %p\n", ev_cq);
				return 1;
			}

			if (ibv_req_notify_cq(ctx->cq, 0)) {
				fprintf(stderr, "Couldn't request CQ notification\n");
				return 1;
			}
		}

		{
			struct ibv_wc wc[2];
			int ne, i;

			do {
				ne = ibv_poll_cq(ctx->cq, 2, wc);
				if (ne < 0) {
					fprintf(stderr, "poll CQ failed %d\n", ne);
					return 1;
				}
			} while (!use_event && ne < 1);

			for (i = 0; i < ne; ++i) {
				if (wc[i].status != IBV_WC_SUCCESS) {
					fprintf(stderr, "Failed status %s (%d) for wr_id %d\n",
						ibv_wc_status_str(wc[i].status),
						wc[i].status, (int) wc[i].wr_id);
					return 1;
				}

				switch ((int) wc[i].wr_id) {
				case PINGPONG_SEND_WRID:
					++scnt;
					break;

				case PINGPONG_RECV_WRID:
					if (--routs <= 1) {
						routs += pp_post_recv(ctx, ctx->rx_depth - routs);
						if (routs < ctx->rx_depth) {
							fprintf(stderr,
								"Couldn't post receive (%d)\n",
								routs);
							return 1;
						}
					}

					++rcnt;
					break;

				default:
					fprintf(stderr, "Completion for unknown wr_id %d\n",
						(int) wc[i].wr_id);
					return 1;
				}

				ctx->pending &= ~(int) wc[i].wr_id;
				if (scnt < iters && !ctx->pending) {
					if (pp_post_send(ctx, rem_dest->qpn)) {
						fprintf(stderr, "Couldn't post send\n");
						return 1;
					}
					ctx->pending = PINGPONG_RECV_WRID |
						       PINGPONG_SEND_WRID;
				}
			}
		}
	}

	if (!QueryPerformanceCounter(&end) ||
		!QueryPerformanceFrequency(&freq)) {
		perror("QueryPerformanceCounter/Frequency");
		return 1;
	}

	{
		double sec = (double) (end.QuadPart - start.QuadPart) / (double) freq.QuadPart;
		long long bytes = (long long) size * iters * 2;

		printf("%I64d bytes in %.2f seconds = %.2f Mbit/sec\n",
		       bytes, sec, bytes * 8. / 1000000. / sec);
		printf("%d iters in %.2f seconds = %.2f usec/iter\n",
		       iters, sec, sec * 1000000. / iters);
	}

	ibv_ack_cq_events(ctx->cq, num_cq_events);

	if (pp_close_ctx(ctx))
		return 1;

	ibv_free_device_list(dev_list);
	free(rem_dest);

	return 0;
}
<file_sep>/src/status.h
#ifndef _NVDS_STATUS_H_
#define _NVDS_STATUS_H_

#include "common.h"

namespace nvds {

  enum class Status : uint8_t {
    OK, ERROR, NO_MEM,
  };

} // namespace nvds

#endif // _NVDS_STATUS_H_
<file_sep>/src/server_main.cc
/*
 * Startup module of server.
 */

#include "server.h"

using namespace nvds;

static Server* server;

static void SigInt(int signo) {
  std::cout << std::endl << "num_recv: " << server->num_recv() << std::endl;
  std::cout << "alloc measurement: " << std::endl;
  server->alloc_measurement.Print();
  std::cout << std::endl;

  std::cout << "sync measurement: " << std::endl;
  server->sync_measurement.Print();
  std::cout << std::endl;

  std::cout << "thread measurement: " << std::endl;
  server->thread_measurement.Print();
  std::cout << std::endl;

  std::cout << "send measurement: " << std::endl;
  server->send_measurement.Print();
  std::cout << std::endl;

  std::cout << "recv measurement: " << std::endl;
  server->recv_measurement.Print();
  std::cout << std::endl;

  std::cout << std::flush;
  exit(0);
}

static void Usage(int argc, const char* argv[]) {
    std::cout << "Usage:" << std::endl
              << "    " << argv[0] << " <port> <coord addr>" << std::endl;
}

int main(int argc, const char* argv[]) {
  signal(SIGINT, SigInt);
  // -2. Argument parsing.
  if (argc < 3) {
    Usage(argc, argv);
    return -1;
  }

  uint16_t server_port = std::stoi(argv[1]);
  std::string coord_addr = argv[2];

  // Step 0, self initialization, including formatting nvm storage.
  // DRAM emulated NVM.
  auto nvm = AcquireNVM<NVMDevice>(kNVMDeviceSize);
  if (nvm == nullptr) {
    NVDS_ERR("acquire nvm failed: size = %PRIu64", kNVMDeviceSize);
    return -1;
  }
  try {
    Server s(server_port, nvm, kNVMDeviceSize);
    server = &s;
    // Step 1: request to the coordinator for joining in.
    if (!s.Join(coord_addr)) {
      NVDS_ERR("join cluster failed");
      return -1;
    }

    // Step 2: get the arrangement from coordinator, contacting servers
    //         that will be affected, performing data migration.

    // Step 3: acknowledge the coordinator of complemention

    // Step 4: serving request
    NVDS_LOG("Server startup");
    NVDS_LOG("Listening at: %u", server_port);
    NVDS_LOG("......");
    
    s.Run();
  } catch (boost::system::system_error& e) {
    NVDS_ERR(e.what());
  } catch (TransportException& e) {
    NVDS_ERR(e.msg().c_str());
    NVDS_LOG("initialize infiniband devices failed");
  }
  return 0;
}
<file_sep>/src/infiniband_old.cc
#include "infiniband.h"

#include <algorithm>
#include <malloc.h>

namespace nvds {

Infiniband::Infiniband(const char* device_name)
    : dev_(device_name), pd_(dev_) {
  std::fill(addr_handlers_.begin(), addr_handlers_.end(), nullptr);
}

Infiniband::~Infiniband() {
  for (auto iter = addr_handlers_.rbegin(); iter != addr_handlers_.rend(); ++iter) {
    auto ah = *iter;
    if (ah != nullptr)
      ibv_destroy_ah(ah);
  }
}

void Infiniband::QueuePairInfo::Print() const {
  std::cout << "lid: " << lid << std::endl;
  std::cout << "qpn: " << qpn << std::endl;
  std::cout << "psn: " << psn << std::endl;
  std::cout << "rkey: " << rkey << std::endl;
  std::cout << "vaddr: " << vaddr << std::endl;
}

Infiniband::DeviceList::DeviceList()
    : dev_list_(ibv_get_device_list(nullptr)) {
  if (dev_list_ == nullptr) {
    throw TransportException(HERE, "opening infiniband device failed", errno);
  }
}

ibv_device* Infiniband::DeviceList::Lookup(const char* name) {
  if (name == nullptr)
    return dev_list_[0];
  for (int i = 0; dev_list_[i] != nullptr; ++i) {
    if (strcmp(dev_list_[i]->name, name) == 0)
      return dev_list_[i];
  }
  return nullptr;
}

Infiniband::Device::Device(const char* name) {
  DeviceList dev_list;
  auto dev = dev_list.Lookup(name);
  if (dev == nullptr) {
    throw TransportException(HERE,
        Format("failed to find infiniband devide: %s", name ? name: "any"));
  }
  
  ctx_ = ibv_open_device(dev);
  if (ctx_ == nullptr) {
    throw TransportException(HERE, 
        Format("failed to find infiniband devide: %s", name ? name: "any"),
        errno);
  }
}

Infiniband::QueuePair::QueuePair(Infiniband& ib, ibv_qp_type type,
    int ib_port, ibv_srq* srq, ibv_cq* scq, ibv_cq* rcq,
    uint32_t max_send, uint32_t max_recv)
    : ib(ib), type(type), ctx(ib.dev().ctx()),
      ib_port(ib_port), pd(ib.pd().pd()), srq(srq),
      qp(nullptr), scq(scq), rcq(rcq), psn(0), peer_lid(0), sin() {
  assert(type == IBV_QPT_UD || type == IBV_QPT_RC);

  ibv_qp_init_attr qpia;
  memset(&qpia, 0, sizeof(qpia));
  qpia.send_cq = scq;
  qpia.recv_cq = rcq;
  qpia.srq = srq;
  qpia.cap.max_send_wr = max_send;
  qpia.cap.max_recv_wr = max_recv;
  // FIXME(wgtdkp): what is the `max_send_sge` for replication `qp`?
  qpia.cap.max_send_sge = type == IBV_QPT_UD ? 1 : 8;
  qpia.cap.max_recv_sge = type == IBV_QPT_UD ? 1 : 8;
  qpia.cap.max_inline_data = kMaxInlineData;
  qpia.qp_type = type;
  qpia.sq_sig_all = 0; // Only generate CQEs on requested WQEs

  qp = ibv_create_qp(pd, &qpia);
  if (qp == nullptr) {
    throw TransportException(HERE, "failed to create queue pair");
  }

  // Set queue pair to status:INIT
  ibv_qp_attr qpa;
  memset(&qpa, 0, sizeof(qpa));
  qpa.qp_state = IBV_QPS_INIT;
  qpa.pkey_index = 0;
  qpa.port_num = static_cast<uint8_t>(ib_port);
  qpa.qp_access_flags = IBV_ACCESS_REMOTE_READ | IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_LOCAL_WRITE;
  qpa.qkey = kQkey;

  int mask = IBV_QP_STATE | IBV_QP_PORT;
  switch (type) {
  case IBV_QPT_RC:
    mask |= IBV_QP_ACCESS_FLAGS;
    mask |= IBV_QP_PKEY_INDEX;
    break;
  case IBV_QPT_UD:
    mask |= IBV_QP_QKEY;
    mask |= IBV_QP_PKEY_INDEX;
    break;
  default:
    assert(false);
  }

  int ret = ibv_modify_qp(qp, &qpa, mask);
  if (ret != 0) {
    ibv_destroy_qp(qp);
    throw TransportException(HERE, ret);
  }
}

uint32_t Infiniband::QueuePair::GetPeerQPNum() const {
  ibv_qp_attr qpa;
  ibv_qp_init_attr qpia;

  auto err = ibv_query_qp(qp, &qpa, IBV_QP_DEST_QPN, &qpia);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
  return qpa.dest_qp_num;
}

int Infiniband::QueuePair::GetState() const {
  ibv_qp_attr qpa;
  ibv_qp_init_attr qpia;

  int err = ibv_query_qp(qp, &qpa, IBV_QP_STATE, &qpia);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
  return qpa.qp_state;
}

// Called only on RC queue pair
void Infiniband::QueuePair::Plumb(enum ibv_qp_state state,
                                  const QueuePairInfo& qpi) {
  assert(type == IBV_QPT_RC);
  assert(GetState() == IBV_QPS_INIT);
  assert(state == IBV_QPS_RTR || state == IBV_QPS_RTS);

  ibv_qp_attr qpa;
  memset(&qpa, 0, sizeof(qpa));
  // We must set qp into RTR state before RTS
  qpa.qp_state = IBV_QPS_RTR;
  qpa.path_mtu = IBV_MTU_4096;
  qpa.dest_qp_num = qpi.qpn;
  qpa.rq_psn = qpi.psn;
  qpa.max_dest_rd_atomic = 1;
  qpa.min_rnr_timer = 12;
  qpa.ah_attr.is_global = 0;
  qpa.ah_attr.dlid = qpi.lid;
  qpa.ah_attr.sl = 0;
  qpa.ah_attr.src_path_bits = 0;
  qpa.ah_attr.port_num = static_cast<uint8_t>(ib_port);

  auto err = ibv_modify_qp(qp, &qpa,
      IBV_QP_STATE | IBV_QP_AV | IBV_QP_PATH_MTU | IBV_QP_DEST_QPN |
      IBV_QP_RQ_PSN | IBV_QP_MIN_RNR_TIMER | IBV_QP_MAX_DEST_RD_ATOMIC);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
  if (state == IBV_QPS_RTR) {
    return;
  }

  memset(&qpa, 0, sizeof(qpa));
  qpa.qp_state = IBV_QPS_RTS;

  // How long to wait before retrying if packet lost or server dead.
  // Supposedly the timeout is 4.096us*2^timeout.  However, the actual
  // timeout appears to be 4.096us*2^(timeout+1), so the setting
  // below creates a 135ms timeout.
  qpa.timeout = 14;

  // How many times to retry after timeouts before giving up.
  qpa.retry_cnt = 7;

  // How many times to retry after RNR (receiver not ready) condition
  // before giving up. Occurs when the remote side has not yet posted
  // a receive request.
  qpa.rnr_retry = 7; // 7 is infinite retry.
  qpa.sq_psn = psn;
  qpa.max_rd_atomic = 1;

  err = ibv_modify_qp(qp, &qpa,
      IBV_QP_STATE |  IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT |
      IBV_QP_RNR_RETRY | IBV_QP_SQ_PSN | IBV_QP_MAX_QP_RD_ATOMIC);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
  peer_lid = qpi.lid;
}

// Bring UD queue pair into RTS status
void Infiniband::QueuePair::Activate() {
  assert(type == IBV_QPT_UD);
  assert(GetState() == IBV_QPS_INIT);

  ibv_qp_attr qpa;
  memset(&qpa, 0, sizeof(qpa));
  qpa.qp_state = IBV_QPS_RTR;
  auto err = ibv_modify_qp(qp, &qpa, IBV_QP_STATE);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
  
  qpa.qp_state = IBV_QPS_RTS;
  qpa.sq_psn = psn;
  err = ibv_modify_qp(qp, &qpa, IBV_QP_STATE | IBV_QP_SQ_PSN);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
}

std::string Infiniband::Address::ToString() const {
  return Format("%u:%u", lid, qpn);
}

Infiniband::RegisteredBuffers::RegisteredBuffers(ProtectionDomain& pd,
    uint32_t buf_size, uint32_t buf_num, bool is_recv)
    : buf_size_(buf_size), buf_num_(buf_num),
      ptr_(nullptr), bufs_(nullptr), root_(nullptr) {
  const size_t bytes = buf_size * buf_num;
  ptr_ = memalign(kPageSize, bytes);
  assert(ptr_ != nullptr);
  auto mr = ibv_reg_mr(pd.pd(), ptr_, bytes,
                       IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_LOCAL_WRITE);
  if (mr == nullptr) {
    throw TransportException(HERE, "failed to register buffer", errno);
  }
  bufs_ = new Buffer[buf_num_];
  char* p = static_cast<char*>(ptr_);
  for (uint32_t i = 0; i < buf_num; ++i) {
    new (&bufs_[i]) Buffer(p, buf_size_, mr, is_recv);
    bufs_[i].next = i + 1 < buf_num ? &bufs_[i+1] : nullptr;
    p += buf_size_;
  }
  root_ = bufs_;
}

Infiniband::RegisteredBuffers::~RegisteredBuffers() {
  // Deregsiter memory region
  ibv_dereg_mr(bufs_[0].mr);
  free(ptr_);
  delete[] bufs_;
}

/*
Infiniband::QueuePair* Infiniband::CreateQP(ibv_qp_type type, int ib_port,
    ibv_srq* srq, ibv_cq* scq, ibv_cq* rcq, uint32_t max_send,
    uint32_t max_recv, uint32_t qkey) {
  return new QueuePair(*this, type, ib_port, srq, scq, rcq,
                       max_send, max_recv, qkey);
}
*/
uint16_t Infiniband::GetLid(int32_t port) {
  ibv_port_attr port_attr;
  auto ret = ibv_query_port(dev_.ctx(), static_cast<uint8_t>(port), &port_attr);
  if (ret != 0) {
    throw TransportException(HERE, ret);
  }
  return port_attr.lid;
}

// May blocking
Infiniband::Buffer* Infiniband::Receive(QueuePair* qp) {
  Buffer* b;
  while ((b = TryReceive(qp)) == nullptr) {}
  return b;
}

Infiniband::Buffer* Infiniband::PollCQ(ibv_cq* cq, int num_entries) {
  ibv_wc wc;
  int r = ibv_poll_cq(cq, num_entries, &wc);
  if (r == 0) {
    return NULL;
  } else if (r < 0) {
    throw TransportException(HERE, r);
  } else if (wc.status != IBV_WC_SUCCESS) {
    throw TransportException(HERE, wc.status);
  }
  // wr_id is used as buffer address
  auto b = reinterpret_cast<Buffer*>(wc.wr_id);
  b->msg_len = wc.byte_len;
  // FIXME(wgtdkp):
  b->peer_addr = {Infiniband::kPort/*qp->ib_port*/, wc.slid, wc.src_qp};
  return b;
}

void Infiniband::PostReceive(QueuePair* qp, Buffer* b) {
  ibv_sge sge {
    reinterpret_cast<uint64_t>(b->buf),
    b->size,
    b->mr->lkey
  };
  ibv_recv_wr rwr;
  memset(&rwr, 0, sizeof(rwr));
  rwr.wr_id = reinterpret_cast<uint64_t>(b);
  rwr.next = nullptr;
  rwr.sg_list = &sge;
  rwr.num_sge = 1;

  ibv_recv_wr* bad_rwr;
  auto err = ibv_post_recv(qp->qp, &rwr, &bad_rwr);
  if (err != 0) {
    throw TransportException(HERE, "ibv_post_recv failed", err);
  }
}

void Infiniband::PostSRQReceive(ibv_srq* srq, Buffer* b) {
  ibv_sge sge {
    reinterpret_cast<uint64_t>(b->buf),
    b->size,
    b->mr->lkey
  };
  ibv_recv_wr rwr;
  memset(&rwr, 0, sizeof(rwr));
  rwr.wr_id = reinterpret_cast<uint64_t>(b);
  rwr.next = nullptr;
  rwr.sg_list = &sge;
  rwr.num_sge = 1;

  ibv_recv_wr* bad_rwr;
  auto err = ibv_post_srq_recv(srq, &rwr, &bad_rwr);
  if (err != 0) {
    throw TransportException(HERE, "ibv_post_srq_recv failed", err);
  }
}

void Infiniband::PostSend(QueuePair* qp, Buffer* b,
    uint32_t len, const Address* peer_addr, uint32_t peer_qkey) {
  if (qp->type == IBV_QPT_UD) {
      assert(peer_addr != nullptr);
  } else {
      assert(peer_addr == nullptr);
      assert(peer_qkey == 0);
  }

  ibv_sge sge {
    reinterpret_cast<uint64_t>(b->buf),
    len,
    b->mr->lkey
  };
  ibv_send_wr swr;
  memset(&swr, 0, sizeof(swr));
  swr.wr_id = reinterpret_cast<uint64_t>(b);
  if (qp->type == IBV_QPT_UD) {
    swr.wr.ud.ah = GetAddrHandler(*peer_addr);
    swr.wr.ud.remote_qpn = peer_addr->qpn;
    swr.wr.ud.remote_qkey = peer_qkey;
    assert(swr.wr.ud.ah != nullptr);
  }
  swr.next = nullptr;
  swr.sg_list = &sge;
  swr.num_sge = 1;
  swr.opcode = IBV_WR_SEND;
  swr.send_flags = IBV_SEND_SIGNALED;
  if (len <= kMaxInlineData) {
    swr.send_flags |= IBV_SEND_INLINE;
  }

  ibv_send_wr* bad_swr;
  auto err = ibv_post_send(qp->qp, &swr, &bad_swr);
  if (err != 0) {
    throw TransportException(HERE, "ibv_post_send failed", err);
  }
}

void Infiniband::PostSendAndWait(QueuePair* qp, Buffer* b,
    uint32_t len, const Address* peer_addr, uint32_t peer_qkey) {
  assert(qp->type == IBV_QPT_UD);

  PostSend(qp, b, len, peer_addr, peer_qkey);

  ibv_wc wc;
  while (ibv_poll_cq(qp->scq, 1, &wc) < 1) {}
  // FIXME(wgtdkp):
  // How can we make sure that the completed work we polled is the one
  // we have just post?
  if (wc.status != IBV_WC_SUCCESS) {
    throw TransportException(HERE, "PostSendAndWait failed", wc.status);
  }
}

ibv_ah* Infiniband::GetAddrHandler(const Address& addr) {
  auto& ah = addr_handlers_[addr.lid];
  if (ah != nullptr) {
    return ah;
  }
  
  ibv_ah_attr attr;
  memset(&attr, 0, sizeof(attr));
  attr.is_global = 0;
  attr.dlid = addr.lid;
  attr.sl = 1;
  attr.src_path_bits = 0;
  attr.port_num = kPort;
  return ah = ibv_create_ah(pd_.pd(), &attr);
}

} // namespace nvds
<file_sep>/src/modification.h
#ifndef _NVDS_MODIFICATION_H_
#define _NVDS_MODIFICATION_H_

#include "common.h"

namespace nvds {

struct Modification {
  uint32_t des;
  uint32_t len;
  uint64_t src;
  Modification(uint32_t des, uint64_t src, uint32_t len)
      : des(des), len(len), src(src) {}
  // For STL
  Modification() : des(0), len(0), src(0) {}
  bool operator<(const Modification& other) const {
    return des < other.des;
  }
  // DEBUG
  void Print() const {
    std::cout << "des: " << des << "; ";
    std::cout << "len: " << len << "; ";
    std::cout << "src: " << src << std::endl;
  }
};
using ModificationList = std::vector<Modification>;

struct Position {
  uint32_t offset;
  uint32_t len;
  Position() = delete;
};

struct ModificationLog {
  uint32_t cnt;
  Position positions[0];
  ModificationLog() = delete;
};

} // namespace nvds

#endif // _NVDS_MODIFICATION_H_
<file_sep>/src/test_memcached.cc
#include <libmemcached/memcached.h>

#include <cassert>
#include <vector>
#include <string>
#include <iostream>
#include <chrono>
#include <random>

using VecStr = std::vector<std::string>;

static VecStr GenRandomStrings(size_t len, size_t n) {
  VecStr ans;
  std::default_random_engine g;
  std::uniform_int_distribution<int> dist('a', 'z');
  for (size_t i = 0; i < n; ++i) {
    ans.emplace_back(len, 0);
    for (size_t j = 0; j < len; ++j) {
      ans[i][j] = dist(g);
    }
  }
  return ans;
}

void test(size_t n) {
  const char* config_string = "--SERVER=127.0.0.1";
  memcached_st* memc = memcached(config_string, strlen(config_string));

  auto keys = GenRandomStrings(16, n);
  auto vals = VecStr(n, std::string(16, 'a'));
  auto begin = std::chrono::high_resolution_clock::now();
  for (size_t i = 0; i < n; ++i) {
    auto& key = keys[i];
    auto& val = vals[i];
    memcached_return_t rc = memcached_set(memc, key.c_str(), key.size(), val.c_str(), val.size(), 0, 0);
    assert(rc == MEMCACHED_SUCCESS);
  }
  auto end = std::chrono::high_resolution_clock::now();
  double t = std::chrono::duration_cast<std::chrono::duration<double>>(end - begin).count();
  double qps = keys.size() / t;

  std::cout << "time: " << t << std::endl;
  std::cout << "qps:  " << qps << std::endl;

  memcached_free(memc);
}

int main() {
  test(10000);
  return 0;
}
<file_sep>/src/allocator.cc
#include "allocator.h"

#include <cstring>

namespace nvds {

void Allocator::Format() {
  ModificationList modifications;
  set_modifications(&modifications);

  memset(flm_, 0, kSize);
  uint32_t blk = sizeof(FreeListManager);
  Write(GetLastFreeList(), blk);

  // 'The block before first block' is not free
  // The last 4 bytes is kept for the free tag
  uint32_t blk_size = (kSize - blk - sizeof(uint32_t)) &
                      ~static_cast<uint32_t>(0x0f);
  Write(blk, blk_size & ~BlockHeader::kFreeMask);
  Write(blk + blk_size - sizeof(uint32_t), blk_size);
  SetTheFreeTag(blk, blk_size);
  // TODO(wgtdkp): setting `next` and `prev` nullptr.(unnecessary if called `memset`)
}

uint32_t Allocator::AllocBlock(uint32_t blk_size) {
  auto free_list = GetFreeListByBlockSize(blk_size);
  uint32_t head;
  while (free_list < kNumFreeLists * sizeof(uint32_t) &&
         (head = Read<uint32_t>(free_list)) == 0) {
    free_list += sizeof(uint32_t);
  }
  if (head == 0) {
    return 0;
  }
  auto head_size = ReadTheSizeTag(head);
  assert(ReadTheFreeTag(head, head_size) != 0);
  assert(head_size >= blk_size);
  if (head_size == blk_size) {
    auto next_blk = Read<uint32_t>(head + offsetof(BlockHeader, next));
    ResetTheFreeTag(head, head_size);
    Write(free_list, next_blk);
    Write(next_blk + offsetof(BlockHeader, prev),
          static_cast<uint32_t>(0));
    return head;
  }
  return SplitBlock(head, head_size, blk_size);
}

void Allocator::FreeBlock(uint32_t blk) {
  auto blk_size      = ReadTheSizeTag(blk);
  auto blk_size_old  = blk_size;
  auto prev_blk_size = Read<uint32_t>(blk - sizeof(uint32_t));
  auto prev_blk      = blk - prev_blk_size;
  auto next_blk      = blk + blk_size;
  auto next_blk_size = ReadTheSizeTag(next_blk);

  if (ReadThePrevFreeTag(blk) &&
      blk_size <= kMaxBlockSize &&
      prev_blk_size <= kMaxBlockSize) {
    RemoveBlock(prev_blk, prev_blk_size);
    blk = prev_blk;
    blk_size += prev_blk_size;
  }
  if (ReadTheFreeTag(next_blk, next_blk_size) &&
      blk_size <= kMaxBlockSize &&
      next_blk_size <= kMaxBlockSize) {
    RemoveBlock(next_blk, next_blk_size);
    blk_size += next_blk_size;
  }

  // If the blk_size changed, write it.
  if (blk_size != blk_size_old) {
    Write(blk, ReadThePrevFreeTag(blk) | blk_size);
    Write(blk + blk_size - sizeof(uint32_t), blk_size);
  }
  SetTheFreeTag(blk, blk_size);

  auto free_list = GetFreeListByBlockSize(blk_size);
  auto head = Read<uint32_t>(free_list);
  Write(free_list, blk);
  Write(blk + offsetof(BlockHeader, prev), static_cast<uint32_t>(0));
  Write(blk + offsetof(BlockHeader, next), head);
  if (head != 0) {
    Write(head + offsetof(BlockHeader, prev), blk);
  }
}

void Allocator::RemoveBlock(uint32_t blk, uint32_t blk_size) {
  assert(ReadTheFreeTag(blk, blk_size) != 0);
  auto prev_blk = Read<uint32_t>(blk + offsetof(BlockHeader, prev));
  auto next_blk = Read<uint32_t>(blk + offsetof(BlockHeader, next));
  if (prev_blk != 0) {
    Write(prev_blk + offsetof(BlockHeader, next),
          Read<uint32_t>(blk + offsetof(BlockHeader, next)));
  } else {
    auto free_list = GetFreeListByBlockSize(blk_size);
    Write(free_list, Read<uint32_t>(blk + offsetof(BlockHeader, next)));
  }
  if (next_blk != 0) {
    Write(next_blk + offsetof(BlockHeader, prev),
          Read<uint32_t>(blk + offsetof(BlockHeader, prev)));
  }
}

uint32_t Allocator::SplitBlock(uint32_t blk, uint32_t blk_size,
                               uint32_t needed_size) {
  auto next_blk  = Read<uint32_t>(blk + offsetof(BlockHeader, next));
  auto free_list = GetFreeListByBlockSize(blk_size);
  assert(blk_size > needed_size);
  auto new_size  = blk_size - needed_size;

  Write(free_list, next_blk);
  if (next_blk != 0) {
    Write(next_blk + offsetof(BlockHeader, prev), static_cast<uint32_t>(0));
  }

  // Update size
  Write(blk, ReadThePrevFreeTag(blk) | new_size);
  Write(blk + new_size - sizeof(uint32_t), new_size);

  free_list = GetFreeListByBlockSize(new_size);
  auto head = Read<uint32_t>(free_list);
  Write(blk + offsetof(BlockHeader, prev), static_cast<uint32_t>(0));
  Write(blk + offsetof(BlockHeader, next), head);
  Write(free_list, blk);
  if (head != 0) {
    Write(head + offsetof(BlockHeader, prev), blk);
  }

  // The rest smaller block is still free
  Write(blk + new_size, BlockHeader::kFreeMask | needed_size);
  Write(blk + blk_size - sizeof(uint32_t), needed_size);

  // Piece splitted is not free now.
  ResetTheFreeTag(blk + new_size, needed_size);
  return blk + new_size;
}

} // namespace nvds
<file_sep>/src/session.h
#ifndef _NVDS_SESSION_H_
#define _NVDS_SESSION_H_

#include "message.h"

#include <boost/asio.hpp>
#include <boost/function.hpp>

namespace nvds {

class Session;
using MessageHandler = boost::function<void(std::shared_ptr<Session>,
                                            std::shared_ptr<Message>)>;
using boost::asio::ip::tcp;

class Session : public std::enable_shared_from_this<Session> {
 public:
  explicit Session(tcp::socket conn_sock,
                   MessageHandler recv_msg_handler=0,
                   MessageHandler send_msg_handler=0)
      : conn_sock_(std::move(conn_sock)),
        recv_msg_handler_(recv_msg_handler),
        send_msg_handler_(send_msg_handler) {}
  ~Session() { /* Close socket automatically */ }

  // Start async message send/recv
  void Start();
  void AsyncRecvMessage(std::shared_ptr<Message> msg);
  void AsyncSendMessage(std::shared_ptr<Message> msg);
  // Throw exception `boost::system::system_error` on failure
  Message RecvMessage();
  // Throw exception `boost::system::system_error` on failure
  void SendMessage(const Message& msg);

  std::string GetPeerAddr() const {
    return conn_sock_.remote_endpoint().address().to_string();
  }

 private:
  tcp::socket conn_sock_;
  MessageHandler recv_msg_handler_;
  MessageHandler send_msg_handler_;
};

} // namespace nvds

#endif // _NVDS_SESSION_H_
<file_sep>/test/test_json.cc
#include "../json/src/json.hpp"

#include <fstream>
#include <string>

using json = nlohmann::json;

int main(int argc, const char* argv[]) {
  if (argc < 2) {
    std::cout << "need parameter" << std::endl;
    return -1;
  }

  json cfg_json;
  try {
    cfg_json = json::parse(std::ifstream(argv[1]));
  } catch (std::invalid_argument& inv_arg) {
    std::cout << "invalid argument" << std::endl;
    return -2;
  }
  std::cout << static_cast<int>(cfg_json["coordinator"]["tcp_port"]) << std::endl;
  return 0;
}
<file_sep>/src/infiniband_old.h
#ifndef _NVDS_INFINIBAND_H_
#define _NVDS_INFINIBAND_H_

#include "common.h"
#include "spinlock.h"
#include "transport.h"

#include <cstring>
#include <mutex>
#include <infiniband/verbs.h>
#include <netinet/ip.h>

namespace nvds {

struct Request;
struct Response;

// Derived from RAMCloud Infiniband.h
class Infiniband {
 public:
  static const uint16_t kPort = 1;
 	static const uint32_t kMaxInlineData = 400;
	// If the device name is not specified, choose the first one in device list
	explicit Infiniband(const char* device_name=nullptr);
	~Infiniband();

	// This class includes informations for constructing a connection
	struct QueuePairInfo {
    uint16_t lid;   // Local id
		uint32_t qpn;   // Queue pair number
		uint32_t psn;   // Packet sequence number
		uint32_t rkey;  // Memory region key
    uint64_t vaddr; // Virtual address of memory regions
    // DEBUG
    void Print() const;
	};

	class DeviceList {
	 public:
		DeviceList();
		~DeviceList() { ibv_free_device_list(dev_list_); }
		DISALLOW_COPY_AND_ASSIGN(DeviceList);
		ibv_device* Lookup(const char* name);

	 private:
		ibv_device** const dev_list_;
	};

	class Device {
	 public:
	 	explicit Device(const char* name=nullptr);
		~Device() { ibv_close_device(ctx_); }
		DISALLOW_COPY_AND_ASSIGN(Device);
		
		ibv_context* ctx() { return ctx_; }
		const ibv_context* ctx() const { return ctx_; }

	 private:
	 	ibv_context* ctx_;
	};

	class ProtectionDomain {
	 public:
	 	explicit ProtectionDomain(Device& device)
		 		: pd_(ibv_alloc_pd(device.ctx())) {
		  if (pd_ == nullptr) {
				throw TransportException(HERE,
					  "failed to allocate infiniband protection domain", errno);
			}
		}
		~ProtectionDomain() { ibv_dealloc_pd(pd_); }
		DISALLOW_COPY_AND_ASSIGN(ProtectionDomain);
		operator ibv_pd() { return *pd_; }
		ibv_pd* pd() { return pd_; }
		const ibv_pd* pd() const { return pd_; }

	 private:
	 	ibv_pd* const pd_;
	};

	struct QueuePair {
		Infiniband& 	ib;       // this QP belongs to
		ibv_qp_type 	type;	    // QP type
		ibv_context* 	ctx;      // Device context
		int 					ib_port;  // Physical port
		ibv_pd* 			pd;
		ibv_srq* 			srq;
		ibv_qp* 			qp;
		ibv_cq* 			scq;
		ibv_cq*				rcq;
		uint32_t 			psn;
		uint16_t 	    peer_lid;
		sockaddr_in 	sin;

	 	QueuePair(Infiniband& ib, ibv_qp_type type,
		 					int ib_port, ibv_srq* srq, ibv_cq* scq,
							ibv_cq* rcq, uint32_t max_send,
							uint32_t max_recv);
		~QueuePair() { ibv_destroy_qp(qp); }
		DISALLOW_COPY_AND_ASSIGN(QueuePair);
		uint32_t GetLocalQPNum() const { return qp->qp_num; }
		uint32_t GetPeerQPNum() const;
		int GetState() const;
    // Bring a queue pair into RTS state
    void Plumb(enum ibv_qp_state state, const QueuePairInfo& qpi);
		void Activate();
   //private:
   // ibv_qp* Create(uint32_t max_send, uint32_t max_recv, );
	};

	struct Address {
		int32_t ib_port;
		uint16_t lid;
		uint32_t qpn;
    // `ah` is moved to class `Infiniband`
    // ibv_ah* ah; // Could be nullptr
		std::string ToString() const;
    bool operator==(const Address& other) const {
      return ib_port == other.ib_port &&
             lid == other.lid &&
             qpn == other.qpn;
    }
    bool operator!=(const Address& other) const {
      return !(*this == other);
    }
	};

	struct Buffer {
    Buffer*   next;
		char* 		buf;				
		uint32_t 	size;		    // Buffer size
		uint32_t 	msg_len;    // message length
		ibv_mr*		mr;         // IBV memory region
		Address   peer_addr;  // Peer address of this request
		bool 			is_recv;    // Is this a recv buffer

		Buffer(char* b, uint32_t size, ibv_mr* mr, bool is_recv=false)
				: next(nullptr), buf(b), size(size), mr(mr), is_recv(is_recv) {}
		Buffer() : next(nullptr), buf(nullptr), size(0), msg_len(0),
               mr(nullptr), is_recv(false) {}
		DISALLOW_COPY_AND_ASSIGN(Buffer);
    Request* MakeRequest() {
      return reinterpret_cast<Request*>(buf + kIBUDPadding);
    }
    Response* MakeResponse() {
      // TODO(wgtdkp): are you sure ?
      return reinterpret_cast<Response*>(buf + kIBUDPadding);
    }
	};

  // A registered buffer pool
	class RegisteredBuffers {
	 public:
	  RegisteredBuffers(ProtectionDomain& pd,
		    uint32_t buf_size, uint32_t buf_num, bool is_recv=false);
		~RegisteredBuffers();
		DISALLOW_COPY_AND_ASSIGN(RegisteredBuffers);
    // Used as buffer pool
    Buffer* Alloc() {
      std::lock_guard<Spinlock> _(spinlock_);
      auto ret = root_;
      if (root_ != nullptr) {
        root_ = root_->next;
      }
      return ret;
    }
    void Free(Buffer* b) {
      std::lock_guard<Spinlock> _(spinlock_);
      b->next = root_;
      root_ = b;
    }
		Buffer& GetBuffer(const void* pos) {
			auto idx = (static_cast<const char*>(pos) - static_cast<const char*>(ptr_)) / buf_size_;
			return bufs_[idx];
		}
    Buffer& operator[](size_t idx) { return bufs_[idx]; }
    const Buffer& operator[](size_t idx) const {
      return const_cast<RegisteredBuffers*>(this)->operator[](idx);
    }
		Buffer* begin() { return bufs_; }
		Buffer* end() { return bufs_ + buf_num_; }
    const Buffer* root() const { return root_; }
	 private:
	 	uint32_t buf_size_;
		uint32_t buf_num_;
		void* ptr_;
		Buffer* bufs_;
    Buffer* root_;
    Spinlock spinlock_;
	};

  uint16_t GetLid(int32_t port);
	//QueuePair* CreateQP(ibv_qp_type type, int ib_port,
	//										ibv_srq* srq, ibv_cq* scq,
	//										ibv_cq* rcq, uint32_t max_send,
	//										uint32_t max_recv, uint32_t qkey=0);
	Buffer* TryReceive(QueuePair* qp) {
    return PollCQ(qp->rcq, 1);
  }
  Buffer* TrySend(QueuePair* qp) {
    return PollCQ(qp->scq, 1);
  }
	Buffer* Receive(QueuePair* qp);
	void PostReceive(QueuePair* qp, Buffer* b);
	void PostSRQReceive(ibv_srq* srq, Buffer* b);
	void PostSend(QueuePair* qp, Buffer* b, uint32_t len,
								const Address* peer_addr=nullptr,
								uint32_t peer_qkey=0);
  void PostSendAndWait(QueuePair* qp, Buffer* b, uint32_t len,
											 const Address* peer_addr=nullptr,
											 uint32_t peer_qkey=0);
  ibv_cq* CreateCQ(int min_entries) {
		return ibv_create_cq(dev_.ctx(), min_entries, nullptr, nullptr, 0);
	}
  void DestroyCQ(ibv_cq* cq) {
    ibv_destroy_cq(cq);
  }
	ibv_srq* CreateSRQ(uint32_t max_wr, uint32_t max_sge) {
		ibv_srq_init_attr sia;
		memset(&sia, 0, sizeof(sia));
		sia.srq_context = dev_.ctx();
		sia.attr.max_wr = max_wr;
		sia.attr.max_sge = max_sge;
		return ibv_create_srq(pd_.pd(), &sia);
	}
  void DestroySRQ(ibv_srq* srq) {
    ibv_destroy_srq(srq);
  }
	Buffer* PollCQ(ibv_cq* cq, int num_entries);

  ibv_ah* GetAddrHandler(const Address& addr);

	Device& dev() { return dev_; }
	const Device& dev() const { return dev_; }
	ProtectionDomain& pd() { return pd_; }
	const ProtectionDomain& pd() const { return pd_; }

 private:
 	Device dev_;
	ProtectionDomain pd_;
  std::array<ibv_ah*, UINT16_MAX + 1> addr_handlers_;
};

} // namespace nvds

#endif // _NVDS_INFINIBAND_H_
<file_sep>/script/stop.sh
#!/bin/bash
killall -9 -q coordinator

while read line; do
    server=($line)
    ssh root@${server[0]} 'killall -9 -q server'
done < servers.txt
<file_sep>/src/test_client.cc
/*
 * Compile: g++ -std=c++11 -Wall -g test_client.cc -lnvds -lboost_system -lboost_thread -pthread -libverbs
 */

#include "nvds/client.h"

#include <iostream>
#include <string>

int main() {
  try {
    nvds::Client c {"192.168.99.14"};
    c.Put("hello", "world");
    assert(!c.Add("hello", "NVDS"));
    std::cout << "hello: " << c.Get("hello") << std::endl;
    c.Del("hello");
    assert(c.Add("hello", "world"));
    c.Del("hello");
    assert(c.Get("hello").size() == 0);
  } catch (boost::system::system_error& e) {
    NVDS_ERR(e.what());
  } catch (nvds::TransportException& e) {
    NVDS_ERR(e.msg().c_str());
    NVDS_LOG("initialize infiniband devices failed");
  }
  return 0;
}
<file_sep>/src/coordinator.cc
#include "coordinator.h"

#include "json.hpp"
#include "message.h"
#include "session.h"

namespace nvds {

using nlohmann::json;

Coordinator::Coordinator()
    : BasicServer(kCoordPort) {
}

void Coordinator::Run() {
  Accept(std::bind(&Coordinator::HandleRecvMessage, this,
                   std::placeholders::_1, std::placeholders::_2),
         std::bind(&Coordinator::HandleSendMessage, this,
                   std::placeholders::_1, std::placeholders::_2));
  RunService();
}

void Coordinator::HandleRecvMessage(std::shared_ptr<Session> session,
                                    std::shared_ptr<Message> msg) {
  switch (msg->sender_type()) {
  case Message::SenderType::SERVER:
    HandleMessageFromServer(session, msg);
    break;
  case Message::SenderType::CLIENT:
    HandleMessageFromClient(session, msg);
    break;
  case Message::SenderType::COORDINATOR:
    assert(false);
  }
}

void Coordinator::HandleSendMessage(std::shared_ptr<Session> session,
                                    std::shared_ptr<Message> msg) {
  // Receive message
  session->AsyncRecvMessage(std::make_shared<Message>());
}

void Coordinator::HandleMessageFromServer(std::shared_ptr<Session> session,
                                          std::shared_ptr<Message> msg) {
  switch (msg->type()) {
  case Message::Type::REQ_JOIN:
    //NVDS_LOG("join message: %s", msg->body().c_str());
    HandleServerRequestJoin(session, msg);
    break;
  case Message::Type::REQ_LEAVE:
    break;
  case Message::Type::ACK_REJECT:
    break;
  case Message::Type::ACK_ERROR:
    break;
  case Message::Type::ACK_OK:
    break;
  default:
    assert(false);
  }
}

void Coordinator::HandleServerRequestJoin(std::shared_ptr<Session> session, 
                                          std::shared_ptr<Message> req) {
  auto body = json::parse(req->body());
  uint64_t nvm_size = body["size"];

  NVDS_LOG("join request from server: [%s]", session->GetPeerAddr().c_str());
  if (num_servers_ >= kNumServers) {
    NVDS_LOG("too much servers, ignored");
    return;
  }

  index_manager_.AddServer(session->GetPeerAddr(), body);
  sessions_.emplace_back(session);
  ++num_servers_;
  total_storage_ += nvm_size;

  assert(num_servers_ <= kNumServers);
  if (num_servers_ == kNumServers) {
    NVDS_LOG("all servers' join request received. [total servers = %d]",
             kNumServers);
    ResponseAllJoins();
  } else {
    NVDS_LOG("[%d/%d] join requests received", num_servers_, kNumServers);
  }
}

void Coordinator::ResponseAllJoins() {
  assert(sessions_.size() == kNumServers);
  for (size_t i = 0; i < sessions_.size(); ++i) {
    json msg_body {
      {"id", i},
      {"index_manager", index_manager_}
    };
    sessions_[i]->AsyncSendMessage(std::make_shared<Message>(
        Message::Header {Message::SenderType::COORDINATOR,
                         Message::Type::RES_JOIN},
        msg_body.dump()));
  }
  sessions_.clear();
}

void Coordinator::HandleMessageFromClient(std::shared_ptr<Session> session,
                                          std::shared_ptr<Message> msg) {
  switch (msg->type()) {
  case Message::Type::REQ_JOIN:
    HandleClientRequestJoin(session, msg);
    break;
  case Message::Type::REQ_LEAVE:
    break;
  case Message::Type::ACK_REJECT:
    break;
  case Message::Type::ACK_ERROR:
    break;
  case Message::Type::ACK_OK:
    break;
  default:
    assert(false);
  }
}

void Coordinator::HandleClientRequestJoin(std::shared_ptr<Session> session,
                                          std::shared_ptr<Message> msg) {
  // TODO(wgtdkp): handle this error!
  assert(num_servers_ == kNumServers);
  assert(msg->sender_type() == Message::SenderType::CLIENT);
  assert(msg->type() == Message::Type::REQ_JOIN);
  NVDS_LOG("join request from client: [%s]", session->GetPeerAddr().c_str());
  
  json j_body {
    {"index_manager", index_manager_}
  };
  session->AsyncSendMessage(std::make_shared<Message>(
      Message::Header {Message::SenderType::COORDINATOR,
                        Message::Type::RES_JOIN},
      j_body.dump()));
}

} // namespace nvds
<file_sep>/src/server.h
#ifndef _NVDS_SERVER_H_
#define _NVDS_SERVER_H_

#include "basic_server.h"
#include "common.h"
#include "index.h"
#include "infiniband.h"
#include "measurement.h"
#include "message.h"
#include "spinlock.h"
#include "tablet.h"

#include <boost/asio.hpp>
#include <boost/function.hpp>
#include <condition_variable>
#include <mutex>
#include <thread>

#define ENABLE_MEASUREMENT

namespace nvds {

/*
 * The abstraction of NVM device. No instantiation is allowed.
 * For each server, there is only one NVMDevice object which is
 * mapped to the address passed when the server is started.
 */
struct NVMDevice {

  // The server id of the machine.
  ServerId server_id;
  // The size in byte of the NVM device
  uint64_t size;
  // The key hash range
  KeyHash key_begin;
  KeyHash key_end;
  // The number of tablets
  uint32_t tablet_num;
  // The begin of tablets
  NVMTablet tablets[0];
  NVMDevice() = delete;
};

static const uint64_t kNVMDeviceSize = sizeof(NVMDevice) +
                                       kNVMTabletSize * kNumTabletAndBackupsPerServer;

class Server : public BasicServer {
 // For convenience
 public:
  Measurement alloc_measurement;
  Measurement sync_measurement;
  Measurement thread_measurement;
  Measurement send_measurement;
  Measurement recv_measurement;

 public:
  // Workers & tablets
  friend class Worker;
  using Work = Infiniband::Buffer;
  Server(uint16_t port, NVMPtr<NVMDevice> nvm, uint64_t nvm_size);
  ~Server();
  DISALLOW_COPY_AND_ASSIGN(Server);

  ServerId id() const { return id_; }
  bool active() const { return active_; }  
  uint64_t nvm_size() const { return nvm_size_; }
  NVMPtr<NVMDevice> nvm() const { return nvm_; }
  size_t num_recv() const { return num_recv_; }

  void Run() override;
  bool Join(const std::string& coord_addr);
  void Leave();
  void Listening();
  void Poll();
  // Dispatch the `work` to specific `worker`, load balance considered.
  void Dispatch(Work* work);

 private:
  void HandleRecvMessage(std::shared_ptr<Session> session,
                         std::shared_ptr<Message> msg);
  void HandleSendMessage(std::shared_ptr<Session> session,
                         std::shared_ptr<Message> msg);

  // The WorkQueue will be enqueued by the dispatch,
  // and dequeued by the worker. Thus, it must be made thread safe.
  class WorkQueue {
   public:
    WorkQueue() {}
    DISALLOW_COPY_AND_ASSIGN(WorkQueue);
    // Made Thread safe
    void Enqueue(Work* work) {
      std::lock_guard<Spinlock> _(spinlock_);
      if (head_ == nullptr) {
        //std::lock_guard<Spinlock> _(spinlock_);
        tail_ = work;
        head_ = work;
      } else {
        tail_->next = work;
        tail_ = work;
      }
    }
    // Made Thread safe
    Work* Dequeue() {
      std::lock_guard<Spinlock> _(spinlock_);
      if (head_ == nullptr) {
        return nullptr;
      } else if (head_ == tail_) {
        auto ans = head_;
        tail_ = nullptr;
        head_ = nullptr;
        return const_cast<Work*>(ans);
      } else {
        auto ans = head_;
        head_ = head_->next;
        return const_cast<Work*>(ans);
      }
    }

    Work* TryPollWork() {
      Measurement m;
      m.begin();
      auto ans = Dequeue();
      while (ans == nullptr && m.cur_period() < 50) {
        ans = Dequeue();
      }
      return const_cast<Work*>(ans);
    }

   private:
    volatile Work* head_ = nullptr;
    Work* tail_ = nullptr;
    Spinlock spinlock_;
  };

  class Worker {
   public:
    Worker(Server* server, Tablet* tablet)
        : server_(server), tablet_(tablet),
          slave_(std::bind(&Worker::Serve, this)) {}
    void Enqueue(Work* work) {
      std::unique_lock<std::mutex> lock(mtx_);
      wq_.Enqueue(work);
      #ifdef ENABLE_MEASUREMENT
        server_->thread_measurement.begin();
      #endif
      cond_var_.notify_one();
    }

   private:
    void Serve();
    WorkQueue wq_;
    Server* server_;
    Tablet* tablet_;

    std::mutex mtx_;
    std::condition_variable cond_var_;
    std::thread slave_;
  };

  static const uint32_t kMaxIBQueueDepth = 128;
  ServerId id_;
  bool active_;
  uint64_t nvm_size_;  
  NVMPtr<NVMDevice> nvm_;

  IndexManager index_manager_;

  // Infiniband
  Infiniband ib_;
  Infiniband::Address ib_addr_;
  Infiniband::RegisteredBuffers send_bufs_;
  Infiniband::RegisteredBuffers recv_bufs_;
  Infiniband::QueuePair* qp_;

  // Worker
  std::array<Worker*, kNumTabletsPerServer> workers_;
  std::array<Tablet*, kNumTabletAndBackupsPerServer> tablets_;

  // Statistic
  size_t num_recv_ {0};
};

} // namespace nvds

#endif // _NVDS_SERVER_H_
<file_sep>/test/Makefile
TARGET = test
CC = gcc -std=c11
SRCS = ../src/consistent_hash.c	test_consistent_hash.c	\
	   ../src/nvds.c			test_nvds.c				\
	   ../src/rbtree.c			test_rbtree.c			\
	   test.c

CFLAGS = -g -Wall -D_XOPEN_SOURCE -D_GNU_SOURCE -I./ -I../src/
CFLAGS_MD5 = -lssl -lcrypto

OBJS_DIR = build/
OBJS = $(OBJS_DIR)consistent_hash.o	$(OBJS_DIR)test_consistent_hash.o	\
	   $(OBJS_DIR)nvds.o			$(OBJS_DIR)test_nvds.o				\
	   $(OBJS_DIR)rbtree.o			$(OBJS_DIR)test_rbtree.o			\
	   $(OBJS_DIR)test.o

default: all

all:
	@mkdir -p $(OBJS_DIR)
	@make $(TARGET)

$(TARGET): $(OBJS)
	$(CC) -o $(OBJS_DIR)$@ $^ $(CFLAGS_MD5)

$(OBJS_DIR)consistent_hash.o: ../src/consistent_hash.c ../src/consistent_hash.h
	$(CC) $(CFLAGS) -o $@ -c $<

$(OBJS_DIR)nvds.o: ../src/nvds.c ../src/nvds.h
	$(CC) $(CFLAGS) -o $@ -c $<

$(OBJS_DIR)rbtree.o: ../src/rbtree.c ../src/rbtree.h
	$(CC) $(CFLAGS) -o $@ -c $<

$(OBJS_DIR)test_consistent_hash.o: test_consistent_hash.c 
	$(CC) $(CFLAGS) -o $@ -c $<

$(OBJS_DIR)nvds.o: ../src/nvds.c ../src/nvds.h
	$(CC) $(CFLAGS) -o $@ -c $<

$(OBJS_DIR)rbtree.o: ../src/rbtree.c ../src/rbtree.h
	$(CC) $(CFLAGS) -o $@ -c $<

test_consistent_hash: $(OBJS_DIR)test_consistent_hash.o $(OBJS_DIR)consistent_hash.o $(OBJS_DIR)rbtree.o $(OBJS_DIR)test.o
	$(CC) -o $(OBJS_DIR)$@ $^ $(CFLAGS_MD5)

.PHONY: clean

clean:
	@rm -rf $(OBJS_DIR)
<file_sep>/src/nvm.h
#ifndef _NVDS_NVM_H_
#define _NVDS_NVM_H_

#include <cstdlib>
#include <cstddef>

namespace nvds {

// using nvmoffset_t = uint32_t;

/*
 * A pointer that points to objects allocated on NVM. As are 
 * persistent on NVM, They should be Allocated and freed by
 * the Allocator. This class is for an abstraction of accessing
 * NVM.
 */
template<typename T>
class NVMPtr {
 public:
  explicit NVMPtr(T* ptr) : ptr_(ptr) {}
  ~NVMPtr() {}
  T* ptr() { return ptr_; }
  const T* ptr() const { return ptr_; }
  T* operator->() { return ptr_; }
  const T* operator->() const { return ptr_; }
  T& operator*() { return *ptr_; }
  const T& operator*() const { return *ptr_; }
  bool operator==(const NVMPtr& other) const {
    return ptr_ == other.ptr_;
  }
  bool operator!=(const NVMPtr& other) const {
    return !(*this == other);
  }
  bool operator==(const std::nullptr_t& null) const {
    return ptr_ == null;
  }
  bool operator!=(const std::nullptr_t& null) const {
    return !(*this == nullptr);
  }
  NVMPtr<T> operator+(std::ptrdiff_t d) const {
    return NVMPtr<T>(ptr_ + d);
  }
  NVMPtr<T> operator-(std::ptrdiff_t d) const {
    return NVMPtr<T>(ptr_ - d);
  }
  std::ptrdiff_t operator-(NVMPtr<T> other) const {
    return ptr_ - other.ptr_;
  }
  template<typename RT>
  const NVMPtr<T>& operator=(NVMPtr<RT> rhs) {
    ptr_ = rhs.ptr_;
    return *this;
  }

 private:
  T* ptr_;
};

/*
 * Get nvm with specified size
 */
template <typename T>
NVMPtr<T> AcquireNVM(size_t size) {
  return NVMPtr<T>(static_cast<T*>(malloc(size)));
}

} // namespace nvds

#endif // _NVDS_NVM_H_
<file_sep>/src/experiments/write_rc.c
#include "write_rc.h"

#include <assert.h>
#include <stdint.h>
#include <time.h>

#define RDMA_WRITE_ID   (3)
#define RDMA_READ_ID    (4)
#define RDMA_MAX_LEN  (128 * 1024)
//#define RDMA_WRITE_LEN  (64)
//#define RDMA_READ_LEN   RDMA_WRITE_LEN

static int page_size;

static void nvds_server_exch_info(nvds_data_t* data);
static void nvds_client_exch_info(nvds_data_t* data);
static void nvds_run_server(nvds_context_t* ctx, nvds_data_t* data);
static void nvds_run_client(nvds_context_t* ctx, nvds_data_t* data);
static void nvds_init_local_ib_connection(nvds_context_t* ctx,
                                          nvds_data_t* data);
static void nvds_init_ctx(nvds_context_t* ctx, nvds_data_t* data);
static void nvds_set_qp_state_init(struct ibv_qp* qp, nvds_data_t* data);
static void nvds_set_qp_state_rtr(struct ibv_qp* qp, nvds_data_t* data);
static void nvds_set_qp_state_rts(struct ibv_qp* qp, nvds_data_t* data);

static void nvds_rdma_write(nvds_context_t* ctx, nvds_data_t* data, uint32_t len);
//static void nvds_rdma_read(nvds_context_t* ctx, nvds_data_t* data);
static void nvds_poll_send(nvds_context_t* ctx);
//static void nvds_poll_recv(nvds_context_t* ctx);
//static void nvds_post_send(nvds_context_t* ctx, nvds_data_t* data);
//static void nvds_post_recv(nvds_context_t* ctx, nvds_data_t* data);

static void nvds_print_ib_connection(nvds_ib_connection_t* c) {
  printf("lid: %d\n", c->lid);
  printf("qpn: %d\n", c->qpn);
  printf("psn: %d\n", c->psn);
  printf("rkey: %u\n", c->rkey);
  printf("vaddr: %llu\n", c->vaddr);
}

static void nvds_set_qp_state_init(struct ibv_qp* qp, nvds_data_t* data) {
  struct ibv_qp_attr attr = {
    .qp_state = IBV_QPS_INIT,
    .pkey_index = 0,
    .port_num = data->ib_port,
    .qp_access_flags = IBV_ACCESS_LOCAL_WRITE |
                       IBV_ACCESS_REMOTE_READ |
                       IBV_ACCESS_REMOTE_WRITE
  };

  int modify_flags = IBV_QP_STATE | IBV_QP_PKEY_INDEX |
                     IBV_QP_PORT | IBV_QP_ACCESS_FLAGS;
  nvds_expect(ibv_modify_qp(qp, &attr, modify_flags) == 0,
              "ibv_modify_qp() failed");
}

static void nvds_set_qp_state_rtr(struct ibv_qp* qp, nvds_data_t* data) {
  struct ibv_qp_attr attr = {
    .qp_state           = IBV_QPS_RTR,
    .path_mtu           = IBV_MTU_2048,
    .dest_qp_num        = data->remote_conn.qpn,
    .rq_psn             = data->remote_conn.psn,
    .max_dest_rd_atomic = 1,
    .min_rnr_timer      = 12,
    .ah_attr            = {
      .is_global      = 0,
      .dlid           = data->remote_conn.lid,
      .sl             = 1,
      .src_path_bits  = 0,
      .port_num       = data->ib_port
    }
  };
  int modify_flags = IBV_QP_STATE               |
                     IBV_QP_AV                  |
                     IBV_QP_PATH_MTU            |
                     IBV_QP_DEST_QPN            |
                     IBV_QP_RQ_PSN              |
                     IBV_QP_MAX_DEST_RD_ATOMIC  |
                     IBV_QP_MIN_RNR_TIMER;
  nvds_expect(ibv_modify_qp(qp, &attr, modify_flags) == 0,
              "ibv_modify_qp() failed");
}

static void nvds_set_qp_state_rts(struct ibv_qp* qp, nvds_data_t* data) {
	// first the qp state has to be changed to rtr
	nvds_set_qp_state_rtr(qp, data);

	struct ibv_qp_attr attr = {
	  .qp_state       = IBV_QPS_RTS,
    .timeout        = 14,
    .retry_cnt      = 7,
    .rnr_retry      = 7,
    .sq_psn         = data->local_conn.psn,
    .max_rd_atomic  = 1
  };
  int modify_flags = IBV_QP_STATE            |
                     IBV_QP_TIMEOUT          |
                     IBV_QP_RETRY_CNT        |
                     IBV_QP_RNR_RETRY        |
                     IBV_QP_SQ_PSN           |
                     IBV_QP_MAX_QP_RD_ATOMIC;
  nvds_expect(ibv_modify_qp(qp, &attr, modify_flags) == 0,
              "ibv_modify_qp() failed");
}

static void nvds_server_exch_info(nvds_data_t* data) {
int listen_fd;
  struct sockaddr_in server_addr;

  listen_fd = socket(AF_INET, SOCK_STREAM, 0);
  nvds_expect(listen_fd >= 0, "socket() failed");

  int on = 1;
  nvds_expect(setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR,
                         (const char*)&on, sizeof(on)) != -1,
              "sotsockopt() failed");
  memset(&server_addr, 0, sizeof(server_addr));
  server_addr.sin_family = AF_INET;
  server_addr.sin_addr.s_addr = INADDR_ANY;
  server_addr.sin_port = htons(data->port);

  nvds_expect(bind(listen_fd, (struct sockaddr*)&server_addr,
                   sizeof(server_addr)) >= 0,
              "bind() failed");
  printf("server listening at port %d\n", data->port);
  listen(listen_fd, 1024);

  int fd = accept(listen_fd, NULL, NULL);
  nvds_expect(fd >= 0, "accept() failed");

  int len = read(fd, &data->remote_conn, sizeof(data->remote_conn));
  nvds_expect(len == sizeof(data->remote_conn), "read() failed");

  len = write(fd, &data->local_conn, sizeof(data->local_conn));
  nvds_expect(len == sizeof(data->local_conn), "write() failed");

  close(fd);
  close(listen_fd);
  printf("information exchanged\n");
}

static void nvds_client_exch_info(nvds_data_t* data) {
  int fd = socket(AF_INET, SOCK_STREAM, 0);
  nvds_expect(fd != -1, "socket() failed");

  struct sockaddr_in server_addr;
  memset(&server_addr, 0, sizeof(server_addr));
  server_addr.sin_family = AF_INET;
  server_addr.sin_port = htons(data->port);
  nvds_expect(inet_pton(AF_INET, data->server_name,
                        &server_addr.sin_addr) == 1, 
              "inet_pton() failed");
  nvds_expect(connect(fd, (struct sockaddr*)&server_addr,
                      sizeof(server_addr)) == 0,
              "connect() failed");

  int len = write(fd, &data->local_conn, sizeof(data->local_conn));
  nvds_expect(len == sizeof(data->local_conn), "write() failed");

  len = read(fd, &data->remote_conn, sizeof(data->remote_conn));
  nvds_expect(len == sizeof(data->remote_conn), "read() failed");
  
  close(fd);
  printf("information exchanged\n");
}

static void nvds_run_server(nvds_context_t* ctx, nvds_data_t* data) {
  nvds_server_exch_info(data);

  // Set queue pair state to RTR(Ready to receive)
  nvds_set_qp_state_rtr(ctx->qp, data);
  printf("server side: \n");
  printf("local connection: \n");
  nvds_print_ib_connection(&data->local_conn);
  printf("remote connection: \n");  
  nvds_print_ib_connection(&data->remote_conn);
  // TODO(wgtdkp): poll request from client or do nothing
  while (1) {}
}

/*
static void nvds_multi_rdma_write(nvds_context_t* ctx, nvds_data_t* data) {

}

static void nvds_test_multi_sge(nvds_context_t* ctx, nvds_data_t* data) {
  static const int n = 1000 * 1000;

}
*/

static inline int64_t time_diff(struct timespec* lhs, struct timespec* rhs) {
  return (lhs->tv_sec - rhs->tv_sec) * 1000000000 + 
         (lhs->tv_nsec - rhs->tv_nsec);
}

static void nvds_run_client(nvds_context_t* ctx, nvds_data_t* data) {
  nvds_client_exch_info(data);
  nvds_set_qp_state_rts(ctx->qp, data);
  //printf("client side: \n");
  //printf("local connection: \n");
  //nvds_print_ib_connection(&data->local_conn);
  //printf("remote connection: \n");  
  //nvds_print_ib_connection(&data->remote_conn);

  static const int n = 1000 * 1000;
  for (int32_t len = 1; len <= 1024; len <<= 1) {
    struct timespec begin, end;
    assert(clock_gettime(CLOCK_REALTIME, &begin) == 0);
    for (int i = 0; i < n; ++i) {
      nvds_rdma_write(ctx, data, len);
      nvds_poll_send(ctx);
    }
    assert(clock_gettime(CLOCK_REALTIME, &end) == 0);
    int64_t t = time_diff(&end, &begin);

    printf("%d, %f\n", len, t / 1000000.0 / 1000);
  }
  exit(0);
}

static void nvds_rdma_write(nvds_context_t* ctx, nvds_data_t* data, uint32_t len) {
  ctx->sge.addr    = (uintptr_t)ctx->buf;
  ctx->sge.length  = len;
  ctx->sge.lkey    = ctx->mr->lkey;

  // The address write to
  ctx->wr.wr.rdma.remote_addr = data->remote_conn.vaddr;
  ctx->wr.wr.rdma.rkey        = data->remote_conn.rkey;
  ctx->wr.wr_id               = RDMA_WRITE_ID;
  ctx->wr.sg_list             = &ctx->sge;
  ctx->wr.num_sge             = 1;
  ctx->wr.opcode              = IBV_WR_RDMA_WRITE;
  ctx->wr.send_flags          = IBV_SEND_SIGNALED;
  ctx->wr.next                = NULL;

  struct ibv_send_wr* bad_wr;
  nvds_expect(ibv_post_send(ctx->qp, &ctx->wr, &bad_wr) == 0,
              "ibv_post_send() failed");
}

/*
static void nvds_rdma_read(nvds_context_t* ctx, nvds_data_t* data) {
  ctx->sge.addr    = (uintptr_t)ctx->buf + RDMA_WRITE_LEN;
  ctx->sge.length  = RDMA_READ_LEN;
  ctx->sge.lkey    = ctx->mr->lkey;

  // The address read from
  ctx->wr.wr.rdma.remote_addr = data->remote_conn.vaddr;
  ctx->wr.wr.rdma.rkey        = data->remote_conn.rkey;
  ctx->wr.wr_id               = RDMA_READ_ID;
  ctx->wr.sg_list             = &ctx->sge;
  ctx->wr.num_sge             = 1;
  ctx->wr.opcode              = IBV_WR_RDMA_READ;
  ctx->wr.send_flags          = IBV_SEND_SIGNALED;
  ctx->wr.next                = NULL;

  struct ibv_send_wr* bad_wr;
  nvds_expect(ibv_post_send(ctx->qp, &ctx->wr, &bad_wr) == 0,
              "ibv_post_send() failed");
}
*/
static void nvds_poll_send(nvds_context_t* ctx) {
  struct ibv_wc wc;
  while (ibv_poll_cq(ctx->scq, 1, &wc) != 1) {}
  nvds_expect(wc.status == IBV_WC_SUCCESS, "rdma write failed");
  nvds_expect(wc.wr_id == ctx->wr.wr_id, "wr id not matched");
}

/*
static void nvds_poll_recv(nvds_context_t* ctx) {
  struct ibv_wc wc;
  int cnt = ibv_poll_cq(ctx->rcq, 1, &wc);
  nvds_expect(cnt == 1, "ibv_poll_cq() failed");
  nvds_expect(wc.status == IBV_WC_SUCCESS, "rdma read failed");
  nvds_expect(wc.wr_id == ctx->wr.wr_id, "wr id not matched");
}
*/

/*
static void nvds_post_send(nvds_context_t* ctx, nvds_data_t* data) {

}

static void nvds_post_recv(nvds_context_t* ctx, nvds_data_t* data) {

}
*/

static void nvds_init_local_ib_connection(nvds_context_t* ctx,
                                          nvds_data_t* data) {
  struct ibv_port_attr attr;
  nvds_expect(ibv_query_port(ctx->context, data->ib_port, &attr) == 0,
              "ibv_query_port() failed");
  data->local_conn.lid = attr.lid;
  data->local_conn.qpn = ctx->qp->qp_num;
  data->local_conn.psn = 33; //lrand48() & 0xffffff;
  data->local_conn.rkey = ctx->mr->rkey;
  data->local_conn.vaddr = (uintptr_t)ctx->buf;
}

static void nvds_init_ctx(nvds_context_t* ctx, nvds_data_t* data) {
  memset(ctx, 0, sizeof(nvds_context_t));
  ctx->size = data->size;
  ctx->tx_depth = data->tx_depth;

  // Get IB device
  struct ibv_device** dev_list = ibv_get_device_list(NULL);
  nvds_expect(dev_list, "ibv_get_device_list() failed");
  data->ib_dev = dev_list[0];
  
  // Open IB device
  ctx->context = ibv_open_device(data->ib_dev);
  nvds_expect(ctx->context, "ibv_open_device() failed");

  // Alloc protection domain
  ctx->pd = ibv_alloc_pd(ctx->context);
  nvds_expect(ctx->pd, "ibv_alloc_pd() failed");

  // Alloc buffer for write
  ctx->buf = memalign(page_size, ctx->size);
  memset(ctx->buf, 0, ctx->size);

  // Register memory region
  int access = IBV_ACCESS_LOCAL_WRITE |
               IBV_ACCESS_REMOTE_READ |
				       IBV_ACCESS_REMOTE_WRITE;
  ctx->mr = ibv_reg_mr(ctx->pd, ctx->buf, ctx->size, access);
  nvds_expect(ctx->mr, "ibv_reg_mr() failed");

  // Create completion channel
  //ctx->ch = ibv_create_comp_channel(ctx->context);
  //nvds_expect(ctx->ch, "ibv_create_comp_channel() failed");

  // Create complete queue
  ctx->rcq = ibv_create_cq(ctx->context, 1, NULL, NULL, 0);
  nvds_expect(ctx->rcq, "ibv_create_cq() failed");
  //ctx->scq = ibv_create_cq(ctx->context, ctx->tx_depth, ctx, ctx->ch, 0);
  ctx->scq = ibv_create_cq(ctx->context, ctx->tx_depth, NULL, NULL, 0);
  nvds_expect(ctx->rcq, "ibv_create_cq() failed");  

  // Create & init queue apir
  struct ibv_qp_init_attr qp_init_attr = {
    .send_cq = ctx->scq,
    .recv_cq = ctx->rcq,
    .qp_type = QP_TYPE,
    .sq_sig_all = 0,
    .cap = {
      .max_send_wr = ctx->tx_depth,
      .max_recv_wr = 1,
      .max_send_sge = 1,
      .max_recv_sge = 1,
      .max_inline_data = 0
    }
  };
  ctx->qp = ibv_create_qp(ctx->pd, &qp_init_attr);
  nvds_expect(ctx->qp, "ibv_create_qp() failed");
  nvds_set_qp_state_init(ctx->qp, data);
}

int main(int argc, const char* argv[]) {
  nvds_context_t ctx;
  nvds_data_t data = {
    .port         = 5500,
    .ib_port      = 1,
    .size         = RDMA_MAX_LEN,
    .tx_depth     = 100,
    .server_name  = NULL,
    .ib_dev       = NULL
  };

  nvds_expect(argc <= 2, "too many arguments");
  int is_client = 0;
  if (argc == 2) {
    is_client = 1;
    data.server_name = argv[1];
  }

  pid_t pid = getpid();
  srand48(pid * time(NULL));
  page_size = sysconf(_SC_PAGESIZE);
  nvds_init_ctx(&ctx, &data);
  nvds_init_local_ib_connection(&ctx, &data);

  if (is_client) {
    nvds_run_client(&ctx, &data);
  } else {
    nvds_run_server(&ctx, &data);
  }

  return 0;
}
<file_sep>/README.md
[Google Test]: https://github.com/google/googletest
[json]: https://github.com/nlohmann/json

# NVDS

Fast replication over RDMA fabric for NVM-based storage system.

## DEPENDENCY

+ [Google Test]
+ [json]

## BUILD

+ Get the source code:
```shell
$ git clone https://github.com/wgtdkp/nvds --recursive
```

+ Checkout the `master` branch of submodule `json`:
```shell
$ cd nvds/json && git checkout master
```

+ Build nvds from source:
```shell
$ cd ../src && make all
```

## INSTALL
```shell
$ make install # root required
```

## RUN
Navigate to directory `script`:
```shell
script$ ./startup.sh
```

## CONFIGURE
### cluster
Parameters below are defined in header file `common.h`, recompilation and reinstallation are needed for changes to take effect.

| parameter    | default | range | description |
| :--------:   | :---:   | :---: | :---------: |
| kNumReplicas |    1    | [1, ] | the number of replications in primary backup |
| kNumServers  |    2    | [1, ] | the number of servers in this cluster |
| kNumTabletsPerServer | 1 | [1, 16] | the number of tablets per server |

The volume of the whole cluster equals to: kNumServers * kNumTabletsPerServer * 64MB;

### servers
Specify server addresses in file `script/servers.txt`. The total number of servers
should equals to parameter `kNumServers`. The file format: a line for a server address(ip address and port number separated by space). An example of two servers:

```text
192.168.99.14 5050
192.168.99.14 5051
```

## PROGRAMMING
To connect to a nvds cluster, a client only needs to include header file `nvds/client.h` and link nvds's static library.
A tutorial snippet below shows how to put a key/value pair to the cluster and fetch it later:

```c++
#include "nvds/client.h"

#include <iostream>
#include <string>

int main() {
  try {
    nvds::Client c {"192.168.99.14"};
    c.Put("hello", "world");
    std::cout << "hello: " << c.Get("hello") << std::endl;
  } catch (boost::system::system_error& e) {
    NVDS_ERR(e.what());
  } catch (nvds::TransportException& e) {
    NVDS_ERR(e.msg().c_str());
    NVDS_LOG("initialize infiniband devices failed");
  }
  return 0;
}
```

Compile:

```bash
g++ -std=c++11 -Wall test_client.cc -lnvds -lboost_system -lboost_thread -pthread -libverbs
```

The whole code can be obtained in source file `src/test_client.cc`.

## TROUBLESHOOTING

### enable UD
 If `ibv_create_ah` failed with `"Cannot allocate memory"`, first, checkout the infiniband port status by typing command `ibstatus`; it should show that the port state is `INIT`; it's caused by disabling of `opensm`, start `opensm` by command: `/etc/init.d/opensmd start`; make sure the port state transformed to `ACTIVE`. It should be fixed then. This problem usually occurs after reboot, just enable _opensm_ once started.

### setting up IPoIB
Setting up IPoIB for testing of memcached latency.
Do below steps on both client and server machine.
1. `lsmod | grep ib_ipoib` check if ipoib module is listed. If not, `modprobe ib_ipoib`;
2. `ifconfig ib0 10.0.0.7/24` setting route and address (the address is for    `ddst7`, you may specify an address as you like) of `ib0`;
3. make sure `opensm` service is enabled, you can restart it by `service opensmd restart`;
4. `ping 10.0.0.4` ping peer machine (this address must be 10.0.0.*) to make sure IPoIB works for you.

Have fun then!
<file_sep>/src/measurement.h
#ifndef _NVDS_MEASUREMENT_H_
#define _NVDS_MEASUREMENT_H_

#include <chrono>
#include <iostream>
#include <ratio>

namespace nvds {

class Measurement {
 public:
  void begin() {
    using namespace std::chrono;
    begin_clock_ = high_resolution_clock::now();
  }
  void end() {
    using namespace std::chrono;
    end_clock_ = high_resolution_clock::now();
    ++num_repetitions_;
    total_time_ += duration_cast<duration<double, std::micro>>(end_clock_ - begin_clock_).count();
  }

  double total_time() const { return total_time_; }
  size_t num_repetitions() const { return num_repetitions_; }
  double average_time() const { return total_time_ / num_repetitions_; }
  double cur_period() const {
    using namespace std::chrono;
    auto now = high_resolution_clock::now();
    return duration_cast<duration<double, std::micro>>(now - begin_clock_).count();
  }
  // DEBUG
  void Print() const {
    std::cout << "total time: " << total_time() << std::endl;
    std::cout << "repetitions: " << num_repetitions() << std::endl;
    std::cout << "average time: " << average_time() << std::endl;
  }

 private:
  std::chrono::time_point<std::chrono::high_resolution_clock> begin_clock_;
  std::chrono::time_point<std::chrono::high_resolution_clock> end_clock_;
  double total_time_ {0};
  size_t num_repetitions_ {0};
};

} // namespace nvds

#endif // _NVDS_MESSAGE_
<file_sep>/src/coordinator_main.cc
/*
 * Startup module of coordinator. For a storage cluster,
 * it's the coordinator that starts up first, load configs
 * from the config file, 
 */

#include "coordinator.h"

using namespace nvds;

/*
static void Usage(int argc, const char* argv[]) {
  std::cout << "Usage:" << std::endl
            << "    " << argv[0] << " <config_file>" << std::endl;
}
*/

static std::string GetLocalAddr() {
  using boost::asio::ip::tcp;
  boost::asio::io_service io_service;
  tcp::resolver resolver(io_service);
  tcp::resolver::query query(boost::asio::ip::host_name(), "");
  auto iter = resolver.resolve(query);
  decltype(iter) end;
  while (iter != end) {
    tcp::endpoint ep = *iter++;
    return ep.address().to_string();
  }
  assert(false);
  return "";
}

int main(int argc, const char* argv[]) {
  // -1. Argument parsing
  //if (argc < 2) {
  //  Usage(argc, argv);
  //  return -1;
  //}

  // 1. Self initialization
  Coordinator coordinator;

  // 2. Listening for joining request from nodes and request from clients
  NVDS_LOG("Coordinator at: %s", GetLocalAddr().c_str());
  NVDS_LOG("Listening at: %u", kCoordPort);
  NVDS_LOG("......");

  coordinator.Run();

  return 0;
}
<file_sep>/src/test.h
#ifndef _NVDS_TEST_H_
#define _NVDS_TEST_H_

namespace nvds {

} // namespace nvds

#endif // _NVDS_TEST_H_
<file_sep>/src/infiniband.cc
#include "infiniband.h"

#include <algorithm>
#include <malloc.h>

namespace nvds {

Infiniband::Infiniband(uint32_t registered_mem_size)
    : ctx_(nullptr), pd_(nullptr), mr_(nullptr),
      raw_mem_(nullptr), raw_mem_size_(registered_mem_size) {
  struct ibv_device** dev_list = ibv_get_device_list(NULL);
  assert(dev_list != nullptr);
  ctx_ = ibv_open_device(dev_list[0]);
  assert(ctx_ != nullptr);
  ibv_free_device_list(dev_list);

  pd_ = ibv_alloc_pd(ctx_);
  assert(pd_ != nullptr);

  raw_mem_ = reinterpret_cast<char*>(memalign(4096, raw_mem_size_));
  assert(raw_mem_ != nullptr);

  int access = IBV_ACCESS_LOCAL_WRITE |
               IBV_ACCESS_REMOTE_READ |
               IBV_ACCESS_REMOTE_WRITE;
  mr_ = ibv_reg_mr(pd_, raw_mem_, raw_mem_size_, access);
  assert(mr_ != nullptr);

  std::fill(addr_handlers_.begin(), addr_handlers_.end(), nullptr);
}

Infiniband::~Infiniband() {
  // Release in reverse order
  int err;
  for (auto ah : addr_handlers_) {
    if (ah != nullptr) {
      err = ibv_destroy_ah(ah);
      assert(err == 0);
    }
  }
  err = ibv_dereg_mr(mr_);
  assert(err == 0);
  free(raw_mem_);
  err = ibv_dealloc_pd(pd_);
  assert(err == 0);
  err = ibv_close_device(ctx_);
  assert(err == 0);
}

void Infiniband::QueuePairInfo::Print() const {
  std::cout << "lid: " << lid << std::endl;
  std::cout << "qpn: " << qpn << std::endl;
  std::cout << "psn: " << psn << std::endl;
  std::cout << "rkey: " << rkey << std::endl;
  std::cout << "vaddr: " << vaddr << std::endl;
}

Infiniband::QueuePair::QueuePair(Infiniband& ib, ibv_qp_type type,
                                 uint32_t max_send, uint32_t max_recv)
    : ib(ib), type(type), qp(nullptr),
      scq(nullptr), rcq(nullptr), psn(kDefaultPsn) {
  assert(type == IBV_QPT_RC || type == IBV_QPT_UD);

  scq = ibv_create_cq(ib.ctx(), max_send, nullptr, nullptr, 0);
  rcq = ibv_create_cq(ib.ctx(), max_recv, nullptr, nullptr, 0);
  assert(scq != nullptr && rcq != nullptr);

  ibv_qp_init_attr init_attr;
  memset(&init_attr, 0, sizeof(init_attr));
  init_attr.srq     = nullptr;
  init_attr.send_cq = scq;
  init_attr.recv_cq = rcq;
  init_attr.qp_type = type;
  init_attr.cap.max_send_wr = max_send;
  init_attr.cap.max_recv_wr = max_recv;
  init_attr.cap.max_send_sge = Infiniband::kMaxSendSge;
  init_attr.cap.max_recv_sge = Infiniband::kMaxRecvSge;
  init_attr.cap.max_inline_data = Infiniband::kMaxInlineData;
  init_attr.sq_sig_all = 0; // Related to unsignaled completion

  qp = ibv_create_qp(ib.pd(), &init_attr);
  if (qp == nullptr) {
    throw TransportException(HERE, "create queue pair failed", errno);
  }

  ibv_qp_attr attr;
  attr.qp_state = IBV_QPS_INIT;
  attr.pkey_index = 0;
  attr.port_num = Infiniband::kPort;
  attr.qkey = 0;
  attr.qp_access_flags = IBV_ACCESS_LOCAL_WRITE |
                         IBV_ACCESS_REMOTE_READ |
                         IBV_ACCESS_REMOTE_WRITE;
  int modify = IBV_QP_STATE | IBV_QP_PORT | IBV_QP_PKEY_INDEX;
  modify |= type == IBV_QPT_RC ? IBV_QP_ACCESS_FLAGS : IBV_QP_QKEY;
  int ret = ibv_modify_qp(qp, &attr, modify);
  if (ret != 0) {
    ibv_destroy_qp(qp);
    throw TransportException(HERE, ret);
  }
}

Infiniband::QueuePair::~QueuePair() {
  int err = ibv_destroy_qp(qp);
  assert(err == 0);
  err = ibv_destroy_cq(rcq);
  assert(err == 0);
  err = ibv_destroy_cq(scq);
  assert(err == 0);
}

uint32_t Infiniband::QueuePair::GetPeerQPNum() const {
  ibv_qp_attr attr;
  ibv_qp_init_attr init_attr;
  auto err = ibv_query_qp(qp, &attr, IBV_QP_DEST_QPN, &init_attr);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
  return attr.dest_qp_num;
}

int Infiniband::QueuePair::GetState() const {
  ibv_qp_attr attr;
  ibv_qp_init_attr init_attr;
  int err = ibv_query_qp(qp, &attr, IBV_QP_STATE, &init_attr);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
  return attr.qp_state;
}

// Bring UD queue pair into RTS status
void Infiniband::QueuePair::Activate() {
  assert(type == IBV_QPT_UD);
  assert(GetState() == IBV_QPS_INIT);

  ibv_qp_attr attr;
  memset(&attr, 0, sizeof(attr));
  attr.qp_state = IBV_QPS_RTR;
  auto err = ibv_modify_qp(qp, &attr, IBV_QP_STATE);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
  
  attr.qp_state = IBV_QPS_RTS;
  attr.sq_psn = psn;
  err = ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_SQ_PSN);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
}

void Infiniband::QueuePair::SetStateRTR(const QueuePairInfo& peer_info) {
  assert(type == IBV_QPT_RC);
  assert(GetState() == IBV_QPS_INIT);

  struct ibv_qp_attr attr;
  // We must set qp into RTR state before RTS
  attr.qp_state           = IBV_QPS_RTR;
  attr.path_mtu           = IBV_MTU_4096;
  attr.dest_qp_num        = peer_info.qpn;
  attr.rq_psn             = peer_info.psn;
  attr.max_dest_rd_atomic = 1;
  attr.min_rnr_timer      = 12;
  attr.ah_attr.is_global  = 0;
  attr.ah_attr.dlid       = peer_info.lid;
  attr.ah_attr.sl         = 0;
  attr.ah_attr.src_path_bits = 0;
  attr.ah_attr.port_num      = Infiniband::kPort;

  int modify = IBV_QP_STATE               |
               IBV_QP_AV                  |
               IBV_QP_PATH_MTU            |
               IBV_QP_DEST_QPN            |
               IBV_QP_RQ_PSN              |
               IBV_QP_MAX_DEST_RD_ATOMIC  |
               IBV_QP_MIN_RNR_TIMER;

  int err = ibv_modify_qp(qp, &attr, modify);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
}

void Infiniband::QueuePair::SetStateRTS(const QueuePairInfo& peer_info) {
  SetStateRTR(peer_info);

  struct ibv_qp_attr attr;
  attr.qp_state   = IBV_QPS_RTS;
  attr.timeout    = 14;
  attr.retry_cnt  = 7;
  attr.rnr_retry  = 7;
  attr.sq_psn     = psn;
  attr.max_rd_atomic = 1;

  int modify = IBV_QP_STATE            |
               IBV_QP_TIMEOUT          |
               IBV_QP_RETRY_CNT        |
               IBV_QP_RNR_RETRY        |
               IBV_QP_SQ_PSN           |
               IBV_QP_MAX_QP_RD_ATOMIC;

  int err = ibv_modify_qp(qp, &attr, modify);
  if (err != 0) {
    throw TransportException(HERE, err);
  }
}

Infiniband::RegisteredBuffers::RegisteredBuffers(ibv_pd* pd,
    uint32_t buf_size, uint32_t buf_num, bool is_recv)
    : buf_size_(buf_size), buf_num_(buf_num),
      ptr_(nullptr), bufs_(nullptr), root_(nullptr) {
  const size_t bytes = buf_size * buf_num;
  ptr_ = memalign(sysconf(_SC_PAGESIZE), bytes);
  assert(ptr_ != nullptr);
  int access = IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_LOCAL_WRITE;
  auto mr = ibv_reg_mr(pd, ptr_, bytes, access);
  if (mr == nullptr) {
    throw TransportException(HERE, "failed to register buffers", errno);
  }
  bufs_ = new Buffer[buf_num_];
  char* p = static_cast<char*>(ptr_);
  for (uint32_t i = 0; i < buf_num; ++i) {
    new (&bufs_[i]) Buffer(p, buf_size_, mr, is_recv);
    bufs_[i].next = i + 1 < buf_num ? &bufs_[i+1] : nullptr;
    p += buf_size_;
  }
  root_ = bufs_;
}

Infiniband::RegisteredBuffers::~RegisteredBuffers() {
  // Deregsiter memory region
  ibv_dereg_mr(bufs_[0].mr);
  free(ptr_);
  delete[] bufs_;
}

uint16_t Infiniband::GetLid(uint16_t port) {
  ibv_port_attr port_attr;
  auto ret = ibv_query_port(ctx_, port, &port_attr);
  if (ret != 0) {
    throw TransportException(HERE, ret);
  }
  return port_attr.lid;
}

Infiniband::Buffer* Infiniband::PollCQ(ibv_cq* cq, int num_entries) {
  ibv_wc wc;
  int r = ibv_poll_cq(cq, num_entries, &wc);
  if (r == 0) {
    return NULL;
  } else if (r < 0) {
    throw TransportException(HERE, r);
  } else if (wc.status != IBV_WC_SUCCESS) {
    throw TransportException(HERE, wc.status);
  }
  // wr_id is used as buffer address
  auto b = reinterpret_cast<Buffer*>(wc.wr_id);
  b->msg_len = wc.byte_len;
  // FIXME(wgtdkp):
  b->peer_addr = {Infiniband::kPort/*qp->ib_port*/, wc.slid, wc.src_qp};
  return b;
}

// May blocking
Infiniband::Buffer* Infiniband::Receive(QueuePair* qp) {
  Buffer* b;
  while ((b = TryReceive(qp)) == nullptr) {}
  return b;
}

void Infiniband::PostReceive(QueuePair* qp, Buffer* b) {
  ibv_sge sge {
    reinterpret_cast<uint64_t>(b->buf),
    b->size,
    b->mr->lkey
  };
  ibv_recv_wr rwr;
  memset(&rwr, 0, sizeof(rwr));
  rwr.wr_id = reinterpret_cast<uint64_t>(b);
  rwr.next = nullptr;
  rwr.sg_list = &sge;
  rwr.num_sge = 1;

  ibv_recv_wr* bad_rwr;
  auto err = ibv_post_recv(qp->qp, &rwr, &bad_rwr);
  if (err != 0) {
    throw TransportException(HERE, "ibv_post_recv failed", err);
  }
}

void Infiniband::PostSend(QueuePair* qp, Buffer* b,
    uint32_t len, const Address* peer_addr) {
  assert(qp->type == IBV_QPT_UD);

  ibv_sge sge {
    reinterpret_cast<uint64_t>(b->buf),
    len,
    b->mr->lkey
  };
  ibv_send_wr swr;
  memset(&swr, 0, sizeof(swr));
  swr.wr_id = reinterpret_cast<uint64_t>(b);
  if (qp->type == IBV_QPT_UD) {
    swr.wr.ud.ah = GetAddrHandler(*peer_addr);
    swr.wr.ud.remote_qpn = peer_addr->qpn;
    swr.wr.ud.remote_qkey = 0;
    assert(swr.wr.ud.ah != nullptr);
  }
  swr.next = nullptr;
  swr.sg_list = &sge;
  swr.num_sge = 1;
  swr.opcode = IBV_WR_SEND;
  swr.send_flags = IBV_SEND_SIGNALED;
  if (len <= kMaxInlineData) {
    swr.send_flags |= IBV_SEND_INLINE;
  }

  ibv_send_wr* bad_swr;
  auto err = ibv_post_send(qp->qp, &swr, &bad_swr);
  if (err != 0) {
    throw TransportException(HERE, "ibv_post_send failed", err);
  }
}

void Infiniband::PostSendAndWait(QueuePair* qp, Buffer* b,
    uint32_t len, const Address* peer_addr) {
  assert(qp->type == IBV_QPT_UD);

  PostSend(qp, b, len, peer_addr);

  ibv_wc wc;
  while (ibv_poll_cq(qp->scq, 1, &wc) < 1) {}
  // FIXME(wgtdkp):
  // How can we make sure that the completed work we polled is the one
  // we have just post?
  if (wc.status != IBV_WC_SUCCESS) {
    throw TransportException(HERE, "PostSendAndWait failed", wc.status);
  }
}

ibv_ah* Infiniband::GetAddrHandler(const Address& addr) {
  auto& ah = addr_handlers_[addr.lid];
  if (ah != nullptr) {
    return ah;
  }
  
  ibv_ah_attr attr;
  attr.is_global = 0;
  attr.dlid = addr.lid;
  attr.sl = 1;
  attr.src_path_bits = 0;
  attr.port_num = kPort;
  ah = ibv_create_ah(pd_, &attr);
  assert(ah != nullptr);
  return ah;
}

void Infiniband::PostWrite(QueuePair& qp, QueuePairInfo& peer_info) {
  struct ibv_sge sge;
  sge.addr   = reinterpret_cast<uint64_t>(raw_mem_);
  sge.length = 64;
  sge.lkey   = mr_->lkey;

  struct ibv_send_wr swr;
  swr.wr.rdma.remote_addr = peer_info.vaddr;
  swr.wr.rdma.rkey        = peer_info.rkey;
  swr.wr_id               = 3;
  swr.sg_list             = &sge;
  swr.num_sge             = 1;
  swr.opcode              = IBV_WR_RDMA_WRITE;
  swr.send_flags          = IBV_SEND_SIGNALED;
  swr.next                = nullptr;

  struct ibv_send_wr* bad_wr;
  int err = ibv_post_send(qp.qp, &swr, &bad_wr);
  if (err != 0) {
    throw TransportException(HERE, "post rdma write failed", err);
  }
}

void Infiniband::PostWriteAndWait(QueuePair& qp, QueuePairInfo& peer_info) {
  PostWrite(qp, peer_info);

  struct ibv_wc wc;
  int r;
  while ((r = ibv_poll_cq(qp.scq, 1, &wc)) != 1) {}
  if (wc.status != IBV_WC_SUCCESS) {
    throw TransportException(HERE, "poll completion queue failed", wc.status);
  }
}

} // namespace nvds
<file_sep>/src/experiments/write_rc_multi.h
#ifndef _NVDS_COMMON_H_
#define _NVDS_COMMON_H_

#include <ctype.h>
#include <errno.h>
#include <malloc.h>
#include <memory.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include <arpa/inet.h>
#include <infiniband/verbs.h>
#include <sys/socket.h>
#include <sys/time.h>
// Interprocess communication access
#include <sys/ipc.h>
// Shared memory facility
#include <sys/shm.h>

#define nvds_error(format, ...) {                         \
  fprintf(stderr, "error: %s: %d: ", __FILE__, __LINE__); \
  fprintf(stderr, format, ## __VA_ARGS__);                \
  fprintf(stderr, "\n");                                  \
  exit(-1);                                               \
}

#define nvds_expect(cond, err, ...) { \
  if (!(cond)) {                      \
    nvds_error(err, ## __VA_ARGS__);  \
  }                                   \
}

#ifndef NUM_QP
#define NUM_QP  (1024)
#endif
#define QP_TYPE IBV_QPT_RC

typedef struct nvds_context {
  struct ibv_context*       context;
	struct ibv_pd*            pd;
	struct ibv_mr*            mr;
	struct ibv_cq*            rcq;
	struct ibv_cq*            scq;
	struct ibv_qp*            qp;
	struct ibv_comp_channel*  ch;
	char*                     buf;
	unsigned            	    buf_size;
	int                 	    tx_depth;
	struct ibv_sge      	    sge;
	struct ibv_send_wr  	    wr;
} nvds_context_t;

typedef struct nvds_ib_connection {
  int             	  lid;
  int            	    qpn;
  int             	  psn;
	unsigned 			      rkey;
	unsigned long long 	vaddr;
} nvds_ib_connection_t;

typedef struct nvds_data {
	int									  port;
	int									  ib_port;
	unsigned           	  size;
	int                		tx_depth;
	int 		    					sockfd;
	const char*           server_name;
	nvds_ib_connection_t	local_conn;
	nvds_ib_connection_t  remote_conn;
	struct ibv_device*    ib_dev;
} nvds_data_t;

#endif
<file_sep>/src/server.cc
#include "server.h"

#include "json.hpp"
#include "request.h"

#include <chrono>
#include <thread>

namespace nvds {

using json = nlohmann::json;

Server::Server(uint16_t port, NVMPtr<NVMDevice> nvm, uint64_t nvm_size)
    : BasicServer(port), id_(0),
      active_(false), nvm_size_(nvm_size), nvm_(nvm),
      send_bufs_(ib_.pd(), kSendBufSize, kMaxIBQueueDepth, false),
      recv_bufs_(ib_.pd(), kRecvBufSize, kMaxIBQueueDepth, true) {
  // Infiniband
  qp_ = new Infiniband::QueuePair(ib_, IBV_QPT_UD,
      kMaxIBQueueDepth, kMaxIBQueueDepth);
  qp_->Activate();
  
  ib_addr_ = {
    Infiniband::kPort,
    ib_.GetLid(Infiniband::kPort),
    qp_->GetLocalQPNum()
  };

  // Tablets
  for (uint32_t i = 0; i < kNumTabletAndBackupsPerServer; ++i) {
    auto ptr = reinterpret_cast<char*>(&nvm_->tablets) + i * kNVMTabletSize;
    bool is_backup = i >= kNumTabletsPerServer;
    tablets_[i] = new Tablet(index_manager_,
        NVMPtr<NVMTablet>(reinterpret_cast<NVMTablet*>(ptr)), is_backup);
    if (i < kNumTabletsPerServer) {
      workers_[i] = new Worker(this, tablets_[i]);
    }
  }
}

Server::~Server() {
  // Destruct elements in reverse order
  for (int64_t i = kNumTabletAndBackupsPerServer - 1; i >= 0; --i) {
    delete tablets_[i];
    if (i < kNumTabletsPerServer) {
      delete workers_[i];
    }
  }
  delete qp_;
}

void Server::Run() {
  std::thread poller(&Server::Poll, this);
  
  Accept(std::bind(&Server::HandleRecvMessage, this,
                   std::placeholders::_1, std::placeholders::_2),
         std::bind(&Server::HandleSendMessage, this,
                   std::placeholders::_1, std::placeholders::_2));
  RunService();
  
  poller.join();
}

bool Server::Join(const std::string& coord_addr) {
  using boost::asio::ip::tcp;
  tcp::resolver resolver(tcp_service_);
  tcp::resolver::query query(coord_addr, std::to_string(kCoordPort));
  auto ep = resolver.resolve(query);
  try {
    boost::asio::connect(conn_sock_, ep);
    Session session_join(std::move(conn_sock_));

    Message::Header header {Message::SenderType::SERVER,
                            Message::Type::REQ_JOIN, 0};

    std::vector<Infiniband::QueuePairInfo> qpis;
    for (uint32_t i = 0; i < kNumTabletAndBackupsPerServer; ++i) {
      for (uint32_t j = 0; j < kNumReplicas; ++j) {
        qpis.emplace_back(tablets_[i]->info().qpis[j]);
      }
    }
    json body {
      {"size", nvm_size_},
      {"ib_addr", ib_addr_},
      {"tablet_qpis", qpis},
    };
    Message msg(header, body.dump());

    try {
      session_join.SendMessage(msg);
    } catch (boost::system::system_error& err) {
      NVDS_ERR("send join request to coordinator failed: %s", err.what());
      return false;
    }
    try {
      msg = session_join.RecvMessage();
      assert(msg.sender_type() == Message::SenderType::COORDINATOR);
      assert(msg.type() == Message::Type::RES_JOIN);
      auto j_body = json::parse(msg.body());
      id_ = j_body["id"];
      index_manager_ = j_body["index_manager"];
      active_ = true;

      auto& server_info = index_manager_.GetServer(id_);
      for (uint32_t i = 0; i < kNumTabletAndBackupsPerServer; ++i) {
        auto tablet_id = server_info.tablets[i];
        // Both master and backup tablets
        tablets_[i]->SettingupQPConnect(tablet_id, index_manager_);
      }
    } catch (boost::system::system_error& err) {
      NVDS_ERR("receive join response from coordinator failed: %s",
               err.what());
      return false;
    }
  } catch (boost::system::system_error& err) {
    NVDS_ERR("connect to coordinator: %s: %" PRIu16 " failed",
             coord_addr.c_str(), kCoordPort);
    return false;
  }
  return true;
}

void Server::Leave() {
  assert(false);
}

void Server::Listening() {
  assert(false);
}

void Server::HandleRecvMessage(std::shared_ptr<Session> session,
                               std::shared_ptr<Message> msg) {
  assert(false);
}

void Server::HandleSendMessage(std::shared_ptr<Session> session,
                               std::shared_ptr<Message> msg) {
  assert(false);
}

void Server::Poll() {
  // This has a great
  // Fill the receive queue
  for (size_t i = 0; i < kMaxIBQueueDepth; ++i) {
    auto b = recv_bufs_.Alloc();
    assert(b);
    ib_.PostReceive(qp_, b);
  }

  using namespace std::chrono;
  //uint64_t cnt = 0;
  high_resolution_clock::time_point begin;
  while (true) {
    auto b = ib_.TryReceive(qp_);
    if (b != nullptr) {
      #ifdef ENABLE_MEASUREMENT
        static bool enable = false;
        if (enable) {
          recv_measurement.end();
        }
        enable = true;
      #endif
      // Dispatch to worker's queue, then worker get work from it's queue.
      // The buffer `b` will be freed by the worker. Thus, the buffer pool
      // must made thread safe.
      Dispatch(b);
    }
    if ((b = ib_.TrySend(qp_)) != nullptr) {
      #ifdef ENABLE_MEASUREMENT
        send_measurement.end();
        recv_measurement.begin();
      #endif
      send_bufs_.Free(b);
    }

    // The receive queue is not full, if there are free buffers.
    if ((b = recv_bufs_.Alloc()) != nullptr) {
      assert(b);
      // Try posting receive if we could
      ib_.PostReceive(qp_, b);
    }
  }
}

void Server::Dispatch(Work* work) {
  auto r = work->MakeRequest();
  auto id = index_manager_.GetTabletId(r->key_hash);
  workers_[id % kNumTabletAndBackupsPerServer]->Enqueue(work);
  ++num_recv_;
}

void Server::Worker::Serve() {
  ModificationList modifications;
  while (true) {
    // Get request
    auto work = wq_.TryPollWork();
    Infiniband::Buffer* sb = nullptr;
    if (work == nullptr) {
      std::unique_lock<std::mutex> lock(mtx_);
      cond_var_.wait(lock, [&work, &sb, this]() -> bool {
          return (work || (work = wq_.Dequeue()));
        });
    }
    #ifdef ENABLE_MEASUREMENT
      server_->thread_measurement.end();
    #endif
    sb = server_->send_bufs_.Alloc();
    assert(work != nullptr && sb != nullptr);

    // Do the work
    auto r = work->MakeRequest();
    auto resp = Response::New(sb, r->type, Status::OK);
    modifications.clear();
    
    #ifdef ENABLE_MEASUREMENT
      server_->alloc_measurement.begin();
    #endif
    switch (r->type) {
    case Request::Type::PUT:
      resp->status = tablet_->Put(r, modifications);
      break;
    case Request::Type::ADD:
      resp->status = tablet_->Add(r, modifications);
      break;
    case Request::Type::DEL:
      resp->status = tablet_->Del(r, modifications);
      break;
    case Request::Type::GET:
      resp->status = tablet_->Get(resp, r, modifications);
      break;
    }
    #ifdef ENABLE_MEASUREMENT
      server_->alloc_measurement.end();
    #endif

    try {
      #ifdef ENABLE_MEASUREMENT
        server_->sync_measurement.begin();
      #endif
      tablet_->Sync(modifications);
      #ifdef ENABLE_MEASUREMENT
        server_->sync_measurement.end();
      #endif
    } catch (TransportException& e) {
      r->Print();
      tablet_->info().Print();
      server_->index_manager_.PrintTablets();
      NVDS_ERR(e.ToString().c_str());
    }

    #ifdef ENABLE_MEASUREMENT
      server_->send_measurement.begin();
    #endif
    server_->ib_.PostSend(server_->qp_, sb, resp->Len(), &work->peer_addr);
    server_->recv_bufs_.Free(work);
  }
}

} // namespace nvds
<file_sep>/src/nvds.h
#ifndef _NVDS_NVDS_H_
#define _NVDS_NVDS_H_

#include <ctype.h>
#include <errno.h>
#include <malloc.h>
#include <memory.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include <arpa/inet.h>
#include <infiniband/verbs.h>
#include <sys/socket.h>
#include <sys/time.h>
// Interprocess communication access
#include <sys/ipc.h>
// Shared memory facility
#include <sys/shm.h>

namespace nvds {

#define nvds_error(format, ...) {                         \
  fprintf(stderr, "error: %s: %d: ", __FILE__, __LINE__); \
  fprintf(stderr, format, ## __VA_ARGS__);                \
  fprintf(stderr, "\n");                                  \
  exit(-1);                                               \
}

#define nvds_expect(cond, err, ...) { \
  if (!(cond)) {                      \
    nvds_error(err, ## __VA_ARGS__);  \
  }                                   \
}

} // namespace nvds

#endif // _NVDS_NVDS_H_
<file_sep>/src/test_infiniband.cc
#include "infiniband.h"
#include <gtest/gtest.h>

#include <cstdlib>
#include <functional>
#include <iostream>
#include <thread>
#include <vector>

#include <ctype.h>
#include <errno.h>
#include <malloc.h>
#include <memory.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include <arpa/inet.h>
#include <infiniband/verbs.h>
#include <sys/socket.h>
#include <sys/time.h>
// Interprocess communication access
#include <sys/ipc.h>
// Shared memory facility
#include <sys/shm.h>

using namespace std;
using namespace nvds;

#define nvds_error(format, ...) {                         \
  fprintf(stderr, "error: %s: %d: ", __FILE__, __LINE__); \
  fprintf(stderr, format, ## __VA_ARGS__);                \
  fprintf(stderr, "\n");                                  \
  exit(-1);                                               \
}

#define nvds_expect(cond, err, ...) { \
  if (!(cond)) {                      \
    nvds_error(err, ## __VA_ARGS__);  \
  }                                   \
}

#define BUF_SIZE    (1024)
#define QP_TYPE     IBV_QPT_RC
#define IB_PORT     (1)
#define DEFAULT_PSN (33)

typedef struct nvds_context {
  struct ibv_context*       context;
	struct ibv_pd*            pd;
	struct ibv_mr*            mr;
	struct ibv_cq*            rcq;
	struct ibv_cq*            scq;
	struct ibv_qp*            qp;
	struct ibv_comp_channel*  ch;
	char*                     buf;
} nvds_context_t;

typedef struct nvds_ib_connection {
  int             	  lid;
  int            	    qpn;
  int             	  psn;
	unsigned 			      rkey;
	unsigned long long 	vaddr;
} nvds_ib_connection_t;

#define RDMA_WRITE_ID   (3)
#define RDMA_READ_ID    (4)
#define RDMA_WRITE_LEN  (64)
#define RDMA_READ_LEN   RDMA_WRITE_LEN

static int page_size;
nvds_ib_connection_t	conn[2];

static void nvds_run_server(nvds_context_t* ctx, nvds_ib_connection_t* remote_conn);
static void nvds_run_client(nvds_context_t* ctx, nvds_ib_connection_t* remote_conn);
static void nvds_init_local_ib_connection(nvds_context_t* ctx,
                                          nvds_ib_connection_t* local_conn);
static void nvds_init_ctx(nvds_context_t* ctx);
static void nvds_set_qp_state_init(struct ibv_qp* qp);
static void nvds_set_qp_state_rtr(struct ibv_qp* qp, nvds_ib_connection_t* remote_conn);
static void nvds_set_qp_state_rts(struct ibv_qp* qp, nvds_ib_connection_t* remote_conn);

static void nvds_rdma_write(nvds_context_t* ctx, nvds_ib_connection_t* remote_conn);
static void nvds_poll_send(nvds_context_t* ctx);

static void copy_info_to_conn(nvds_ib_connection_t* conn, Infiniband::QueuePairInfo* info) {
  conn->lid = info->lid;
  conn->qpn = info->qpn;
  conn->psn = info->psn;
  conn->rkey = info->rkey;
  conn->vaddr = info->vaddr;
}

static void copy_conn_to_info(Infiniband::QueuePairInfo* info, nvds_ib_connection_t* conn) {
  info->lid = conn->lid;
  info->qpn = conn->qpn;
  info->psn = conn->psn;
  info->rkey = conn->rkey;
  info->vaddr = conn->vaddr;
}

static void nvds_print_ib_connection(nvds_ib_connection_t* c) {
  printf("lid: %d\n", c->lid);
  printf("qpn: %d\n", c->qpn);
  printf("psn: %d\n", c->psn);
  printf("rkey: %u\n", c->rkey);
  printf("vaddr: %llu\n", c->vaddr);
}

static void nvds_set_qp_state_init(struct ibv_qp* qp) {
  struct ibv_qp_attr attr;
  attr.qp_state = IBV_QPS_INIT;
  attr.pkey_index = 0;
  attr.port_num = Infiniband::kPort;
  attr.qp_access_flags = IBV_ACCESS_LOCAL_WRITE |
                         IBV_ACCESS_REMOTE_READ |
                         IBV_ACCESS_REMOTE_WRITE;

  int modify_flags = IBV_QP_STATE | IBV_QP_PKEY_INDEX |
                     IBV_QP_PORT | IBV_QP_ACCESS_FLAGS;
  nvds_expect(ibv_modify_qp(qp, &attr, modify_flags) == 0,
              "ibv_modify_qp() failed");
}

static void nvds_set_qp_state_rtr(struct ibv_qp* qp, nvds_ib_connection_t* remote_conn) {
  struct ibv_qp_attr attr;
  attr.qp_state           = IBV_QPS_RTR;
  attr.path_mtu           = IBV_MTU_4096;
  attr.dest_qp_num        = remote_conn->qpn;
  attr.rq_psn             = remote_conn->psn;
  attr.max_dest_rd_atomic = 1;
  attr.min_rnr_timer      = 12;
  attr.ah_attr.is_global  = 0;
  attr.ah_attr.dlid       = remote_conn->lid;
  attr.ah_attr.sl             = 0;//1;
  attr.ah_attr.src_path_bits  = 0;
  attr.ah_attr.port_num       = 1;

  int modify_flags = IBV_QP_STATE               |
                     IBV_QP_AV                  |
                     IBV_QP_PATH_MTU            |
                     IBV_QP_DEST_QPN            |
                     IBV_QP_RQ_PSN              |
                     IBV_QP_MAX_DEST_RD_ATOMIC  |
                     IBV_QP_MIN_RNR_TIMER;
  nvds_expect(ibv_modify_qp(qp, &attr, modify_flags) == 0,
              "ibv_modify_qp() failed");
}

static void nvds_set_qp_state_rts(struct ibv_qp* qp, nvds_ib_connection_t* remote_conn) {
	// first the qp state has to be changed to rtr
	nvds_set_qp_state_rtr(qp, remote_conn);

	struct ibv_qp_attr attr;
  attr.qp_state       = IBV_QPS_RTS;
  attr.timeout        = 14;
  attr.retry_cnt      = 7;
  attr.rnr_retry      = 7;
  attr.sq_psn         = DEFAULT_PSN;
  attr.max_rd_atomic  = 1;

  int modify_flags = IBV_QP_STATE            |
                     IBV_QP_TIMEOUT          |
                     IBV_QP_RETRY_CNT        |
                     IBV_QP_RNR_RETRY        |
                     IBV_QP_SQ_PSN           |
                     IBV_QP_MAX_QP_RD_ATOMIC;
  nvds_expect(ibv_modify_qp(qp, &attr, modify_flags) == 0,
              "ibv_modify_qp() failed");
}

static void nvds_rdma_write(nvds_context_t* ctx, nvds_ib_connection_t* remote_conn) {
  struct ibv_sge sge;
  sge.addr    = (uintptr_t)ctx->buf;
  sge.length  = RDMA_WRITE_LEN;
  sge.lkey    = ctx->mr->lkey;

  struct ibv_send_wr wr;
  // The address write to
  wr.wr.rdma.remote_addr = remote_conn->vaddr;
  wr.wr.rdma.rkey        = remote_conn->rkey;
  wr.wr_id               = RDMA_WRITE_ID;
  wr.sg_list             = &sge;
  wr.num_sge             = 1;
  wr.opcode              = IBV_WR_RDMA_WRITE;
  wr.send_flags          = IBV_SEND_SIGNALED;
  wr.next                = NULL;

  struct ibv_send_wr* bad_wr;
  nvds_expect(ibv_post_send(ctx->qp, &wr, &bad_wr) == 0,
              "ibv_post_send() failed");
}

static void nvds_poll_send(nvds_context_t* ctx) {
  struct ibv_wc wc;
  int r;
  while ((r = ibv_poll_cq(ctx->scq, 1, &wc)) != 1) {}
  //nvds_expect(wc.status == IBV_WC_SUCCESS, "rdma write failed");
  if (wc.status != IBV_WC_SUCCESS) {
    printf("rdma write failed, status = %d\n", wc.status);
    perror("ibv_poll_cq");
    abort();
  }
}

static void nvds_init_local_ib_connection(nvds_context_t* ctx,
                                          nvds_ib_connection_t* local_conn) {
  struct ibv_port_attr attr;
  nvds_expect(ibv_query_port(ctx->context, IB_PORT, &attr) == 0,
              "ibv_query_port() failed");
  local_conn->lid = attr.lid;
  local_conn->qpn = ctx->qp->qp_num;
  local_conn->psn = Infiniband::QueuePair::kDefaultPsn; //lrand48() & 0xffffff;
  local_conn->rkey = ctx->mr->rkey;
  local_conn->vaddr = (uintptr_t)ctx->buf;
}

static void nvds_init_ctx(nvds_context_t* ctx) {
  memset(ctx, 0, sizeof(nvds_context_t));

  // Get IB device
  struct ibv_device** dev_list = ibv_get_device_list(NULL);
  nvds_expect(dev_list, "ibv_get_device_list() failed");

  // Open IB device
  ctx->context = ibv_open_device(dev_list[0]);
  nvds_expect(ctx->context, "ibv_open_device() failed");

  // Alloc protection domain
  ctx->pd = ibv_alloc_pd(ctx->context);
  nvds_expect(ctx->pd, "ibv_alloc_pd() failed");

  // Alloc buffer for write
  ctx->buf = (char*)memalign(4096, BUF_SIZE);
  memset(ctx->buf, 0, BUF_SIZE);

  // Register memory region
  int access = IBV_ACCESS_LOCAL_WRITE |
               IBV_ACCESS_REMOTE_READ |
				       IBV_ACCESS_REMOTE_WRITE;
  ctx->mr = ibv_reg_mr(ctx->pd, ctx->buf, BUF_SIZE, access);
  nvds_expect(ctx->mr, "ibv_reg_mr() failed");

  // Create completion channel
  //ctx->ch = ibv_create_comp_channel(ctx->context);
  //nvds_expect(ctx->ch, "ibv_create_comp_channel() failed");

  // Create complete queue
  ctx->rcq = ibv_create_cq(ctx->context, 128, NULL, NULL, 0);
  nvds_expect(ctx->rcq, "ibv_create_cq() failed");
  //ctx->scq = ibv_create_cq(ctx->context, ctx->tx_depth, ctx, ctx->ch, 0);
  ctx->scq = ibv_create_cq(ctx->context, 128, NULL, NULL, 0);
  nvds_expect(ctx->rcq, "ibv_create_cq() failed");  

  // Create & init queue apir
  struct ibv_qp_init_attr qp_init_attr;
  qp_init_attr.send_cq = ctx->scq;
  qp_init_attr.recv_cq = ctx->rcq;
  qp_init_attr.qp_type = QP_TYPE;
  qp_init_attr.cap.max_send_wr = 128;
  qp_init_attr.cap.max_recv_wr = 128;
  qp_init_attr.cap.max_send_sge = Infiniband::kMaxSendSge;
  qp_init_attr.cap.max_recv_sge = Infiniband::kMaxRecvSge;
  qp_init_attr.cap.max_inline_data = Infiniband::kMaxInlineData;

  ctx->qp = ibv_create_qp(ctx->pd, &qp_init_attr);
  nvds_expect(ctx->qp, "ibv_create_qp() failed");
  nvds_set_qp_state_init(ctx->qp);
}

static void nvds_run_server(nvds_context_t* ctx, nvds_ib_connection_t* remote_conn) {
  //nvds_server_exch_info(data);
  sleep(1); // Waitting for peer

  // Set queue pair state to RTR(Ready to receive)
  nvds_set_qp_state_rtr(ctx->qp, remote_conn);
  printf("server side: \n");
  //printf("\n");
  //nvds_print_qp_attr(ctx->qp);
  //printf("\n\n");

  /*
  printf("local connection: \n");
  nvds_print_ib_connection(&data->local_conn);
  printf("remote connection: \n");  
  nvds_print_ib_connection(&data->remote_conn);
  */
  // TODO(wgtdkp): poll request from client or do nothing
  while (1) {}
}

static void nvds_run_client(nvds_context_t* ctx, nvds_ib_connection_t* remote_conn) {
  sleep(2); // Waitting for peer

  nvds_set_qp_state_rts(ctx->qp, remote_conn);
  //nvds_set_qp_state_rtr(ctx->qp, data);
  printf("client side: \n");
  //printf("\n");
  //nvds_print_qp_attr(ctx->qp);
  //printf("\n\n");

  /*
  printf("local connection: \n");
  nvds_print_ib_connection(&data->local_conn);
  printf("remote connection: \n");  
  nvds_print_ib_connection(&data->remote_conn);
  */
  snprintf(ctx->buf, RDMA_WRITE_LEN, "hello rdma\n");

  static const int n = 1000 * 1000;  
  clock_t begin;
  double t;
  
  // Test write latency
  begin = clock();
  for (int i = 0; i < n; ++i) {
    // Step 1: RDMA write to server
    nvds_rdma_write(ctx, remote_conn);
    // Step 2: polling if RDMA write done
    nvds_poll_send(ctx);
  }
  t = (clock() - begin) * 1.0 / CLOCKS_PER_SEC;
  printf("time: %fs\n", t);
  printf("write latency: %fus\n", t / n * 1000 * 1000);

  abort();
  // Dump statistic info
}

Infiniband::QueuePairInfo qpis[2];

int nvds_main(int argc, const char* argv[]) {
  nvds_context_t ctx;

  nvds_expect(argc <= 2, "too many arguments");
  int is_client = 0;
  if (argc == 2) {
    is_client = 1;
  }

  page_size = sysconf(_SC_PAGESIZE);
  nvds_init_ctx(&ctx);
  if (is_client) {
    nvds_init_local_ib_connection(&ctx, &conn[0]);
    copy_conn_to_info(&qpis[0], &conn[0]);
    nvds_run_client(&ctx, &conn[1]);
  } else {
    nvds_init_local_ib_connection(&ctx, &conn[1]);
    copy_conn_to_info(&qpis[1], &conn[1]);
    nvds_run_server(&ctx, &conn[0]);
  }

  return 0;
}


static void RunClient() {
  try {
    Infiniband ib {1024};
    Infiniband::QueuePair qp(ib, IBV_QPT_RC, 128, 128);

    qpis[0] = {
      ib.GetLid(Infiniband::kPort),
      qp.GetLocalQPNum(),
      qp.psn,
      ib.mr()->rkey,
      reinterpret_cast<uint64_t>(ib.raw_mem())
    };
    copy_info_to_conn(&conn[0], &qpis[0]);
    
    sleep(2); // Waitting for peer
    printf("client side\n");
    qp.SetStateRTS(qpis[1]);
    assert(qp.GetState() == IBV_QPS_RTS);
    
    snprintf(ib.raw_mem(), RDMA_WRITE_LEN, "hello rdma\n");
    
    for (size_t i = 0; i < 10000; ++i) {
      ib.PostWriteAndWait(qp, qpis[1]);
    }

    std::cout << "finally, you bitch!" << std::endl;
  } catch (TransportException& e) {
    NVDS_ERR(e.ToString().c_str());
  }
}

static void RunServer() {
  try {
    Infiniband ib {1024};
    Infiniband::QueuePair qp(ib, IBV_QPT_RC, 128, 128);

    qpis[1] = {
      ib.GetLid(Infiniband::kPort),
      qp.GetLocalQPNum(),
      qp.psn,
      ib.mr()->rkey,
      reinterpret_cast<uint64_t>(ib.raw_mem())
    };
    copy_info_to_conn(&conn[1], &qpis[1]);
    
    sleep(1); // Waitting for peer
    printf("server side\n");
    qp.SetStateRTS(qpis[0]);

    assert(qp.GetState() == IBV_QPS_RTS);

    while (1) {}
  } catch (TransportException& e) {
    NVDS_ERR(e.ToString().c_str());
  }
}

///*
TEST (InfinibandTest, RDMAWrite) {
  const char* argv[] {"a.out", "192.168.99.14"};
  //thread server(bind(nvds_main, 1, argv));
  thread server(RunServer);
  //thread client(bind(nvds_main, 2, argv));
  thread client(RunClient);
  
  client.join();
  server.join();
}
//*/

/*
TEST (InfinibandTest, RDMAWrite) {
  thread server(RunServer);
  thread client(RunClient);
  client.join();
  server.join(); 
}
*/

/*
TEST (InfinibandTest, RDMAWrite) {
  try {
    Infiniband ib;
    char bufs[2][1024] {{0}};
    ibv_mr* mr = ibv_reg_mr(ib.pd().pd(), bufs, sizeof(bufs),
        IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
    assert(mr != nullptr);

    auto scq0 = ib.CreateCQ(128);
    auto rcq0 = ib.CreateCQ(128);
    Infiniband::QueuePair qp0(ib, IBV_QPT_RC, 1,
        nullptr, scq0, rcq0, 128, 128);
    auto scq1 = ib.CreateCQ(128);
    auto rcq1 = ib.CreateCQ(128);
    Infiniband::QueuePair qp1(ib, IBV_QPT_RC, 1,
        nullptr, scq1, rcq1, 128, 128);

    Infiniband::QueuePairInfo qpi0 {
      ib.GetLid(1),
      qp0.GetLocalQPNum(),
      33,
      mr->rkey,
      reinterpret_cast<uint64_t>(bufs[0])
    };
    Infiniband::QueuePairInfo qpi1 {
      ib.GetLid(1),
      qp1.GetLocalQPNum(),
      33,
      mr->rkey,
      reinterpret_cast<uint64_t>(bufs[1])
    };
    
    qp0.Plumb(IBV_QPS_RTS, qpi1);
    qp1.Plumb(IBV_QPS_RTR, qpi0);

    struct ibv_sge sge {qpi0.vaddr, 64, mr->lkey};
    struct ibv_send_wr swr;
    swr.wr.rdma.remote_addr = qpi1.vaddr;
    swr.wr.rdma.rkey        = qpi1.rkey;
    swr.wr_id               = 1;
    swr.sg_list             = &sge;
    swr.num_sge             = 1;
    swr.opcode              = IBV_WR_RDMA_WRITE;
    swr.send_flags          = IBV_SEND_SIGNALED;
    swr.next                = nullptr;

    struct ibv_send_wr* bad_wr;
    int err = ibv_post_send(qp0.qp, &swr, &bad_wr);
    assert(err == 0);

    ibv_wc wc;
    int r;
    while ((r = ibv_poll_cq(qp0.scq, 1, &wc)) != 1) {}
    if (wc.status != IBV_WC_SUCCESS) {
      throw TransportException(HERE, wc.status);
    } else {
      std::clog << "success" << std::endl << std::flush;
    }
  } catch (TransportException& e) {
    NVDS_ERR(e.ToString().c_str());
  }
}
*/

int main(int argc, char* argv[]) {
  testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}
<file_sep>/src/basic_server.h
#ifndef _NVDS_BASIC_SERVER_H_
#define _NVDS_BASIC_SERVER_H_

#include "session.h"

#include <boost/asio.hpp>

namespace nvds {

class BasicServer {
 public:
  BasicServer(uint16_t port);
  virtual ~BasicServer() {}

  virtual void Run() = 0;
  void Accept(MessageHandler recv_msg_handler,
              MessageHandler send_msg_handler);

 protected:
  void RunService() { tcp_service_.run(); }
  boost::asio::io_service tcp_service_;
  boost::asio::ip::tcp::acceptor tcp_acceptor_;
  boost::asio::ip::tcp::socket conn_sock_;
};

} // namespace nvds

#endif // _NVDS_BASIC_SERVER_H_
<file_sep>/src/test_allocator.cc
#include "allocator.h"
#include "measurement.h"

#include <gtest/gtest.h>

#include <cstdlib>
#include <iostream>
#include <vector>

using namespace std;
using namespace nvds;

TEST (AllocatorTest, Alloc) {
  auto base = malloc(Allocator::kSize);
  assert(base != nullptr);

  Allocator a(base);
  Allocator::ModificationList modifications;
  modifications.reserve(20);
  a.set_modifications(&modifications);

  std::vector<uint32_t> collector;

  Measurement alloc_measurement;
  alloc_measurement.begin();
  for (size_t i = 0; i < 1000 * 1000; ++i) {
    modifications.clear();
    auto ptr = a.Alloc(40);
    collector.push_back(ptr);
  }
  alloc_measurement.end();
  alloc_measurement.Print();
  free(base);
}

/*
TEST (AllocatorTest, Init) {
  auto mem = malloc(Allocator::kSize);

  uintptr_t base = reinterpret_cast<uintptr_t>(mem);
  Allocator a(base);
  free(mem);
}

TEST (AllocatorTest, Format) {
  auto mem = malloc(Allocator::kSize);

  uintptr_t base = reinterpret_cast<uintptr_t>(mem);
  Allocator a(base);

  free(mem);
}

TEST (AllocatorTest, Alloc) {
  auto mem = malloc(Allocator::kSize);

  uintptr_t base = reinterpret_cast<uintptr_t>(mem);
  Allocator a(base);

  uintptr_t blk;
  vector<uintptr_t> blks;
  while ((blk = a.Alloc(12)) != 0) {
    blks.push_back(blk);
  }
  ASSERT_EQ(4194285, blks.size());
  cout << "average cnt_writes: " << a.cnt_writes() * 1.0 / blks.size() << endl;
  free(mem);
}

TEST (AllocatorTest, AllocFree) {
  auto mem = malloc(Allocator::kSize);

  uintptr_t base = reinterpret_cast<uintptr_t>(mem);
  Allocator a(base);

  vector<uintptr_t> blks;
  uintptr_t blk;
  while ((blk = a.Alloc(12)) != 0) {
    blks.push_back(blk);
  }
  ASSERT_EQ(4194285, blks.size());

  while (blks.size() > 0) {
    auto blk = blks.back();
    blks.pop_back();
    a.Free(blk);
  }

  while ((blk = a.Alloc(12)) != 0) {
    blks.push_back(blk);
  }
  ASSERT_EQ(4194285, blks.size());

  free(mem);  
}

TEST (AllocatorTest, Random) {
  auto mem = malloc(Allocator::kSize);

  uintptr_t base = reinterpret_cast<uintptr_t>(mem);
  Allocator a(base);

  vector<uintptr_t> blks;
  uintptr_t blk;
  auto g = default_random_engine();
  size_t cnt = 0;
  clock_t begin = clock();
  for (size_t i = 0, j = 0, k = 0; i < 64 * 1024 * 1024; ++i) {
    int op = g() % 2;
    if (op == 0) {
      blk = a.Alloc(g() % (Allocator::kMaxBlockSize - sizeof(uint32_t)));
      if (blk == 0)
        break;
      blks.push_back(blk);
      ++k;
      ++cnt;
    } else if (j < blks.size()) {
      a.Free(blks[j++]);
      ++cnt;
    }
  }
  auto total_time = (clock() - begin) * 1.0 / CLOCKS_PER_SEC;
  cout << "cnt: " << cnt << endl;
  cout << "total time: " << total_time << endl;
  cout << "average time: " << total_time / cnt << endl;
  cout << "average cnt_writes: " << a.cnt_writes() * 1.0 / cnt << endl;

  free(mem);
}

TEST (AllocatorTest, UsageRate) {
  auto mem = malloc(Allocator::kSize);

  uintptr_t base = reinterpret_cast<uintptr_t>(mem);
  Allocator a(base);

  vector<uintptr_t> blks;
  vector<uint32_t> blk_sizes;
  uintptr_t blk;
  auto g = default_random_engine();
  uint32_t total_size = 0;
  size_t j = 0;
  while (true) {
    int op = g() % 3;
    if (op != 0) {
      uint32_t size = g() % (Allocator::kMaxBlockSize - sizeof(uint32_t));
      blk = a.Alloc(size);
      if (blk == 0)
        break;
      blks.push_back(blk);
      blk_sizes.push_back(size);
      total_size += size;
    } else if (j < blks.size()) {
      total_size -= blk_sizes[j];
      a.Free(blks[j++]);
    }
  }

  cout << "usage rate: " << total_size * 1.0 / Allocator::kSize << endl;

  free(mem);
}
*/

int main(int argc, char* argv[]) {
  testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}
<file_sep>/src/client.h
#ifndef _NVDS_CLIENT_H_
#define _NVDS_CLIENT_H_

#include "common.h"
#include "index.h"
#include "infiniband.h"
#include "request.h"
#include "response.h"
#include "session.h"

namespace nvds {

class Client {
  using Buffer = Infiniband::Buffer;
 public:
  Client(const std::string& coord_addr);
  ~Client();
  DISALLOW_COPY_AND_ASSIGN(Client);

  // Get value by the key, return empty string if error occurs.
  // Throw: TransportException
  std::string Get(const std::string& key) {
    return Get(key.c_str(), key.size());
  }
  std::string Get(const char* key, size_t key_len) {
    auto rb = RequestAndWait(key, key_len, nullptr, 0, Request::Type::GET);
    auto resp = rb->MakeResponse();
    std::string ans {resp->val, resp->val_len};
    recv_bufs_.Free(rb);
    return ans;
  }

  // Insert key/value pair to the cluster, return if operation succeed.
  // Throw: TransportException
  bool Put(const std::string& key, const std::string& val) {
    return Put(key.c_str(), key.size(), val.c_str(), val.size());
  }
  bool Put(const char* key, size_t key_len, const char* val, size_t val_len) {
    auto rb = RequestAndWait(key, key_len, val, val_len, Request::Type::PUT);
    bool ans = rb->MakeResponse()->status == Status::OK;
    recv_bufs_.Free(rb);
    return ans;
  }

  // Add key/value pair to the cluster,
  // return false if there is already the same key;
  // Throw: TransportException
  bool Add(const std::string& key, const std::string& val) {
    return Add(key.c_str(), key.size(), val.c_str(), val.size());
  }
  bool Add(const char* key, size_t key_len, const char* val, size_t val_len) {
    auto rb = RequestAndWait(key, key_len, val, val_len, Request::Type::ADD);
    // FIXME(wgtdkp): what about Status::NO_MEM?
    bool ans = rb->MakeResponse()->status == Status::OK;
    recv_bufs_.Free(rb);
    return ans;
  }

  // Delete item indexed by the key, return if operation succeed.
  // Throw: TransportException
  bool Del(const std::string& key) {
    return Del(key.c_str(), key.size());
  }
  bool Del(const char* key, size_t key_len) {
    auto rb = RequestAndWait(key, key_len, nullptr, 0, Request::Type::DEL);
    bool ans = rb->MakeResponse()->status == Status::OK;
    recv_bufs_.Free(rb);
    return ans;
  }

  // Statistic
  size_t num_send() const { return num_send_; }

 private:
  static const uint32_t kMaxIBQueueDepth = 128;
  static const uint32_t kSendBufSize = 1024 * 2 + 128;
  static const uint32_t kRecvBufSize = 1024 + 128;
  // May throw exception `boost::system::system_error`
  tcp::socket Connect(const std::string& coord_addr);
  void Close() {}
  void Join();
  Buffer* RequestAndWait(const char* key, size_t key_len,
      const char* val, size_t val_len, Request::Type type);

  boost::asio::io_service tcp_service_;
  Session session_;
  IndexManager index_manager_;

  // Infiniband
  Infiniband ib_;
  Infiniband::RegisteredBuffers send_bufs_;
  Infiniband::RegisteredBuffers recv_bufs_;
  Infiniband::QueuePair* qp_;

  // Statistic 
  size_t num_send_ {0};
};

} // namespace nvds

#endif // _NVDS_CLIENT_H_
<file_sep>/src/experiments/Makefile
CC = gcc -std=c11
CFLAGS = -Wall -g -D_GNU_SOURCE
LDFLAGS = -libverbs -lrt

all:
	@make write-rc write-rc-multi write-rc-unsignaled send-rc send-ud

write-rc: write_rc.o
	${CC} -o $@ $^ ${LDFLAGS}

write_rc.o: write_rc.c write_rc.h
	${CC} ${CFLAGS} -o $@ -c $<

write-rc-multi: write_rc_multi.o
	${CC} -o $@ $^ ${LDFLAGS}

write_rc_multi.o: write_rc_multi.c write_rc_multi.h
	${CC} ${CFLAGS} -o $@ -c $<

write-rc-unsignaled: write_rc_unsignaled.o
	${CC} -o $@ $^ ${LDFLAGS}

write_rc_unsignaled.o: write_rc_unsignaled.c write_rc_unsignaled.h
	${CC} ${CFLAGS} -o $@ -c $<

send-rc: send_rc.o
	${CC} -o $@ $^ ${LDFLAGS}

send_rc.o: send_rc.c send_rc.h
	${CC} ${CFLAGS} -o $@ -c $<

send-ud: send_ud.o
	${CC} -o $@ $^ ${LDFLAGS}

send_ud.o: send_ud.c send_ud.h
	${CC} ${CFLAGS} -o $@ -c $<

clean:
	rm -f *.o write-rc send-rc send-ud
