Initial commit

Author: mwidmoser

.clang-format

@@ -0,0 +1,17 @@
+---
+BasedOnStyle: Google
+IndentWidth: 2
+---
+Language: Cpp
+AccessModifierOffset: -2
+AlignTrailingComments: false
+AllowAllParametersOfDeclarationOnNextLine: false
+AllowShortIfStatementsOnASingleLine: false
+AllowShortLoopsOnASingleLine: false
+AlwaysBreakBeforeMultilineStrings: false
+BinPackArguments: false
+BinPackParameters: false
+ContinuationIndentWidth: 2
+DerivePointerAlignment: false
+IndentCaseLabels: false
+TabWidth: 2

.gitignore

@@ -0,0 +1,12 @@
+*.gch
+*.o
+*.out
+*.cmake
+!IBVerbsConfig.cmake
+build/
+data/
+cmake-build-debug/
+cmake-build-release/
+.DS_Store
+.cache
+.idea

CMakeCompilerConfig.txt

@@ -0,0 +1,12 @@
+set(CMAKE_CXX_COMPILER clang++)
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+add_compile_options(-Wall -Wextra -Wpedantic -Wfatal-errors)
+
+# add_compile_options(-DVERIFY)       # checks whether the insert queries did not mess up the index
+# add_compile_options(-DNOHUGEPAGES)  # disables hugepages
+# add_compile_options(-DDEV_DEBUG)    # outputs additional debugging information
+
+set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -DDEBUG -g -O0")
+set(CMAKE_CXX_FLAGS_RELEASE "${CMAKE_CXX_FLAGS_RELEASE} -O3")
+set(CMAKE_EXPORT_COMPILE_COMMANDS ON)

CMakeLists.txt

@@ -0,0 +1,27 @@
+cmake_minimum_required(VERSION 3.0)
+if (APPLE)
+    project(rdma_inverted_index)
+    include("CMakeCompilerConfig.txt")
+else ()
+    include("CMakeCompilerConfig.txt")
+    project(rdma_inverted_index)
+endif ()
+
+# set module path
+set(CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR})
+include_directories(src)
+add_subdirectory(rdma-library)
+add_subdirectory(src/data_processing)
+
+# executables
+add_executable(term_index src/index/term_based/main.cc src/timing/timing.cc)
+target_link_libraries(term_index rdma_library)
+
+add_executable(document_index src/index/document_based/main.cc src/timing/timing.cc)
+target_link_libraries(document_index rdma_library)
+
+add_executable(block_index src/index/block_based/main.cc src/timing/timing.cc)
+target_link_libraries(block_index rdma_library)
+
+add_executable(dynamic_block_index src/index/block_based_dynamic/main.cc src/timing/timing.cc)
+target_link_libraries(dynamic_block_index rdma_library)

README.md

@@ -0,0 +1,16 @@
+cmake_minimum_required(VERSION 3.0)
+
+set(IBVerbs_DIR .)
+set(THREADS_PREFER_PTHREAD_FLAG TRUE)
+
+# required packages
+find_package(IBVerbs REQUIRED)
+find_package(Boost COMPONENTS program_options REQUIRED)
+find_package(Threads REQUIRED)
+find_package(TBB REQUIRED)
+
+# our library
+file(GLOB_RECURSE LIBRARY_CC library/**.cc)
+add_library(rdma_library STATIC ${LIBRARY_CC})
+target_include_directories(rdma_library PUBLIC .)
+target_link_libraries(rdma_library ${IBVERBS_LIBRARY} ${Boost_LIBRARIES} TBB::tbb Threads::Threads)

rdma-library/CMakeLists.txt

@@ -0,0 +1,45 @@
+FIND_PATH(IBVERBS_INCLUDE_DIR infiniband/verbs.h
+        PATHS
+        $ENV{IBVERBS_HOME}
+        NO_DEFAULT_PATH
+        PATH_SUFFIXES include
+        )
+
+FIND_PATH(IBVERBS_INCLUDE_DIR infiniband/verbs.h
+        PATHS
+        /usr/local/include
+        /usr/include
+        /sw/include # Fink
+        /opt/local/include # DarwinPorts
+        /opt/csw/include # Blastwave
+        /opt/include
+        )
+
+FIND_LIBRARY(IBVERBS_LIBRARY
+        NAMES ibverbs
+        PATHS $ENV{IBVERBS_HOME}
+        NO_DEFAULT_PATH
+        PATH_SUFFIXES lib64 lib
+        )
+
+FIND_LIBRARY(IBVERBS_LIBRARY
+        NAMES ibverbs
+        PATHS
+        /usr/local
+        /usr
+        /sw
+        /opt/local
+        /opt/csw
+        /opt
+        /usr/freeware
+        PATH_SUFFIXES lib64 lib
+        )
+SET(IBVERBS_FOUND FALSE)
+IF (IBVERBS_LIBRARY AND IBVERBS_INCLUDE_DIR)
+    SET(IBVERBS_FOUND TRUE)
+ENDIF (IBVERBS_LIBRARY AND IBVERBS_INCLUDE_DIR)
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(IBVerbs DEFAULT_MSG IBVERBS_LIBRARY IBVERBS_INCLUDE_DIR)
+
+mark_as_advanced(IBVERBS_INCLUDE_DIR IBVERBS_LIBRARIES)

rdma-library/IBVerbsConfig.cmake

@@ -0,0 +1,14 @@
+# RDMA Library
+
+"High-level" library to connect machines, connect queue pairs, register memory regions, post RDMA verbs, etc.
+The goal of this library is to conveniently wrap
+the [ibverbs library](https://github.com/linux-rdma/rdma-core/tree/master/libibverbs).
+
+[TODO: public library interface and namespaces...]
+
+## Required C++ Libraries
+
+* ibverbs
+* Boost (for CLI parsing)
+* pthreads (for multithreading)
+* oneTBB (for concurrent data structures)

rdma-library/README.md

@@ -0,0 +1,73 @@
+#ifndef RDMA_LIBRARY_BATCHED_READ_HH
+#define RDMA_LIBRARY_BATCHED_READ_HH
+
+#include "queue_pair.hh"
+#include "utils.hh"
+
+struct BatchedREAD {
+  const u32 max_size;
+  u32 requests{0};
+  u64 total_size{0};
+
+  vec<ibv_send_wr> work_requests;
+  vec<ibv_sge> scatter_gather_entries;
+  ibv_send_wr* bad_work_request{nullptr};
+
+  explicit BatchedREAD(size_t max_batch_size)
+      : max_size(max_batch_size),
+        work_requests(max_batch_size),
+        scatter_gather_entries(max_batch_size) {}
+
+  void add_to_batch(u64 local_address,
+                    u64 remote_address,
+                    u32 length,
+                    u32 lkey,
+                    u32 rkey,
+                    u64 wr_id,
+                    bool signaled = true) {
+    lib_assert(length > 0, "Cannot READ 0 bytes");
+    lib_assert(requests < max_size, "Batch exceeds maximum batch size");
+
+    auto& sge = scatter_gather_entries[requests];
+    auto& wr = work_requests[requests];
+
+    sge.addr = local_address;
+    sge.length = length;
+    sge.lkey = lkey;
+
+    wr.opcode = IBV_WR_RDMA_READ;
+    wr.send_flags = signaled ? IBV_SEND_SIGNALED : 0;
+    wr.wr_id = wr_id;
+    wr.sg_list = &sge;
+    wr.num_sge = 1;
+
+    wr.next = nullptr;
+    if (requests > 0) {
+      work_requests[requests - 1].next = &wr;
+    }
+
+    wr.wr.rdma.remote_addr = remote_address;
+    wr.wr.rdma.rkey = rkey;
+
+    ++requests;
+    total_size += length;
+  }
+
+  void post_batch(QP& qp) {
+    lib_assert(requests > 0, "Empty READ batch");
+
+    // send final WR signaled in any case
+    work_requests[requests - 1].send_flags = IBV_SEND_SIGNALED;
+
+    lib_assert(
+      ibv_post_send(
+        qp->get_ibv_qp(), work_requests.data(), &bad_work_request) == 0,
+      "Cannot post send request");
+
+    // reset batch
+    requests = 0;
+    total_size = 0;
+  }
+};
+
+#endif  // RDMA_LIBRARY_BATCHED_READ_HH

rdma-library/library/batched_read.hh

@@ -0,0 +1,143 @@
+#include "configuration.hh"
+
+#include <boost/program_options.hpp>
+#include <iomanip>
+#include <iostream>
+
+namespace configuration {
+
+Configuration::Configuration() { create_rdma_options(); }
+
+Configuration::Configuration(int argc, char** argv) : Configuration() {
+  process_program_options(argc, argv);
+  operator<<(std::cerr, *this);
+}
+
+void Configuration::create_rdma_options() {
+  desc.add_options()("help,h", "Show help message")(
+    "is-server,s",
+    po::bool_switch(&is_server)->default_value(is_server),
+    "Program acts as server if set")(
+    "servers",
+    po::value<vec<str>>(&server_nodes)->multitoken(),
+    "A list of server nodes to which a client connects, e.g., \"cluster3\"")(
+    "clients",
+    po::value<vec<str>>(&client_nodes)->multitoken(),
+    "A list of client nodes to which the initiator connects, e.g., "
+    "\"cluster4 cluster5\"")(
+    "initiator,i",
+    po::bool_switch(&is_initiator)->default_value(is_initiator),
+    "Program acts as initiating client if set")(
+    "num-clients,c",
+    po::value<u32>(&num_clients)->default_value(num_clients),
+    "Number of clients that connect to each server (relevant only for "
+    "server nodes)");
+
+  // configuration options
+  desc.add_options()(
+    "port", po::value<u32>(&port)->default_value(port), "TCP port")(
+    "ib-port",
+    po::value<u32>(&device_port)->default_value(device_port),
+    "Port of infiniband device")(
+    "max-poll-cqes",
+    po::value<i32>(&max_poll_cqes)->default_value(max_poll_cqes),
+    "Number of outstanding RDMA operations allowed (hardware-specific)")(
+    "max-send-wrs",
+    po::value<i32>(&max_send_queue_wr)->default_value(max_send_queue_wr),
+    "Maximum number of outstanding send work requests")(
+    "max-receive-wrs",
+    po::value<i32>(&max_recv_queue_wr)->default_value(max_recv_queue_wr),
+    "Maximum number of outstanding receive work requests");
+}
+
+void Configuration::exit_with_help_message(char** argv) {
+  std::cerr << "Try " << argv[0] << " --help" << std::endl;
+  std::exit(EXIT_FAILURE);
+}
+
+void Configuration::process_program_options(int argc, char** argv) {
+  try {
+    po::variables_map vm;
+    po::store(po::parse_command_line(argc, argv, desc), vm);
+
+    if (vm.count("help")) {
+      std::cerr << desc << std::endl;
+      std::exit(EXIT_FAILURE);
+    }
+
+    po::notify(vm);
+
+    if (!is_server && server_nodes.empty()) {
+      std::cerr << "[ERROR]: --servers <arg-list> must be given if "
+                   "--server is not set"
+                << std::endl;
+      exit_with_help_message(argv);
+    }
+
+    if (is_server && is_initiator) {
+      std::cerr << "[ERROR]: a server cannot be the initiator" << std::endl;
+      exit_with_help_message(argv);
+    }
+
+    if (!is_initiator && !client_nodes.empty()) {
+      std::cerr << "[ERROR]: --clients <arg-list> is only required by the "
+                   "initiating client"
+                << std::endl;
+      exit_with_help_message(argv);
+    }
+
+  } catch (const std::exception& e) {
+    std::cerr << "[ERROR]: " << e.what() << std::endl;
+    exit_with_help_message(argv);
+  }
+}
+
+std::ostream& operator<<(std::ostream& os, const Configuration& config) {
+  const int32_t width = 30;
+  const int32_t max_width = width * 2;
+  const char filler = '=';
+
+  os << std::setfill(filler) << std::setw(max_width) << "" << std::endl
+     << std::setfill(' ') << std::setw(width)
+     << (config.is_server ? "SERVER" : "CLIENT") << std::endl
+     << std::setfill(filler) << std::setw(max_width) << "" << std::endl;
+
+  os << std::left << std::setfill(' ');
+  if (!config.is_server) {
+    os << std::setw(width) << "connect to: "
+       << "[";
+    for (const str& node : config.server_nodes) {
+      os << node << ", ";
+    }
+    os << "\b\b]" << std::endl;
+    os << std::boolalpha << std::setw(width)
+       << "is initiator: " << config.is_initiator << std::endl;
+    if (config.is_initiator) {
+      if (!config.client_nodes.empty()) {
+        os << std::setw(width) << "client nodes: "
+           << "[";
+        for (const str& node : config.client_nodes) {
+          os << node << ", ";
+        }
+        os << "\b\b]" << std::endl;
+      }
+    }
+  } else {
+    os << std::setw(width) << "num clients: " << config.num_clients
+       << std::endl;
+  }
+  os << std::setw(width) << "TCP port: " << config.port << std::endl
+     << std::setw(width) << "IB port: " << config.device_port << std::endl
+     << std::setw(width) << "max outstanding CQEs: " << config.max_poll_cqes
+     << std::endl
+     << std::setw(width)
+     << "max send work requests: " << config.max_send_queue_wr << std::endl
+     << std::setw(width)
+     << "max receive work requests: " << config.max_recv_queue_wr << std::endl;
+
+  os << std::setfill(filler) << std::setw(max_width) << "" << std::endl;
+
+  return os;
+}
+
+}  // namespace configuration