WIP Separate data connections

Add minsize_1 test
Support zero-copy send in fio_sec_osd to allow testing it
2021-12-30 01:57:13 +03:00 · 2021-12-28 10:54:36 +03:00 · 2021-12-27 02:12:44 +03:00 · 2021-12-27 02:12:44 +03:00 · 2021-12-26 21:11:24 +03:00 · 2021-12-25 11:38:13 +03:00
44 changed files with 677 additions and 422 deletions
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -2,6 +2,6 @@ cmake_minimum_required(VERSION 2.8)

 project(vitastor)

-set(VERSION "0.6.10")
+set(VERSION "0.6.11")

 add_subdirectory(src)
--- a/csi/Makefile
+++ b/csi/Makefile
@@ -1,4 +1,4 @@
-VERSION ?= v0.6.10
+VERSION ?= v0.6.11

 all: build push

--- a/csi/deploy/004-csi-nodeplugin.yaml
+++ b/csi/deploy/004-csi-nodeplugin.yaml
@@ -49,7 +49,7 @@ spec:
            capabilities:
              add: ["SYS_ADMIN"]
            allowPrivilegeEscalation: true
-          image: vitalif/vitastor-csi:v0.6.10
+          image: vitalif/vitastor-csi:v0.6.11
          args:
            - "--node=$(NODE_ID)"
            - "--endpoint=$(CSI_ENDPOINT)"
--- a/csi/deploy/007-csi-provisioner.yaml
+++ b/csi/deploy/007-csi-provisioner.yaml
@@ -116,7 +116,7 @@ spec:
            privileged: true
            capabilities:
              add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v0.6.10
+          image: vitalif/vitastor-csi:v0.6.11
          args:
            - "--node=$(NODE_ID)"
            - "--endpoint=$(CSI_ENDPOINT)"
--- a/csi/src/config.go
+++ b/csi/src/config.go
@@ -5,7 +5,7 @@ package vitastor

 const (
    vitastorCSIDriverName    = "csi.vitastor.io"
-    vitastorCSIDriverVersion = "0.6.10"
+    vitastorCSIDriverVersion = "0.6.11"
 )

 // Config struct fills the parameters of request or user input
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,4 +1,4 @@
-vitastor (0.6.10-1) unstable; urgency=medium
+vitastor (0.6.11-1) unstable; urgency=medium

  * RDMA support
  * Bugfixes
--- a/debian/vitastor.Dockerfile
+++ b/debian/vitastor.Dockerfile
@@ -33,8 +33,8 @@ 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.10; \
-    cd vitastor-0.6.10; \
+    cp -r /root/vitastor vitastor-0.6.11; \
+    cd vitastor-0.6.11; \
    ln -s /root/fio-build/fio-*/ ./fio; \
    FIO=$(head -n1 fio/debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
    ls /usr/include/linux/raw.h || cp ./debian/raw.h /usr/include/linux/raw.h; \
@@ -47,8 +47,8 @@ RUN set -e -x; \
    rm -rf a b; \
    echo "dep:fio=$FIO" > debian/fio_version; \
    cd /root/packages/vitastor-$REL; \
-    tar --sort=name --mtime='2020-01-01' --owner=0 --group=0 --exclude=debian -cJf vitastor_0.6.10.orig.tar.xz vitastor-0.6.10; \
-    cd vitastor-0.6.10; \
+    tar --sort=name --mtime='2020-01-01' --owner=0 --group=0 --exclude=debian -cJf vitastor_0.6.11.orig.tar.xz vitastor-0.6.11; \
+    cd vitastor-0.6.11; \
    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; \
--- a/patches/cinder-vitastor.py
+++ b/patches/cinder-vitastor.py
@@ -50,7 +50,7 @@ from cinder.volume import configuration
 from cinder.volume import driver
 from cinder.volume import volume_utils

-VERSION = '0.6.10'
+VERSION = '0.6.11'

 LOG = logging.getLogger(__name__)

--- a/rpm/build-tarball.sh
+++ b/rpm/build-tarball.sh
@@ -25,4 +25,4 @@ rm fio
 mv fio-copy fio
 FIO=`rpm -qi fio | 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
-tar --transform 's#^#vitastor-0.6.10/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.6.10$(rpm --eval '%dist').tar.gz *
+tar --transform 's#^#vitastor-0.6.11/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.6.11$(rpm --eval '%dist').tar.gz *
--- a/rpm/vitastor-el7.Dockerfile
+++ b/rpm/vitastor-el7.Dockerfile
@@ -34,7 +34,7 @@ ADD . /root/vitastor
 RUN set -e; \
    cd /root/vitastor/rpm; \
    sh build-tarball.sh; \
-    cp /root/vitastor-0.6.10.el7.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.6.11.el7.tar.gz ~/rpmbuild/SOURCES; \
    cp vitastor-el7.spec ~/rpmbuild/SPECS/vitastor.spec; \
    cd ~/rpmbuild/SPECS/; \
    rpmbuild -ba vitastor.spec; \
--- a/rpm/vitastor-el7.spec
+++ b/rpm/vitastor-el7.spec
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        0.6.10
+Version:        0.6.11
 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.10.el7.tar.gz
+Source0:        vitastor-0.6.11.el7.tar.gz

 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel
--- a/rpm/vitastor-el8.Dockerfile
+++ b/rpm/vitastor-el8.Dockerfile
@@ -33,7 +33,7 @@ ADD . /root/vitastor
 RUN set -e; \
    cd /root/vitastor/rpm; \
    sh build-tarball.sh; \
-    cp /root/vitastor-0.6.10.el8.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.6.11.el8.tar.gz ~/rpmbuild/SOURCES; \
    cp vitastor-el8.spec ~/rpmbuild/SPECS/vitastor.spec; \
    cd ~/rpmbuild/SPECS/; \
    rpmbuild -ba vitastor.spec; \
--- a/rpm/vitastor-el8.spec
+++ b/rpm/vitastor-el8.spec
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        0.6.10
+Version:        0.6.11
 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.10.el8.tar.gz
+Source0:        vitastor-0.6.11.el8.tar.gz

 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -15,7 +15,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 	set(CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/${CMAKE_INSTALL_LIBDIR}")
 endif()

-add_definitions(-DVERSION="0.6.10")
+add_definitions(-DVERSION="0.6.11")
 add_definitions(-Wall -Wno-sign-compare -Wno-comment -Wno-parentheses -Wno-pointer-arith -fdiagnostics-color=always -I ${CMAKE_SOURCE_DIR}/src)
 if (${WITH_ASAN})
 	add_definitions(-fsanitize=address -fno-omit-frame-pointer)
@@ -88,8 +88,8 @@ 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
+	epoll_manager.cpp etcd_state_client.cpp messenger.cpp addr_util.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 ${MSGR_RDMA}
 )
 target_compile_options(vitastor_common PUBLIC -fPIC)
@@ -112,6 +112,7 @@ if (${WITH_FIO})
 	add_library(fio_vitastor_sec SHARED
 		fio_sec_osd.cpp
 		rw_blocking.cpp
+		addr_util.cpp
 	)
 	target_link_libraries(fio_vitastor_sec
 		tcmalloc_minimal
@@ -189,11 +190,11 @@ endif (${WITH_QEMU})
 ### Test stubs

 # stub_osd, stub_bench, osd_test
-add_executable(stub_osd stub_osd.cpp rw_blocking.cpp)
+add_executable(stub_osd stub_osd.cpp rw_blocking.cpp addr_util.cpp)
 target_link_libraries(stub_osd tcmalloc_minimal)
-add_executable(stub_bench stub_bench.cpp rw_blocking.cpp)
+add_executable(stub_bench stub_bench.cpp rw_blocking.cpp addr_util.cpp)
 target_link_libraries(stub_bench tcmalloc_minimal)
-add_executable(osd_test osd_test.cpp rw_blocking.cpp)
+add_executable(osd_test osd_test.cpp rw_blocking.cpp addr_util.cpp)
 target_link_libraries(osd_test tcmalloc_minimal)

 # osd_rmw_test
--- a/src/addr_util.cpp
+++ b/src/addr_util.cpp
@@ -0,0 +1,60 @@
+#include <arpa/inet.h>
+#include <string.h>
+#include <stdio.h>
+
+#include <stdexcept>
+
+#include "addr_util.h"
+
+bool string_to_addr(std::string str, bool parse_port, int default_port, struct sockaddr *addr)
+{
+    if (parse_port)
+    {
+        int p = str.rfind(':');
+        if (p != std::string::npos && !(str.length() > 0 && str[p-1] == ']')) // "[ipv6]" which contains ':'
+        {
+            char null_byte = 0;
+            int n = sscanf(str.c_str()+p+1, "%d%c", &default_port, &null_byte);
+            if (n != 1 || default_port >= 0x10000)
+                return false;
+            str = str.substr(0, p);
+        }
+    }
+    if (inet_pton(AF_INET, str.c_str(), &((struct sockaddr_in*)addr)->sin_addr) == 1)
+    {
+        addr->sa_family = AF_INET;
+        ((struct sockaddr_in*)addr)->sin_port = htons(default_port);
+        return true;
+    }
+    if (str.length() >= 2 && str[0] == '[' && str[str.length()-1] == ']')
+        str = str.substr(1, str.length()-2);
+    if (inet_pton(AF_INET6, str.c_str(), &((struct sockaddr_in6*)addr)->sin6_addr) == 1)
+    {
+        addr->sa_family = AF_INET6;
+        ((struct sockaddr_in6*)addr)->sin6_port = htons(default_port);
+        return true;
+    }
+    return false;
+}
+
+std::string addr_to_string(const sockaddr &addr)
+{
+    char peer_str[256];
+    bool ok = false;
+    int port;
+    if (addr.sa_family == AF_INET)
+    {
+        ok = !!inet_ntop(AF_INET, &((sockaddr_in*)&addr)->sin_addr, peer_str, 256);
+        port = ntohs(((sockaddr_in*)&addr)->sin_port);
+    }
+    else if (addr.sa_family == AF_INET6)
+    {
+        ok = !!inet_ntop(AF_INET6, &((sockaddr_in6*)&addr)->sin6_addr, peer_str, 256);
+        port = ntohs(((sockaddr_in6*)&addr)->sin6_port);
+    }
+    else
+        throw std::runtime_error("Unknown address family "+std::to_string(addr.sa_family));
+    if (!ok)
+        throw std::runtime_error(std::string("inet_ntop: ") + strerror(errno));
+    return std::string(peer_str)+":"+std::to_string(port);
+}
--- a/src/addr_util.h
+++ b/src/addr_util.h
@@ -0,0 +1,7 @@
+#pragma once
+
+#include <sys/socket.h>
+#include <string>
+
+bool string_to_addr(std::string str, bool parse_port, int default_port, struct sockaddr *addr);
+std::string addr_to_string(const sockaddr &addr);
--- a/src/blockstore_impl.cpp
+++ b/src/blockstore_impl.cpp
@@ -235,6 +235,12 @@ void blockstore_impl_t::loop()
        {
            throw std::runtime_error(std::string("io_uring_submit: ") + strerror(-ret));
        }
+        for (auto s: journal.submitting_sectors)
+        {
+            // Mark journal sector writes as submitted
+            journal.sector_info[s].submit_id = 0;
+        }
+        journal.submitting_sectors.clear();
        if ((initial_ring_space - ringloop->space_left()) > 0)
        {
            live = true;
--- a/src/blockstore_impl.h
+++ b/src/blockstore_impl.h
@@ -54,6 +54,14 @@
 #define IS_BIG_WRITE(st) (((st) & 0x0F) == BS_ST_BIG_WRITE)
 #define IS_DELETE(st) (((st) & 0x0F) == BS_ST_DELETE)

+#define BS_SUBMIT_CHECK_SQES(n) \
+    if (ringloop->space_left() < (n))\
+    {\
+        /* Pause until there are more requests available */\
+        PRIV(op)->wait_for = WAIT_SQE;\
+        return 0;\
+    }
+
 #define BS_SUBMIT_GET_SQE(sqe, data) \
    BS_SUBMIT_GET_ONLY_SQE(sqe); \
    struct ring_data_t *data = ((ring_data_t*)sqe->user_data)
@@ -170,7 +178,7 @@ struct blockstore_op_private_t
    std::vector<fulfill_read_t> read_vec;

    // Sync, write
-    uint64_t min_flushed_journal_sector, max_flushed_journal_sector;
+    int min_flushed_journal_sector, max_flushed_journal_sector;

    // Write
    struct iovec iov_zerofill[3];
@@ -283,6 +291,10 @@ class blockstore_impl_t
    void open_journal();
    uint8_t* get_clean_entry_bitmap(uint64_t block_loc, int offset);

+    // Journaling
+    void prepare_journal_sector_write(int sector, blockstore_op_t *op);
+    void handle_journal_write(ring_data_t *data, uint64_t flush_id);
+
    // Asynchronous init
    int initialized;
    int metadata_buf_size;
@@ -310,21 +322,18 @@ class blockstore_impl_t

    // Sync
    int continue_sync(blockstore_op_t *op, bool queue_has_in_progress_sync);
-    void handle_sync_event(ring_data_t *data, blockstore_op_t *op);
    void ack_sync(blockstore_op_t *op);

    // Stabilize
    int dequeue_stable(blockstore_op_t *op);
    int continue_stable(blockstore_op_t *op);
    void mark_stable(const obj_ver_id & ov, bool forget_dirty = false);
-    void handle_stable_event(ring_data_t *data, blockstore_op_t *op);
    void stabilize_object(object_id oid, uint64_t max_ver);

    // Rollback
    int dequeue_rollback(blockstore_op_t *op);
    int continue_rollback(blockstore_op_t *op);
    void mark_rolled_back(const obj_ver_id & ov);
-    void handle_rollback_event(ring_data_t *data, blockstore_op_t *op);
    void erase_dirty(blockstore_dirty_db_t::iterator dirty_start, blockstore_dirty_db_t::iterator dirty_end, uint64_t clean_loc);

    // List
--- a/src/blockstore_journal.cpp
+++ b/src/blockstore_journal.cpp
@@ -153,22 +153,73 @@ journal_entry* prefill_single_journal_entry(journal_t & journal, uint16_t type,
    return je;
 }

-void prepare_journal_sector_write(journal_t & journal, int cur_sector, io_uring_sqe *sqe, std::function<void(ring_data_t*)> cb)
+void blockstore_impl_t::prepare_journal_sector_write(int cur_sector, blockstore_op_t *op)
 {
+    // Don't submit the same sector twice in the same batch
+    if (!journal.sector_info[cur_sector].submit_id)
+    {
+        io_uring_sqe *sqe = get_sqe();
+        // Caller must ensure availability of an SQE
+        assert(sqe != NULL);
+        ring_data_t *data = ((ring_data_t*)sqe->user_data);
+        journal.sector_info[cur_sector].written = true;
+        journal.sector_info[cur_sector].submit_id = ++journal.submit_id;
+        journal.submitting_sectors.push_back(cur_sector);
+        journal.sector_info[cur_sector].flush_count++;
+        data->iov = (struct iovec){
+            (journal.inmemory
+                ? journal.buffer + journal.sector_info[cur_sector].offset
+                : journal.sector_buf + journal.block_size*cur_sector),
+            journal.block_size
+        };
+        data->callback = [this, flush_id = journal.submit_id](ring_data_t *data) { handle_journal_write(data, flush_id); };
+        my_uring_prep_writev(
+            sqe, journal.fd, &data->iov, 1, journal.offset + journal.sector_info[cur_sector].offset
+        );
+    }
    journal.sector_info[cur_sector].dirty = false;
-    journal.sector_info[cur_sector].written = true;
-    journal.sector_info[cur_sector].flush_count++;
-    ring_data_t *data = ((ring_data_t*)sqe->user_data);
-    data->iov = (struct iovec){
-        (journal.inmemory
-            ? journal.buffer + journal.sector_info[cur_sector].offset
-            : journal.sector_buf + journal.block_size*cur_sector),
-        journal.block_size
-    };
-    data->callback = cb;
-    my_uring_prep_writev(
-        sqe, journal.fd, &data->iov, 1, journal.offset + journal.sector_info[cur_sector].offset
-    );
+    // But always remember that this operation has to wait until this exact journal write is finished
+    journal.flushing_ops.insert((pending_journaling_t){
+        .flush_id = journal.sector_info[cur_sector].submit_id,
+        .sector = cur_sector,
+        .op = op,
+    });
+    auto priv = PRIV(op);
+    priv->pending_ops++;
+    if (!priv->min_flushed_journal_sector)
+        priv->min_flushed_journal_sector = 1+cur_sector;
+    priv->max_flushed_journal_sector = 1+cur_sector;
+}
+
+void blockstore_impl_t::handle_journal_write(ring_data_t *data, uint64_t flush_id)
+{
+    live = true;
+    if (data->res != data->iov.iov_len)
+    {
+        // FIXME: our state becomes corrupted after a write error. maybe do something better than just die
+        throw std::runtime_error(
+            "journal write failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
+            "). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
+        );
+    }
+    auto fl_it = journal.flushing_ops.upper_bound((pending_journaling_t){ .flush_id = flush_id });
+    if (fl_it != journal.flushing_ops.end() && fl_it->flush_id == flush_id)
+    {
+        journal.sector_info[fl_it->sector].flush_count--;
+    }
+    while (fl_it != journal.flushing_ops.end() && fl_it->flush_id == flush_id)
+    {
+        auto priv = PRIV(fl_it->op);
+        priv->pending_ops--;
+        assert(priv->pending_ops >= 0);
+        if (priv->pending_ops == 0)
+        {
+            release_journal_sectors(fl_it->op);
+            priv->op_state++;
+            ringloop->wakeup();
+        }
+        journal.flushing_ops.erase(fl_it++);
+    }
 }

 journal_t::~journal_t()
--- a/src/blockstore_journal.h
+++ b/src/blockstore_journal.h
@@ -4,6 +4,7 @@
 #pragma once

 #include "crc32c.h"
+#include <set>

 #define MIN_JOURNAL_SIZE 4*1024*1024
 #define JOURNAL_MAGIC 0x4A33
@@ -145,8 +146,21 @@ struct journal_sector_info_t
    uint64_t flush_count;
    bool written;
    bool dirty;
+    uint64_t submit_id;
 };

+struct pending_journaling_t
+{
+    uint64_t flush_id;
+    int sector;
+    blockstore_op_t *op;
+};
+
+inline bool operator < (const pending_journaling_t & a, const pending_journaling_t & b)
+{
+    return a.flush_id < b.flush_id || a.flush_id == b.flush_id && a.op < b.op;
+}
+
 struct journal_t
 {
    int fd;
@@ -172,6 +186,9 @@ struct journal_t
    bool no_same_sector_overwrites = false;
    int cur_sector = 0;
    int in_sector_pos = 0;
+    std::vector<int> submitting_sectors;
+    std::set<pending_journaling_t> flushing_ops;
+    uint64_t submit_id = 0;

    // Used sector map
    // May use ~ 80 MB per 1 GB of used journal space in the worst case
@@ -200,5 +217,3 @@ struct blockstore_journal_check_t
 };

 journal_entry* prefill_single_journal_entry(journal_t & journal, uint16_t type, uint32_t size);
-
-void prepare_journal_sector_write(journal_t & journal, int sector, io_uring_sqe *sqe, std::function<void(ring_data_t*)> cb);
--- a/src/blockstore_open.cpp
+++ b/src/blockstore_open.cpp
@@ -306,6 +306,10 @@ static void check_size(int fd, uint64_t *size, uint64_t *sectsize, std::string n
    if (S_ISREG(st.st_mode))
    {
        *size = st.st_size;
+        if (sectsize)
+        {
+            *sectsize = st.st_blksize;
+        }
    }
    else if (S_ISBLK(st.st_mode))
    {
--- a/src/blockstore_rollback.cpp
+++ b/src/blockstore_rollback.cpp
@@ -74,24 +74,17 @@ skip_ov:
    {
        return 0;
    }
-    // There is sufficient space. Get SQEs
-    struct io_uring_sqe *sqe[space_check.sectors_to_write];
-    for (i = 0; i < space_check.sectors_to_write; i++)
-    {
-        BS_SUBMIT_GET_SQE_DECL(sqe[i]);
-    }
+    // There is sufficient space. Check SQEs
+    BS_SUBMIT_CHECK_SQES(space_check.sectors_to_write);
    // Prepare and submit journal entries
-    auto cb = [this, op](ring_data_t *data) { handle_rollback_event(data, op); };
-    int s = 0, cur_sector = -1;
+    int s = 0;
    for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
    {
        if (!journal.entry_fits(sizeof(journal_entry_rollback)) &&
            journal.sector_info[journal.cur_sector].dirty)
        {
-            if (cur_sector == -1)
-                PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-            prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
-            cur_sector = journal.cur_sector;
+            prepare_journal_sector_write(journal.cur_sector, op);
+            s++;
        }
        journal_entry_rollback *je = (journal_entry_rollback*)
            prefill_single_journal_entry(journal, JE_ROLLBACK, sizeof(journal_entry_rollback));
@@ -100,12 +93,9 @@ skip_ov:
        je->crc32 = je_crc32((journal_entry*)je);
        journal.crc32_last = je->crc32;
    }
-    prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
+    prepare_journal_sector_write(journal.cur_sector, op);
+    s++;
    assert(s == space_check.sectors_to_write);
-    if (cur_sector == -1)
-        PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-    PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
-    PRIV(op)->pending_ops = s;
    PRIV(op)->op_state = 1;
    return 1;
 }
@@ -114,30 +104,23 @@ int blockstore_impl_t::continue_rollback(blockstore_op_t *op)
 {
    if (PRIV(op)->op_state == 2)
        goto resume_2;
-    else if (PRIV(op)->op_state == 3)
-        goto resume_3;
-    else if (PRIV(op)->op_state == 5)
-        goto resume_5;
+    else if (PRIV(op)->op_state == 4)
+        goto resume_4;
    else
        return 1;
 resume_2:
-    // Release used journal sectors
-    release_journal_sectors(op);
-resume_3:
    if (!disable_journal_fsync)
    {
-        io_uring_sqe *sqe;
-        BS_SUBMIT_GET_SQE_DECL(sqe);
-        ring_data_t *data = ((ring_data_t*)sqe->user_data);
+        BS_SUBMIT_GET_SQE(sqe, data);
        my_uring_prep_fsync(sqe, journal.fd, IORING_FSYNC_DATASYNC);
        data->iov = { 0 };
-        data->callback = [this, op](ring_data_t *data) { handle_rollback_event(data, op); };
+        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
        PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
        PRIV(op)->pending_ops = 1;
-        PRIV(op)->op_state = 4;
+        PRIV(op)->op_state = 3;
        return 1;
    }
-resume_5:
+resume_4:
    obj_ver_id* v;
    int i;
    for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
@@ -196,24 +179,6 @@ void blockstore_impl_t::mark_rolled_back(const obj_ver_id & ov)
    }
 }

-void blockstore_impl_t::handle_rollback_event(ring_data_t *data, blockstore_op_t *op)
-{
-    live = true;
-    if (data->res != data->iov.iov_len)
-    {
-        throw std::runtime_error(
-            "write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
-            "). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
-        );
-    }
-    PRIV(op)->pending_ops--;
-    if (PRIV(op)->pending_ops == 0)
-    {
-        PRIV(op)->op_state++;
-        ringloop->wakeup();
-    }
-}
-
 void blockstore_impl_t::erase_dirty(blockstore_dirty_db_t::iterator dirty_start, blockstore_dirty_db_t::iterator dirty_end, uint64_t clean_loc)
 {
    if (dirty_end == dirty_start)
--- a/src/blockstore_stable.cpp
+++ b/src/blockstore_stable.cpp
@@ -97,25 +97,18 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
    {
        return 0;
    }
-    // There is sufficient space. Get SQEs
-    struct io_uring_sqe *sqe[space_check.sectors_to_write];
-    for (i = 0; i < space_check.sectors_to_write; i++)
-    {
-        BS_SUBMIT_GET_SQE_DECL(sqe[i]);
-    }
+    // There is sufficient space. Check SQEs
+    BS_SUBMIT_CHECK_SQES(space_check.sectors_to_write);
    // Prepare and submit journal entries
-    auto cb = [this, op](ring_data_t *data) { handle_stable_event(data, op); };
-    int s = 0, cur_sector = -1;
+    int s = 0;
    for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
    {
        // FIXME: Only stabilize versions that aren't stable yet
        if (!journal.entry_fits(sizeof(journal_entry_stable)) &&
            journal.sector_info[journal.cur_sector].dirty)
        {
-            if (cur_sector == -1)
-                PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-            prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
-            cur_sector = journal.cur_sector;
+            prepare_journal_sector_write(journal.cur_sector, op);
+            s++;
        }
        journal_entry_stable *je = (journal_entry_stable*)
            prefill_single_journal_entry(journal, JE_STABLE, sizeof(journal_entry_stable));
@@ -124,12 +117,9 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
        je->crc32 = je_crc32((journal_entry*)je);
        journal.crc32_last = je->crc32;
    }
-    prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
+    prepare_journal_sector_write(journal.cur_sector, op);
+    s++;
    assert(s == space_check.sectors_to_write);
-    if (cur_sector == -1)
-        PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-    PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
-    PRIV(op)->pending_ops = s;
    PRIV(op)->op_state = 1;
    return 1;
 }
@@ -138,30 +128,23 @@ int blockstore_impl_t::continue_stable(blockstore_op_t *op)
 {
    if (PRIV(op)->op_state == 2)
        goto resume_2;
-    else if (PRIV(op)->op_state == 3)
-        goto resume_3;
-    else if (PRIV(op)->op_state == 5)
-        goto resume_5;
+    else if (PRIV(op)->op_state == 4)
+        goto resume_4;
    else
        return 1;
 resume_2:
-    // Release used journal sectors
-    release_journal_sectors(op);
-resume_3:
    if (!disable_journal_fsync)
    {
-        io_uring_sqe *sqe;
-        BS_SUBMIT_GET_SQE_DECL(sqe);
-        ring_data_t *data = ((ring_data_t*)sqe->user_data);
+        BS_SUBMIT_GET_SQE(sqe, data);
        my_uring_prep_fsync(sqe, journal.fd, IORING_FSYNC_DATASYNC);
        data->iov = { 0 };
-        data->callback = [this, op](ring_data_t *data) { handle_stable_event(data, op); };
+        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
        PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
        PRIV(op)->pending_ops = 1;
-        PRIV(op)->op_state = 4;
+        PRIV(op)->op_state = 3;
        return 1;
    }
-resume_5:
+resume_4:
    // Mark dirty_db entries as stable, acknowledge op completion
    obj_ver_id* v;
    int i;
@@ -257,21 +240,3 @@ void blockstore_impl_t::mark_stable(const obj_ver_id & v, bool forget_dirty)
        unstable_writes.erase(unstab_it);
    }
 }
-
-void blockstore_impl_t::handle_stable_event(ring_data_t *data, blockstore_op_t *op)
-{
-    live = true;
-    if (data->res != data->iov.iov_len)
-    {
-        throw std::runtime_error(
-            "write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
-            "). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
-        );
-    }
-    PRIV(op)->pending_ops--;
-    if (PRIV(op)->pending_ops == 0)
-    {
-        PRIV(op)->op_state++;
-        ringloop->wakeup();
-    }
-}
--- a/src/blockstore_sync.cpp
+++ b/src/blockstore_sync.cpp
@@ -44,10 +44,8 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op, bool queue_has_in_prog
        if (journal.sector_info[journal.cur_sector].dirty)
        {
            // Write out the last journal sector if it happens to be dirty
-            BS_SUBMIT_GET_ONLY_SQE(sqe);
-            prepare_journal_sector_write(journal, journal.cur_sector, sqe, [this, op](ring_data_t *data) { handle_sync_event(data, op); });
-            PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
-            PRIV(op)->pending_ops = 1;
+            BS_SUBMIT_CHECK_SQES(1);
+            prepare_journal_sector_write(journal.cur_sector, op);
            PRIV(op)->op_state = SYNC_JOURNAL_WRITE_SENT;
            return 1;
        }
@@ -64,7 +62,7 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op, bool queue_has_in_prog
            BS_SUBMIT_GET_SQE(sqe, data);
            my_uring_prep_fsync(sqe, data_fd, IORING_FSYNC_DATASYNC);
            data->iov = { 0 };
-            data->callback = [this, op](ring_data_t *data) { handle_sync_event(data, op); };
+            data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
            PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
            PRIV(op)->pending_ops = 1;
            PRIV(op)->op_state = SYNC_DATA_SYNC_SENT;
@@ -85,24 +83,18 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op, bool queue_has_in_prog
        {
            return 0;
        }
-        // Get SQEs. Don't bother about merging, submit each journal sector as a separate request
-        struct io_uring_sqe *sqe[space_check.sectors_to_write];
-        for (int i = 0; i < space_check.sectors_to_write; i++)
-        {
-            BS_SUBMIT_GET_SQE_DECL(sqe[i]);
-        }
+        // Check SQEs. Don't bother about merging, submit each journal sector as a separate request
+        BS_SUBMIT_CHECK_SQES(space_check.sectors_to_write);
        // Prepare and submit journal entries
        auto it = PRIV(op)->sync_big_writes.begin();
-        int s = 0, cur_sector = -1;
+        int s = 0;
        while (it != PRIV(op)->sync_big_writes.end())
        {
            if (!journal.entry_fits(sizeof(journal_entry_big_write) + clean_entry_bitmap_size) &&
                journal.sector_info[journal.cur_sector].dirty)
            {
-                if (cur_sector == -1)
-                    PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-                prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], [this, op](ring_data_t *data) { handle_sync_event(data, op); });
-                cur_sector = journal.cur_sector;
+                prepare_journal_sector_write(journal.cur_sector, op);
+                s++;
            }
            auto & dirty_entry = dirty_db.at(*it);
            journal_entry_big_write *je = (journal_entry_big_write*)prefill_single_journal_entry(
@@ -129,12 +121,9 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op, bool queue_has_in_prog
            journal.crc32_last = je->crc32;
            it++;
        }
-        prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], [this, op](ring_data_t *data) { handle_sync_event(data, op); });
+        prepare_journal_sector_write(journal.cur_sector, op);
+        s++;
        assert(s == space_check.sectors_to_write);
-        if (cur_sector == -1)
-            PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-        PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
-        PRIV(op)->pending_ops = s;
        PRIV(op)->op_state = SYNC_JOURNAL_WRITE_SENT;
        return 1;
    }
@@ -145,7 +134,7 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op, bool queue_has_in_prog
            BS_SUBMIT_GET_SQE(sqe, data);
            my_uring_prep_fsync(sqe, journal.fd, IORING_FSYNC_DATASYNC);
            data->iov = { 0 };
-            data->callback = [this, op](ring_data_t *data) { handle_sync_event(data, op); };
+            data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
            PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
            PRIV(op)->pending_ops = 1;
            PRIV(op)->op_state = SYNC_JOURNAL_SYNC_SENT;
@@ -164,42 +153,6 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op, bool queue_has_in_prog
    return 1;
 }

-void blockstore_impl_t::handle_sync_event(ring_data_t *data, blockstore_op_t *op)
-{
-    live = true;
-    if (data->res != data->iov.iov_len)
-    {
-        throw std::runtime_error(
-            "write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
-            "). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
-        );
-    }
-    PRIV(op)->pending_ops--;
-    if (PRIV(op)->pending_ops == 0)
-    {
-        // Release used journal sectors
-        release_journal_sectors(op);
-        // Handle states
-        if (PRIV(op)->op_state == SYNC_DATA_SYNC_SENT)
-        {
-            PRIV(op)->op_state = SYNC_DATA_SYNC_DONE;
-        }
-        else if (PRIV(op)->op_state == SYNC_JOURNAL_WRITE_SENT)
-        {
-            PRIV(op)->op_state = SYNC_JOURNAL_WRITE_DONE;
-        }
-        else if (PRIV(op)->op_state == SYNC_JOURNAL_SYNC_SENT)
-        {
-            PRIV(op)->op_state = SYNC_DONE;
-        }
-        else
-        {
-            throw std::runtime_error("BUG: unexpected sync op state");
-        }
-        ringloop->wakeup();
-    }
-}
-
 void blockstore_impl_t::ack_sync(blockstore_op_t *op)
 {
    // Handle states
--- a/src/blockstore_write.cpp
+++ b/src/blockstore_write.cpp
@@ -268,18 +268,8 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
            cancel_all_writes(op, dirty_it, -ENOSPC);
            return 2;
        }
-#ifndef NDEBUG
-        // Double-check to not overwrite anything if easily possible
-        if (inmemory_meta)
-        {
-            uint64_t sector = (loc / (meta_block_size / clean_entry_size)) * meta_block_size;
-            uint64_t pos = (loc % (meta_block_size / clean_entry_size));
-            clean_disk_entry *meta_entry = (clean_disk_entry*)(metadata_buffer + sector + pos*clean_entry_size);
-            assert(!meta_entry->oid.inode);
-        }
-#endif
-        write_iodepth++;
        BS_SUBMIT_GET_SQE(sqe, data);
+        write_iodepth++;
        dirty_it->second.location = loc << block_order;
        dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_SUBMITTED;
 #ifdef BLOCKSTORE_DEBUG
@@ -334,29 +324,21 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
        {
            return 0;
        }
-        write_iodepth++;
-        // There is sufficient space. Get SQE(s)
-        struct io_uring_sqe *sqe1 = NULL;
-        if (immediate_commit != IMMEDIATE_NONE ||
-            !journal.entry_fits(sizeof(journal_entry_small_write) + clean_entry_bitmap_size))
-        {
+        // There is sufficient space. Check SQE(s)
+        BS_SUBMIT_CHECK_SQES(
            // Write current journal sector only if it's dirty and full, or in the immediate_commit mode
-            BS_SUBMIT_GET_SQE_DECL(sqe1);
-        }
-        struct io_uring_sqe *sqe2 = NULL;
-        if (op->len > 0)
-        {
-            BS_SUBMIT_GET_SQE_DECL(sqe2);
-        }
+            (immediate_commit != IMMEDIATE_NONE ||
+                !journal.entry_fits(sizeof(journal_entry_small_write) + clean_entry_bitmap_size) ? 1 : 0) +
+            (op->len > 0 ? 1 : 0)
+        );
+        write_iodepth++;
        // Got SQEs. Prepare previous journal sector write if required
        auto cb = [this, op](ring_data_t *data) { handle_write_event(data, op); };
        if (immediate_commit == IMMEDIATE_NONE)
        {
-            if (sqe1)
+            if (!journal.entry_fits(sizeof(journal_entry_small_write) + clean_entry_bitmap_size))
            {
-                prepare_journal_sector_write(journal, journal.cur_sector, sqe1, cb);
-                PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
-                PRIV(op)->pending_ops++;
+                prepare_journal_sector_write(journal.cur_sector, op);
            }
            else
            {
@@ -390,9 +372,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
        journal.crc32_last = je->crc32;
        if (immediate_commit != IMMEDIATE_NONE)
        {
-            prepare_journal_sector_write(journal, journal.cur_sector, sqe1, cb);
-            PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
-            PRIV(op)->pending_ops++;
+            prepare_journal_sector_write(journal.cur_sector, op);
        }
        if (op->len > 0)
        {
@@ -402,7 +382,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
                // Copy data
                memcpy(journal.buffer + journal.next_free, op->buf, op->len);
            }
-            ring_data_t *data2 = ((ring_data_t*)sqe2->user_data);
+            BS_SUBMIT_GET_SQE(sqe2, data2);
            data2->iov = (struct iovec){ op->buf, op->len };
            data2->callback = cb;
            my_uring_prep_writev(
@@ -451,13 +431,12 @@ int blockstore_impl_t::continue_write(blockstore_op_t *op)
 resume_2:
    // Only for the immediate_commit mode: prepare and submit big_write journal entry
    {
+        BS_SUBMIT_CHECK_SQES(1);
        auto dirty_it = dirty_db.find((obj_ver_id){
            .oid = op->oid,
            .version = op->version,
        });
        assert(dirty_it != dirty_db.end());
-        io_uring_sqe *sqe = NULL;
-        BS_SUBMIT_GET_SQE_DECL(sqe);
        journal_entry_big_write *je = (journal_entry_big_write*)prefill_single_journal_entry(
            journal, op->opcode == BS_OP_WRITE_STABLE ? JE_BIG_WRITE_INSTANT : JE_BIG_WRITE,
            sizeof(journal_entry_big_write) + clean_entry_bitmap_size
@@ -479,10 +458,7 @@ resume_2:
        memcpy((void*)(je+1), (clean_entry_bitmap_size > sizeof(void*) ? dirty_it->second.bitmap : &dirty_it->second.bitmap), clean_entry_bitmap_size);
        je->crc32 = je_crc32((journal_entry*)je);
        journal.crc32_last = je->crc32;
-        prepare_journal_sector_write(journal, journal.cur_sector, sqe,
-            [this, op](ring_data_t *data) { handle_write_event(data, op); });
-        PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
-        PRIV(op)->pending_ops = 1;
+        prepare_journal_sector_write(journal.cur_sector, op);
        PRIV(op)->op_state = 3;
        return 1;
    }
@@ -597,6 +573,7 @@ void blockstore_impl_t::handle_write_event(ring_data_t *data, blockstore_op_t *o
        );
    }
    PRIV(op)->pending_ops--;
+    assert(PRIV(op)->pending_ops >= 0);
    if (PRIV(op)->pending_ops == 0)
    {
        release_journal_sectors(op);
@@ -614,7 +591,6 @@ void blockstore_impl_t::release_journal_sectors(blockstore_op_t *op)
        uint64_t s = PRIV(op)->min_flushed_journal_sector;
        while (1)
        {
-            journal.sector_info[s-1].flush_count--;
            if (s != (1+journal.cur_sector) && journal.sector_info[s-1].flush_count == 0)
            {
                // We know for sure that we won't write into this sector anymore
@@ -654,23 +630,19 @@ int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
        return 0;
    }
    write_iodepth++;
-    io_uring_sqe *sqe = NULL;
-    if (immediate_commit != IMMEDIATE_NONE ||
-        (journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_del) &&
-        journal.sector_info[journal.cur_sector].dirty)
-    {
-        // Write current journal sector only if it's dirty and full, or in the immediate_commit mode
-        BS_SUBMIT_GET_SQE_DECL(sqe);
-    }
-    auto cb = [this, op](ring_data_t *data) { handle_write_event(data, op); };
+    // Write current journal sector only if it's dirty and full, or in the immediate_commit mode
+    BS_SUBMIT_CHECK_SQES(
+        (immediate_commit != IMMEDIATE_NONE ||
+            (journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_del) &&
+            journal.sector_info[journal.cur_sector].dirty) ? 1 : 0
+    );
    // Prepare journal sector write
    if (immediate_commit == IMMEDIATE_NONE)
    {
-        if (sqe)
+        if ((journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_del) &&
+            journal.sector_info[journal.cur_sector].dirty)
        {
-            prepare_journal_sector_write(journal, journal.cur_sector, sqe, cb);
-            PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
-            PRIV(op)->pending_ops++;
+            prepare_journal_sector_write(journal.cur_sector, op);
        }
        else
        {
@@ -697,9 +669,7 @@ int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
    dirty_it->second.state = BS_ST_DELETE | BS_ST_SUBMITTED;
    if (immediate_commit != IMMEDIATE_NONE)
    {
-        prepare_journal_sector_write(journal, journal.cur_sector, sqe, cb);
-        PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
-        PRIV(op)->pending_ops++;
+        prepare_journal_sector_write(journal.cur_sector, op);
    }
    if (!PRIV(op)->pending_ops)
    {
--- a/src/cli.cpp
+++ b/src/cli.cpp
@@ -305,6 +305,10 @@ void cli_tool_t::run(json11::Json cfg)
        fprintf(stderr, "unknown command: %s\n", cmd[0].string_value().c_str());
        exit(1);
    }
+    if (action_cb == NULL)
+    {
+        return;
+    }
    color = !cfg["no-color"].bool_value();
    json_output = cfg["json"].bool_value();
    iodepth = cfg["iodepth"].uint64_value();
--- a/src/fio_sec_osd.cpp
+++ b/src/fio_sec_osd.cpp
@@ -28,16 +28,23 @@
 #include <vector>
 #include <unordered_map>

+#include "addr_util.h"
 #include "rw_blocking.h"
 #include "osd_ops.h"
 #include "fio_headers.h"

+struct op_buf_t
+{
+    osd_any_op_t buf;
+    io_u* fio_op;
+};
+
 struct sec_data
 {
    int connect_fd;
    /* block_size = 1 << block_order (128KB by default) */
    uint64_t block_order = 17, block_size = 1 << 17;
-    std::unordered_map<uint64_t, io_u*> queue;
+    std::unordered_map<uint64_t, op_buf_t*> queue;
    bool last_sync = false;
    /* The list of completed io_u structs. */
    std::vector<io_u*> completed;
@@ -52,6 +59,7 @@ struct sec_options
    int single_primary = 0;
    int trace = 0;
    int block_order = 17;
+    int zerocopy_send = 0;
 };

 static struct fio_option options[] = {
@@ -102,6 +110,16 @@ static struct fio_option options[] = {
        .category = FIO_OPT_C_ENGINE,
        .group  = FIO_OPT_G_FILENAME,
    },
+    {
+        .name   = "zerocopy_send",
+        .lname  = "Use zero-copy send",
+        .type   = FIO_OPT_BOOL,
+        .off1   = offsetof(struct sec_options, zerocopy_send),
+        .help   = "Use zero-copy send (MSG_ZEROCOPY)",
+        .def    = "0",
+        .category = FIO_OPT_C_ENGINE,
+        .group  = FIO_OPT_G_FILENAME,
+    },
    {
        .name = NULL,
    },
@@ -152,17 +170,14 @@ static int sec_init(struct thread_data *td)
    bsd->block_order = o->block_order == 0 ? 17 : o->block_order;
    bsd->block_size = 1 << o->block_order;

-    struct sockaddr_in addr;
-    int r;
-    if ((r = inet_pton(AF_INET, o->host ? o->host : "127.0.0.1", &addr.sin_addr)) != 1)
+    sockaddr addr;
+    if (!string_to_addr(std::string(o->host ? o->host : "127.0.0.1"), false, o->port > 0 ? o->port : 11203, &addr))
    {
-        fprintf(stderr, "server address: %s%s\n", o->host ? o->host : "127.0.0.1", r == 0 ? " is not valid" : ": no ipv4 support");
+        fprintf(stderr, "server address: %s is not valid\n", o->host ? o->host : "127.0.0.1");
        return 1;
    }
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(o->port ? o->port : 11203);

-    bsd->connect_fd = socket(AF_INET, SOCK_STREAM, 0);
+    bsd->connect_fd = socket(addr.sa_family, SOCK_STREAM, 0);
    if (bsd->connect_fd < 0)
    {
        perror("socket");
@@ -175,6 +190,14 @@ static int sec_init(struct thread_data *td)
    }
    int one = 1;
    setsockopt(bsd->connect_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
+    if (o->zerocopy_send)
+    {
+        if (setsockopt(bsd->connect_fd, SOL_SOCKET, SO_ZEROCOPY, &one, sizeof(one)) < 0)
+        {
+            perror("setsockopt zerocopy");
+            return 1;
+        }
+    }

    // FIXME: read config (block size) from OSD

@@ -195,7 +218,9 @@ static enum fio_q_status sec_queue(struct thread_data *td, struct io_u *io)
    }

    io->engine_data = bsd;
-    osd_any_op_t op = { 0 };
+    op_buf_t *op_buf = new op_buf_t;
+    op_buf->fio_op = io;
+    osd_any_op_t &op = op_buf->buf;

    op.hdr.magic = SECONDARY_OSD_OP_MAGIC;
    op.hdr.id = n;
@@ -271,19 +296,18 @@ static enum fio_q_status sec_queue(struct thread_data *td, struct io_u *io)
    io->error = 0;
    bsd->inflight++;
    bsd->op_n++;
-    bsd->queue[n] = io;
+    bsd->queue[n] = op_buf;

    iovec iov[2] = { { .iov_base = op.buf, .iov_len = OSD_PACKET_SIZE } };
    int iovcnt = 1, wtotal = OSD_PACKET_SIZE;
    if (io->ddir == DDIR_WRITE)
    {
-        iov[1] = { .iov_base = io->xfer_buf, .iov_len = io->xfer_buflen };
+        iov[iovcnt++] = { .iov_base = io->xfer_buf, .iov_len = io->xfer_buflen };
        wtotal += io->xfer_buflen;
-        iovcnt++;
    }
-    if (writev_blocking(bsd->connect_fd, iov, iovcnt) != wtotal)
+    if (sendv_blocking(bsd->connect_fd, iov, iovcnt, opt->zerocopy_send ? MSG_ZEROCOPY : 0) != wtotal)
    {
-        perror("writev");
+        perror("sendmsg");
        exit(1);
    }

@@ -312,7 +336,8 @@ static int sec_getevents(struct thread_data *td, unsigned int min, unsigned int
            fprintf(stderr, "bad reply: op id %lx missing in local queue\n", reply.hdr.id);
            exit(1);
        }
-        io_u* io = it->second;
+        io_u* io = it->second->fio_op;
+        delete it->second;
        bsd->queue.erase(it);
        if (io->ddir == DDIR_READ)
        {
@@ -321,7 +346,23 @@ static int sec_getevents(struct thread_data *td, unsigned int min, unsigned int
                fprintf(stderr, "Short read: retval = %ld instead of %llu\n", reply.hdr.retval, io->xfer_buflen);
                exit(1);
            }
-            read_blocking(bsd->connect_fd, io->xfer_buf, io->xfer_buflen);
+            // Support bitmap
+            uint64_t bitmap = 0;
+            int iovcnt = 0;
+            iovec iov[2];
+            if (reply.sec_rw.attr_len > 0)
+            {
+                if (reply.sec_rw.attr_len <= 8)
+                    iov[iovcnt++] = { .iov_base = &bitmap, .iov_len = reply.sec_rw.attr_len };
+                else
+                    iov[iovcnt++] = { .iov_base = (void*)(bitmap = (uint64_t)malloc(reply.sec_rw.attr_len)), .iov_len = reply.sec_rw.attr_len };
+            }
+            iov[iovcnt++] = { .iov_base = io->xfer_buf, .iov_len = io->xfer_buflen };
+            readv_blocking(bsd->connect_fd, iov, iovcnt);
+            if (reply.sec_rw.attr_len > 8)
+            {
+                free((void*)bitmap);
+            }
        }
        else if (io->ddir == DDIR_WRITE)
        {
--- a/src/http_client.cpp
+++ b/src/http_client.cpp
@@ -15,13 +15,13 @@

 #include <stdexcept>

+#include "addr_util.h"
 #include "json11/json11.hpp"
 #include "http_client.h"
 #include "timerfd_manager.h"

 #define READ_BUFFER_SIZE 9000

-static int extract_port(std::string & host);
 static std::string trim(const std::string & in);
 static std::string ws_format_frame(int type, uint64_t size);
 static bool ws_parse_frame(std::string & buf, int & type, std::string & res);
@@ -185,19 +185,15 @@ http_co_t::~http_co_t()
 void http_co_t::start_connection()
 {
    stackin();
-    int port = extract_port(host);
-    struct sockaddr_in addr;
-    int r;
-    if ((r = inet_pton(AF_INET, host.c_str(), &addr.sin_addr)) != 1)
+    struct sockaddr addr;
+    if (!string_to_addr(host.c_str(), 1, 80, &addr))
    {
        parsed.error_code = ENXIO;
        stackout();
        end();
        return;
    }
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(port ? port : 80);
-    peer_fd = socket(AF_INET, SOCK_STREAM, 0);
+    peer_fd = socket(addr.sa_family, SOCK_STREAM, 0);
    if (peer_fd < 0)
    {
        parsed.error_code = errno;
@@ -219,7 +215,7 @@ void http_co_t::start_connection()
    }
    epoll_events = 0;
    // Finally call connect
-    r = ::connect(peer_fd, (sockaddr*)&addr, sizeof(addr));
+    int r = ::connect(peer_fd, (sockaddr*)&addr, sizeof(addr));
    if (r < 0 && errno != EINPROGRESS)
    {
        parsed.error_code = errno;
@@ -759,22 +755,6 @@ std::vector<std::string> getifaddr_list(json11::Json mask_cfg, bool include_v6)
    return addresses;
 }

-static int extract_port(std::string & host)
-{
-    int port = 0;
-    int pos = 0;
-    if ((pos = host.find(':')) >= 0)
-    {
-        port = strtoull(host.c_str() + pos + 1, NULL, 10);
-        if (port >= 0x10000)
-        {
-            port = 0;
-        }
-        host = host.substr(0, pos);
-    }
-    return port;
-}
-
 std::string strtolower(const std::string & in)
 {
    std::string s = in;
--- a/src/http_client.h
+++ b/src/http_client.h
@@ -45,7 +45,7 @@ struct websocket_t

 void parse_http_headers(std::string & res, http_response_t *parsed);

-std::vector<std::string> getifaddr_list(json11::Json mask_cfg = json11::Json(), bool include_v6 = false);
+std::vector<std::string> getifaddr_list(json11::Json mask_cfg = json11::Json(), bool include_v6 = true);

 uint64_t stoull_full(const std::string & str, int base = 10);

--- a/src/messenger.cpp
+++ b/src/messenger.cpp
@@ -4,10 +4,13 @@
 #include <unistd.h>
 #include <fcntl.h>
 #include <sys/socket.h>
+#include <sys/random.h>
 #include <sys/epoll.h>
 #include <netinet/tcp.h>
 #include <stdexcept>

+#include "base64.h"
+#include "addr_util.h"
 #include "messenger.h"

 void osd_messenger_t::init()
@@ -193,7 +196,7 @@ void osd_messenger_t::connect_peer(uint64_t peer_osd, json11::Json peer_state)
    try_connect_peer(peer_osd);
 }

-void osd_messenger_t::try_connect_peer(uint64_t peer_osd)
+void osd_messenger_t::try_connect_peer(osd_num_t peer_osd)
 {
    auto wp_it = wanted_peers.find(peer_osd);
    if (wp_it == wanted_peers.end() || wp_it->second.connecting ||
@@ -214,43 +217,75 @@ void osd_messenger_t::try_connect_peer(uint64_t peer_osd)
    wp.cur_addr = wp.address_list[wp.address_index].string_value();
    wp.cur_port = wp.port;
    wp.connecting = true;
-    try_connect_peer_addr(peer_osd, wp.cur_addr.c_str(), wp.cur_port);
+    try_connect_peer_addr(peer_osd, wp.cur_addr.c_str(), wp.cur_port, NULL, [this](osd_num_t peer_osd, int peer_fd)
+    {
+        if (peer_fd >= 0)
+            osd_peer_fds[peer_osd] = peer_fd;
+        on_connect_peer(peer_osd, peer_fd);
+    });
 }

-void osd_messenger_t::try_connect_peer_addr(osd_num_t peer_osd, const char *peer_host, int peer_port)
+static std::string urandom_str(int bytes)
+{
+    std::string str;
+    str.resize(bytes);
+    char *buf = (char*)str.data();
+    while (bytes > 0)
+    {
+        int r = getrandom(buf, bytes, 0);
+        if (r < 0)
+            throw std::runtime_error(std::string("getrandom: ") + strerror(errno));
+        buf += r;
+        bytes -= r;
+    }
+    return str;
+}
+
+void osd_messenger_t::try_connect_peer_addr(osd_num_t peer_osd, const char *peer_host, int peer_port,
+    osd_client_t *meta_cl, std::function<void(osd_num_t, int)> connect_callback)
 {
    assert(peer_osd != this->osd_num);
-    struct sockaddr_in addr;
-    int r;
-    if ((r = inet_pton(AF_INET, peer_host, &addr.sin_addr)) != 1)
+    struct sockaddr addr;
+    if (!meta_cl)
    {
-        on_connect_peer(peer_osd, -EINVAL);
-        return;
+        if (!string_to_addr(peer_host, 0, peer_port, &addr))
+        {
+            connect_callback(peer_osd, -EINVAL);
+            return;
+        }
+    }
+    else
+    {
+        addr = meta_cl->peer_addr;
+    }
+    int peer_fd = socket(addr.sa_family, SOCK_STREAM, 0);
+    if (peer_fd >= 0)
+    {
+        fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) | O_NONBLOCK);
+        int r = connect(peer_fd, (sockaddr*)&addr, sizeof(addr));
+        if (r < 0 && errno != EINPROGRESS)
+        {
+            close(peer_fd);
+            peer_fd = -1;
+        }
    }
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(peer_port ? peer_port : 11203);
-    int peer_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (peer_fd < 0)
    {
-        on_connect_peer(peer_osd, -errno);
-        return;
-    }
-    fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) | O_NONBLOCK);
-    r = connect(peer_fd, (sockaddr*)&addr, sizeof(addr));
-    if (r < 0 && errno != EINPROGRESS)
-    {
-        close(peer_fd);
-        on_connect_peer(peer_osd, -errno);
+        connect_callback(peer_osd, -errno);
        return;
    }
    clients[peer_fd] = new osd_client_t();
    clients[peer_fd]->peer_addr = addr;
-    clients[peer_fd]->peer_port = peer_port;
+    clients[peer_fd]->peer_port = ((struct sockaddr_in*)&addr)->sin_port;
    clients[peer_fd]->peer_fd = peer_fd;
    clients[peer_fd]->peer_state = PEER_CONNECTING;
    clients[peer_fd]->connect_timeout_id = -1;
+    clients[peer_fd]->connect_callback = connect_callback;
    clients[peer_fd]->osd_num = peer_osd;
    clients[peer_fd]->in_buf = malloc_or_die(receive_buffer_size);
+    clients[peer_fd]->data_for = meta_cl ? addr_to_string(meta_cl->peer_addr) : "";
+    clients[peer_fd]->data_connection_cookie = meta_cl
+        ? meta_cl->data_connection_cookie : base64_encode(urandom_str(12));
    tfd->set_fd_handler(peer_fd, true, [this](int peer_fd, int epoll_events)
    {
        // Either OUT (connected) or HUP
@@ -260,10 +295,12 @@ void osd_messenger_t::try_connect_peer_addr(osd_num_t peer_osd, const char *peer
    {
        clients[peer_fd]->connect_timeout_id = tfd->set_timer(1000*peer_connect_timeout, false, [this, peer_fd](int timer_id)
        {
-            osd_num_t peer_osd = clients.at(peer_fd)->osd_num;
+            auto cl = clients.at(peer_fd);
+            auto connect_callback = cl->connect_callback;
+            cl->connect_callback = NULL;
+            osd_num_t peer_osd = cl->osd_num;
            stop_client(peer_fd, true);
-            on_connect_peer(peer_osd, -EPIPE);
-            return;
+            connect_callback(peer_osd, -EPIPE);
        });
    }
 }
@@ -285,8 +322,10 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd)
    }
    if (result != 0)
    {
+        auto connect_callback = cl->connect_callback;
+        cl->connect_callback = NULL;
        stop_client(peer_fd, true);
-        on_connect_peer(peer_osd, -result);
+        connect_callback(peer_osd, -result);
        return;
    }
    int one = 1;
@@ -366,6 +405,11 @@ 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)
 {
+    json11::Json::object payload;
+    if (cl->data_connection_cookie != "")
+    {
+        payload["data_cookie"] = cl->data_connection_cookie;
+    }
    osd_op_t *op = new osd_op_t();
    op->op_type = OSD_OP_OUT;
    op->peer_fd = cl->peer_fd;
@@ -378,24 +422,33 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
            },
        },
    };
-#ifdef WITH_RDMA
-    if (rdma_context)
+    if (cl->data_for == "")
    {
-        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)
+#ifdef WITH_RDMA
+        if (rdma_context)
        {
-            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();
-            op->buf = malloc_or_die(payload_str.size());
-            op->iov.push_back(op->buf, payload_str.size());
-            memcpy(op->buf, payload_str.c_str(), payload_str.size());
+            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)
+            {
+                payload["connect_rdma"] = cl->rdma_conn->addr.to_string();
+                payload["rdma_max_msg"] = cl->rdma_conn->max_msg;
+            }
        }
-    }
 #endif
+    }
+    else
+    {
+        // Mark it as a data connection
+        payload["data_for"] = cl->data_for;
+    }
+    if (payload.size())
+    {
+        std::string payload_str = json11::Json(payload).dump();
+        op->req.show_conf.json_len = payload_str.size();
+        op->buf = malloc_or_die(payload_str.size());
+        op->iov.push_back(op->buf, payload_str.size());
+        memcpy(op->buf, payload_str.c_str(), payload_str.size());
+    }
    op->callback = [this, cl](osd_op_t *op)
    {
        std::string json_err;
@@ -428,18 +481,30 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
                    cl->osd_num, config["protocol_version"].uint64_value(), OSD_PROTOCOL_VERSION
                );
            }
+            else if (cl->data_for != "" && config["data_for"] != cl->data_for)
+            {
+                err = true;
+                fprintf(
+                    stderr, "OSD %lu does not support separate data connections."
+                    " Proceeding with a single connection\n", cl->osd_num
+                );
+            }
        }
        if (err)
        {
            osd_num_t peer_osd = cl->osd_num;
+            auto connect_callback = cl->connect_callback;
+            cl->connect_callback = NULL;
            stop_client(op->peer_fd);
-            on_connect_peer(peer_osd, -1);
+            connect_callback(peer_osd, -EINVAL);
            delete op;
            return;
        }
 #ifdef WITH_RDMA
-        if (config["rdma_address"].is_string())
+        if (rdma_context && cl->rdma_conn && config["rdma_address"].is_string())
        {
+            // Prevent creating data connection - we are trying RDMA
+            cl->data_connection_cookie = "";
            msgr_rdma_address_t addr;
            if (!msgr_rdma_address_t::from_string(config["rdma_address"].string_value().c_str(), &addr) ||
                cl->rdma_conn->connect(&addr) != 0)
@@ -452,8 +517,10 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
                cl->rdma_conn = NULL;
                // FIXME: Keep TCP connection in this case
                osd_num_t peer_osd = cl->osd_num;
+                auto connect_callback = cl->connect_callback;
+                cl->connect_callback = NULL;
                stop_client(cl->peer_fd);
-                on_connect_peer(peer_osd, -1);
+                connect_callback(peer_osd, -EPIPE);
                delete op;
                return;
            }
@@ -475,8 +542,37 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
            }
        }
 #endif
-        osd_peer_fds[cl->osd_num] = cl->peer_fd;
-        on_connect_peer(cl->osd_num, cl->peer_fd);
+        if (cl->data_connection_cookie != "")
+        {
+            // Try to open second connection to the same address
+            try_connect_peer_addr(cl->osd_num, NULL, 0, cl, [this, peer_fd = cl->peer_fd](osd_num_t data_peer, int data_peer_fd)
+            {
+                auto cl_it = clients.find(peer_fd);
+                if (cl_it != clients.end())
+                {
+                    // Proceed with or without the data connection
+                    auto cl = cl_it->second;
+                    if (data_peer_fd >= 0)
+                    {
+                        cl->data_connection_fd = data_peer_fd;
+                        auto data_cl = clients.at(data_peer_fd);
+                        data_cl->meta_connection_fd = cl->peer_fd;
+                    }
+                    osd_peer_fds[cl->osd_num] = cl->peer_fd;
+                    on_connect_peer(cl->osd_num, cl->peer_fd);
+                }
+                else if (data_peer_fd >= 0)
+                {
+                    stop_client(data_peer_fd);
+                }
+            });
+        }
+        else
+        {
+            auto connect_callback = cl->connect_callback;
+            cl->connect_callback = NULL;
+            connect_callback(cl->osd_num, cl->peer_fd);
+        }
        delete op;
    };
    outbox_push(op);
@@ -485,24 +581,24 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
 void osd_messenger_t::accept_connections(int listen_fd)
 {
    // Accept new connections
-    sockaddr_in addr;
+    sockaddr addr;
    socklen_t peer_addr_size = sizeof(addr);
    int peer_fd;
-    while ((peer_fd = accept(listen_fd, (sockaddr*)&addr, &peer_addr_size)) >= 0)
+    while ((peer_fd = accept(listen_fd, &addr, &peer_addr_size)) >= 0)
    {
        assert(peer_fd != 0);
-        char peer_str[256];
-        fprintf(stderr, "[OSD %lu] new client %d: connection from %s port %d\n", this->osd_num, peer_fd,
-            inet_ntop(AF_INET, &addr.sin_addr, peer_str, 256), ntohs(addr.sin_port));
+        fprintf(stderr, "[OSD %lu] new client %d: connection from %s\n", this->osd_num, peer_fd,
+            addr_to_string(addr).c_str());
        fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) | O_NONBLOCK);
        int one = 1;
        setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
        clients[peer_fd] = new osd_client_t();
        clients[peer_fd]->peer_addr = addr;
-        clients[peer_fd]->peer_port = ntohs(addr.sin_port);
+        clients[peer_fd]->peer_port = ntohs(((sockaddr_in*)&addr)->sin_port);
        clients[peer_fd]->peer_fd = peer_fd;
        clients[peer_fd]->peer_state = PEER_CONNECTED;
        clients[peer_fd]->in_buf = malloc_or_die(receive_buffer_size);
+        clients_by_addr[addr_to_string(addr)] = peer_fd;
        // Add FD to epoll
        tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
        {
--- a/src/messenger.h
+++ b/src/messenger.h
@@ -49,7 +49,7 @@ struct osd_client_t
 {
    int refs = 0;

-    sockaddr_in peer_addr;
+    sockaddr peer_addr;
    int peer_port;
    int peer_fd;
    int peer_state;
@@ -57,6 +57,10 @@ struct osd_client_t
    int ping_time_remaining = 0;
    int idle_time_remaining = 0;
    osd_num_t osd_num = 0;
+    std::function<void(osd_num_t, int)> connect_callback;
+
+    int data_connection_fd = -1, meta_connection_fd = -1;
+    std::string data_connection_cookie, data_for;

    void *in_buf = NULL;

@@ -148,6 +152,7 @@ public:
    osd_num_t osd_num;
    uint64_t next_subop_id = 1;
    std::map<int, osd_client_t*> clients;
+    std::map<std::string, int> clients_by_addr;
    std::map<osd_num_t, osd_wanted_peer_t> wanted_peers;
    std::map<uint64_t, int> osd_peer_fds;
    // op statistics
@@ -157,6 +162,7 @@ public:
    void parse_config(const json11::Json & config);
    void connect_peer(uint64_t osd_num, json11::Json peer_state);
    void stop_client(int peer_fd, bool force = false, bool force_delete = false);
+    void break_data_client_pair(osd_client_t *cl);
    void outbox_push(osd_op_t *cur_op);
    std::function<void(osd_op_t*)> exec_op;
    std::function<void(osd_num_t)> repeer_pgs;
@@ -174,7 +180,8 @@ public:

 protected:
    void try_connect_peer(uint64_t osd_num);
-    void try_connect_peer_addr(osd_num_t peer_osd, const char *peer_host, int peer_port);
+    void try_connect_peer_addr(osd_num_t peer_osd, const char *peer_host, int peer_port,
+        osd_client_t *meta_cl, std::function<void(osd_num_t, int)> connect_callback);
    void handle_peer_epoll(int peer_fd, int epoll_events);
    void handle_connect_epoll(int peer_fd);
    void on_connect_peer(osd_num_t peer_osd, int peer_fd);
--- a/src/msgr_stop.cpp
+++ b/src/msgr_stop.cpp
@@ -4,6 +4,7 @@
 #include <unistd.h>
 #include <assert.h>

+#include "addr_util.h"
 #include "messenger.h"

 void osd_messenger_t::cancel_osd_ops(osd_client_t *cl)
@@ -58,7 +59,8 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
    {
        if (cl->osd_num)
        {
-            fprintf(stderr, "[OSD %lu] Stopping client %d (OSD peer %lu)\n", osd_num, peer_fd, cl->osd_num);
+            fprintf(stderr, "[OSD %lu] Stopping client %d (OSD %speer %lu)\n",
+                osd_num, peer_fd, cl->meta_connection_fd >= 0 ? " data" : "", cl->osd_num);
        }
        else
        {
@@ -68,7 +70,7 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
    // First set state to STOPPED so another stop_client() call doesn't try to free it again
    cl->refs++;
    cl->peer_state = PEER_STOPPED;
-    if (cl->osd_num)
+    if (cl->osd_num && cl->meta_connection_fd < 0)
    {
        // ...and forget OSD peer
        osd_peer_fds.erase(cl->osd_num);
@@ -100,9 +102,17 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
 #endif
    if (cl->osd_num)
    {
-        // Then repeer PGs because cancel_op() callbacks can try to perform
-        // some actions and we need correct PG states to not do something silly
-        repeer_pgs(cl->osd_num);
+        if (cl->meta_connection_fd < 0)
+        {
+            // Then repeer PGs because cancel_op() callbacks can try to perform
+            // some actions and we need correct PG states to not do something silly
+            repeer_pgs(cl->osd_num);
+        }
+        else
+        {
+            // FIXME Try to re-establish data connection
+            // Only when the connection is outbound, but here it's always outbound
+        }
    }
    // Then cancel all operations
    if (cl->read_op)
@@ -128,6 +138,7 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
        delete cl->rdma_conn;
    }
 #endif
+    clients_by_addr.erase(addr_to_string(cl->peer_addr));
 #endif
    // Find the item again because it can be invalidated at this point
    it = clients.find(peer_fd);
@@ -135,9 +146,40 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
    {
        clients.erase(it);
    }
+    // Break metadata/data connection pair
+    if (cl->data_connection_fd >= 0)
+    {
+        // No sense to keep data connection when metadata connection is stopped
+        auto dc_it = clients.find(cl->data_connection_fd);
+        cl->data_connection_fd = -1;
+        if (dc_it != clients.end() && dc_it->second->meta_connection_fd == cl->peer_fd)
+        {
+            stop_client(dc_it->second->peer_fd);
+        }
+    }
+    break_data_client_pair(cl);
+    // Refcount and delete
    cl->refs--;
    if (cl->refs <= 0 || force_delete)
    {
        delete cl;
    }
 }
+
+void osd_messenger_t::break_data_client_pair(osd_client_t *cl)
+{
+    if (cl->meta_connection_fd >= 0)
+    {
+        auto dc_it = clients.find(cl->meta_connection_fd);
+        if (dc_it != clients.end() && dc_it->second->data_connection_fd == cl->peer_fd)
+            dc_it->second->data_connection_fd = -1;
+        cl->meta_connection_fd = -1;
+    }
+    if (cl->data_connection_fd >= 0)
+    {
+        auto dc_it = clients.find(cl->data_connection_fd);
+        if (dc_it != clients.end() && dc_it->second->meta_connection_fd == cl->peer_fd)
+            dc_it->second->meta_connection_fd = -1;
+        cl->data_connection_fd = -1;
+    }
+}
--- a/src/osd.cpp
+++ b/src/osd.cpp
@@ -7,6 +7,7 @@
 #include <netinet/tcp.h>
 #include <arpa/inet.h>

+#include "addr_util.h"
 #include "blockstore_impl.h"
 #include "osd_primary.h"
 #include "osd.h"
@@ -156,14 +157,6 @@ void osd_t::parse_config(const json11::Json & config)

 void osd_t::bind_socket()
 {
-    listen_fd = socket(AF_INET, SOCK_STREAM, 0);
-    if (listen_fd < 0)
-    {
-        throw std::runtime_error(std::string("socket: ") + strerror(errno));
-    }
-    int enable = 1;
-    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));
-
    if (config["osd_network"].is_string() ||
        config["osd_network"].is_array())
    {
@@ -173,7 +166,7 @@ void osd_t::bind_socket()
        else
            for (auto v: config["osd_network"].array_items())
                mask.push_back(v.string_value());
-        auto matched_addrs = getifaddr_list(mask, false);
+        auto matched_addrs = getifaddr_list(mask);
        if (matched_addrs.size() > 1)
        {
            fprintf(stderr, "More than 1 address matches requested network(s): %s\n", json11::Json(matched_addrs).dump().c_str());
@@ -192,17 +185,21 @@ void osd_t::bind_socket()

    // FIXME Support multiple listening sockets

-    sockaddr_in addr;
-    int r;
-    if ((r = inet_pton(AF_INET, bind_address.c_str(), &addr.sin_addr)) != 1)
+    sockaddr addr;
+    if (!string_to_addr(bind_address, 0, bind_port, &addr))
    {
-        close(listen_fd);
-        throw std::runtime_error("bind address "+bind_address+(r == 0 ? " is not valid" : ": no ipv4 support"));
+        throw std::runtime_error("bind address "+bind_address+" is not valid");
    }
-    addr.sin_family = AF_INET;

-    addr.sin_port = htons(bind_port);
-    if (bind(listen_fd, (sockaddr*)&addr, sizeof(addr)) < 0)
+    listen_fd = socket(addr.sa_family, SOCK_STREAM, 0);
+    if (listen_fd < 0)
+    {
+        throw std::runtime_error(std::string("socket: ") + strerror(errno));
+    }
+    int enable = 1;
+    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));
+
+    if (bind(listen_fd, &addr, sizeof(addr)) < 0)
    {
        close(listen_fd);
        throw std::runtime_error(std::string("bind: ") + strerror(errno));
@@ -215,7 +212,7 @@ void osd_t::bind_socket()
            close(listen_fd);
            throw std::runtime_error(std::string("getsockname: ") + strerror(errno));
        }
-        listening_port = ntohs(addr.sin_port);
+        listening_port = ntohs(((sockaddr_in*)&addr)->sin_port);
    }
    else
    {
--- a/src/osd.h
+++ b/src/osd.h
@@ -102,7 +102,7 @@ class osd_t
    bool no_rebalance = false;
    bool no_recovery = false;
    std::string bind_address;
-    int bind_port, listen_backlog;
+    int bind_port, listen_backlog = 128;
    // FIXME: Implement client queue depth limit
    int client_queue_depth = 128;
    bool allow_test_ops = false;
--- a/src/osd_secondary.cpp
+++ b/src/osd_secondary.cpp
@@ -178,6 +178,37 @@ void osd_t::exec_show_config(osd_op_t *cur_op)
            }
        }
    }
+    else
+    {
+#endif
+        if (req_json["data_for"].is_string())
+        {
+            auto cli = msgr.clients.at(cur_op->peer_fd);
+            auto md_it = msgr.clients_by_addr.find(req_json["data_for"].string_value());
+            if (md_it != msgr.clients_by_addr.end())
+            {
+                int md_peer_fd = md_it->second;
+                auto md_it = msgr.clients.find(md_peer_fd);
+                if (md_it != msgr.clients.end() && md_it->second->data_connection_cookie != "" &&
+                    req_json["data_cookie"].string_value() == md_it->second->data_connection_cookie)
+                {
+                    // Break previous metadata/data connections for both FDs, if present
+                    msgr.break_data_client_pair(cli);
+                    msgr.break_data_client_pair(md_it->second);
+                    // And setup the new pair
+                    cli->meta_connection_fd = md_it->second->peer_fd;
+                    md_it->second->data_connection_fd = cli->peer_fd;
+                    wire_config["data_for"] = req_json["data_for"];
+                }
+            }
+        }
+        else if (req_json["data_cookie"].is_string())
+        {
+            auto cli = msgr.clients.at(cur_op->peer_fd);
+            cli->data_connection_cookie = req_json["data_cookie"].string_value();
+        }
+#ifdef WITH_RDMA
+    }
 #endif
    if (cur_op->buf)
        free(cur_op->buf);
--- a/src/osd_test.cpp
+++ b/src/osd_test.cpp
@@ -16,6 +16,7 @@

 #include <stdexcept>

+#include "addr_util.h"
 #include "osd_ops.h"
 #include "rw_blocking.h"
 #include "test_pattern.h"
@@ -133,17 +134,14 @@ int main(int narg, char *args[])

 int connect_osd(const char *osd_address, int osd_port)
 {
-    struct sockaddr_in addr;
-    int r;
-    if ((r = inet_pton(AF_INET, osd_address, &addr.sin_addr)) != 1)
+    struct sockaddr addr;
+    if (!string_to_addr(osd_address, 0, osd_port, &addr))
    {
-        fprintf(stderr, "server address: %s%s\n", osd_address, r == 0 ? " is not valid" : ": no ipv4 support");
+        fprintf(stderr, "server address: %s is not valid\n", osd_address);
        return -1;
    }
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(osd_port);

-    int connect_fd = socket(AF_INET, SOCK_STREAM, 0);
+    int connect_fd = socket(addr.sa_family, SOCK_STREAM, 0);
    if (connect_fd < 0)
    {
        perror("socket");
--- a/src/rw_blocking.cpp
+++ b/src/rw_blocking.cpp
@@ -4,6 +4,8 @@
 #include <errno.h>
 #include <stdlib.h>
 #include <stdio.h>
+#include <sys/types.h>
+#include <sys/socket.h>

 #include "rw_blocking.h"

@@ -123,3 +125,41 @@ int writev_blocking(int fd, iovec *iov, int iovcnt)
    }
    return done;
 }
+
+int sendv_blocking(int fd, iovec *iov, int iovcnt, int flags)
+{
+    struct msghdr msg = { 0 };
+    int v = 0;
+    int done = 0;
+    while (v < iovcnt)
+    {
+        msg.msg_iov = iov+v;
+        msg.msg_iovlen = iovcnt-v;
+        ssize_t r = sendmsg(fd, &msg, flags);
+        if (r < 0)
+        {
+            if (errno != EAGAIN && errno != EPIPE)
+            {
+                perror("sendmsg");
+                exit(1);
+            }
+            continue;
+        }
+        done += r;
+        while (v < iovcnt)
+        {
+            if (iov[v].iov_len > r)
+            {
+                iov[v].iov_len -= r;
+                iov[v].iov_base += r;
+                break;
+            }
+            else
+            {
+                r -= iov[v].iov_len;
+                v++;
+            }
+        }
+    }
+    return done;
+}
--- a/src/rw_blocking.h
+++ b/src/rw_blocking.h
@@ -10,3 +10,4 @@ int read_blocking(int fd, void *read_buf, size_t remaining);
 int write_blocking(int fd, void *write_buf, size_t remaining);
 int readv_blocking(int fd, iovec *iov, int iovcnt);
 int writev_blocking(int fd, iovec *iov, int iovcnt);
+int sendv_blocking(int fd, iovec *iov, int iovcnt, int flags);
--- a/src/stub_bench.cpp
+++ b/src/stub_bench.cpp
@@ -21,6 +21,7 @@

 #include <stdexcept>

+#include "addr_util.h"
 #include "rw_blocking.h"
 #include "osd_ops.h"

@@ -66,16 +67,14 @@ int main(int narg, char *args[])

 int connect_stub(const char *server_address, int server_port)
 {
-    struct sockaddr_in addr;
-    int r;
-    if ((r = inet_pton(AF_INET, server_address, &addr.sin_addr)) != 1)
+    struct sockaddr addr;
+    if (!string_to_addr(server_address, 0, server_port, &addr))
    {
-        fprintf(stderr, "server address: %s%s\n", server_address, r == 0 ? " is not valid" : ": no ipv4 support");
+        fprintf(stderr, "server address: %s is not valid\n", server_address);
        return -1;
    }
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(server_port);
-    int connect_fd = socket(AF_INET, SOCK_STREAM, 0);
+
+    int connect_fd = socket(addr.sa_family, SOCK_STREAM, 0);
    if (connect_fd < 0)
    {
        perror("socket");
--- a/src/stub_osd.cpp
+++ b/src/stub_osd.cpp
@@ -37,10 +37,11 @@

 #include <stdexcept>

+#include "addr_util.h"
 #include "rw_blocking.h"
 #include "osd_ops.h"

-int bind_stub(const char *bind_address, int bind_port);
+int bind_stub(std::string bind_address, int bind_port);

 void run_stub(int peer_fd);

@@ -48,13 +49,13 @@ int main(int narg, char *args[])
 {
    int listen_fd = bind_stub("0.0.0.0", 11203);
    // Accept new connections
-    sockaddr_in addr;
+    sockaddr addr;
    socklen_t peer_addr_size = sizeof(addr);
    int peer_fd;
    while (1)
    {
        printf("stub_osd: waiting for 1 client\n");
-        peer_fd = accept(listen_fd, (sockaddr*)&addr, &peer_addr_size);
+        peer_fd = accept(listen_fd, &addr, &peer_addr_size);
        if (peer_fd == -1)
        {
            if (errno == EAGAIN)
@@ -62,9 +63,8 @@ int main(int narg, char *args[])
            else
                throw std::runtime_error(std::string("accept: ") + strerror(errno));
        }
-        char peer_str[256];
-        printf("stub_osd: new client %d: connection from %s port %d\n", peer_fd,
-            inet_ntop(AF_INET, &addr.sin_addr, peer_str, 256), ntohs(addr.sin_port));
+        printf("stub_osd: new client %d: connection from %s\n", peer_fd,
+            addr_to_string(addr).c_str());
        int one = 1;
        setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
        run_stub(peer_fd);
@@ -76,11 +76,17 @@ int main(int narg, char *args[])
    return 0;
 }

-int bind_stub(const char *bind_address, int bind_port)
+int bind_stub(std::string bind_address, int bind_port)
 {
    int listen_backlog = 128;

-    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
+    sockaddr addr;
+    if (!string_to_addr(bind_address, 0, bind_port, &addr))
+    {
+        throw std::runtime_error("bind address "+bind_address+" is not valid");
+    }
+
+    int listen_fd = socket(addr.sa_family, SOCK_STREAM, 0);
    if (listen_fd < 0)
    {
        throw std::runtime_error(std::string("socket: ") + strerror(errno));
@@ -88,17 +94,7 @@ int bind_stub(const char *bind_address, int bind_port)
    int enable = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));

-    sockaddr_in addr;
-    int r;
-    if ((r = inet_pton(AF_INET, bind_address, &addr.sin_addr)) != 1)
-    {
-        close(listen_fd);
-        throw std::runtime_error("bind address "+std::string(bind_address)+(r == 0 ? " is not valid" : ": no ipv4 support"));
-    }
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(bind_port);
-
-    if (bind(listen_fd, (sockaddr*)&addr, sizeof(addr)) < 0)
+    if (bind(listen_fd, &addr, sizeof(addr)) < 0)
    {
        close(listen_fd);
        throw std::runtime_error(std::string("bind: ") + strerror(errno));
--- a/src/stub_uring_osd.cpp
+++ b/src/stub_uring_osd.cpp
@@ -20,11 +20,12 @@

 #include <stdexcept>

+#include "addr_util.h"
 #include "ringloop.h"
 #include "epoll_manager.h"
 #include "messenger.h"

-int bind_stub(const char *bind_address, int bind_port);
+int bind_stub(std::string bind_address, int bind_port);

 void stub_exec_op(osd_messenger_t *msgr, osd_op_t *op);

@@ -66,11 +67,17 @@ int main(int narg, char *args[])
    return 0;
 }

-int bind_stub(const char *bind_address, int bind_port)
+int bind_stub(std::string bind_address, int bind_port)
 {
    int listen_backlog = 128;

-    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
+    sockaddr addr;
+    if (!string_to_addr(bind_address, 0, bind_port, &addr))
+    {
+        throw std::runtime_error("bind address "+bind_address+" is not valid");
+    }
+
+    int listen_fd = socket(addr.sa_family, SOCK_STREAM, 0);
    if (listen_fd < 0)
    {
        throw std::runtime_error(std::string("socket: ") + strerror(errno));
@@ -78,17 +85,7 @@ int bind_stub(const char *bind_address, int bind_port)
    int enable = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));

-    sockaddr_in addr;
-    int r;
-    if ((r = inet_pton(AF_INET, bind_address, &addr.sin_addr)) != 1)
-    {
-        close(listen_fd);
-        throw std::runtime_error("bind address "+std::string(bind_address)+(r == 0 ? " is not valid" : ": no ipv4 support"));
-    }
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(bind_port);
-
-    if (bind(listen_fd, (sockaddr*)&addr, sizeof(addr)) < 0)
+    if (bind(listen_fd, &addr, sizeof(addr)) < 0)
    {
        close(listen_fd);
        throw std::runtime_error(std::string("bind: ") + strerror(errno));
@@ -113,8 +110,10 @@ void stub_exec_op(osd_messenger_t *msgr, osd_op_t *op)
    if (op->req.hdr.opcode == OSD_OP_SEC_READ)
    {
        op->reply.hdr.retval = op->req.sec_rw.len;
-        op->buf = malloc(op->req.sec_rw.len);
+        op->buf = memalign_or_die(MEM_ALIGNMENT, op->req.sec_rw.len);
        op->iov.push_back(op->buf, op->req.sec_rw.len);
+        op->reply.sec_rw.attr_len = 4;
+        op->bitmap = op->buf;
    }
    else if (op->req.hdr.opcode == OSD_OP_SEC_WRITE || op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE)
    {
--- a/src/vitastor.pc.in
+++ b/src/vitastor.pc.in
@@ -6,7 +6,7 @@ includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@

 Name: Vitastor
 Description: Vitastor client library
-Version: 0.6.10
+Version: 0.6.11
 Libs: -L${libdir} -lvitastor_client
 Cflags: -I${includedir}

--- a/tests/run_3osds.sh
+++ b/tests/run_3osds.sh
@@ -5,6 +5,7 @@
 OSD_SIZE=${OSD_SIZE:-1024}
 PG_COUNT=${PG_COUNT:-1}
 PG_SIZE=${PG_SIZE:-3}
+PG_MINSIZE=${PG_MINSIZE:-2}
 OSD_COUNT=${OSD_COUNT:-3}
 SCHEME=${SCHEME:-ec}

@@ -25,9 +26,9 @@ if [ -n "$GLOBAL_CONF" ]; then
 fi

 if [ "$SCHEME" = "replicated" ]; then
-    $ETCDCTL put /vitastor/config/pools '{"1":{"name":"testpool","scheme":"replicated","pg_size":'$PG_SIZE',"pg_minsize":'$((PG_SIZE-1))',"pg_count":'$PG_COUNT',"failure_domain":"osd"}}'
+    $ETCDCTL put /vitastor/config/pools '{"1":{"name":"testpool","scheme":"replicated","pg_size":'$PG_SIZE',"pg_minsize":'$PG_MINSIZE',"pg_count":'$PG_COUNT',"failure_domain":"osd"}}'
 else
-    $ETCDCTL put /vitastor/config/pools '{"1":{"name":"testpool","scheme":"xor","pg_size":'$PG_SIZE',"pg_minsize":'$((PG_SIZE-1))',"parity_chunks":1,"pg_count":'$PG_COUNT',"failure_domain":"osd"}}'
+    $ETCDCTL put /vitastor/config/pools '{"1":{"name":"testpool","scheme":"xor","pg_size":'$PG_SIZE',"pg_minsize":'$PG_MINSIZE',"parity_chunks":1,"pg_count":'$PG_COUNT',"failure_domain":"osd"}}'
 fi

 sleep 2
--- a/tests/test_minsize_1.sh
+++ b/tests/test_minsize_1.sh
@@ -0,0 +1,17 @@
+#!/bin/bash -ex
+
+PG_MINSIZE=1
+SCHEME=replicated
+
+. `dirname $0`/run_3osds.sh
+
+kill -INT $OSD1_PID
+kill -INT $OSD2_PID
+
+sleep 5
+
+if ! ($ETCDCTL get /vitastor/pg/state/1/ --prefix --print-value-only | jq -s -e '[ .[] | select(.state == ["active", "degraded"]) ] | length == '$PG_COUNT); then
+    format_error "FAILED: $PG_COUNT PG(s) NOT ACTIVE+DEGRADED"
+fi
+
+format_green OK
Author	SHA1	Message	Date
Vitaliy Filippov	5fab6fc5ed	WIP Separate data connections	2021-12-30 01:57:13 +03:00
Vitaliy Filippov	ec2852c598	Add minsize_1 test	2021-12-28 10:54:36 +03:00
Vitaliy Filippov	b9f5c2a823	Support zero-copy send in fio_sec_osd to allow testing it Prelimilary results: - CPU usage drops significantly. For example, in T1Q8 128K write test against stub_uring_osd with 10G network and Athlon X4 860k CPU it drops from 100% to 30% - Latency becomes slightly worse. In T1Q1 4K write test in the same environment latency increases from 56 to 63 us. - Small write throughput also becomes slightly worse. In T1Q128 4K write test against stub iops decreases from 138k to ~110k (unstable, fluctuates 100k..120k). Note that this is without io_uring, of course.	2021-12-27 02:12:44 +03:00
Vitaliy Filippov	e9d2f79aa7	Support reading bitmaps in fio_sec_osd	2021-12-27 02:12:44 +03:00
Vitaliy Filippov	0785bdf8b3	Release 0.6.11 - Slightly reduce journaling write amplification (requires no_same_sector_overwrites=false) - Fix listen_backlog (it was 0) because it could more than halve OSD socket send speed - Support IPv6 OSD addresses - Do not try to initialize client in simple-offsets - Fix OSDs sometimes marking PGs incomplete instead of trying to connect with peers - Allow to configure OSD placement in node_placement - Allow to run with 4k sector size block devices. Natural, but it was forbidden	2021-12-26 21:11:24 +03:00
Vitaliy Filippov	b57e44748b	Send 4 byte bitmap in stub_uring_osd	2021-12-25 11:38:13 +03:00
Vitaliy Filippov	1bbe62f29c	Fix uninitialized listen_backlog which was leading to REALLY SLOW send speeds!!!	2021-12-25 11:38:13 +03:00
Vitaliy Filippov	20a4406acc	Support IPv6 OSD addresses	2021-12-19 10:42:17 +03:00
Vitaliy Filippov	f93491bc6c	Implement journal write batching and slightly refactor journal writes Slightly reduces WA. For example, in 4K T1Q128 replicated randwrite tests WA is reduced from ~3.6 to ~3.1, in T1Q64 from ~3.8 to ~3.4. Only effective without no_same_sector_overwrites.	2021-12-16 00:27:17 +03:00
Vitaliy Filippov	999bed8514	Fix opening regular files as blockstore	2021-12-15 02:08:58 +03:00
Vitaliy Filippov	3f33095fd7	Do not try to initialize client in simple-offsets	2021-12-15 02:07:27 +03:00