Double-check for double-alloc bugs

CMakeLists.txt

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

csi/Makefile

@@ -1,4 +1,4 @@
-VERSION ?= v0.6.11
+VERSION ?= v0.6.10
 
 all: build push
 

csi/deploy/004-csi-nodeplugin.yaml

@@ -49,7 +49,7 @@ spec:
             capabilities:
               add: ["SYS_ADMIN"]
             allowPrivilegeEscalation: true
-          image: vitalif/vitastor-csi:v0.6.11
+          image: vitalif/vitastor-csi:v0.6.10
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/deploy/007-csi-provisioner.yaml

@@ -116,7 +116,7 @@ spec:
             privileged: true
             capabilities:
               add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v0.6.11
+          image: vitalif/vitastor-csi:v0.6.10
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/src/config.go

@@ -5,7 +5,7 @@ package vitastor
 
 const (
     vitastorCSIDriverName    = "csi.vitastor.io"
-    vitastorCSIDriverVersion = "0.6.11"
+    vitastorCSIDriverVersion = "0.6.10"
 )
 
 // Config struct fills the parameters of request or user input

debian/changelog

@@ -1,4 +1,4 @@
-vitastor (0.6.11-1) unstable; urgency=medium
+vitastor (0.6.10-1) unstable; urgency=medium
 
   * RDMA support
   * Bugfixes

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.11; \
-    cd vitastor-0.6.11; \
+    cp -r /root/vitastor vitastor-0.6.10; \
+    cd vitastor-0.6.10; \
     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.11.orig.tar.xz vitastor-0.6.11; \
-    cd vitastor-0.6.11; \
+    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; \
     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; \

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.11'
+VERSION = '0.6.10'
 
 LOG = logging.getLogger(__name__)
 

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.11/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.6.11$(rpm --eval '%dist').tar.gz *
+tar --transform 's#^#vitastor-0.6.10/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.6.10$(rpm --eval '%dist').tar.gz *

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.11.el7.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.6.10.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.11
+Version:        0.6.10
 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.11.el7.tar.gz
+Source0:        vitastor-0.6.10.el7.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

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.11.el8.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.6.10.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.11
+Version:        0.6.10
 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.11.el8.tar.gz
+Source0:        vitastor-0.6.10.el8.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

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.11")
+add_definitions(-DVERSION="0.6.10")
 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 addr_util.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 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,7 +112,6 @@ 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
@@ -190,11 +189,11 @@ endif (${WITH_QEMU})
 ### Test stubs
 
 # stub_osd, stub_bench, osd_test
-add_executable(stub_osd stub_osd.cpp rw_blocking.cpp addr_util.cpp)
+add_executable(stub_osd stub_osd.cpp rw_blocking.cpp)
 target_link_libraries(stub_osd tcmalloc_minimal)
-add_executable(stub_bench stub_bench.cpp rw_blocking.cpp addr_util.cpp)
+add_executable(stub_bench stub_bench.cpp rw_blocking.cpp)
 target_link_libraries(stub_bench tcmalloc_minimal)
-add_executable(osd_test osd_test.cpp rw_blocking.cpp addr_util.cpp)
+add_executable(osd_test osd_test.cpp rw_blocking.cpp)
 target_link_libraries(osd_test tcmalloc_minimal)
 
 # osd_rmw_test

src/addr_util.cpp

@@ -1,60 +0,0 @@
-#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);
-}

src/addr_util.h

@@ -1,7 +0,0 @@
-#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);

src/blockstore_impl.cpp

@@ -235,12 +235,6 @@ 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;

src/blockstore_impl.h

@@ -54,14 +54,6 @@
 #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)
@@ -178,7 +170,7 @@ struct blockstore_op_private_t
     std::vector<fulfill_read_t> read_vec;
 
     // Sync, write
-    int min_flushed_journal_sector, max_flushed_journal_sector;
+    uint64_t min_flushed_journal_sector, max_flushed_journal_sector;
 
     // Write
     struct iovec iov_zerofill[3];
@@ -291,10 +283,6 @@ 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;
@@ -322,18 +310,21 @@ 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

src/blockstore_journal.cpp

@@ -153,73 +153,22 @@ journal_entry* prefill_single_journal_entry(journal_t & journal, uint16_t type,
     return je;
 }
 
-void blockstore_impl_t::prepare_journal_sector_write(int cur_sector, blockstore_op_t *op)
+void prepare_journal_sector_write(journal_t & journal, int cur_sector, io_uring_sqe *sqe, std::function<void(ring_data_t*)> cb)
 {
-    // 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;
-    // 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.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
+    );
 }
 
 journal_t::~journal_t()

src/blockstore_journal.h

@@ -4,7 +4,6 @@
 #pragma once
 
 #include "crc32c.h"
-#include <set>
 
 #define MIN_JOURNAL_SIZE 4*1024*1024
 #define JOURNAL_MAGIC 0x4A33
@@ -146,21 +145,8 @@ 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;
@@ -186,9 +172,6 @@ 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
@@ -217,3 +200,5 @@ 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);

src/blockstore_open.cpp

@@ -306,10 +306,6 @@ 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))
     {

src/blockstore_rollback.cpp

@@ -74,17 +74,24 @@ skip_ov:
     {
         return 0;
     }
-    // There is sufficient space. Check SQEs
-    BS_SUBMIT_CHECK_SQES(space_check.sectors_to_write);
+    // 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]);
+    }
     // Prepare and submit journal entries
-    int s = 0;
+    auto cb = [this, op](ring_data_t *data) { handle_rollback_event(data, op); };
+    int s = 0, cur_sector = -1;
     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)
         {
-            prepare_journal_sector_write(journal.cur_sector, op);
-            s++;
+            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;
         }
         journal_entry_rollback *je = (journal_entry_rollback*)
             prefill_single_journal_entry(journal, JE_ROLLBACK, sizeof(journal_entry_rollback));
@@ -93,9 +100,12 @@ skip_ov:
         je->crc32 = je_crc32((journal_entry*)je);
         journal.crc32_last = je->crc32;
     }
-    prepare_journal_sector_write(journal.cur_sector, op);
-    s++;
+    prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
     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;
 }
@@ -104,23 +114,30 @@ 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 == 4)
-        goto resume_4;
+    else if (PRIV(op)->op_state == 3)
+        goto resume_3;
+    else if (PRIV(op)->op_state == 5)
+        goto resume_5;
     else
         return 1;
 resume_2:
+    // Release used journal sectors
+    release_journal_sectors(op);
+resume_3:
     if (!disable_journal_fsync)
     {
-        BS_SUBMIT_GET_SQE(sqe, data);
+        io_uring_sqe *sqe;
+        BS_SUBMIT_GET_SQE_DECL(sqe);
+        ring_data_t *data = ((ring_data_t*)sqe->user_data);
         my_uring_prep_fsync(sqe, journal.fd, IORING_FSYNC_DATASYNC);
         data->iov = { 0 };
-        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
+        data->callback = [this, op](ring_data_t *data) { handle_rollback_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 = 3;
+        PRIV(op)->op_state = 4;
         return 1;
     }
-resume_4:
+resume_5:
     obj_ver_id* v;
     int i;
     for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
@@ -179,6 +196,24 @@ 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)

src/blockstore_stable.cpp

@@ -97,18 +97,25 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
     {
         return 0;
     }
-    // There is sufficient space. Check SQEs
-    BS_SUBMIT_CHECK_SQES(space_check.sectors_to_write);
+    // 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]);
+    }
     // Prepare and submit journal entries
-    int s = 0;
+    auto cb = [this, op](ring_data_t *data) { handle_stable_event(data, op); };
+    int s = 0, cur_sector = -1;
     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)
         {
-            prepare_journal_sector_write(journal.cur_sector, op);
-            s++;
+            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;
         }
         journal_entry_stable *je = (journal_entry_stable*)
             prefill_single_journal_entry(journal, JE_STABLE, sizeof(journal_entry_stable));
@@ -117,9 +124,12 @@ 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.cur_sector, op);
-    s++;
+    prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
     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;
 }
@@ -128,23 +138,30 @@ 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 == 4)
-        goto resume_4;
+    else if (PRIV(op)->op_state == 3)
+        goto resume_3;
+    else if (PRIV(op)->op_state == 5)
+        goto resume_5;
     else
         return 1;
 resume_2:
+    // Release used journal sectors
+    release_journal_sectors(op);
+resume_3:
     if (!disable_journal_fsync)
     {
-        BS_SUBMIT_GET_SQE(sqe, data);
+        io_uring_sqe *sqe;
+        BS_SUBMIT_GET_SQE_DECL(sqe);
+        ring_data_t *data = ((ring_data_t*)sqe->user_data);
         my_uring_prep_fsync(sqe, journal.fd, IORING_FSYNC_DATASYNC);
         data->iov = { 0 };
-        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
+        data->callback = [this, op](ring_data_t *data) { handle_stable_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 = 3;
+        PRIV(op)->op_state = 4;
         return 1;
     }
-resume_4:
+resume_5:
     // Mark dirty_db entries as stable, acknowledge op completion
     obj_ver_id* v;
     int i;
@@ -240,3 +257,21 @@ 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();
+    }
+}

src/blockstore_sync.cpp

@@ -44,8 +44,10 @@ 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_CHECK_SQES(1);
-            prepare_journal_sector_write(journal.cur_sector, op);
+            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;
             PRIV(op)->op_state = SYNC_JOURNAL_WRITE_SENT;
             return 1;
         }
@@ -62,7 +64,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_write_event(data, op); };
+            data->callback = [this, op](ring_data_t *data) { handle_sync_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;
@@ -83,18 +85,24 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op, bool queue_has_in_prog
         {
             return 0;
         }
-        // Check SQEs. Don't bother about merging, submit each journal sector as a separate request
-        BS_SUBMIT_CHECK_SQES(space_check.sectors_to_write);
+        // 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]);
+        }
         // Prepare and submit journal entries
         auto it = PRIV(op)->sync_big_writes.begin();
-        int s = 0;
+        int s = 0, cur_sector = -1;
         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)
             {
-                prepare_journal_sector_write(journal.cur_sector, op);
-                s++;
+                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;
             }
             auto & dirty_entry = dirty_db.at(*it);
             journal_entry_big_write *je = (journal_entry_big_write*)prefill_single_journal_entry(
@@ -121,9 +129,12 @@ 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.cur_sector, op);
-        s++;
+        prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], [this, op](ring_data_t *data) { handle_sync_event(data, op); });
         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;
     }
@@ -134,7 +145,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_write_event(data, op); };
+            data->callback = [this, op](ring_data_t *data) { handle_sync_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;
@@ -153,6 +164,42 @@ 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

src/blockstore_write.cpp

@@ -268,8 +268,18 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             cancel_all_writes(op, dirty_it, -ENOSPC);
             return 2;
         }
-        BS_SUBMIT_GET_SQE(sqe, data);
+#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);
         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
@@ -324,21 +334,29 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         {
             return 0;
         }
-        // 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
-            (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++;
+        // 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))
+        {
+            // 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);
+        }
         // 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 (!journal.entry_fits(sizeof(journal_entry_small_write) + clean_entry_bitmap_size))
+            if (sqe1)
             {
-                prepare_journal_sector_write(journal.cur_sector, op);
+                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++;
             }
             else
             {
@@ -372,7 +390,9 @@ 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.cur_sector, op);
+            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++;
         }
         if (op->len > 0)
         {
@@ -382,7 +402,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
                 // Copy data
                 memcpy(journal.buffer + journal.next_free, op->buf, op->len);
             }
-            BS_SUBMIT_GET_SQE(sqe2, data2);
+            ring_data_t *data2 = ((ring_data_t*)sqe2->user_data);
             data2->iov = (struct iovec){ op->buf, op->len };
             data2->callback = cb;
             my_uring_prep_writev(
@@ -431,12 +451,13 @@ 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
@@ -458,7 +479,10 @@ 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.cur_sector, op);
+        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;
         PRIV(op)->op_state = 3;
         return 1;
     }
@@ -573,7 +597,6 @@ 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);
@@ -591,6 +614,7 @@ 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
@@ -630,19 +654,23 @@ int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
         return 0;
     }
     write_iodepth++;
-    // 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
-    );
+    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); };
     // Prepare journal sector write
     if (immediate_commit == IMMEDIATE_NONE)
     {
-        if ((journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_del) &&
-            journal.sector_info[journal.cur_sector].dirty)
+        if (sqe)
         {
-            prepare_journal_sector_write(journal.cur_sector, op);
+            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++;
         }
         else
         {
@@ -669,7 +697,9 @@ 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.cur_sector, op);
+        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++;
     }
     if (!PRIV(op)->pending_ops)
     {

src/cli.cpp

@@ -305,10 +305,6 @@ 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();

src/fio_sec_osd.cpp

@@ -28,7 +28,6 @@
 #include <vector>
 #include <unordered_map>
 
-#include "addr_util.h"
 #include "rw_blocking.h"
 #include "osd_ops.h"
 #include "fio_headers.h"
@@ -153,14 +152,17 @@ 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;
 
-    sockaddr addr;
-    if (!string_to_addr(std::string(o->host ? o->host : "127.0.0.1"), false, o->port > 0 ? o->port : 11203, &addr))
+    struct sockaddr_in addr;
+    int r;
+    if ((r = inet_pton(AF_INET, o->host ? o->host : "127.0.0.1", &addr.sin_addr)) != 1)
     {
-        fprintf(stderr, "server address: %s is not valid\n", o->host ? o->host : "127.0.0.1");
+        fprintf(stderr, "server address: %s%s\n", o->host ? o->host : "127.0.0.1", r == 0 ? " is not valid" : ": no ipv4 support");
         return 1;
     }
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(o->port ? o->port : 11203);
 
-    bsd->connect_fd = socket(addr.sa_family, SOCK_STREAM, 0);
+    bsd->connect_fd = socket(AF_INET, SOCK_STREAM, 0);
     if (bsd->connect_fd < 0)
     {
         perror("socket");

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,15 +185,19 @@ http_co_t::~http_co_t()
 void http_co_t::start_connection()
 {
     stackin();
-    struct sockaddr addr;
-    if (!string_to_addr(host.c_str(), 1, 80, &addr))
+    int port = extract_port(host);
+    struct sockaddr_in addr;
+    int r;
+    if ((r = inet_pton(AF_INET, host.c_str(), &addr.sin_addr)) != 1)
     {
         parsed.error_code = ENXIO;
         stackout();
         end();
         return;
     }
-    peer_fd = socket(addr.sa_family, SOCK_STREAM, 0);
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(port ? port : 80);
+    peer_fd = socket(AF_INET, SOCK_STREAM, 0);
     if (peer_fd < 0)
     {
         parsed.error_code = errno;
@@ -215,7 +219,7 @@ void http_co_t::start_connection()
     }
     epoll_events = 0;
     // Finally call connect
-    int r = ::connect(peer_fd, (sockaddr*)&addr, sizeof(addr));
+    r = ::connect(peer_fd, (sockaddr*)&addr, sizeof(addr));
     if (r < 0 && errno != EINPROGRESS)
     {
         parsed.error_code = errno;
@@ -755,6 +759,22 @@ 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;

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 = true);
+std::vector<std::string> getifaddr_list(json11::Json mask_cfg = json11::Json(), bool include_v6 = false);
 
 uint64_t stoull_full(const std::string & str, int base = 10);
 

src/messenger.cpp

@@ -8,7 +8,6 @@
 #include <netinet/tcp.h>
 #include <stdexcept>
 
-#include "addr_util.h"
 #include "messenger.h"
 
 void osd_messenger_t::init()
@@ -221,20 +220,23 @@ void osd_messenger_t::try_connect_peer(uint64_t peer_osd)
 void osd_messenger_t::try_connect_peer_addr(osd_num_t peer_osd, const char *peer_host, int peer_port)
 {
     assert(peer_osd != this->osd_num);
-    struct sockaddr addr;
-    if (!string_to_addr(peer_host, 0, peer_port, &addr))
+    struct sockaddr_in addr;
+    int r;
+    if ((r = inet_pton(AF_INET, peer_host, &addr.sin_addr)) != 1)
     {
         on_connect_peer(peer_osd, -EINVAL);
         return;
     }
-    int peer_fd = socket(addr.sa_family, SOCK_STREAM, 0);
+    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);
-    int r = connect(peer_fd, (sockaddr*)&addr, sizeof(addr));
+    r = connect(peer_fd, (sockaddr*)&addr, sizeof(addr));
     if (r < 0 && errno != EINPROGRESS)
     {
         close(peer_fd);
@@ -483,20 +485,21 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
 void osd_messenger_t::accept_connections(int listen_fd)
 {
     // Accept new connections
-    sockaddr addr;
+    sockaddr_in addr;
     socklen_t peer_addr_size = sizeof(addr);
     int peer_fd;
-    while ((peer_fd = accept(listen_fd, &addr, &peer_addr_size)) >= 0)
+    while ((peer_fd = accept(listen_fd, (sockaddr*)&addr, &peer_addr_size)) >= 0)
     {
         assert(peer_fd != 0);
-        fprintf(stderr, "[OSD %lu] new client %d: connection from %s\n", this->osd_num, peer_fd,
-            addr_to_string(addr).c_str());
+        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));
         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(((sockaddr_in*)&addr)->sin_port);
+        clients[peer_fd]->peer_port = ntohs(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);

src/messenger.h

@@ -49,7 +49,7 @@ struct osd_client_t
 {
     int refs = 0;
 
-    sockaddr peer_addr;
+    sockaddr_in peer_addr;
     int peer_port;
     int peer_fd;
     int peer_state;

src/osd.cpp

@@ -7,7 +7,6 @@
 #include <netinet/tcp.h>
 #include <arpa/inet.h>
 
-#include "addr_util.h"
 #include "blockstore_impl.h"
 #include "osd_primary.h"
 #include "osd.h"
@@ -157,6 +156,14 @@ 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())
     {
@@ -166,7 +173,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);
+        auto matched_addrs = getifaddr_list(mask, false);
         if (matched_addrs.size() > 1)
         {
             fprintf(stderr, "More than 1 address matches requested network(s): %s\n", json11::Json(matched_addrs).dump().c_str());
@@ -185,21 +192,17 @@ void osd_t::bind_socket()
 
     // FIXME Support multiple listening sockets
 
-    sockaddr addr;
-    if (!string_to_addr(bind_address, 0, bind_port, &addr))
+    sockaddr_in addr;
+    int r;
+    if ((r = inet_pton(AF_INET, bind_address.c_str(), &addr.sin_addr)) != 1)
     {
-        throw std::runtime_error("bind address "+bind_address+" is not valid");
+        close(listen_fd);
+        throw std::runtime_error("bind address "+bind_address+(r == 0 ? " is not valid" : ": no ipv4 support"));
     }
+    addr.sin_family = AF_INET;
 
-    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)
+    addr.sin_port = htons(bind_port);
+    if (bind(listen_fd, (sockaddr*)&addr, sizeof(addr)) < 0)
     {
         close(listen_fd);
         throw std::runtime_error(std::string("bind: ") + strerror(errno));
@@ -212,7 +215,7 @@ void osd_t::bind_socket()
             close(listen_fd);
             throw std::runtime_error(std::string("getsockname: ") + strerror(errno));
         }
-        listening_port = ntohs(((sockaddr_in*)&addr)->sin_port);
+        listening_port = ntohs(addr.sin_port);
     }
     else
     {

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 = 128;
+    int bind_port, listen_backlog;
     // FIXME: Implement client queue depth limit
     int client_queue_depth = 128;
     bool allow_test_ops = false;

src/osd_test.cpp

@@ -16,7 +16,6 @@
 
 #include <stdexcept>
 
-#include "addr_util.h"
 #include "osd_ops.h"
 #include "rw_blocking.h"
 #include "test_pattern.h"
@@ -134,14 +133,17 @@ int main(int narg, char *args[])
 
 int connect_osd(const char *osd_address, int osd_port)
 {
-    struct sockaddr addr;
-    if (!string_to_addr(osd_address, 0, osd_port, &addr))
+    struct sockaddr_in addr;
+    int r;
+    if ((r = inet_pton(AF_INET, osd_address, &addr.sin_addr)) != 1)
     {
-        fprintf(stderr, "server address: %s is not valid\n", osd_address);
+        fprintf(stderr, "server address: %s%s\n", osd_address, r == 0 ? " is not valid" : ": no ipv4 support");
         return -1;
     }
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(osd_port);
 
-    int connect_fd = socket(addr.sa_family, SOCK_STREAM, 0);
+    int connect_fd = socket(AF_INET, SOCK_STREAM, 0);
     if (connect_fd < 0)
     {
         perror("socket");

src/stub_bench.cpp

@@ -21,7 +21,6 @@
 
 #include <stdexcept>
 
-#include "addr_util.h"
 #include "rw_blocking.h"
 #include "osd_ops.h"
 
@@ -67,14 +66,16 @@ int main(int narg, char *args[])
 
 int connect_stub(const char *server_address, int server_port)
 {
-    struct sockaddr addr;
-    if (!string_to_addr(server_address, 0, server_port, &addr))
+    struct sockaddr_in addr;
+    int r;
+    if ((r = inet_pton(AF_INET, server_address, &addr.sin_addr)) != 1)
     {
-        fprintf(stderr, "server address: %s is not valid\n", server_address);
+        fprintf(stderr, "server address: %s%s\n", server_address, r == 0 ? " is not valid" : ": no ipv4 support");
         return -1;
     }
-
-    int connect_fd = socket(addr.sa_family, SOCK_STREAM, 0);
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(server_port);
+    int connect_fd = socket(AF_INET, SOCK_STREAM, 0);
     if (connect_fd < 0)
     {
         perror("socket");

src/stub_osd.cpp

@@ -37,11 +37,10 @@
 
 #include <stdexcept>
 
-#include "addr_util.h"
 #include "rw_blocking.h"
 #include "osd_ops.h"
 
-int bind_stub(std::string bind_address, int bind_port);
+int bind_stub(const char *bind_address, int bind_port);
 
 void run_stub(int peer_fd);
 
@@ -49,13 +48,13 @@ int main(int narg, char *args[])
 {
     int listen_fd = bind_stub("0.0.0.0", 11203);
     // Accept new connections
-    sockaddr addr;
+    sockaddr_in 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, &addr, &peer_addr_size);
+        peer_fd = accept(listen_fd, (sockaddr*)&addr, &peer_addr_size);
         if (peer_fd == -1)
         {
             if (errno == EAGAIN)
@@ -63,8 +62,9 @@ int main(int narg, char *args[])
             else
                 throw std::runtime_error(std::string("accept: ") + strerror(errno));
         }
-        printf("stub_osd: new client %d: connection from %s\n", peer_fd,
-            addr_to_string(addr).c_str());
+        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));
         int one = 1;
         setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
         run_stub(peer_fd);
@@ -76,17 +76,11 @@ int main(int narg, char *args[])
     return 0;
 }
 
-int bind_stub(std::string bind_address, int bind_port)
+int bind_stub(const char *bind_address, int bind_port)
 {
     int listen_backlog = 128;
 
-    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);
+    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
     if (listen_fd < 0)
     {
         throw std::runtime_error(std::string("socket: ") + strerror(errno));
@@ -94,7 +88,17 @@ int bind_stub(std::string bind_address, int bind_port)
     int enable = 1;
     setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));
 
-    if (bind(listen_fd, &addr, sizeof(addr)) < 0)
+    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)
     {
         close(listen_fd);
         throw std::runtime_error(std::string("bind: ") + strerror(errno));

src/stub_uring_osd.cpp

@@ -20,12 +20,11 @@
 
 #include <stdexcept>
 
-#include "addr_util.h"
 #include "ringloop.h"
 #include "epoll_manager.h"
 #include "messenger.h"
 
-int bind_stub(std::string bind_address, int bind_port);
+int bind_stub(const char *bind_address, int bind_port);
 
 void stub_exec_op(osd_messenger_t *msgr, osd_op_t *op);
 
@@ -67,17 +66,11 @@ int main(int narg, char *args[])
     return 0;
 }
 
-int bind_stub(std::string bind_address, int bind_port)
+int bind_stub(const char *bind_address, int bind_port)
 {
     int listen_backlog = 128;
 
-    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);
+    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
     if (listen_fd < 0)
     {
         throw std::runtime_error(std::string("socket: ") + strerror(errno));
@@ -85,7 +78,17 @@ int bind_stub(std::string bind_address, int bind_port)
     int enable = 1;
     setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));
 
-    if (bind(listen_fd, &addr, sizeof(addr)) < 0)
+    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)
     {
         close(listen_fd);
         throw std::runtime_error(std::string("bind: ") + strerror(errno));
@@ -110,10 +113,8 @@ 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 = memalign_or_die(MEM_ALIGNMENT, op->req.sec_rw.len);
+        op->buf = malloc(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)
     {

src/vitastor.pc.in

@@ -6,7 +6,7 @@ includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@
 
 Name: Vitastor
 Description: Vitastor client library
-Version: 0.6.11
+Version: 0.6.10
 Libs: -L${libdir} -lvitastor_client
 Cflags: -I${includedir}