Release 0.6.3

- RDMA support
- Client performance optimisations (4k randread ~120k -> ~180k on 1 core)
- JSON configuration file (/etc/vitastor/vitastor.conf) support
- Bug fixes

CMakeLists.txt

@@ -2,6 +2,6 @@ cmake_minimum_required(VERSION 2.8)
 
 project(vitastor)
 
-set(VERSION "0.6.2")
+set(VERSION "0.6.3")
 
 add_subdirectory(src)

README-ru.md

@@ -49,6 +49,7 @@ Vitastor на данный момент находится в статусе п
 - Именование инодов через хранение их метаданных в etcd
 - Снапшоты и copy-on-write клоны
 - Сглаживание производительности случайной записи в SSD+HDD конфигурациях
+- Поддержка RDMA/RoCEv2 через libibverbs
 
 ## Планы развития
 
@@ -60,7 +61,7 @@ Vitastor на данный момент находится в статусе п
 - Фоновая проверка целостности без контрольных сумм (сверка реплик)
 - Контрольные суммы
 - Поддержка SSD-кэширования (tiered storage)
-- Поддержка RDMA и NVDIMM
+- Поддержка NVDIMM
 - Web-интерфейс
 - Возможно, сжатие
 - Возможно, поддержка кэширования данных через системный page cache

README.md

@@ -43,6 +43,7 @@ breaking changes in the future. However, the following is implemented:
 - Inode metadata storage in etcd
 - Snapshots and copy-on-write image clones
 - Write throttling to smooth random write workloads in SSD+HDD configurations
+- RDMA/RoCEv2 support via libibverbs
 
 ## Roadmap
 
@@ -54,7 +55,7 @@ breaking changes in the future. However, the following is implemented:
 - Scrubbing without checksums (verification of replicas)
 - Checksums
 - Tiered storage
-- RDMA and NVDIMM support
+- NVDIMM support
 - Web GUI
 - Compression (possibly)
 - Read caching using system page cache (possibly)

debian/changelog

@@ -1,8 +1,18 @@
-vitastor (0.6.2-1) unstable; urgency=medium
+vitastor (0.6.3-1) unstable; urgency=medium
 
+  * RDMA support
   * Bugfixes
 
- -- Vitaliy Filippov <vitalif@yourcmc.ru>  Tue, 02 Feb 2021 23:01:24 +0300
+ -- Vitaliy Filippov <vitalif@yourcmc.ru>  Sat, 01 May 2021 18:46:10 +0300
+
+vitastor (0.6.0-1) unstable; urgency=medium
+
+  * Snapshots and Copy-on-Write clones
+  * Image metadata in etcd (name, size)
+  * Image I/O and space statistics in etcd
+  * Write throttling for smoothing random write workloads in SSD+HDD configurations
+
+ -- Vitaliy Filippov <vitalif@yourcmc.ru>  Sun, 11 Apr 2021 00:49:18 +0300
 
 vitastor (0.5.1-1) unstable; urgency=medium
 

debian/control

@@ -2,7 +2,7 @@ Source: vitastor
 Section: admin
 Priority: optional
 Maintainer: Vitaliy Filippov <vitalif@yourcmc.ru>
-Build-Depends: debhelper, liburing-dev (>= 0.6), g++ (>= 8), libstdc++6 (>= 8), linux-libc-dev, libgoogle-perftools-dev, libjerasure-dev, libgf-complete-dev
+Build-Depends: debhelper, liburing-dev (>= 0.6), g++ (>= 8), libstdc++6 (>= 8), linux-libc-dev, libgoogle-perftools-dev, libjerasure-dev, libgf-complete-dev, libibverbs-dev
 Standards-Version: 4.5.0
 Homepage: https://vitastor.io/
 Rules-Requires-Root: no

debian/vitastor.Dockerfile

@@ -22,7 +22,7 @@ RUN apt-get -y build-dep qemu
 RUN apt-get -y build-dep fio
 RUN apt-get --download-only source qemu
 RUN apt-get --download-only source fio
-RUN apt-get -y install libjerasure-dev cmake
+RUN apt-get update && apt-get -y install libjerasure-dev cmake libibverbs-dev
 
 ADD . /root/vitastor
 RUN set -e -x; \
@@ -40,10 +40,10 @@ RUN set -e -x; \
     mkdir -p /root/packages/vitastor-$REL; \
     rm -rf /root/packages/vitastor-$REL/*; \
     cd /root/packages/vitastor-$REL; \
-    cp -r /root/vitastor vitastor-0.6.2; \
+    cp -r /root/vitastor vitastor-0.6.3; \
-    ln -s /root/packages/qemu-$REL/qemu-*/ vitastor-0.6.2/qemu; \
+    ln -s /root/packages/qemu-$REL/qemu-*/ vitastor-0.6.3/qemu; \
-    ln -s /root/fio-build/fio-*/ vitastor-0.6.2/fio; \
+    ln -s /root/fio-build/fio-*/ vitastor-0.6.3/fio; \
-    cd vitastor-0.6.2; \
+    cd vitastor-0.6.3; \
     FIO=$(head -n1 fio/debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
     QEMU=$(head -n1 qemu/debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
     sh copy-qemu-includes.sh; \
@@ -59,8 +59,8 @@ RUN set -e -x; \
     echo "dep:fio=$FIO" > debian/substvars; \
     echo "dep:qemu=$QEMU" >> debian/substvars; \
     cd /root/packages/vitastor-$REL; \
-    tar --sort=name --mtime='2020-01-01' --owner=0 --group=0 --exclude=debian -cJf vitastor_0.6.2.orig.tar.xz vitastor-0.6.2; \
+    tar --sort=name --mtime='2020-01-01' --owner=0 --group=0 --exclude=debian -cJf vitastor_0.6.3.orig.tar.xz vitastor-0.6.3; \
-    cd vitastor-0.6.2; \
+    cd vitastor-0.6.3; \
     V=$(head -n1 debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
     DEBFULLNAME="Vitaliy Filippov <vitalif@yourcmc.ru>" dch -D $REL -v "$V""$REL" "Rebuild for $REL"; \
     DEB_BUILD_OPTIONS=nocheck dpkg-buildpackage --jobs=auto -sa; \

mon/mon.js

@@ -41,6 +41,12 @@ const etcd_allow = new RegExp('^'+[
 const etcd_tree = {
     config: {
         /* global: {
+            // WARNING: NOT ALL OF THESE ARE ACTUALLY CONFIGURABLE HERE
+            // THIS IS JUST A POOR'S MAN CONFIG DOCUMENTATION
+            // etcd connection
+            config_path: "/etc/vitastor/vitastor.conf",
+            etcd_address: "10.0.115.10:2379/v3",
+            etcd_prefix: "/vitastor",
             // mon
             etcd_mon_ttl: 30, // min: 10
             etcd_mon_timeout: 1000, // ms. min: 0
@@ -50,7 +56,17 @@ const etcd_tree = {
             osd_out_time: 600, // seconds. min: 0
             placement_levels: { datacenter: 1, rack: 2, host: 3, osd: 4, ... },
             // client and osd
+            tcp_header_buffer_size: 65536,
             use_sync_send_recv: false,
+            use_rdma: true,
+            rdma_device: null, // for example, "rocep5s0f0"
+            rdma_port_num: 1,
+            rdma_gid_index: 0,
+            rdma_mtu: 4096,
+            rdma_max_sge: 128,
+            rdma_max_send: 32,
+            rdma_max_recv: 8,
+            rdma_max_msg: 1048576,
             log_level: 0,
             block_size: 131072,
             disk_alignment: 4096,

rpm/build-tarball.sh

@@ -48,4 +48,4 @@ FIO=`rpm -qi fio | perl -e 'while(<>) { /^Epoch[\s:]+(\S+)/ && print "$1:"; /^Ve
 QEMU=`rpm -qi qemu qemu-kvm | perl -e 'while(<>) { /^Epoch[\s:]+(\S+)/ && print "$1:"; /^Version[\s:]+(\S+)/ && print $1; /^Release[\s:]+(\S+)/ && print "-$1"; }'`
 perl -i -pe 's/(Requires:\s*fio)([^\n]+)?/$1 = '$FIO'/' $VITASTOR/rpm/vitastor-el$EL.spec
 perl -i -pe 's/(Requires:\s*qemu(?:-kvm)?)([^\n]+)?/$1 = '$QEMU'/' $VITASTOR/rpm/vitastor-el$EL.spec
-tar --transform 's#^#vitastor-0.6.2/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.6.2$(rpm --eval '%dist').tar.gz *
+tar --transform 's#^#vitastor-0.6.3/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.6.3$(rpm --eval '%dist').tar.gz *

rpm/vitastor-el7.Dockerfile

@@ -17,6 +17,7 @@ RUN rpm --nomd5 -i fio*.src.rpm
 RUN rm -f /etc/yum.repos.d/CentOS-Media.repo
 RUN cd ~/rpmbuild/SPECS && yum-builddep -y --enablerepo='*' --disablerepo=centos-sclo-rh --disablerepo=centos-sclo-rh-source --disablerepo=centos-sclo-sclo-testing qemu-kvm.spec
 RUN cd ~/rpmbuild/SPECS && yum-builddep -y --enablerepo='*' --disablerepo=centos-sclo-rh --disablerepo=centos-sclo-rh-source --disablerepo=centos-sclo-sclo-testing fio.spec
+RUN yum -y install rdma-core-devel
 
 ADD https://vitastor.io/rpms/liburing-el7/liburing-0.7-2.el7.src.rpm /root
 
@@ -37,7 +38,7 @@ ADD . /root/vitastor
 RUN set -e; \
     cd /root/vitastor/rpm; \
     sh build-tarball.sh; \
-    cp /root/vitastor-0.6.2.el7.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.6.3.el7.tar.gz ~/rpmbuild/SOURCES; \
     cp vitastor-el7.spec ~/rpmbuild/SPECS/vitastor.spec; \
     cd ~/rpmbuild/SPECS/; \
     rpmbuild -ba vitastor.spec; \

rpm/vitastor-el7.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        0.6.2
+Version:        0.6.3
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-0.6.2.el7.tar.gz
+Source0:        vitastor-0.6.3.el7.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel
@@ -14,6 +14,7 @@ BuildRequires:  rh-nodejs12
 BuildRequires:  rh-nodejs12-npm
 BuildRequires:  jerasure-devel
 BuildRequires:  gf-complete-devel
+BuildRequires:  libibverbs-devel
 BuildRequires:  cmake
 Requires:       fio = 3.7-1.el7
 Requires:       qemu-kvm = 2.0.0-1.el7.6
@@ -61,8 +62,8 @@ cp -r mon %buildroot/usr/lib/vitastor/mon
 %_libdir/libfio_vitastor.so
 %_libdir/libfio_vitastor_blk.so
 %_libdir/libfio_vitastor_sec.so
-%_libdir/libvitastor_blk.so
+%_libdir/libvitastor_blk.so*
-%_libdir/libvitastor_client.so
+%_libdir/libvitastor_client.so*
 /usr/lib/vitastor
 
 

rpm/vitastor-el8.Dockerfile

@@ -15,6 +15,7 @@ RUN rpm --nomd5 -i qemu*.src.rpm
 RUN rpm --nomd5 -i fio*.src.rpm
 RUN cd ~/rpmbuild/SPECS && dnf builddep -y --enablerepo=powertools --spec qemu-kvm.spec
 RUN cd ~/rpmbuild/SPECS && dnf builddep -y --enablerepo=powertools --spec fio.spec && dnf install -y cmake
+RUN yum -y install libibverbs-devel
 
 ADD https://vitastor.io/rpms/liburing-el7/liburing-0.7-2.el7.src.rpm /root
 
@@ -35,7 +36,7 @@ ADD . /root/vitastor
 RUN set -e; \
     cd /root/vitastor/rpm; \
     sh build-tarball.sh; \
-    cp /root/vitastor-0.6.2.el8.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.6.3.el8.tar.gz ~/rpmbuild/SOURCES; \
     cp vitastor-el8.spec ~/rpmbuild/SPECS/vitastor.spec; \
     cd ~/rpmbuild/SPECS/; \
     rpmbuild -ba vitastor.spec; \

rpm/vitastor-el8.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        0.6.2
+Version:        0.6.3
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-0.6.2.el8.tar.gz
+Source0:        vitastor-0.6.3.el8.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel
@@ -13,6 +13,7 @@ BuildRequires:  gcc-toolset-9-gcc-c++
 BuildRequires:  nodejs >= 10
 BuildRequires:  jerasure-devel
 BuildRequires:  gf-complete-devel
+BuildRequires:  libibverbs-devel
 BuildRequires:  cmake
 Requires:       fio = 3.7-3.el8
 Requires:       qemu-kvm = 4.2.0-29.el8.6
@@ -58,8 +59,8 @@ cp -r mon %buildroot/usr/lib/vitastor
 %_libdir/libfio_vitastor.so
 %_libdir/libfio_vitastor_blk.so
 %_libdir/libfio_vitastor_sec.so
-%_libdir/libvitastor_blk.so
+%_libdir/libvitastor_blk.so*
-%_libdir/libvitastor_client.so
+%_libdir/libvitastor_client.so*
 /usr/lib/vitastor
 
 

src/CMakeLists.txt

@@ -13,7 +13,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 	set(CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/${CMAKE_INSTALL_LIBDIR}")
 endif()
 
-add_definitions(-DVERSION="0.6.2")
+add_definitions(-DVERSION="0.6.3")
 add_definitions(-Wall -Wno-sign-compare -Wno-comment -Wno-parentheses -Wno-pointer-arith -I ${CMAKE_SOURCE_DIR}/src)
 if (${WITH_ASAN})
 	add_definitions(-fsanitize=address -fno-omit-frame-pointer)
@@ -72,14 +72,15 @@ target_link_libraries(fio_vitastor_blk
 )
 
 # libvitastor_common.a
+set(MSGR_RDMA "")
+if (IBVERBS_LIBRARIES)
+	set(MSGR_RDMA "msgr_rdma.cpp")
+endif (IBVERBS_LIBRARIES)
 add_library(vitastor_common STATIC
 	epoll_manager.cpp etcd_state_client.cpp
 	messenger.cpp msgr_stop.cpp msgr_op.cpp msgr_send.cpp msgr_receive.cpp ringloop.cpp ../json11/json11.cpp
-	http_client.cpp osd_ops.cpp pg_states.cpp timerfd_manager.cpp base64.cpp
+	http_client.cpp osd_ops.cpp pg_states.cpp timerfd_manager.cpp base64.cpp ${MSGR_RDMA}
 )
-if (IBVERBS_LIBRARIES)
-	target_sources(vitastor_common PRIVATE msgr_rdma.cpp)
-endif (IBVERBS_LIBRARIES)
 target_compile_options(vitastor_common PUBLIC -fPIC)
 
 # vitastor-osd

src/blockstore.cpp

@@ -43,11 +43,6 @@ int blockstore_t::read_bitmap(object_id oid, uint64_t target_version, void *bitm
     return impl->read_bitmap(oid, target_version, bitmap, result_version);
 }
 
-std::unordered_map<object_id, uint64_t> & blockstore_t::get_unstable_writes()
-{
-    return impl->unstable_writes;
-}
-
 std::map<uint64_t, uint64_t> & blockstore_t::get_inode_space_stats()
 {
     return impl->inode_space_stats;

src/blockstore.h

@@ -183,9 +183,6 @@ public:
     // Simplified synchronous operation: get object bitmap & current version
     int read_bitmap(object_id oid, uint64_t target_version, void *bitmap, uint64_t *result_version = NULL);
 
-    // Unstable writes are added here (map of object_id -> version)
-    std::unordered_map<object_id, uint64_t> & get_unstable_writes();
-
     // Get per-inode space usage statistics
     std::map<uint64_t, uint64_t> & get_inode_space_stats();
 

src/cluster_client.cpp

@@ -16,6 +16,8 @@
 
 cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json & config)
 {
+    config = osd_messenger_t::read_config(config);
+
     this->ringloop = ringloop;
     this->tfd = tfd;
     this->config = config;
@@ -952,13 +954,14 @@ int cluster_client_t::continue_sync(cluster_op_t *op)
         return 1;
     }
     // Check that all OSD connections are still alive
-    for (auto sync_osd: dirty_osds)
+    for (auto do_it = dirty_osds.begin(); do_it != dirty_osds.end(); )
     {
+        osd_num_t sync_osd = *do_it;
         auto peer_it = msgr.osd_peer_fds.find(sync_osd);
         if (peer_it == msgr.osd_peer_fds.end())
-        {
+            dirty_osds.erase(do_it++);
-            return 0;
+        else
-        }
+            do_it++;
     }
     // Post sync to affected OSDs
     for (auto & prev_op: dirty_buffers)

src/etcd_state_client.cpp

@@ -50,6 +50,11 @@ void etcd_state_client_t::etcd_txn(json11::Json txn, int timeout, std::function<
 
 void etcd_state_client_t::etcd_call(std::string api, json11::Json payload, int timeout, std::function<void(std::string, json11::Json)> callback)
 {
+    if (!etcd_addresses.size())
+    {
+        fprintf(stderr, "etcd_address is missing in Vitastor configuration\n");
+        exit(1);
+    }
     std::string etcd_address = etcd_addresses[rand() % etcd_addresses.size()];
     std::string etcd_api_path;
     int pos = etcd_address.find('/');
@@ -85,7 +90,7 @@ void etcd_state_client_t::add_etcd_url(std::string addr)
     }
 }
 
-void etcd_state_client_t::parse_config(json11::Json & config)
+void etcd_state_client_t::parse_config(const json11::Json & config)
 {
     this->etcd_addresses.clear();
     if (config["etcd_address"].is_string())
@@ -122,6 +127,11 @@ void etcd_state_client_t::parse_config(json11::Json & config)
 
 void etcd_state_client_t::start_etcd_watcher()
 {
+    if (!etcd_addresses.size())
+    {
+        fprintf(stderr, "etcd_address is missing in Vitastor configuration\n");
+        exit(1);
+    }
     std::string etcd_address = etcd_addresses[rand() % etcd_addresses.size()];
     std::string etcd_api_path;
     int pos = etcd_address.find('/');
@@ -342,7 +352,7 @@ void etcd_state_client_t::load_pgs()
     });
 }
 #else
-void etcd_state_client_t::parse_config(json11::Json & config)
+void etcd_state_client_t::parse_config(const json11::Json & config)
 {
 }
 

src/etcd_state_client.h

@@ -106,7 +106,7 @@ public:
     void load_global_config();
     void load_pgs();
     void parse_state(const etcd_kv_t & kv);
-    void parse_config(json11::Json & config);
+    void parse_config(const json11::Json & config);
     inode_watch_t* watch_inode(std::string name);
     void close_watch(inode_watch_t* watch);
     ~etcd_state_client_t();

src/fio_cluster.cpp

@@ -24,7 +24,6 @@
 #include <netinet/tcp.h>
 
 #include <vector>
-#include <unordered_map>
 
 #include "epoll_manager.h"
 #include "cluster_client.h"
@@ -46,6 +45,7 @@ struct sec_data
 struct sec_options
 {
     int __pad;
+    char *config_path = NULL;
     char *etcd_host = NULL;
     char *etcd_prefix = NULL;
     char *image = NULL;
@@ -54,12 +54,22 @@ struct sec_options
     int cluster_log = 0;
     int trace = 0;
     int use_rdma = 0;
+    char *rdma_device = NULL;
     int rdma_port_num = 0;
     int rdma_gid_index = 0;
     int rdma_mtu = 0;
 };
 
 static struct fio_option options[] = {
+    {
+        .name   = "conf",
+        .lname  = "Vitastor config path",
+        .type   = FIO_OPT_STR_STORE,
+        .off1   = offsetof(struct sec_options, config_path),
+        .help   = "Vitastor config path",
+        .category = FIO_OPT_C_ENGINE,
+        .group  = FIO_OPT_G_FILENAME,
+    },
     {
         .name   = "etcd",
         .lname  = "etcd address",
@@ -127,10 +137,49 @@ static struct fio_option options[] = {
     },
     {
         .name   = "use_rdma",
-        .lname  = "OSD trace",
+        .lname  = "Use RDMA",
         .type   = FIO_OPT_BOOL,
         .off1   = offsetof(struct sec_options, use_rdma),
         .help   = "Use RDMA",
+        .def    = "-1",
+        .category = FIO_OPT_C_ENGINE,
+        .group  = FIO_OPT_G_FILENAME,
+    },
+    {
+        .name   = "rdma_device",
+        .lname  = "RDMA device name",
+        .type   = FIO_OPT_STR_STORE,
+        .off1   = offsetof(struct sec_options, rdma_device),
+        .help   = "RDMA device name",
+        .category = FIO_OPT_C_ENGINE,
+        .group  = FIO_OPT_G_FILENAME,
+    },
+    {
+        .name   = "rdma_port_num",
+        .lname  = "RDMA port number",
+        .type   = FIO_OPT_INT,
+        .off1   = offsetof(struct sec_options, rdma_port_num),
+        .help   = "RDMA port number",
+        .def    = "0",
+        .category = FIO_OPT_C_ENGINE,
+        .group  = FIO_OPT_G_FILENAME,
+    },
+    {
+        .name   = "rdma_gid_index",
+        .lname  = "RDMA gid index",
+        .type   = FIO_OPT_INT,
+        .off1   = offsetof(struct sec_options, rdma_gid_index),
+        .help   = "RDMA gid index",
+        .def    = "0",
+        .category = FIO_OPT_C_ENGINE,
+        .group  = FIO_OPT_G_FILENAME,
+    },
+    {
+        .name   = "rdma_mtu",
+        .lname  = "RDMA path MTU",
+        .type   = FIO_OPT_INT,
+        .off1   = offsetof(struct sec_options, rdma_mtu),
+        .help   = "RDMA path MTU",
         .def    = "0",
         .category = FIO_OPT_C_ENGINE,
         .group  = FIO_OPT_G_FILENAME,
@@ -145,12 +194,6 @@ static int sec_setup(struct thread_data *td)
     sec_options *o = (sec_options*)td->eo;
     sec_data *bsd;
 
-    if (!o->etcd_host)
-    {
-        td_verror(td, EINVAL, "etcd address is missing");
-        return 1;
-    }
-
     bsd = new sec_data;
     if (!bsd)
     {
@@ -166,12 +209,26 @@ static int sec_setup(struct thread_data *td)
         td->o.open_files++;
     }
 
-    json11::Json cfg = json11::Json::object {
+    json11::Json::object cfg;
-        { "etcd_address", std::string(o->etcd_host) },
+    if (o->config_path)
-        { "etcd_prefix", std::string(o->etcd_prefix ? o->etcd_prefix : "/vitastor") },
+        cfg["config_path"] = std::string(o->config_path);
-        { "log_level", o->cluster_log },
+    if (o->etcd_host)
-        { "use_rdma", o->use_rdma },
+        cfg["etcd_address"] = std::string(o->etcd_host);
-    };
+    if (o->etcd_prefix)
+        cfg["etcd_prefix"] = std::string(o->etcd_prefix);
+    if (o->rdma_device)
+        cfg["rdma_device"] = std::string(o->rdma_device);
+    if (o->rdma_port_num)
+        cfg["rdma_port_num"] = o->rdma_port_num;
+    if (o->rdma_gid_index)
+        cfg["rdma_gid_index"] = o->rdma_gid_index;
+    if (o->rdma_mtu)
+        cfg["rdma_mtu"] = o->rdma_mtu;
+    if (o->cluster_log)
+        cfg["log_level"] = o->cluster_log;
+    if (o->use_rdma != -1)
+        cfg["use_rdma"] = o->use_rdma;
+    json11::Json cfg_json(cfg);
 
     if (!o->image)
     {
@@ -196,7 +253,7 @@ static int sec_setup(struct thread_data *td)
     }
     bsd->ringloop = new ring_loop_t(512);
     bsd->epmgr = new epoll_manager_t(bsd->ringloop);
-    bsd->cli = new cluster_client_t(bsd->ringloop, bsd->epmgr->tfd, cfg);
+    bsd->cli = new cluster_client_t(bsd->ringloop, bsd->epmgr->tfd, cfg_json);
     if (o->image)
     {
         while (!bsd->cli->is_ready())

src/messenger.cpp

@@ -25,6 +25,8 @@ void osd_messenger_t::init()
         }
         else
         {
+            rdma_max_sge = rdma_max_sge < rdma_context->attrx.orig_attr.max_sge
+                ? rdma_max_sge : rdma_context->attrx.orig_attr.max_sge;
             printf("[OSD %lu] RDMA initialized successfully\n", osd_num);
             fcntl(rdma_context->channel->fd, F_SETFL, fcntl(rdma_context->channel->fd, F_GETFL, 0) | O_NONBLOCK);
             tfd->set_fd_handler(rdma_context->channel->fd, false, [this](int notify_fd, int epoll_events)
@@ -129,39 +131,46 @@ void osd_messenger_t::parse_config(const json11::Json & config)
 {
 #ifdef WITH_RDMA
     if (!config["use_rdma"].is_null())
+    {
+        // RDMA is on by default in RDMA-enabled builds
         this->use_rdma = config["use_rdma"].bool_value() || config["use_rdma"].uint64_value() != 0;
+    }
     this->rdma_device = config["rdma_device"].string_value();
     this->rdma_port_num = (uint8_t)config["rdma_port_num"].uint64_value();
     if (!this->rdma_port_num)
         this->rdma_port_num = 1;
     this->rdma_gid_index = (uint8_t)config["rdma_gid_index"].uint64_value();
     this->rdma_mtu = (uint32_t)config["rdma_mtu"].uint64_value();
+    this->rdma_max_sge = config["rdma_max_sge"].uint64_value();
+    if (!this->rdma_max_sge)
+        this->rdma_max_sge = 128;
+    this->rdma_max_send = config["rdma_max_send"].uint64_value();
+    if (!this->rdma_max_send)
+        this->rdma_max_send = 32;
+    this->rdma_max_recv = config["rdma_max_recv"].uint64_value();
+    if (!this->rdma_max_recv)
+        this->rdma_max_recv = 8;
+    this->rdma_max_msg = config["rdma_max_msg"].uint64_value();
+    if (!this->rdma_max_msg || this->rdma_max_msg > 128*1024*1024)
+        this->rdma_max_msg = 1024*1024;
 #endif
-    this->bs_bitmap_granularity = strtoull(config["bitmap_granularity"].string_value().c_str(), NULL, 10);
+    this->receive_buffer_size = (uint32_t)config["tcp_header_buffer_size"].uint64_value();
-    if (!this->bs_bitmap_granularity)
+    if (!this->receive_buffer_size || this->receive_buffer_size > 1024*1024*1024)
-        this->bs_bitmap_granularity = DEFAULT_BITMAP_GRANULARITY;
+        this->receive_buffer_size = 65536;
     this->use_sync_send_recv = config["use_sync_send_recv"].bool_value() ||
         config["use_sync_send_recv"].uint64_value();
     this->peer_connect_interval = config["peer_connect_interval"].uint64_value();
     if (!this->peer_connect_interval)
-    {
+        this->peer_connect_interval = 5;
-        this->peer_connect_interval = DEFAULT_PEER_CONNECT_INTERVAL;
-    }
     this->peer_connect_timeout = config["peer_connect_timeout"].uint64_value();
     if (!this->peer_connect_timeout)
-    {
+        this->peer_connect_timeout = 5;
-        this->peer_connect_timeout = DEFAULT_PEER_CONNECT_TIMEOUT;
-    }
     this->osd_idle_timeout = config["osd_idle_timeout"].uint64_value();
     if (!this->osd_idle_timeout)
-    {
+        this->osd_idle_timeout = 5;
-        this->osd_idle_timeout = DEFAULT_OSD_PING_TIMEOUT;
-    }
     this->osd_ping_timeout = config["osd_ping_timeout"].uint64_value();
     if (!this->osd_ping_timeout)
-    {
+        this->osd_ping_timeout = 5;
-        this->osd_ping_timeout = DEFAULT_OSD_PING_TIMEOUT;
-    }
     this->log_level = config["log_level"].uint64_value();
 }
 
@@ -356,9 +365,6 @@ void osd_messenger_t::on_connect_peer(osd_num_t peer_osd, int peer_fd)
 
 void osd_messenger_t::check_peer_config(osd_client_t *cl)
 {
-#ifdef WITH_RDMA
-    msgr_rdma_connection_t *rdma_conn = NULL;
-#endif
     osd_op_t *op = new osd_op_t();
     op->op_type = OSD_OP_OUT;
     op->peer_fd = cl->peer_fd;
@@ -374,11 +380,12 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
 #ifdef WITH_RDMA
     if (rdma_context)
     {
-        cl->rdma_conn = msgr_rdma_connection_t::create(rdma_context, max_rdma_send, max_rdma_recv, max_rdma_sge);
+        cl->rdma_conn = msgr_rdma_connection_t::create(rdma_context, rdma_max_send, rdma_max_recv, rdma_max_sge, rdma_max_msg);
         if (cl->rdma_conn)
         {
             json11::Json payload = json11::Json::object {
                 { "connect_rdma", cl->rdma_conn->addr.to_string() },
+                { "rdma_max_msg", cl->rdma_conn->max_msg },
             };
             std::string payload_str = payload.dump();
             op->req.show_conf.json_len = payload_str.size();
@@ -388,11 +395,7 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
         }
     }
 #endif
-    op->callback = [this, cl
+    op->callback = [this, cl](osd_op_t *op)
-#ifdef WITH_RDMA
-    , rdma_conn
-#endif
-    ](osd_op_t *op)
     {
         std::string json_err;
         json11::Json config;
@@ -455,7 +458,15 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
             }
             else
             {
-                printf("Connected to OSD %lu using RDMA\n", cl->osd_num);
+                uint64_t server_max_msg = config["rdma_max_msg"].uint64_value();
+                if (cl->rdma_conn->max_msg > server_max_msg)
+                {
+                    cl->rdma_conn->max_msg = server_max_msg;
+                }
+                if (log_level > 0)
+                {
+                    printf("Connected to OSD %lu using RDMA\n", cl->osd_num);
+                }
                 cl->peer_state = PEER_RDMA;
                 tfd->set_fd_handler(cl->peer_fd, false, NULL);
                 // Add the initial receive request
@@ -509,3 +520,52 @@ bool osd_messenger_t::is_rdma_enabled()
 {
     return rdma_context != NULL;
 }
+
+json11::Json osd_messenger_t::read_config(const json11::Json & config)
+{
+    const char *config_path = config["config_path"].string_value() != ""
+        ? config["config_path"].string_value().c_str() : VITASTOR_CONFIG_PATH;
+    int fd = open(config_path, O_RDONLY);
+    if (fd < 0)
+    {
+        if (errno != ENOENT)
+            fprintf(stderr, "Error reading %s: %s\n", config_path, strerror(errno));
+        return config;
+    }
+    struct stat st;
+    if (fstat(fd, &st) != 0)
+    {
+        fprintf(stderr, "Error reading %s: %s\n", config_path, strerror(errno));
+        close(fd);
+        return config;
+    }
+    std::string buf;
+    buf.resize(st.st_size);
+    int done = 0;
+    while (done < st.st_size)
+    {
+        int r = read(fd, (void*)buf.data()+done, st.st_size-done);
+        if (r < 0)
+        {
+            fprintf(stderr, "Error reading %s: %s\n", config_path, strerror(errno));
+            close(fd);
+            return config;
+        }
+        done += r;
+    }
+    close(fd);
+    std::string json_err;
+    json11::Json::object file_config = json11::Json::parse(buf, json_err).object_items();
+    if (json_err != "")
+    {
+        fprintf(stderr, "Invalid JSON in %s: %s\n", config_path, json_err.c_str());
+        return config;
+    }
+    file_config.erase("config_path");
+    file_config.erase("osd_num");
+    for (auto kv: config.object_items())
+    {
+        file_config[kv.first] = kv.second;
+    }
+    return file_config;
+}

src/messenger.h

@@ -33,10 +33,8 @@
 #define PEER_RDMA 4
 #define PEER_STOPPED 5
 
-#define DEFAULT_PEER_CONNECT_INTERVAL 5
-#define DEFAULT_PEER_CONNECT_TIMEOUT 5
-#define DEFAULT_OSD_PING_TIMEOUT 5
 #define DEFAULT_BITMAP_GRANULARITY 4096
+#define VITASTOR_CONFIG_PATH "/etc/vitastor/vitastor.conf"
 
 #define MSGR_SENDP_HDR 1
 #define MSGR_SENDP_FREE 2
@@ -122,13 +120,11 @@ struct osd_messenger_t
 protected:
     int keepalive_timer_id = -1;
 
-    // FIXME: make receive_buffer_size configurable
+    uint32_t receive_buffer_size = 0;
-    int receive_buffer_size = 64*1024;
+    int peer_connect_interval = 0;
-    int peer_connect_interval = DEFAULT_PEER_CONNECT_INTERVAL;
+    int peer_connect_timeout = 0;
-    int peer_connect_timeout = DEFAULT_PEER_CONNECT_TIMEOUT;
+    int osd_idle_timeout = 0;
-    int osd_idle_timeout = DEFAULT_OSD_PING_TIMEOUT;
+    int osd_ping_timeout = 0;
-    int osd_ping_timeout = DEFAULT_OSD_PING_TIMEOUT;
-    uint32_t bs_bitmap_granularity = 0;
     int log_level = 0;
     bool use_sync_send_recv = false;
 
@@ -137,7 +133,8 @@ protected:
     std::string rdma_device;
     uint64_t rdma_port_num = 1, rdma_gid_index = 0, rdma_mtu = 0;
     msgr_rdma_context_t *rdma_context = NULL;
-    int max_rdma_sge = 128, max_rdma_send = 32, max_rdma_recv = 32;
+    uint64_t rdma_max_sge = 0, rdma_max_send = 0, rdma_max_recv = 8;
+    uint64_t rdma_max_msg = 0;
 #endif
 
     std::vector<int> read_ready_clients;
@@ -168,9 +165,11 @@ public:
     void accept_connections(int listen_fd);
     ~osd_messenger_t();
 
+    static json11::Json read_config(const json11::Json & config);
+
 #ifdef WITH_RDMA
     bool is_rdma_enabled();
-    bool connect_rdma(int peer_fd, std::string rdma_address);
+    bool connect_rdma(int peer_fd, std::string rdma_address, uint64_t client_max_msg);
 #endif
 
 protected:
@@ -188,6 +187,7 @@ protected:
     void handle_send(int result, osd_client_t *cl);
 
     bool handle_read(int result, osd_client_t *cl);
+    bool handle_read_buffer(osd_client_t *cl, void *curbuf, int remain);
     bool handle_finished_read(osd_client_t *cl);
     void handle_op_hdr(osd_client_t *cl);
     bool handle_reply_hdr(osd_client_t *cl);

src/mock/messenger.cpp

@@ -42,3 +42,8 @@ void osd_messenger_t::read_requests()
 void osd_messenger_t::send_replies()
 {
 }
+
+json11::Json osd_messenger_t::read_config(const json11::Json & config)
+{
+    return config;
+}

src/msgr_rdma.cpp

@@ -1,3 +1,6 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
+
 #include <stdio.h>
 #include <stdlib.h>
 #include "msgr_rdma.h"
@@ -163,7 +166,8 @@ cleanup:
     return NULL;
 }
 
-msgr_rdma_connection_t *msgr_rdma_connection_t::create(msgr_rdma_context_t *ctx, uint32_t max_send, uint32_t max_recv, uint32_t max_sge)
+msgr_rdma_connection_t *msgr_rdma_connection_t::create(msgr_rdma_context_t *ctx, uint32_t max_send,
+    uint32_t max_recv, uint32_t max_sge, uint32_t max_msg)
 {
     msgr_rdma_connection_t *conn = new msgr_rdma_connection_t;
 
@@ -173,6 +177,7 @@ msgr_rdma_connection_t *msgr_rdma_connection_t::create(msgr_rdma_context_t *ctx,
     conn->max_send = max_send;
     conn->max_recv = max_recv;
     conn->max_sge = max_sge;
+    conn->max_msg = max_msg;
 
     ctx->used_max_cqe += max_send+max_recv;
     if (ctx->used_max_cqe > ctx->max_cqe)
@@ -293,13 +298,17 @@ int msgr_rdma_connection_t::connect(msgr_rdma_address_t *dest)
     return 0;
 }
 
-bool osd_messenger_t::connect_rdma(int peer_fd, std::string rdma_address)
+bool osd_messenger_t::connect_rdma(int peer_fd, std::string rdma_address, uint64_t client_max_msg)
 {
     // Try to connect to the peer using RDMA
     msgr_rdma_address_t addr;
     if (msgr_rdma_address_t::from_string(rdma_address.c_str(), &addr))
     {
-        auto rdma_conn = msgr_rdma_connection_t::create(rdma_context, max_rdma_send, max_rdma_recv, max_rdma_sge);
+        if (client_max_msg > rdma_max_msg)
+        {
+            client_max_msg = rdma_max_msg;
+        }
+        auto rdma_conn = msgr_rdma_connection_t::create(rdma_context, rdma_max_send, rdma_max_recv, rdma_max_sge, client_max_msg);
         if (rdma_conn)
         {
             int r = rdma_conn->connect(&addr);
@@ -351,58 +360,34 @@ bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
         // Only send one batch at a time
         return true;
     }
-    int op_size = 0, op_sge = 0, op_max = rc->max_sge*bs_bitmap_granularity;
+    uint64_t op_size = 0, op_sge = 0;
-    // FIXME: rc->max_sge should be negotiated between client & server
     ibv_sge sge[rc->max_sge];
     while (rc->send_pos < cl->send_list.size())
     {
         iovec & iov = cl->send_list[rc->send_pos];
-        if (cl->outbox[rc->send_pos].flags & MSGR_SENDP_HDR)
+        if (op_size >= rc->max_msg || op_sge >= rc->max_sge)
         {
-            if (op_sge > 0)
+            try_send_rdma_wr(cl, sge, op_sge);
-            {
+            op_sge = 0;
-                try_send_rdma_wr(cl, sge, op_sge);
+            op_size = 0;
-                op_sge = 0;
-                op_size = 0;
-                if (rc->cur_send >= rc->max_send)
-                    break;
-            }
-            assert(rc->send_buf_pos == 0);
-            sge[0] = {
-                .addr = (uintptr_t)iov.iov_base,
-                .length = (uint32_t)iov.iov_len,
-                .lkey = rc->ctx->mr->lkey,
-            };
-            try_send_rdma_wr(cl, sge, 1);
-            rc->send_pos++;
             if (rc->cur_send >= rc->max_send)
+            {
                 break;
+            }
         }
-        else
+        uint32_t len = (uint32_t)(op_size+iov.iov_len-rc->send_buf_pos < rc->max_msg
+            ? iov.iov_len-rc->send_buf_pos : rc->max_msg-op_size);
+        sge[op_sge++] = {
+            .addr = (uintptr_t)(iov.iov_base+rc->send_buf_pos),
+            .length = len,
+            .lkey = rc->ctx->mr->lkey,
+        };
+        op_size += len;
+        rc->send_buf_pos += len;
+        if (rc->send_buf_pos >= iov.iov_len)
         {
-            if (op_size >= op_max || op_sge >= rc->max_sge)
+            rc->send_pos++;
-            {
+            rc->send_buf_pos = 0;
-                try_send_rdma_wr(cl, sge, op_sge);
-                op_sge = 0;
-                op_size = 0;
-                if (rc->cur_send >= rc->max_send)
-                    break;
-            }
-            // Fragment all messages into parts no longer than (max_sge*4k) = 120k on ConnectX-4
-            // Otherwise the client may not be able to receive them in small parts
-            uint32_t len = (uint32_t)(op_size+iov.iov_len-rc->send_buf_pos < op_max ? iov.iov_len-rc->send_buf_pos : op_max-op_size);
-            sge[op_sge++] = {
-                .addr = (uintptr_t)(iov.iov_base+rc->send_buf_pos),
-                .length = len,
-                .lkey = rc->ctx->mr->lkey,
-            };
-            op_size += len;
-            rc->send_buf_pos += len;
-            if (rc->send_buf_pos >= iov.iov_len)
-            {
-                rc->send_pos++;
-                rc->send_buf_pos = 0;
-            }
         }
     }
     if (op_sge > 0)
@@ -432,53 +417,16 @@ static void try_recv_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
 bool osd_messenger_t::try_recv_rdma(osd_client_t *cl)
 {
     auto rc = cl->rdma_conn;
-    if (rc->cur_recv > 0)
+    while (rc->cur_recv < rc->max_recv)
     {
-        return true;
+        void *buf = malloc_or_die(rc->max_msg);
-    }
+        rc->recv_buffers.push_back(buf);
-    if (!cl->recv_list.get_size())
+        ibv_sge sge = {
-    {
+            .addr = (uintptr_t)buf,
-        cl->recv_list.reset();
+            .length = (uint32_t)rc->max_msg,
-        cl->read_op = new osd_op_t;
-        cl->read_op->peer_fd = cl->peer_fd;
-        cl->read_op->op_type = OSD_OP_IN;
-        cl->recv_list.push_back(cl->read_op->req.buf, OSD_PACKET_SIZE);
-        cl->read_remaining = OSD_PACKET_SIZE;
-        cl->read_state = CL_READ_HDR;
-    }
-    int op_size = 0, op_sge = 0, op_max = rc->max_sge*bs_bitmap_granularity;
-    iovec *segments = cl->recv_list.get_iovec();
-    // FIXME: rc->max_sge should be negotiated between client & server
-    ibv_sge sge[rc->max_sge];
-    while (rc->recv_pos < cl->recv_list.get_size())
-    {
-        iovec & iov = segments[rc->recv_pos];
-        if (op_size >= op_max || op_sge >= rc->max_sge)
-        {
-            try_recv_rdma_wr(cl, sge, op_sge);
-            op_sge = 0;
-            op_size = 0;
-            if (rc->cur_recv >= rc->max_recv)
-                break;
-        }
-        // Receive in identical (max_sge*4k) fragments
-        uint32_t len = (uint32_t)(op_size+iov.iov_len-rc->recv_buf_pos < op_max ? iov.iov_len-rc->recv_buf_pos : op_max-op_size);
-        sge[op_sge++] = {
-            .addr = (uintptr_t)(iov.iov_base+rc->recv_buf_pos),
-            .length = len,
             .lkey = rc->ctx->mr->lkey,
         };
-        op_size += len;
+        try_recv_rdma_wr(cl, &sge, 1);
-        rc->recv_buf_pos += len;
-        if (rc->recv_buf_pos >= iov.iov_len)
-        {
-            rc->recv_pos++;
-            rc->recv_buf_pos = 0;
-        }
-    }
-    if (op_sge > 0)
-    {
-        try_recv_rdma_wr(cl, sge, op_sge);
     }
     return true;
 }
@@ -529,24 +477,10 @@ void osd_messenger_t::handle_rdma_events()
             if (!is_send)
             {
                 cl->rdma_conn->cur_recv--;
-                if (!cl->rdma_conn->cur_recv)
+                handle_read_buffer(cl, cl->rdma_conn->recv_buffers[0], wc[i].byte_len);
-                {
+                free(cl->rdma_conn->recv_buffers[0]);
-                    cl->recv_list.done += cl->rdma_conn->recv_pos;
+                cl->rdma_conn->recv_buffers.erase(cl->rdma_conn->recv_buffers.begin(), cl->rdma_conn->recv_buffers.begin()+1);
-                    cl->rdma_conn->recv_pos = 0;
+                try_recv_rdma(cl);
-                    if (!cl->recv_list.get_size())
-                    {
-                        cl->read_remaining = 0;
-                        if (handle_finished_read(cl))
-                        {
-                            try_recv_rdma(cl);
-                        }
-                    }
-                    else
-                    {
-                        // Continue to receive data
-                        try_recv_rdma(cl);
-                    }
-                }
             }
             else
             {

src/msgr_rdma.h

@@ -1,3 +1,6 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
+
 #pragma once
 #include <infiniband/verbs.h>
 #include <string>
@@ -43,11 +46,13 @@ struct msgr_rdma_connection_t
     msgr_rdma_address_t addr;
     int max_send = 0, max_recv = 0, max_sge = 0;
     int cur_send = 0, cur_recv = 0;
+    uint64_t max_msg = 0;
 
     int send_pos = 0, send_buf_pos = 0;
     int recv_pos = 0, recv_buf_pos = 0;
+    std::vector<void*> recv_buffers;
 
     ~msgr_rdma_connection_t();
-    static msgr_rdma_connection_t *create(msgr_rdma_context_t *ctx, uint32_t max_send, uint32_t max_recv, uint32_t max_sge);
+    static msgr_rdma_connection_t *create(msgr_rdma_context_t *ctx, uint32_t max_send, uint32_t max_recv, uint32_t max_sge, uint32_t max_msg);
     int connect(msgr_rdma_address_t *dest);
 };

src/msgr_receive.cpp

@@ -91,48 +91,9 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl)
     {
         if (cl->read_iov.iov_base == cl->in_buf)
         {
-            // Compose operation(s) from the buffer
+            if (!handle_read_buffer(cl, cl->in_buf, result))
-            int remain = result;
-            void *curbuf = cl->in_buf;
-            while (remain > 0)
             {
-                if (!cl->read_op)
+                goto fin;
-                {
-                    cl->read_op = new osd_op_t;
-                    cl->read_op->peer_fd = cl->peer_fd;
-                    cl->read_op->op_type = OSD_OP_IN;
-                    cl->recv_list.push_back(cl->read_op->req.buf, OSD_PACKET_SIZE);
-                    cl->read_remaining = OSD_PACKET_SIZE;
-                    cl->read_state = CL_READ_HDR;
-                }
-                while (cl->recv_list.done < cl->recv_list.count && remain > 0)
-                {
-                    iovec* cur = cl->recv_list.get_iovec();
-                    if (cur->iov_len > remain)
-                    {
-                        memcpy(cur->iov_base, curbuf, remain);
-                        cl->read_remaining -= remain;
-                        cur->iov_len -= remain;
-                        cur->iov_base += remain;
-                        remain = 0;
-                    }
-                    else
-                    {
-                        memcpy(cur->iov_base, curbuf, cur->iov_len);
-                        curbuf += cur->iov_len;
-                        cl->read_remaining -= cur->iov_len;
-                        remain -= cur->iov_len;
-                        cur->iov_len = 0;
-                        cl->recv_list.done++;
-                    }
-                }
-                if (cl->recv_list.done >= cl->recv_list.count)
-                {
-                    if (!handle_finished_read(cl))
-                    {
-                        goto fin;
-                    }
-                }
             }
         }
         else
@@ -159,6 +120,52 @@ fin:
     return ret;
 }
 
+bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, void *curbuf, int remain)
+{
+    // Compose operation(s) from the buffer
+    while (remain > 0)
+    {
+        if (!cl->read_op)
+        {
+            cl->read_op = new osd_op_t;
+            cl->read_op->peer_fd = cl->peer_fd;
+            cl->read_op->op_type = OSD_OP_IN;
+            cl->recv_list.push_back(cl->read_op->req.buf, OSD_PACKET_SIZE);
+            cl->read_remaining = OSD_PACKET_SIZE;
+            cl->read_state = CL_READ_HDR;
+        }
+        while (cl->recv_list.done < cl->recv_list.count && remain > 0)
+        {
+            iovec* cur = cl->recv_list.get_iovec();
+            if (cur->iov_len > remain)
+            {
+                memcpy(cur->iov_base, curbuf, remain);
+                cl->read_remaining -= remain;
+                cur->iov_len -= remain;
+                cur->iov_base += remain;
+                remain = 0;
+            }
+            else
+            {
+                memcpy(cur->iov_base, curbuf, cur->iov_len);
+                curbuf += cur->iov_len;
+                cl->read_remaining -= cur->iov_len;
+                remain -= cur->iov_len;
+                cur->iov_len = 0;
+                cl->recv_list.done++;
+            }
+        }
+        if (cl->recv_list.done >= cl->recv_list.count)
+        {
+            if (!handle_finished_read(cl))
+            {
+                return false;
+            }
+        }
+    }
+    return true;
+}
+
 bool osd_messenger_t::handle_finished_read(osd_client_t *cl)
 {
     cl->recv_list.reset();

src/msgr_send.cpp

@@ -272,7 +272,10 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
         {
             // FIXME: Do something better than just forgetting the FD
             // FIXME: Ignore pings during RDMA state transition
-            printf("Successfully connected with client %d using RDMA\n", cl->peer_fd);
+            if (log_level > 0)
+            {
+                printf("Successfully connected with client %d using RDMA\n", cl->peer_fd);
+            }
             cl->peer_state = PEER_RDMA;
             tfd->set_fd_handler(cl->peer_fd, false, NULL);
             // Add the initial receive request

src/nbd_proxy.cpp

@@ -116,7 +116,7 @@ public:
             "Vitastor NBD proxy\n"
             "(c) Vitaliy Filippov, 2020-2021 (VNPL-1.1)\n\n"
             "USAGE:\n"
-            "  %s map --etcd_address <etcd_address> (--image <image> | --pool <pool> --inode <inode> --size <size in bytes>)\n"
+            "  %s map [--etcd_address <etcd_address>] (--image <image> | --pool <pool> --inode <inode> --size <size in bytes>)\n"
             "  %s unmap /dev/nbd0\n"
             "  %s list [--json]\n",
             exe_name, exe_name, exe_name
@@ -146,11 +146,6 @@ public:
     void start(json11::Json cfg)
     {
         // Check options
-        if (cfg["etcd_address"].string_value() == "")
-        {
-            fprintf(stderr, "etcd_address is missing\n");
-            exit(1);
-        }
         if (cfg["image"].string_value() != "")
         {
             // Use image name

src/osd.cpp

@@ -10,31 +10,39 @@
 #include "osd.h"
 #include "http_client.h"
 
-osd_t::osd_t(blockstore_config_t & config, ring_loop_t *ringloop)
+static blockstore_config_t json_to_bs(const json11::Json::object & config)
 {
-    bs_block_size = strtoull(config["block_size"].c_str(), NULL, 10);
+    blockstore_config_t bs;
-    bs_bitmap_granularity = strtoull(config["bitmap_granularity"].c_str(), NULL, 10);
+    for (auto kv: config)
-    if (!bs_block_size)
+    {
-        bs_block_size = DEFAULT_BLOCK_SIZE;
+        if (kv.second.is_string())
-    if (!bs_bitmap_granularity)
+            bs[kv.first] = kv.second.string_value();
-        bs_bitmap_granularity = DEFAULT_BITMAP_GRANULARITY;
+        else
-    clean_entry_bitmap_size = bs_block_size / bs_bitmap_granularity / 8;
+            bs[kv.first] = kv.second.dump();
+    }
+    return bs;
+}
 
+osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop)
+{
     zero_buffer_size = 1<<20;
     zero_buffer = malloc_or_die(zero_buffer_size);
     memset(zero_buffer, 0, zero_buffer_size);
 
-    this->config = config;
     this->ringloop = ringloop;
 
+    this->config = msgr.read_config(config).object_items();
+    if (this->config.find("log_level") == this->config.end())
+        this->config["log_level"] = 1;
+    parse_config(this->config);
+
     epmgr = new epoll_manager_t(ringloop);
     // FIXME: Use timerfd_interval based directly on io_uring
     this->tfd = epmgr->tfd;
 
     // FIXME: Create Blockstore from on-disk superblock config and check it against the OSD cluster config
-    this->bs = new blockstore_t(config, ringloop, tfd);
+    auto bs_cfg = json_to_bs(this->config);
-
+    this->bs = new blockstore_t(bs_cfg, ringloop, tfd);
-    parse_config(config);
 
     this->tfd->set_timer(print_stats_interval*1000, true, [this](int timer_id)
     {
@@ -66,63 +74,71 @@ osd_t::~osd_t()
     free(zero_buffer);
 }
 
-void osd_t::parse_config(blockstore_config_t & config)
+void osd_t::parse_config(const json11::Json & config)
 {
-    if (config.find("log_level") == config.end())
+    st_cli.parse_config(config);
-        config["log_level"] = "1";
+    msgr.parse_config(config);
-    log_level = strtoull(config["log_level"].c_str(), NULL, 10);
+    // OSD number
-    // Initial startup configuration
+    osd_num = config["osd_num"].uint64_value();
-    json11::Json json_config = json11::Json(config);
-    st_cli.parse_config(json_config);
-    etcd_report_interval = strtoull(config["etcd_report_interval"].c_str(), NULL, 10);
-    if (etcd_report_interval <= 0)
-        etcd_report_interval = 30;
-    osd_num = strtoull(config["osd_num"].c_str(), NULL, 10);
     if (!osd_num)
         throw std::runtime_error("osd_num is required in the configuration");
     msgr.osd_num = osd_num;
+    // Vital Blockstore parameters
+    bs_block_size = config["block_size"].uint64_value();
+    if (!bs_block_size)
+        bs_block_size = DEFAULT_BLOCK_SIZE;
+    bs_bitmap_granularity = config["bitmap_granularity"].uint64_value();
+    if (!bs_bitmap_granularity)
+        bs_bitmap_granularity = DEFAULT_BITMAP_GRANULARITY;
+    clean_entry_bitmap_size = bs_block_size / bs_bitmap_granularity / 8;
+    // Bind address
+    bind_address = config["bind_address"].string_value();
+    if (bind_address == "")
+        bind_address = "0.0.0.0";
+    bind_port = config["bind_port"].uint64_value();
+    if (bind_port <= 0 || bind_port > 65535)
+        bind_port = 0;
+    // OSD configuration
+    log_level = config["log_level"].uint64_value();
+    etcd_report_interval = config["etcd_report_interval"].uint64_value();
+    if (etcd_report_interval <= 0)
+        etcd_report_interval = 30;
+    readonly = config["readonly"] == "true" || config["readonly"] == "1" || config["readonly"] == "yes";
     run_primary = config["run_primary"] != "false" && config["run_primary"] != "0" && config["run_primary"] != "no";
     no_rebalance = config["no_rebalance"] == "true" || config["no_rebalance"] == "1" || config["no_rebalance"] == "yes";
     no_recovery = config["no_recovery"] == "true" || config["no_recovery"] == "1" || config["no_recovery"] == "yes";
     allow_test_ops = config["allow_test_ops"] == "true" || config["allow_test_ops"] == "1" || config["allow_test_ops"] == "yes";
-    // Cluster configuration
-    bind_address = config["bind_address"];
-    if (bind_address == "")
-        bind_address = "0.0.0.0";
-    bind_port = stoull_full(config["bind_port"]);
-    if (bind_port <= 0 || bind_port > 65535)
-        bind_port = 0;
     if (config["immediate_commit"] == "all")
         immediate_commit = IMMEDIATE_ALL;
     else if (config["immediate_commit"] == "small")
         immediate_commit = IMMEDIATE_SMALL;
-    if (config.find("autosync_interval") != config.end())
+    else
+        immediate_commit = IMMEDIATE_NONE;
+    if (!config["autosync_interval"].is_null())
     {
-        autosync_interval = strtoull(config["autosync_interval"].c_str(), NULL, 10);
+        // Allow to set it to 0
+        autosync_interval = config["autosync_interval"].uint64_value();
         if (autosync_interval > MAX_AUTOSYNC_INTERVAL)
             autosync_interval = DEFAULT_AUTOSYNC_INTERVAL;
     }
-    if (config.find("client_queue_depth") != config.end())
+    if (!config["client_queue_depth"].is_null())
     {
-        client_queue_depth = strtoull(config["client_queue_depth"].c_str(), NULL, 10);
+        client_queue_depth = config["client_queue_depth"].uint64_value();
         if (client_queue_depth < 128)
             client_queue_depth = 128;
     }
-    recovery_queue_depth = strtoull(config["recovery_queue_depth"].c_str(), NULL, 10);
+    recovery_queue_depth = config["recovery_queue_depth"].uint64_value();
     if (recovery_queue_depth < 1 || recovery_queue_depth > MAX_RECOVERY_QUEUE)
         recovery_queue_depth = DEFAULT_RECOVERY_QUEUE;
-    recovery_sync_batch = strtoull(config["recovery_sync_batch"].c_str(), NULL, 10);
+    recovery_sync_batch = config["recovery_sync_batch"].uint64_value();
     if (recovery_sync_batch < 1 || recovery_sync_batch > MAX_RECOVERY_QUEUE)
         recovery_sync_batch = DEFAULT_RECOVERY_BATCH;
-    if (config["readonly"] == "true" || config["readonly"] == "1" || config["readonly"] == "yes")
+    print_stats_interval = config["print_stats_interval"].uint64_value();
-        readonly = true;
-    print_stats_interval = strtoull(config["print_stats_interval"].c_str(), NULL, 10);
     if (!print_stats_interval)
         print_stats_interval = 3;
-    slow_log_interval = strtoull(config["slow_log_interval"].c_str(), NULL, 10);
+    slow_log_interval = config["slow_log_interval"].uint64_value();
     if (!slow_log_interval)
         slow_log_interval = 10;
-    msgr.parse_config(json_config);
 }
 
 void osd_t::bind_socket()

src/osd.h

@@ -92,7 +92,7 @@ class osd_t
 {
     // config
 
-    blockstore_config_t config;
+    json11::Json::object config;
     int etcd_report_interval = 30;
 
     bool readonly = false;
@@ -167,7 +167,7 @@ class osd_t
     uint64_t recovery_stat_bytes[2][2] = { 0 };
 
     // cluster connection
-    void parse_config(blockstore_config_t & config);
+    void parse_config(const json11::Json & config);
     void init_cluster();
     void on_change_osd_state_hook(osd_num_t peer_osd);
     void on_change_pg_history_hook(pool_id_t pool_id, pg_num_t pg_num);
@@ -268,7 +268,7 @@ class osd_t
     }
 
 public:
-    osd_t(blockstore_config_t & config, ring_loop_t *ringloop);
+    osd_t(const json11::Json & config, ring_loop_t *ringloop);
     ~osd_t();
     void force_stop(int exitcode);
     bool shutdown();

src/osd_cluster.cpp

@@ -21,7 +21,7 @@ void osd_t::init_cluster()
         {
             // Test version of clustering code with 1 pool, 1 PG and 2 peers
             // Example: peers = 2:127.0.0.1:11204,3:127.0.0.1:11205
-            std::string peerstr = config["peers"];
+            std::string peerstr = config["peers"].string_value();
             while (peerstr.size())
             {
                 int pos = peerstr.find(',');
@@ -340,21 +340,10 @@ void osd_t::on_change_pg_history_hook(pool_id_t pool_id, pg_num_t pg_num)
 
 void osd_t::on_load_config_hook(json11::Json::object & global_config)
 {
-    blockstore_config_t osd_config = this->config;
+    json11::Json::object osd_config = this->config;
-    for (auto & cfg_var: global_config)
+    for (auto & kv: global_config)
-    {
+        if (osd_config.find(kv.first) == osd_config.end())
-        if (this->config.find(cfg_var.first) == this->config.end())
+            osd_config[kv.first] = kv.second;
-        {
-            if (cfg_var.second.is_string())
-            {
-                osd_config[cfg_var.first] = cfg_var.second.string_value();
-            }
-            else
-            {
-                osd_config[cfg_var.first] = cfg_var.second.dump();
-            }
-        }
-    }
     parse_config(osd_config);
     bind_socket();
     acquire_lease();
@@ -380,7 +369,7 @@ void osd_t::acquire_lease()
         etcd_lease_id = data["ID"].string_value();
         create_osd_state();
     });
-    printf("[OSD %lu] reporting to etcd at %s every %d seconds\n", this->osd_num, config["etcd_address"].c_str(), etcd_report_interval);
+    printf("[OSD %lu] reporting to etcd at %s every %d seconds\n", this->osd_num, config["etcd_address"].string_value().c_str(), etcd_report_interval);
     tfd->set_timer(etcd_report_interval*1000, true, [this](int timer_id)
     {
         renew_lease();

src/osd_main.cpp

@@ -29,13 +29,13 @@ int main(int narg, char *args[])
         perror("BUG: too small packet size");
         return 1;
     }
-    blockstore_config_t config;
+    json11::Json::object config;
     for (int i = 1; i < narg; i++)
     {
         if (args[i][0] == '-' && args[i][1] == '-' && i < narg-1)
         {
             char *opt = args[i]+2;
-            config[opt] = args[++i];
+            config[std::string(opt)] = std::string(args[++i]);
         }
     }
     signal(SIGINT, handle_sigint);

src/osd_secondary.cpp

@@ -169,11 +169,12 @@ void osd_t::exec_show_config(osd_op_t *cur_op)
         if (req_json["connect_rdma"].is_string())
         {
             // Peer is trying to connect using RDMA, try to satisfy him
-            bool ok = msgr.connect_rdma(cur_op->peer_fd, req_json["connect_rdma"].string_value());
+            bool ok = msgr.connect_rdma(cur_op->peer_fd, req_json["connect_rdma"].string_value(), req_json["rdma_max_msg"].uint64_value());
             if (ok)
             {
-                wire_config["rdma_connected"] = true;
+                auto rc = msgr.clients.at(cur_op->peer_fd)->rdma_conn;
-                wire_config["rdma_address"] = msgr.clients.at(cur_op->peer_fd)->rdma_conn->addr.to_string();
+                wire_config["rdma_address"] = rc->addr.to_string();
+                wire_config["rdma_max_msg"] = rc->max_msg;
             }
         }
     }

src/qemu_driver.c

@@ -40,6 +40,7 @@ typedef struct VitastorClient
 {
     void *proxy;
     void *watch;
+    char *config_path;
     char *etcd_host;
     char *etcd_prefix;
     char *image;
@@ -47,6 +48,10 @@ typedef struct VitastorClient
     uint64_t pool;
     uint64_t size;
     long readonly;
+    char *rdma_device;
+    int rdma_port_num;
+    int rdma_gid_index;
+    int rdma_mtu;
     QemuMutex mutex;
 } VitastorClient;
 
@@ -95,7 +100,8 @@ static void qemu_rbd_unescape(char *src)
 }
 
 // vitastor[:key=value]*
-// vitastor:etcd_host=127.0.0.1:inode=1:pool=1
+// vitastor[:etcd_host=127.0.0.1]:inode=1:pool=1[:rdma_gid_index=3]
+// vitastor:config_path=/etc/vitastor/vitastor.conf:image=testimg
 static void vitastor_parse_filename(const char *filename, QDict *options, Error **errp)
 {
     const char *start;
@@ -123,7 +129,12 @@ static void vitastor_parse_filename(const char *filename, QDict *options, Error
         qemu_rbd_unescape(name);
         value = qemu_rbd_next_tok(p, ':', &p);
         qemu_rbd_unescape(value);
-        if (!strcmp(name, "inode") || !strcmp(name, "pool") || !strcmp(name, "size"))
+        if (!strcmp(name, "inode") ||
+            !strcmp(name, "pool") ||
+            !strcmp(name, "size") ||
+            !strcmp(name, "rdma_port_num") ||
+            !strcmp(name, "rdma_gid_index") ||
+            !strcmp(name, "rdma_mtu"))
         {
             unsigned long long num_val;
             if (parse_uint_full(value, &num_val, 0))
@@ -157,11 +168,6 @@ static void vitastor_parse_filename(const char *filename, QDict *options, Error
             goto out;
         }
     }
-    if (!qdict_get_str(options, "etcd_host"))
-    {
-        error_setg(errp, "etcd_host is missing");
-        goto out;
-    }
 
 out:
     g_free(buf);
@@ -189,9 +195,17 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
     VitastorClient *client = bs->opaque;
     int64_t ret = 0;
     qemu_mutex_init(&client->mutex);
+    client->config_path = g_strdup(qdict_get_try_str(options, "config_path"));
     client->etcd_host = g_strdup(qdict_get_try_str(options, "etcd_host"));
     client->etcd_prefix = g_strdup(qdict_get_try_str(options, "etcd_prefix"));
-    client->proxy = vitastor_proxy_create(bdrv_get_aio_context(bs), client->etcd_host, client->etcd_prefix);
+    client->rdma_device = g_strdup(qdict_get_try_str(options, "rdma_device"));
+    client->rdma_port_num = qdict_get_try_int(options, "rdma_port_num", 0);
+    client->rdma_gid_index = qdict_get_try_int(options, "rdma_gid_index", 0);
+    client->rdma_mtu = qdict_get_try_int(options, "rdma_mtu", 0);
+    client->proxy = vitastor_proxy_create(
+        bdrv_get_aio_context(bs), client->config_path, client->etcd_host, client->etcd_prefix,
+        client->rdma_device, client->rdma_port_num, client->rdma_gid_index, client->rdma_mtu
+    );
     client->image = g_strdup(qdict_get_try_str(options, "image"));
     client->readonly = (flags & BDRV_O_RDWR) ? 1 : 0;
     if (client->image)
@@ -241,6 +255,11 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
     }
     bs->total_sectors = client->size / BDRV_SECTOR_SIZE;
     //client->aio_context = bdrv_get_aio_context(bs);
+    qdict_del(options, "rdma_mtu");
+    qdict_del(options, "rdma_gid_index");
+    qdict_del(options, "rdma_port_num");
+    qdict_del(options, "rdma_device");
+    qdict_del(options, "config_path");
     qdict_del(options, "etcd_host");
     qdict_del(options, "etcd_prefix");
     qdict_del(options, "image");
@@ -255,7 +274,10 @@ static void vitastor_close(BlockDriverState *bs)
     VitastorClient *client = bs->opaque;
     vitastor_proxy_destroy(client->proxy);
     qemu_mutex_destroy(&client->mutex);
-    g_free(client->etcd_host);
+    if (client->config_path)
+        g_free(client->config_path);
+    if (client->etcd_host)
+        g_free(client->etcd_host);
     if (client->etcd_prefix)
         g_free(client->etcd_prefix);
     if (client->image)
@@ -478,6 +500,7 @@ static QEMUOptionParameter vitastor_create_opts[] = {
 static const char *vitastor_strong_runtime_opts[] = {
     "inode",
     "pool",
+    "config_path",
     "etcd_host",
     "etcd_prefix",
 

src/qemu_proxy.cpp

@@ -34,15 +34,28 @@ public:
     cluster_client_t *cli;
     AioContext *ctx;
 
-    QemuProxy(AioContext *ctx, const char *etcd_host, const char *etcd_prefix)
+    QemuProxy(AioContext *ctx, const char *config_path, const char *etcd_host, const char *etcd_prefix,
+        const char *rdma_device, int rdma_port_num, int rdma_gid_index, int rdma_mtu)
     {
         this->ctx = ctx;
-        json11::Json cfg = json11::Json::object {
+        json11::Json::object cfg;
-            { "etcd_address", std::string(etcd_host) },
+        if (config_path)
-            { "etcd_prefix", std::string(etcd_prefix ? etcd_prefix : "/vitastor") },
+            cfg["config_path"] = std::string(config_path);
-        };
+        if (etcd_host)
+            cfg["etcd_address"] = std::string(etcd_host);
+        if (etcd_prefix)
+            cfg["etcd_prefix"] = std::string(etcd_prefix);
+        if (rdma_device)
+            cfg["rdma_device"] = std::string(rdma_device);
+        if (rdma_port_num)
+            cfg["rdma_port_num"] = rdma_port_num;
+        if (rdma_gid_index)
+            cfg["rdma_gid_index"] = rdma_gid_index;
+        if (rdma_mtu)
+            cfg["rdma_mtu"] = rdma_mtu;
+        json11::Json cfg_json(cfg);
         tfd = new timerfd_manager_t([this](int fd, bool wr, std::function<void(int, int)> callback) { set_fd_handler(fd, wr, callback); });
-        cli = new cluster_client_t(NULL, tfd, cfg);
+        cli = new cluster_client_t(NULL, tfd, cfg_json);
     }
 
     ~QemuProxy()
@@ -80,9 +93,10 @@ public:
 
 extern "C" {
 
-void* vitastor_proxy_create(AioContext *ctx, const char *etcd_host, const char *etcd_prefix)
+void* vitastor_proxy_create(AioContext *ctx, const char *config_path, const char *etcd_host, const char *etcd_prefix,
+    const char *rdma_device, int rdma_port_num, int rdma_gid_index, int rdma_mtu)
 {
-    QemuProxy *p = new QemuProxy(ctx, etcd_host, etcd_prefix);
+    QemuProxy *p = new QemuProxy(ctx, config_path, etcd_host, etcd_prefix, rdma_device, rdma_port_num, rdma_gid_index, rdma_mtu);
     return p;
 }
 

src/qemu_proxy.h

@@ -16,7 +16,8 @@ extern "C" {
 
 // Our exports
 typedef void VitastorIOHandler(long retval, void *opaque);
-void* vitastor_proxy_create(AioContext *ctx, const char *etcd_host, const char *etcd_prefix);
+void* vitastor_proxy_create(AioContext *ctx, const char *config_path, const char *etcd_host, const char *etcd_prefix,
+    const char *rdma_device, int rdma_port_num, int rdma_gid_index, int rdma_mtu);
 void vitastor_proxy_destroy(void *client);
 void vitastor_proxy_rw(int write, void *client, uint64_t inode, uint64_t offset, uint64_t len,
     struct iovec *iov, int iovcnt, VitastorIOHandler cb, void *opaque);

src/rm_inode.cpp

@@ -87,7 +87,7 @@ public:
             "Vitastor inode removal tool\n"
             "(c) Vitaliy Filippov, 2020 (VNPL-1.1)\n\n"
             "USAGE:\n"
-            "  %s --etcd_address <etcd_address> --pool <pool> --inode <inode> [--wait-list]\n",
+            "  %s [--etcd_address <etcd_address>] --pool <pool> --inode <inode> [--wait-list]\n",
             exe_name
         );
         exit(0);
@@ -95,11 +95,6 @@ public:
 
     void run(json11::Json cfg)
     {
-        if (cfg["etcd_address"].string_value() == "")
-        {
-            fprintf(stderr, "etcd_address is missing\n");
-            exit(1);
-        }
         inode = cfg["inode"].uint64_value();
         pool_id = cfg["pool"].uint64_value();
         if (pool_id)

tests/test_vm_start.sh

@@ -46,7 +46,7 @@ $ETCDCTL put /vitastor/config/inode/1/1 '{"name":"debian9@0","size":'$((2048*102
 $ETCDCTL put /vitastor/config/inode/1/2 '{"parent_id":1,"name":"debian9","size":'$((2048*1024*1024))'}'
 
 qemu-system-x86_64 -enable-kvm -m 1024 \
-    -drive 'file=vitastor:etcd_host=127.0.0.1\:$ETCD_PORT/v3:image=debian9',format=raw,if=none,id=drive-virtio-disk0,cache=none \
+    -drive 'file=vitastor:etcd_host=127.0.0.1\:'$ETCD_PORT'/v3:image=debian9',format=raw,if=none,id=drive-virtio-disk0,cache=none \
     -device virtio-blk-pci,scsi=off,bus=pci.0,addr=0x5,drive=drive-virtio-disk0,id=virtio-disk0,bootindex=1,write-cache=off,physical_block_size=4096,logical_block_size=512 \
     -vnc 0.0.0.0:0