Release 0.8.7

This release includes a bunch of important bugfixes for erasure-coded setups
with disabled immediate_commit. After these fixes, "test_heal" OSD killing test
now passes fine with EC:

- Fix cluster write stalls with "Error while doing flush on OSD xx: -16 (Device or resource busy)"
  in OSD logs possible in EC setups with disabled immediate_commit by selectively
  syncing nonsynced objects on STABILIZE/ROLLBACK (https://github.com/vitalif/vitastor/issues/51)
- Fix other EC + disabled immediate_commit problems:
  - Fix "opcode=5 retval=-2" errors happening on SYNC retries
  - Fix non-working "pagination" during PG dirty object flushing
  - Fix write operations not continued correctly after dirty object flushing
- Fix incorrect parity read-modify-write calculation when writing into a lost chunk
- Fix OSDs losing left_on_dead PG state of non-clean PGs and thus not removing junk data in the cluster
- Fix a small memory leak caused by bad indexing of EC recovery matrices
- Fix a rare use-after-free in cluster_client caused by a reenterability issue
- Fix vitastor-cli create command syntax in the CSI driver
- Allow to start OSDs without local store for tests
- Fix memory allocation error in disk_tool_meta for non-standard metadata block sizes
- Fix delete operations received before loading pool metadata crashing OSDs with "null pointer exception"
- Improve "theoretical performance" Russian documentation

New features:

- Implement online configuration update for some parameters. Documentation is coming soon :)

CMakeLists.txt

@@ -2,6 +2,6 @@ cmake_minimum_required(VERSION 2.8.12)
 
 project(vitastor)
 
-set(VERSION "0.8.6")
+set(VERSION "0.8.7")
 
 add_subdirectory(src)

csi/Makefile

@@ -1,4 +1,4 @@
-VERSION ?= v0.8.6
+VERSION ?= v0.8.7
 
 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.8.6
+          image: vitalif/vitastor-csi:v0.8.7
           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.8.6
+          image: vitalif/vitastor-csi:v0.8.7
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/src/config.go

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

debian/changelog

@@ -1,10 +1,10 @@
-vitastor (0.8.6-1) unstable; urgency=medium
+vitastor (0.8.7-1) unstable; urgency=medium
 
   * Bugfixes
 
  -- Vitaliy Filippov <vitalif@yourcmc.ru>  Fri, 03 Jun 2022 02:09:44 +0300
 
-vitastor (0.8.6-1) unstable; urgency=medium
+vitastor (0.8.7-1) unstable; urgency=medium
 
   * Implement NFS proxy
   * Add documentation

debian/vitastor.Dockerfile

@@ -34,8 +34,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.8.6; \
-    cd vitastor-0.8.6; \
+    cp -r /root/vitastor vitastor-0.8.7; \
+    cd vitastor-0.8.7; \
     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; \
@@ -48,8 +48,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.8.6.orig.tar.xz vitastor-0.8.6; \
-    cd vitastor-0.8.6; \
+    tar --sort=name --mtime='2020-01-01' --owner=0 --group=0 --exclude=debian -cJf vitastor_0.8.7.orig.tar.xz vitastor-0.8.7; \
+    cd vitastor-0.8.7; \
     V=$(head -n1 debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
     DEBFULLNAME="Vitaliy Filippov <vitalif@yourcmc.ru>" dch -D $REL -v "$V""$REL" "Rebuild for $REL"; \
     DEB_BUILD_OPTIONS=nocheck dpkg-buildpackage --jobs=auto -sa; \

mon/mon.js

@@ -51,8 +51,9 @@ const etcd_tree = {
             // THIS IS JUST A POOR MAN'S CONFIG DOCUMENTATION
             // etcd connection
             config_path: "/etc/vitastor/vitastor.conf",
-            etcd_address: "10.0.115.10:2379/v3",
             etcd_prefix: "/vitastor",
+            // etcd connection - configurable online
+            etcd_address: "10.0.115.10:2379/v3",
             // mon
             etcd_mon_ttl: 30, // min: 10
             etcd_mon_timeout: 1000, // ms. min: 0
@@ -73,11 +74,12 @@ const etcd_tree = {
             rdma_max_send: 8,
             rdma_max_recv: 16,
             rdma_max_msg: 132096,
-            log_level: 0,
             block_size: 131072,
             disk_alignment: 4096,
             bitmap_granularity: 4096,
             immediate_commit: false, // 'all' or 'small'
+            // client and osd - configurable online
+            log_level: 0,
             client_dirty_limit: 33554432,
             peer_connect_interval: 5, // seconds. min: 1
             peer_connect_timeout: 5, // seconds. min: 1
@@ -95,18 +97,19 @@ const etcd_tree = {
             osd_network: null, // "192.168.7.0/24" or an array of masks
             bind_address: "0.0.0.0",
             bind_port: 0,
+            readonly: false,
+            osd_memlock: false,
+            // osd - configurable online
             autosync_interval: 5,
             autosync_writes: 128,
             client_queue_depth: 128, // unused
             recovery_queue_depth: 4,
             recovery_sync_batch: 16,
-            readonly: false,
             no_recovery: false,
             no_rebalance: false,
             print_stats_interval: 3,
             slow_log_interval: 10,
             inode_vanish_time: 60,
-            osd_memlock: false,
             // blockstore - fixed in superblock
             block_size,
             disk_alignment,
@@ -125,14 +128,15 @@ const etcd_tree = {
             meta_offset,
             disable_meta_fsync,
             disable_device_lock,
-            // blockstore - configurable
-            max_write_iodepth,
-            min_flusher_count: 1,
-            max_flusher_count: 256,
+            // blockstore - configurable offline
             inmemory_metadata,
             inmemory_journal,
             journal_sector_buffer_count,
             journal_no_same_sector_overwrites,
+            // blockstore - configurable online
+            max_write_iodepth,
+            min_flusher_count: 1,
+            max_flusher_count: 256,
             throttle_small_writes: false,
             throttle_target_iops: 100,
             throttle_target_mbs: 100,

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.8.6'
+VERSION = '0.8.7'
 
 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.8.6/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.8.6$(rpm --eval '%dist').tar.gz *
+tar --transform 's#^#vitastor-0.8.7/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.8.7$(rpm --eval '%dist').tar.gz *

rpm/vitastor-el7.Dockerfile

@@ -35,7 +35,7 @@ ADD . /root/vitastor
 RUN set -e; \
     cd /root/vitastor/rpm; \
     sh build-tarball.sh; \
-    cp /root/vitastor-0.8.6.el7.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.8.7.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.8.6
+Version:        0.8.7
 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.8.6.el7.tar.gz
+Source0:        vitastor-0.8.7.el7.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

rpm/vitastor-el8.Dockerfile

@@ -35,7 +35,7 @@ ADD . /root/vitastor
 RUN set -e; \
     cd /root/vitastor/rpm; \
     sh build-tarball.sh; \
-    cp /root/vitastor-0.8.6.el8.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.8.7.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.8.6
+Version:        0.8.7
 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.8.6.el8.tar.gz
+Source0:        vitastor-0.8.7.el8.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

src/CMakeLists.txt

@@ -16,7 +16,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 	set(CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/${CMAKE_INSTALL_LIBDIR}")
 endif()
 
-add_definitions(-DVERSION="0.8.6")
+add_definitions(-DVERSION="0.8.7")
 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)

src/blockstore.cpp

@@ -13,6 +13,11 @@ blockstore_t::~blockstore_t()
     delete impl;
 }
 
+void blockstore_t::parse_config(blockstore_config_t & config)
+{
+    impl->parse_config(config, false);
+}
+
 void blockstore_t::loop()
 {
     impl->loop();

src/blockstore.h

@@ -107,7 +107,7 @@ Input:
 - buf = pre-allocated obj_ver_id array <len> units long
 
 Output:
-- retval = 0 or negative error number (-EINVAL, -ENOENT if no such version or -EBUSY if not synced)
+- retval = 0 or negative error number (-ENOENT if no such version for stabilize)
 
 ## BS_OP_SYNC_STAB_ALL
 
@@ -165,6 +165,9 @@ public:
     blockstore_t(blockstore_config_t & config, ring_loop_t *ringloop, timerfd_manager_t *tfd);
     ~blockstore_t();
 
+    // Update configuration
+    void parse_config(blockstore_config_t & config);
+
     // Event loop
     void loop();
 

src/blockstore_impl.cpp

@@ -11,7 +11,7 @@ blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_t *
     ring_consumer.loop = [this]() { loop(); };
     ringloop->register_consumer(&ring_consumer);
     initialized = 0;
-    parse_config(config);
+    parse_config(config, true);
     zero_object = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.data_block_size);
     try
     {
@@ -171,7 +171,7 @@ void blockstore_impl_t::loop()
                     // Can't submit SYNC before previous writes
                     continue;
                 }
-                wr_st = continue_sync(op, false);
+                wr_st = continue_sync(op);
                 if (wr_st != 2)
                 {
                     has_writes = wr_st > 0 ? 1 : 2;
@@ -371,13 +371,18 @@ void blockstore_impl_t::enqueue_op(blockstore_op_t *op)
         ringloop->set_immediate([op]() { std::function<void (blockstore_op_t*)>(op->callback)(op); });
         return;
     }
+    init_op(op);
+    submit_queue.push_back(op);
+    ringloop->wakeup();
+}
+
+void blockstore_impl_t::init_op(blockstore_op_t *op)
+{
     // Call constructor without allocating memory. We'll call destructor before returning op back
     new ((void*)op->private_data) blockstore_op_private_t;
     PRIV(op)->wait_for = 0;
     PRIV(op)->op_state = 0;
     PRIV(op)->pending_ops = 0;
-    submit_queue.push_back(op);
-    ringloop->wakeup();
 }
 
 static bool replace_stable(object_id oid, uint64_t version, int search_start, int search_end, obj_ver_id* list)

src/blockstore_impl.h

@@ -216,6 +216,11 @@ struct pool_shard_settings_t
     uint32_t pg_stripe_size;
 };
 
+#define STAB_SPLIT_DONE 1
+#define STAB_SPLIT_WAIT 2
+#define STAB_SPLIT_SYNC 3
+#define STAB_SPLIT_TODO 4
+
 class blockstore_impl_t
 {
     blockstore_disk_t dsk;
@@ -277,7 +282,6 @@ class blockstore_impl_t
     friend class journal_flusher_t;
     friend class journal_flusher_co;
 
-    void parse_config(blockstore_config_t & config);
     void calc_lengths();
     void open_data();
     void open_meta();
@@ -299,6 +303,7 @@ class blockstore_impl_t
     blockstore_init_journal* journal_init_reader;
 
     void check_wait(blockstore_op_t *op);
+    void init_op(blockstore_op_t *op);
 
     // Read
     int dequeue_read(blockstore_op_t *read_op);
@@ -318,7 +323,7 @@ class blockstore_impl_t
     void handle_write_event(ring_data_t *data, blockstore_op_t *op);
 
     // Sync
-    int continue_sync(blockstore_op_t *op, bool queue_has_in_progress_sync);
+    int continue_sync(blockstore_op_t *op);
     void ack_sync(blockstore_op_t *op);
 
     // Stabilize
@@ -326,6 +331,8 @@ class blockstore_impl_t
     int continue_stable(blockstore_op_t *op);
     void mark_stable(const obj_ver_id & ov, bool forget_dirty = false);
     void stabilize_object(object_id oid, uint64_t max_ver);
+    blockstore_op_t* selective_sync(blockstore_op_t *op);
+    int split_stab_op(blockstore_op_t *op, std::function<int(obj_ver_id v)> decider);
 
     // Rollback
     int dequeue_rollback(blockstore_op_t *op);
@@ -341,6 +348,8 @@ public:
     blockstore_impl_t(blockstore_config_t & config, ring_loop_t *ringloop, timerfd_manager_t *tfd);
     ~blockstore_impl_t();
 
+    void parse_config(blockstore_config_t & config, bool init);
+
     // Event loop
     void loop();
 

src/blockstore_open.cpp

@@ -4,8 +4,54 @@
 #include <sys/file.h>
 #include "blockstore_impl.h"
 
-void blockstore_impl_t::parse_config(blockstore_config_t & config)
+void blockstore_impl_t::parse_config(blockstore_config_t & config, bool init)
 {
+    // Online-configurable options:
+    max_flusher_count = strtoull(config["max_flusher_count"].c_str(), NULL, 10);
+    if (!max_flusher_count)
+    {
+        max_flusher_count = strtoull(config["flusher_count"].c_str(), NULL, 10);
+    }
+    min_flusher_count = strtoull(config["min_flusher_count"].c_str(), NULL, 10);
+    max_write_iodepth = strtoull(config["max_write_iodepth"].c_str(), NULL, 10);
+    throttle_small_writes = config["throttle_small_writes"] == "true" || config["throttle_small_writes"] == "1" || config["throttle_small_writes"] == "yes";
+    throttle_target_iops = strtoull(config["throttle_target_iops"].c_str(), NULL, 10);
+    throttle_target_mbs = strtoull(config["throttle_target_mbs"].c_str(), NULL, 10);
+    throttle_target_parallelism = strtoull(config["throttle_target_parallelism"].c_str(), NULL, 10);
+    throttle_threshold_us = strtoull(config["throttle_threshold_us"].c_str(), NULL, 10);
+    if (!max_flusher_count)
+    {
+        max_flusher_count = 256;
+    }
+    if (!min_flusher_count || journal.flush_journal)
+    {
+        min_flusher_count = 1;
+    }
+    if (!max_write_iodepth)
+    {
+        max_write_iodepth = 128;
+    }
+    if (!throttle_target_iops)
+    {
+        throttle_target_iops = 100;
+    }
+    if (!throttle_target_mbs)
+    {
+        throttle_target_mbs = 100;
+    }
+    if (!throttle_target_parallelism)
+    {
+        throttle_target_parallelism = 1;
+    }
+    if (!throttle_threshold_us)
+    {
+        throttle_threshold_us = 50;
+    }
+    if (!init)
+    {
+        return;
+    }
+    // Offline-configurable options:
     // Common disk options
     dsk.parse_config(config);
     // Parse
@@ -44,29 +90,7 @@ void blockstore_impl_t::parse_config(blockstore_config_t & config)
     journal.no_same_sector_overwrites = config["journal_no_same_sector_overwrites"] == "true" ||
         config["journal_no_same_sector_overwrites"] == "1" || config["journal_no_same_sector_overwrites"] == "yes";
     journal.inmemory = config["inmemory_journal"] != "false";
-    max_flusher_count = strtoull(config["max_flusher_count"].c_str(), NULL, 10);
-    if (!max_flusher_count)
-        max_flusher_count = strtoull(config["flusher_count"].c_str(), NULL, 10);
-    min_flusher_count = strtoull(config["min_flusher_count"].c_str(), NULL, 10);
-    max_write_iodepth = strtoull(config["max_write_iodepth"].c_str(), NULL, 10);
-    throttle_small_writes = config["throttle_small_writes"] == "true" || config["throttle_small_writes"] == "1" || config["throttle_small_writes"] == "yes";
-    throttle_target_iops = strtoull(config["throttle_target_iops"].c_str(), NULL, 10);
-    throttle_target_mbs = strtoull(config["throttle_target_mbs"].c_str(), NULL, 10);
-    throttle_target_parallelism = strtoull(config["throttle_target_parallelism"].c_str(), NULL, 10);
-    throttle_threshold_us = strtoull(config["throttle_threshold_us"].c_str(), NULL, 10);
     // Validate
-    if (!max_flusher_count)
-    {
-        max_flusher_count = 256;
-    }
-    if (!min_flusher_count || journal.flush_journal)
-    {
-        min_flusher_count = 1;
-    }
-    if (!max_write_iodepth)
-    {
-        max_write_iodepth = 128;
-    }
     if (journal.sector_count < 2)
     {
         journal.sector_count = 32;
@@ -91,22 +115,6 @@ void blockstore_impl_t::parse_config(blockstore_config_t & config)
     {
         throw std::runtime_error("immediate_commit=all requires disable_journal_fsync and disable_data_fsync");
     }
-    if (!throttle_target_iops)
-    {
-        throttle_target_iops = 100;
-    }
-    if (!throttle_target_mbs)
-    {
-        throttle_target_mbs = 100;
-    }
-    if (!throttle_target_parallelism)
-    {
-        throttle_target_parallelism = 1;
-    }
-    if (!throttle_threshold_us)
-    {
-        throttle_threshold_us = 50;
-    }
     // init some fields
     journal.block_size = dsk.journal_block_size;
     journal.next_free = dsk.journal_block_size;

src/blockstore_rollback.cpp

@@ -9,48 +9,39 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
     {
         return continue_rollback(op);
     }
-    obj_ver_id *v, *nv;
-    int i, todo = op->len;
-    for (i = 0, v = (obj_ver_id*)op->buf, nv = (obj_ver_id*)op->buf; i < op->len; i++, v++, nv++)
+    int r = split_stab_op(op, [this](obj_ver_id ov)
     {
-        if (nv != v)
-        {
-            *nv = *v;
-        }
         // Check that there are some versions greater than v->version (which may be zero),
         // check that they're unstable, synced, and not currently written to
         auto dirty_it = dirty_db.lower_bound((obj_ver_id){
-            .oid = v->oid,
+            .oid = ov.oid,
             .version = UINT64_MAX,
         });
         if (dirty_it == dirty_db.begin())
         {
-skip_ov:
             // Already rolled back, skip this object version
-            todo--;
-            nv--;
-            continue;
+            return STAB_SPLIT_DONE;
         }
         else
         {
             dirty_it--;
-            if (dirty_it->first.oid != v->oid || dirty_it->first.version < v->version)
+            if (dirty_it->first.oid != ov.oid || dirty_it->first.version < ov.version)
             {
-                goto skip_ov;
+                // Already rolled back, skip this object version
+                return STAB_SPLIT_DONE;
             }
-            while (dirty_it->first.oid == v->oid && dirty_it->first.version > v->version)
+            while (dirty_it->first.oid == ov.oid && dirty_it->first.version > ov.version)
             {
                 if (IS_IN_FLIGHT(dirty_it->second.state))
                 {
                     // Object write is still in progress. Wait until the write request completes
-                    return 0;
+                    return STAB_SPLIT_WAIT;
                 }
                 else if (!IS_SYNCED(dirty_it->second.state) ||
                     IS_STABLE(dirty_it->second.state))
                 {
-                    op->retval = -EBUSY;
-                    FINISH_OP(op);
-                    return 2;
+                    // Sync the object
+                    return STAB_SPLIT_SYNC;
                 }
                 if (dirty_it == dirty_db.begin())
                 {
@@ -58,19 +49,16 @@ skip_ov:
                 }
                 dirty_it--;
             }
+            return STAB_SPLIT_TODO;
         }
-    }
-    op->len = todo;
-    if (!todo)
+    });
+    if (r != 1)
     {
-        // Already rolled back
-        op->retval = 0;
-        FINISH_OP(op);
-        return 2;
+        return r;
     }
     // Check journal space
     blockstore_journal_check_t space_check(this);
-    if (!space_check.check_available(op, todo, sizeof(journal_entry_rollback), 0))
+    if (!space_check.check_available(op, op->len, sizeof(journal_entry_rollback), 0))
     {
         return 0;
     }
@@ -78,7 +66,8 @@ skip_ov:
     BS_SUBMIT_CHECK_SQES(space_check.sectors_to_write);
     // Prepare and submit journal entries
     int s = 0;
-    for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
+    auto v = (obj_ver_id*)op->buf;
+    for (int i = 0; i < op->len; i++, v++)
     {
         if (!journal.entry_fits(sizeof(journal_entry_rollback)) &&
             journal.sector_info[journal.cur_sector].dirty)

src/blockstore_stable.cpp

@@ -41,60 +41,309 @@
 // 4) after a while it takes his synced object list and sends stabilize requests
 //    to peers and to its own blockstore, thus freeing the old version
 
-int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
+struct ver_vector_t
 {
-    if (PRIV(op)->op_state)
+    obj_ver_id *items = NULL;
+    uint64_t alloc = 0, size = 0;
+};
+
+static void init_versions(ver_vector_t & vec, obj_ver_id *start, obj_ver_id *end, uint64_t len)
+{
+    if (!vec.items)
     {
-        return continue_stable(op);
+        vec.alloc = len;
+        vec.items = (obj_ver_id*)malloc_or_die(sizeof(obj_ver_id) * vec.alloc);
+        for (auto sv = start; sv < end; sv++)
+        {
+            vec.items[vec.size++] = *sv;
+        }
     }
+}
+
+static void append_version(ver_vector_t & vec, obj_ver_id ov)
+{
+    if (vec.size >= vec.alloc)
+    {
+        vec.alloc = !vec.alloc ? 4 : vec.alloc*2;
+        vec.items = (obj_ver_id*)realloc_or_die(vec.items, sizeof(obj_ver_id) * vec.alloc);
+    }
+    vec.items[vec.size++] = ov;
+}
+
+static bool check_unsynced(std::vector<obj_ver_id> & check, obj_ver_id ov, std::vector<obj_ver_id> & to, int *count)
+{
+    bool found = false;
+    int j = 0, k = 0;
+    while (j < check.size())
+    {
+        if (check[j] == ov)
+            found = true;
+        if (check[j].oid == ov.oid && check[j].version <= ov.version)
+        {
+            to.push_back(check[j++]);
+            if (count)
+                (*count)--;
+        }
+        else
+            check[k++] = check[j++];
+    }
+    check.resize(k);
+    return found;
+}
+
+blockstore_op_t* blockstore_impl_t::selective_sync(blockstore_op_t *op)
+{
+    unsynced_big_write_count -= unsynced_big_writes.size();
+    unsynced_big_writes.swap(PRIV(op)->sync_big_writes);
+    unsynced_big_write_count += unsynced_big_writes.size();
+    unsynced_small_writes.swap(PRIV(op)->sync_small_writes);
+    // Create a sync operation, insert into the end of the queue
+    // And move ourselves into the end too!
+    // Rather hacky but that's what we need...
+    blockstore_op_t *sync_op = new blockstore_op_t;
+    sync_op->opcode = BS_OP_SYNC;
+    sync_op->buf = NULL;
+    sync_op->callback = [this](blockstore_op_t *sync_op)
+    {
+        delete sync_op;
+    };
+    init_op(sync_op);
+    int sync_res = continue_sync(sync_op);
+    if (sync_res != 2)
+    {
+        // Put SYNC into the queue if it's not finished yet
+        submit_queue.push_back(sync_op);
+    }
+    // Restore unsynced_writes
+    unsynced_small_writes.swap(PRIV(op)->sync_small_writes);
+    unsynced_big_write_count -= unsynced_big_writes.size();
+    unsynced_big_writes.swap(PRIV(op)->sync_big_writes);
+    unsynced_big_write_count += unsynced_big_writes.size();
+    if (sync_res == 2)
+    {
+        // Sync is immediately completed
+        return NULL;
+    }
+    return sync_op;
+}
+
+// Returns: 2 = stop processing and dequeue, 0 = stop processing and do not dequeue, 1 = proceed with op itself
+int blockstore_impl_t::split_stab_op(blockstore_op_t *op, std::function<int(obj_ver_id v)> decider)
+{
+    bool add_sync = false;
+    ver_vector_t good_vers, bad_vers;
     obj_ver_id* v;
     int i, todo = 0;
     for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
     {
-        auto dirty_it = dirty_db.find(*v);
-        if (dirty_it == dirty_db.end())
+        int action = decider(*v);
+        if (action < 0)
         {
-            auto & clean_db = clean_db_shard(v->oid);
-            auto clean_it = clean_db.find(v->oid);
-            if (clean_it == clean_db.end() || clean_it->second.version < v->version)
+            // Rollback changes
+            for (auto & ov: PRIV(op)->sync_big_writes)
             {
-                // No such object version
-                op->retval = -ENOENT;
-                FINISH_OP(op);
-                return 2;
+                unsynced_big_writes.push_back(ov);
+                unsynced_big_write_count++;
             }
-            else
+            for (auto & ov: PRIV(op)->sync_small_writes)
             {
-                // Already stable
+                unsynced_small_writes.push_back(ov);
             }
-        }
-        else if (IS_IN_FLIGHT(dirty_it->second.state))
-        {
-            // Object write is still in progress. Wait until the write request completes
-            return 0;
-        }
-        else if (!IS_SYNCED(dirty_it->second.state))
-        {
-            // Object not synced yet. Caller must sync it first
-            op->retval = -EBUSY;
+            free(good_vers.items);
+            good_vers.items = NULL;
+            free(bad_vers.items);
+            bad_vers.items = NULL;
+            // Error
+            op->retval = action;
             FINISH_OP(op);
             return 2;
         }
-        else if (!IS_STABLE(dirty_it->second.state))
+        else if (action == STAB_SPLIT_DONE)
         {
+            // Already done
+            init_versions(good_vers, (obj_ver_id*)op->buf, v, op->len);
+        }
+        else if (action == STAB_SPLIT_WAIT)
+        {
+            // Already in progress, we just have to wait until it finishes
+            init_versions(good_vers, (obj_ver_id*)op->buf, v, op->len);
+            append_version(bad_vers, *v);
+        }
+        else if (action == STAB_SPLIT_SYNC)
+        {
+            // Needs a SYNC, we have to send a SYNC if not already in progress
+            //
+            // If the object is not present in unsynced_(big|small)_writes then
+            // it's currently being synced. If it's present then we can initiate
+            // its sync ourselves.
+            init_versions(good_vers, (obj_ver_id*)op->buf, v, op->len);
+            append_version(bad_vers, *v);
+            if (!add_sync)
+            {
+                PRIV(op)->sync_big_writes.clear();
+                PRIV(op)->sync_small_writes.clear();
+                add_sync = true;
+            }
+            check_unsynced(unsynced_small_writes, *v, PRIV(op)->sync_small_writes, NULL);
+            check_unsynced(unsynced_big_writes, *v, PRIV(op)->sync_big_writes, &unsynced_big_write_count);
+        }
+        else /* if (action == STAB_SPLIT_TODO) */
+        {
+            if (good_vers.items)
+            {
+                // If we're selecting versions then append it
+                // Main idea is that 99% of the time all versions passed to BS_OP_STABLE are synced
+                // And we don't want to select/allocate anything in that optimistic case
+                append_version(good_vers, *v);
+            }
             todo++;
         }
     }
-    if (!todo)
+    // In a pessimistic scenario, an operation may be split into 3:
+    // - Stabilize synced entries
+    // - Sync unsynced entries
+    // - Continue for unsynced entries after sync
+    add_sync = add_sync && (PRIV(op)->sync_big_writes.size() || PRIV(op)->sync_small_writes.size());
+    if (!todo && !bad_vers.size)
     {
         // Already stable
         op->retval = 0;
         FINISH_OP(op);
         return 2;
     }
+    op->retval = 0;
+    if (!todo && !add_sync)
+    {
+        // Only wait for inflight writes or current in-progress syncs
+        return 0;
+    }
+    blockstore_op_t *sync_op = NULL, *split_stab_op = NULL;
+    if (add_sync)
+    {
+        // Initiate a selective sync for PRIV(op)->sync_(big|small)_writes
+        sync_op = selective_sync(op);
+    }
+    if (bad_vers.size)
+    {
+        // Split part of the request into a separate operation
+        split_stab_op = new blockstore_op_t;
+        split_stab_op->opcode = op->opcode;
+        split_stab_op->buf = bad_vers.items;
+        split_stab_op->len = bad_vers.size;
+        init_op(split_stab_op);
+        submit_queue.push_back(split_stab_op);
+    }
+    if (sync_op || split_stab_op || good_vers.items)
+    {
+        void *orig_buf = op->buf;
+        if (good_vers.items)
+        {
+            op->buf = good_vers.items;
+            op->len = good_vers.size;
+        }
+        // Make a wrapped callback
+        int *split_op_counter = (int*)malloc_or_die(sizeof(int));
+        *split_op_counter = (sync_op ? 1 : 0) + (split_stab_op ? 1 : 0) + (todo ? 1 : 0);
+        auto cb = [this, op, good_items = good_vers.items,
+            bad_items = bad_vers.items, split_op_counter,
+            orig_buf, real_cb = op->callback](blockstore_op_t *split_op)
+        {
+            if (split_op->retval != 0)
+                op->retval = split_op->retval;
+            (*split_op_counter)--;
+            assert((*split_op_counter) >= 0);
+            if (op != split_op)
+                delete split_op;
+            if (!*split_op_counter)
+            {
+                free(good_items);
+                free(bad_items);
+                free(split_op_counter);
+                op->buf = orig_buf;
+                real_cb(op);
+            }
+        };
+        if (sync_op)
+        {
+            sync_op->callback = cb;
+        }
+        if (split_stab_op)
+        {
+            split_stab_op->callback = cb;
+        }
+        op->callback = cb;
+    }
+    if (!todo)
+    {
+        // All work is postponed
+        op->callback = NULL;
+        return 2;
+    }
+    return 1;
+}
+
+int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
+{
+    if (PRIV(op)->op_state)
+    {
+        return continue_stable(op);
+    }
+    int r = split_stab_op(op, [this](obj_ver_id ov)
+    {
+        auto dirty_it = dirty_db.find(ov);
+        if (dirty_it == dirty_db.end())
+        {
+            auto & clean_db = clean_db_shard(ov.oid);
+            auto clean_it = clean_db.find(ov.oid);
+            if (clean_it == clean_db.end() || clean_it->second.version < ov.version)
+            {
+                // No such object version
+                printf("Error: %lx:%lx v%lu not found while stabilizing\n", ov.oid.inode, ov.oid.stripe, ov.version);
+                return -ENOENT;
+            }
+            else
+            {
+                // Already stable
+                return STAB_SPLIT_DONE;
+            }
+        }
+        else if (IS_IN_FLIGHT(dirty_it->second.state))
+        {
+            // Object write is still in progress. Wait until the write request completes
+            return STAB_SPLIT_WAIT;
+        }
+        else if (!IS_SYNCED(dirty_it->second.state))
+        {
+            // Object not synced yet - sync it
+            // In previous versions we returned EBUSY here and required
+            // the caller (OSD) to issue a global sync first. But a global sync
+            // waits for all writes in the queue including inflight writes. And
+            // inflight writes may themselves be blocked by unstable writes being
+            // still present in the journal and not flushed away from it.
+            // So we must sync specific objects here.
+            //
+            // Even more, we have to process "stabilize" request in parts. That is,
+            // we must stabilize all objects which are already synced. Otherwise
+            // they may block objects which are NOT synced yet.
+            return STAB_SPLIT_SYNC;
+        }
+        else if (IS_STABLE(dirty_it->second.state))
+        {
+            // Already stable
+            return STAB_SPLIT_DONE;
+        }
+        else
+        {
+            return STAB_SPLIT_TODO;
+        }
+    });
+    if (r != 1)
+    {
+        return r;
+    }
     // Check journal space
     blockstore_journal_check_t space_check(this);
-    if (!space_check.check_available(op, todo, sizeof(journal_entry_stable), 0))
+    if (!space_check.check_available(op, op->len, sizeof(journal_entry_stable), 0))
     {
         return 0;
     }
@@ -102,9 +351,9 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
     BS_SUBMIT_CHECK_SQES(space_check.sectors_to_write);
     // Prepare and submit journal entries
     int s = 0;
-    for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
+    auto v = (obj_ver_id*)op->buf;
+    for (int i = 0; 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)
         {

src/blockstore_sync.cpp

@@ -12,7 +12,7 @@
 #define SYNC_JOURNAL_SYNC_SENT 7
 #define SYNC_DONE 8
 
-int blockstore_impl_t::continue_sync(blockstore_op_t *op, bool queue_has_in_progress_sync)
+int blockstore_impl_t::continue_sync(blockstore_op_t *op)
 {
     if (immediate_commit == IMMEDIATE_ALL)
     {
@@ -145,7 +145,7 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op, bool queue_has_in_prog
             PRIV(op)->op_state = SYNC_DONE;
         }
     }
-    if (PRIV(op)->op_state == SYNC_DONE && !queue_has_in_progress_sync)
+    if (PRIV(op)->op_state == SYNC_DONE)
     {
         ack_sync(op);
         return 2;

src/cli_rm_osd.cpp

@@ -278,7 +278,7 @@ struct rm_osd_t
             if (rsp["response_delete_range"]["deleted"].uint64_value() > 0)
             {
                 // Wait for mon_change_timeout before updating PG history, or the monitor's change will likely interfere with ours
-                retry_wait = parent->cli->merged_config["mon_change_timeout"].uint64_value();
+                retry_wait = parent->cli->config["mon_change_timeout"].uint64_value();
                 if (!retry_wait)
                     retry_wait = 1000;
                 retry_wait += etcd_tx_retry_ms;

src/cli_status.cpp

@@ -198,9 +198,9 @@ resume_2:
             }
             pgs_by_state_str += std::to_string(kv.second)+" "+kv.first;
         }
-        bool readonly = json_is_true(parent->cli->merged_config["readonly"]);
-        bool no_recovery = json_is_true(parent->cli->merged_config["no_recovery"]);
-        bool no_rebalance = json_is_true(parent->cli->merged_config["no_rebalance"]);
+        bool readonly = json_is_true(parent->cli->config["readonly"]);
+        bool no_recovery = json_is_true(parent->cli->config["no_recovery"]);
+        bool no_rebalance = json_is_true(parent->cli->config["no_rebalance"]);
         if (parent->json_output)
         {
             // JSON output

src/cluster_client.cpp

@@ -18,11 +18,12 @@
 
 cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json & config)
 {
-    config = osd_messenger_t::read_config(config);
+    cli_config = config.object_items();
+    file_config = osd_messenger_t::read_config(config);
+    config = osd_messenger_t::merge_configs(cli_config, file_config, etcd_global_config, {});
 
     this->ringloop = ringloop;
     this->tfd = tfd;
-    this->config = config;
 
     msgr.osd_num = 0;
     msgr.tfd = tfd;
@@ -58,7 +59,7 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
         msgr.stop_client(op->peer_fd);
         delete op;
     };
-    msgr.parse_config(this->config);
+    msgr.parse_config(config);
 
     st_cli.tfd = tfd;
     st_cli.on_load_config_hook = [this](json11::Json::object & cfg) { on_load_config_hook(cfg); };
@@ -276,13 +277,10 @@ restart:
     continuing_ops = 0;
 }
 
-void cluster_client_t::on_load_config_hook(json11::Json::object & config)
+void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_config)
 {
-    this->merged_config = config;
-    for (auto & kv: this->config.object_items())
-    {
-        this->merged_config[kv.first] = kv.second;
-    }
+    this->etcd_global_config = etcd_global_config;
+    config = osd_messenger_t::merge_configs(cli_config, file_config, etcd_global_config, {});
     if (config.find("client_max_dirty_bytes") != config.end())
     {
         client_max_dirty_bytes = config["client_max_dirty_bytes"].uint64_value();
@@ -292,14 +290,13 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & config)
         // Old name
         client_max_dirty_bytes = config["client_dirty_limit"].uint64_value();
     }
-    if (config.find("client_max_dirty_ops") != config.end())
-    {
-        client_max_dirty_ops = config["client_max_dirty_ops"].uint64_value();
-    }
+    else
+        client_max_dirty_bytes = 0;
     if (!client_max_dirty_bytes)
     {
         client_max_dirty_bytes = DEFAULT_CLIENT_MAX_DIRTY_BYTES;
     }
+    client_max_dirty_ops = config["client_max_dirty_ops"].uint64_value();
     if (!client_max_dirty_ops)
     {
         client_max_dirty_ops = DEFAULT_CLIENT_MAX_DIRTY_OPS;
@@ -314,7 +311,7 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & config)
         up_wait_retry_interval = 50;
     }
     msgr.parse_config(config);
-    msgr.parse_config(this->config);
+    st_cli.parse_config(config);
     st_cli.load_pgs();
 }
 
@@ -1121,6 +1118,24 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
     if (part->op.reply.hdr.retval != expected)
     {
         // Operation failed, retry
+        part->flags |= PART_ERROR;
+        if (!op->retval || op->retval == -EPIPE)
+        {
+            // Don't overwrite other errors with -EPIPE
+            op->retval = part->op.reply.hdr.retval;
+        }
+        int stop_fd = -1;
+        if (op->retval != -EINTR && op->retval != -EIO)
+        {
+            stop_fd = part->op.peer_fd;
+            fprintf(
+                stderr, "%s operation failed on OSD %lu: retval=%ld (expected %d), dropping connection\n",
+                osd_op_names[part->op.req.hdr.opcode], part->osd_num, part->op.reply.hdr.retval, expected
+            );
+        }
+        // All next things like timer, continue_sync/rw and stop_client may affect the operation again
+        // So do all these things after modifying operation state, otherwise we may hit reenterability bugs
+        // FIXME postpone such things to set_immediate here to avoid bugs
         if (part->op.reply.hdr.retval == -EPIPE)
         {
             // Mark op->up_wait = true before stopping the client
@@ -1134,20 +1149,17 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
                 });
             }
         }
-        if (!op->retval || op->retval == -EPIPE)
+        if (op->inflight_count == 0)
         {
-            // Don't overwrite other errors with -EPIPE
-            op->retval = part->op.reply.hdr.retval;
+            if (op->opcode == OSD_OP_SYNC)
+                continue_sync(op);
+            else
+                continue_rw(op);
         }
-        if (op->retval != -EINTR && op->retval != -EIO)
+        if (stop_fd >= 0)
         {
-            fprintf(
-                stderr, "%s operation failed on OSD %lu: retval=%ld (expected %d), dropping connection\n",
-                osd_op_names[part->op.req.hdr.opcode], part->osd_num, part->op.reply.hdr.retval, expected
-            );
-            msgr.stop_client(part->op.peer_fd);
+            msgr.stop_client(stop_fd);
         }
-        part->flags |= PART_ERROR;
     }
     else
     {
@@ -1161,13 +1173,13 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
             copy_part_bitmap(op, part);
             op->version = op->parts.size() == 1 ? part->op.reply.rw.version : 0;
         }
-    }
-    if (op->inflight_count == 0)
-    {
-        if (op->opcode == OSD_OP_SYNC)
-            continue_sync(op);
-        else
-            continue_rw(op);
+        if (op->inflight_count == 0)
+        {
+            if (op->opcode == OSD_OP_SYNC)
+                continue_sync(op);
+            else
+                continue_rw(op);
+        }
     }
 }
 

src/cluster_client.h

@@ -112,8 +112,8 @@ public:
     osd_messenger_t msgr;
     void init_msgr();
 
-    json11::Json config;
-    json11::Json::object merged_config;
+    json11::Json::object cli_config, file_config, etcd_global_config;
+    json11::Json::object config;
 
     cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json & config);
     ~cluster_client_t();

src/etcd_state_client.cpp

@@ -18,12 +18,8 @@ etcd_state_client_t::~etcd_state_client_t()
     }
     watches.clear();
     etcd_watches_initialised = -1;
-    if (ws_keepalive_timer >= 0)
-    {
-        tfd->clear_timer(ws_keepalive_timer);
-        ws_keepalive_timer = -1;
-    }
 #ifndef __MOCK__
+    stop_ws_keepalive();
     if (etcd_watch_ws)
     {
         http_close(etcd_watch_ws);
@@ -245,6 +241,7 @@ void etcd_state_client_t::parse_config(const json11::Json & config)
         if (this->etcd_keepalive_timeout < 30)
             this->etcd_keepalive_timeout = 30;
     }
+    auto old_etcd_ws_keepalive_interval = this->etcd_ws_keepalive_interval;
     this->etcd_ws_keepalive_interval = config["etcd_ws_keepalive_interval"].uint64_value();
     if (this->etcd_ws_keepalive_interval <= 0)
     {
@@ -265,6 +262,13 @@ void etcd_state_client_t::parse_config(const json11::Json & config)
     {
         this->etcd_quick_timeout = 1000;
     }
+    if (this->etcd_ws_keepalive_interval != old_etcd_ws_keepalive_interval && ws_keepalive_timer >= 0)
+    {
+#ifndef __MOCK__
+        stop_ws_keepalive();
+        start_ws_keepalive();
+#endif
+    }
 }
 
 void etcd_state_client_t::pick_next_etcd()
@@ -478,6 +482,20 @@ void etcd_state_client_t::start_etcd_watcher()
     {
         on_start_watcher_hook(etcd_watch_ws);
     }
+    start_ws_keepalive();
+}
+
+void etcd_state_client_t::stop_ws_keepalive()
+{
+    if (ws_keepalive_timer >= 0)
+    {
+        tfd->clear_timer(ws_keepalive_timer);
+        ws_keepalive_timer = -1;
+    }
+}
+
+void etcd_state_client_t::start_ws_keepalive()
+{
     if (ws_keepalive_timer < 0)
     {
         ws_keepalive_timer = tfd->set_timer(etcd_ws_keepalive_interval*1000, true, [this](int)

src/etcd_state_client.h

@@ -132,6 +132,8 @@ public:
     void etcd_txn(json11::Json txn, int timeout, int retries, int interval, std::function<void(std::string, json11::Json)> callback);
     void etcd_txn_slow(json11::Json txn, std::function<void(std::string, json11::Json)> callback);
     void start_etcd_watcher();
+    void stop_ws_keepalive();
+    void start_ws_keepalive();
     void load_global_config();
     void load_pgs();
     void parse_state(const etcd_kv_t & kv);

src/messenger.cpp

@@ -534,8 +534,9 @@ bool osd_messenger_t::is_rdma_enabled()
 }
 #endif
 
-json11::Json osd_messenger_t::read_config(const json11::Json & config)
+json11::Json::object osd_messenger_t::read_config(const json11::Json & config)
 {
+    json11::Json::object file_config;
     const char *config_path = config["config_path"].string_value() != ""
         ? config["config_path"].string_value().c_str() : VITASTOR_CONFIG_PATH;
     int fd = open(config_path, O_RDONLY);
@@ -543,14 +544,14 @@ json11::Json osd_messenger_t::read_config(const json11::Json & config)
     {
         if (errno != ENOENT)
             fprintf(stderr, "Error reading %s: %s\n", config_path, strerror(errno));
-        return config;
+        return file_config;
     }
     struct stat st;
     if (fstat(fd, &st) != 0)
     {
         fprintf(stderr, "Error reading %s: %s\n", config_path, strerror(errno));
         close(fd);
-        return config;
+        return file_config;
     }
     std::string buf;
     buf.resize(st.st_size);
@@ -562,23 +563,125 @@ json11::Json osd_messenger_t::read_config(const json11::Json & config)
         {
             fprintf(stderr, "Error reading %s: %s\n", config_path, strerror(errno));
             close(fd);
-            return config;
+            return file_config;
         }
         done += r;
     }
     close(fd);
     std::string json_err;
-    json11::Json::object file_config = json11::Json::parse(buf, json_err).object_items();
+    file_config = json11::Json::parse(buf, json_err).object_items();
     if (json_err != "")
     {
         fprintf(stderr, "Invalid JSON in %s: %s\n", config_path, json_err.c_str());
-        return config;
-    }
-    file_config.erase("config_path");
-    file_config.erase("osd_num");
-    for (auto kv: config.object_items())
-    {
-        file_config[kv.first] = kv.second;
     }
     return file_config;
 }
+
+static const char* cli_only_params[] = {
+    // The list has to be sorted
+    "bitmap_granularity",
+    "block_size",
+    "data_device",
+    "data_offset",
+    "data_size",
+    "disable_data_fsync",
+    "disable_device_lock",
+    "disable_journal_fsync",
+    "disable_meta_fsync",
+    "disk_alignment",
+    "flush_journal",
+    "immediate_commit",
+    "inmemory_journal",
+    "inmemory_metadata",
+    "journal_block_size",
+    "journal_device",
+    "journal_no_same_sector_overwrites",
+    "journal_offset",
+    "journal_sector_buffer_count",
+    "journal_size",
+    "meta_block_size",
+    "meta_buf_size",
+    "meta_device",
+    "meta_offset",
+    "osd_num",
+    "readonly",
+};
+
+static const char **cli_only_end = cli_only_params + (sizeof(cli_only_params)/sizeof(cli_only_params[0]));
+
+static const char* local_only_params[] = {
+    // The list has to be sorted
+    "config_path",
+    "rdma_device",
+    "rdma_gid_index",
+    "rdma_max_msg",
+    "rdma_max_recv",
+    "rdma_max_send",
+    "rdma_max_sge",
+    "rdma_mtu",
+    "rdma_port_num",
+    "tcp_header_buffer_size",
+    "use_rdma",
+    "use_sync_send_recv",
+};
+
+static const char **local_only_end = local_only_params + (sizeof(local_only_params)/sizeof(local_only_params[0]));
+
+// Basically could be replaced by std::lower_bound()...
+static int find_str_array(const char **start, const char **end, const std::string & s)
+{
+    int min = 0, max = end-start;
+    while (max-min >= 2)
+    {
+        int mid = (min+max)/2;
+        int r = strcmp(s.c_str(), start[mid]);
+        if (r < 0)
+            max = mid;
+        else if (r > 0)
+            min = mid;
+        else
+            return mid;
+    }
+    if (min < end-start && !strcmp(s.c_str(), start[min]))
+        return min;
+    return -1;
+}
+
+json11::Json::object osd_messenger_t::merge_configs(const json11::Json::object & cli_config,
+    const json11::Json::object & file_config,
+    const json11::Json::object & etcd_global_config,
+    const json11::Json::object & etcd_osd_config)
+{
+    // Priority: most important -> less important:
+    // etcd_osd_config -> cli_config -> etcd_global_config -> file_config
+    json11::Json::object res = file_config;
+    for (auto & kv: file_config)
+    {
+        int cli_only = find_str_array(cli_only_params, cli_only_end, kv.first);
+        if (cli_only < 0)
+        {
+            res[kv.first] = kv.second;
+        }
+    }
+    for (auto & kv: etcd_global_config)
+    {
+        int local_only = find_str_array(local_only_params, local_only_end, kv.first);
+        if (local_only < 0)
+        {
+            res[kv.first] = kv.second;
+        }
+    }
+    for (auto & kv: cli_config)
+    {
+        res[kv.first] = kv.second;
+    }
+    for (auto & kv: etcd_osd_config)
+    {
+        int local_only = find_str_array(local_only_params, local_only_end, kv.first);
+        if (local_only < 0)
+        {
+            res[kv.first] = kv.second;
+        }
+    }
+    return res;
+}

src/messenger.h

@@ -166,7 +166,11 @@ public:
     void accept_connections(int listen_fd);
     ~osd_messenger_t();
 
-    static json11::Json read_config(const json11::Json & config);
+    static json11::Json::object read_config(const json11::Json & config);
+    static json11::Json::object merge_configs(const json11::Json::object & cli_config,
+        const json11::Json::object & file_config,
+        const json11::Json::object & etcd_global_config,
+        const json11::Json::object & etcd_osd_config);
 
 #ifdef WITH_RDMA
     bool is_rdma_enabled();

src/mock/messenger.cpp

@@ -43,7 +43,15 @@ void osd_messenger_t::send_replies()
 {
 }
 
-json11::Json osd_messenger_t::read_config(const json11::Json & config)
+json11::Json::object osd_messenger_t::read_config(const json11::Json & config)
 {
-    return config;
+    return json11::Json::object();
+}
+
+json11::Json::object osd_messenger_t::merge_configs(const json11::Json::object & cli_config,
+    const json11::Json::object & file_config,
+    const json11::Json::object & etcd_global_config,
+    const json11::Json::object & etcd_osd_config)
+{
+    return cli_config;
 }

src/object_id.h

@@ -39,6 +39,11 @@ struct __attribute__((__packed__)) obj_ver_id
     uint64_t version;
 };
 
+inline bool operator == (const obj_ver_id & a, const obj_ver_id & b)
+{
+    return a.oid == b.oid && a.version == b.version;
+}
+
 inline bool operator < (const obj_ver_id & a, const obj_ver_id & b)
 {
     return a.oid < b.oid || a.oid == b.oid && a.version < b.version;

src/osd.cpp

@@ -35,10 +35,9 @@ osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop)
 
     this->ringloop = ringloop;
 
-    this->config = msgr.read_config(config).object_items();
-    if (this->config.find("log_level") == this->config.end())
-        this->config["log_level"] = 1;
-    parse_config(this->config, true);
+    this->cli_config = config.object_items();
+    this->file_config = msgr.read_config(this->cli_config);
+    parse_config(true);
 
     epmgr = new epoll_manager_t(ringloop);
     // FIXME: Use timerfd_interval based directly on io_uring
@@ -68,11 +67,11 @@ osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop)
         }
     }
 
-    this->tfd->set_timer(print_stats_interval*1000, true, [this](int timer_id)
+    print_stats_timer_id = this->tfd->set_timer(print_stats_interval*1000, true, [this](int timer_id)
     {
         print_stats();
     });
-    this->tfd->set_timer(slow_log_interval*1000, true, [this](int timer_id)
+    slow_log_timer_id = this->tfd->set_timer(slow_log_interval*1000, true, [this](int timer_id)
     {
         print_slow();
     });
@@ -92,6 +91,21 @@ osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop)
 
 osd_t::~osd_t()
 {
+    if (slow_log_timer_id >= 0)
+    {
+        tfd->clear_timer(slow_log_timer_id);
+        slow_log_timer_id = -1;
+    }
+    if (print_stats_timer_id >= 0)
+    {
+        tfd->clear_timer(print_stats_timer_id);
+        print_stats_timer_id = -1;
+    }
+    if (autosync_timer_id >= 0)
+    {
+        tfd->clear_timer(autosync_timer_id);
+        autosync_timer_id = -1;
+    }
     ringloop->unregister_consumer(&consumer);
     delete epmgr;
     if (bs)
@@ -100,11 +114,19 @@ osd_t::~osd_t()
     free(zero_buffer);
 }
 
-void osd_t::parse_config(const json11::Json & config, bool allow_disk_params)
+void osd_t::parse_config(bool init)
 {
+    config = msgr.merge_configs(cli_config, file_config, etcd_global_config, etcd_osd_config);
+    if (config.find("log_level") == this->config.end())
+        config["log_level"] = 1;
+    if (bs)
+    {
+        auto bs_cfg = json_to_bs(config);
+        bs->parse_config(bs_cfg);
+    }
     st_cli.parse_config(config);
     msgr.parse_config(config);
-    if (allow_disk_params)
+    if (init)
     {
         // OSD number
         osd_num = config["osd_num"].uint64_value();
@@ -126,24 +148,27 @@ void osd_t::parse_config(const json11::Json & config, bool allow_disk_params)
             immediate_commit = IMMEDIATE_SMALL;
         else
             immediate_commit = IMMEDIATE_NONE;
+        // Bind address
+        bind_address = config["bind_address"].string_value();
+        if (bind_address == "")
+            bind_address = "0.0.0.0";
+        bind_port = config["bind_port"].uint64_value();
+        if (bind_port <= 0 || bind_port > 65535)
+            bind_port = 0;
+        // OSD configuration
+        etcd_report_interval = config["etcd_report_interval"].uint64_value();
+        if (etcd_report_interval <= 0)
+            etcd_report_interval = 5;
+        readonly = json_is_true(config["readonly"]);
+        run_primary = !json_is_false(config["run_primary"]);
+        allow_test_ops = json_is_true(config["allow_test_ops"]);
     }
-    // Bind address
-    bind_address = config["bind_address"].string_value();
-    if (bind_address == "")
-        bind_address = "0.0.0.0";
-    bind_port = config["bind_port"].uint64_value();
-    if (bind_port <= 0 || bind_port > 65535)
-        bind_port = 0;
-    // OSD configuration
     log_level = config["log_level"].uint64_value();
-    etcd_report_interval = config["etcd_report_interval"].uint64_value();
-    if (etcd_report_interval <= 0)
-        etcd_report_interval = 5;
-    readonly = json_is_true(config["readonly"]);
-    run_primary = !json_is_false(config["run_primary"]);
+    auto old_no_rebalance = no_rebalance;
     no_rebalance = json_is_true(config["no_rebalance"]);
+    auto old_no_recovery = no_recovery;
     no_recovery = json_is_true(config["no_recovery"]);
-    allow_test_ops = json_is_true(config["allow_test_ops"]);
+    auto old_autosync_interval = autosync_interval;
     if (!config["autosync_interval"].is_null())
     {
         // Allow to set it to 0
@@ -171,15 +196,46 @@ void osd_t::parse_config(const json11::Json & config, bool allow_disk_params)
     recovery_sync_batch = config["recovery_sync_batch"].uint64_value();
     if (recovery_sync_batch < 1 || recovery_sync_batch > MAX_RECOVERY_QUEUE)
         recovery_sync_batch = DEFAULT_RECOVERY_BATCH;
+    auto old_print_stats_interval = print_stats_interval;
     print_stats_interval = config["print_stats_interval"].uint64_value();
     if (!print_stats_interval)
         print_stats_interval = 3;
+    auto old_slow_log_interval = slow_log_interval;
     slow_log_interval = config["slow_log_interval"].uint64_value();
     if (!slow_log_interval)
         slow_log_interval = 10;
     inode_vanish_time = config["inode_vanish_time"].uint64_value();
     if (!inode_vanish_time)
         inode_vanish_time = 60;
+    if ((old_no_rebalance && !no_rebalance || old_no_recovery && !no_recovery) &&
+        !(peering_state & (OSD_RECOVERING | OSD_FLUSHING_PGS)))
+    {
+        peering_state = peering_state | OSD_RECOVERING;
+    }
+    if (old_autosync_interval != autosync_interval && autosync_timer_id >= 0)
+    {
+        this->tfd->clear_timer(autosync_timer_id);
+        autosync_timer_id = this->tfd->set_timer(autosync_interval*1000, true, [this](int timer_id)
+        {
+            autosync();
+        });
+    }
+    if (old_print_stats_interval != print_stats_interval && print_stats_timer_id >= 0)
+    {
+        tfd->clear_timer(print_stats_timer_id);
+        print_stats_timer_id = this->tfd->set_timer(print_stats_interval*1000, true, [this](int timer_id)
+        {
+            print_stats();
+        });
+    }
+    if (old_slow_log_interval != slow_log_interval && slow_log_timer_id >= 0)
+    {
+        tfd->clear_timer(slow_log_timer_id);
+        slow_log_timer_id = this->tfd->set_timer(slow_log_interval*1000, true, [this](int timer_id)
+        {
+            print_slow();
+        });
+    }
 }
 
 void osd_t::bind_socket()

src/osd.h

@@ -90,7 +90,7 @@ class osd_t
 {
     // config
 
-    json11::Json::object config;
+    json11::Json::object cli_config, file_config, etcd_global_config, etcd_osd_config, config;
     int etcd_report_interval = 5;
 
     bool readonly = false;
@@ -126,6 +126,7 @@ class osd_t
     bool pg_config_applied = false;
     bool etcd_reporting_pg_state = false;
     bool etcd_reporting_stats = false;
+    int autosync_timer_id = -1, print_stats_timer_id = -1, slow_log_timer_id = -1;
 
     // peers and PGs
 
@@ -173,7 +174,7 @@ class osd_t
     uint64_t recovery_stat_bytes[2][2] = {};
 
     // cluster connection
-    void parse_config(const json11::Json & config, bool allow_disk_params);
+    void parse_config(bool init);
     void init_cluster();
     void on_change_osd_state_hook(osd_num_t peer_osd);
     void on_change_pg_history_hook(pool_id_t pool_id, pg_num_t pg_num);

src/osd_cluster.cpp

@@ -75,7 +75,7 @@ void osd_t::init_cluster()
     }
     if (run_primary && autosync_interval > 0)
     {
-        this->tfd->set_timer(autosync_interval*1000, true, [this](int timer_id)
+        autosync_timer_id = this->tfd->set_timer(autosync_interval*1000, true, [this](int timer_id)
         {
             autosync();
         });
@@ -375,7 +375,11 @@ void osd_t::on_change_osd_state_hook(osd_num_t peer_osd)
 
 void osd_t::on_change_etcd_state_hook(std::map<std::string, etcd_kv_t> & changes)
 {
-    // FIXME apply config changes in runtime (maybe, some)
+    if (changes.find(st_cli.etcd_prefix+"/config/global") != changes.end())
+    {
+        etcd_global_config = changes[st_cli.etcd_prefix+"/config/global"].value.object_items();
+        parse_config(false);
+    }
     if (run_primary)
     {
         apply_pg_count();
@@ -385,11 +389,8 @@ void osd_t::on_change_etcd_state_hook(std::map<std::string, etcd_kv_t> & changes
 
 void osd_t::on_load_config_hook(json11::Json::object & global_config)
 {
-    json11::Json::object osd_config = this->config;
-    for (auto & kv: global_config)
-        if (osd_config.find(kv.first) == osd_config.end())
-            osd_config[kv.first] = kv.second;
-    parse_config(osd_config, false);
+    etcd_global_config = global_config;
+    parse_config(true);
     bind_socket();
     acquire_lease();
 }

src/osd_flush.cpp

@@ -64,6 +64,11 @@ void osd_t::submit_pg_flush_ops(pg_t & pg)
 
 void osd_t::handle_flush_op(bool rollback, pool_id_t pool_id, pg_num_t pg_num, pg_flush_batch_t *fb, osd_num_t peer_osd, int retval)
 {
+    if (log_level > 2)
+    {
+        printf("[PG %u/%u] flush batch %lx completed on OSD %lu with result %d\n",
+            pool_id, pg_num, (uint64_t)fb, peer_osd, retval);
+    }
     pool_pg_num_t pg_id = { .pool_id = pool_id, .pg_num = pg_num };
     if (pgs.find(pg_id) == pgs.end() || pgs[pg_id].flush_batch != fb)
     {
@@ -99,10 +104,9 @@ void osd_t::handle_flush_op(bool rollback, pool_id_t pool_id, pg_num_t pg_num, p
         std::vector<osd_op_t*> continue_ops;
         auto & pg = pgs.at(pg_id);
         auto it = pg.flush_actions.begin(), prev_it = it;
-        auto erase_start = it;
         while (1)
         {
-            if (it == pg.flush_actions.end() ||
+            if (it == pg.flush_actions.end() || !it->second.submitted ||
                 it->first.oid.inode != prev_it->first.oid.inode ||
                 (it->first.oid.stripe & ~STRIPE_MASK) != (prev_it->first.oid.stripe & ~STRIPE_MASK))
             {
@@ -116,29 +120,23 @@ void osd_t::handle_flush_op(bool rollback, pool_id_t pool_id, pg_num_t pg_num, p
                 });
                 if (wr_it != pg.write_queue.end())
                 {
+                    if (log_level > 2)
+                    {
+                        printf("[PG %u/%u] continuing write %lx to object %lx:%lx after flush\n",
+                            pool_id, pg_num, (uint64_t)wr_it->second, wr_it->first.inode, wr_it->first.stripe);
+                    }
                     continue_ops.push_back(wr_it->second);
-                    pg.write_queue.erase(wr_it);
                 }
             }
-            if ((it == pg.flush_actions.end() || !it->second.submitted) &&
-                erase_start != it)
-            {
-                pg.flush_actions.erase(erase_start, it);
-            }
-            if (it == pg.flush_actions.end())
+            if (it == pg.flush_actions.end() || !it->second.submitted)
             {
+                if (it != pg.flush_actions.begin())
+                {
+                    pg.flush_actions.erase(pg.flush_actions.begin(), it);
+                }
                 break;
             }
-            prev_it = it;
-            if (!it->second.submitted)
-            {
-                it++;
-                erase_start = it;
-            }
-            else
-            {
-                it++;
-            }
+            prev_it = it++;
         }
         delete fb;
         pg.flush_batch = NULL;
@@ -168,6 +166,18 @@ bool osd_t::submit_flush_op(pool_id_t pool_id, pg_num_t pg_num, pg_flush_batch_t
     // Copy buffer so it gets freed along with the operation
     op->buf = malloc_or_die(sizeof(obj_ver_id) * count);
     memcpy(op->buf, data, sizeof(obj_ver_id) * count);
+    if (log_level > 2)
+    {
+        printf(
+            "[PG %u/%u] flush batch %lx on OSD %lu: %s objects: ",
+            pool_id, pg_num, (uint64_t)fb, peer_osd, rollback ? "rollback" : "stabilize"
+        );
+        for (int i = 0; i < count; i++)
+        {
+            printf(i > 0 ? ", %lx:%lx v%lu" : "%lx:%lx v%lu", data[i].oid.inode, data[i].oid.stripe, data[i].version);
+        }
+        printf("\n");
+    }
     if (peer_osd == this->osd_num)
     {
         // local
@@ -304,9 +314,10 @@ void osd_t::submit_recovery_op(osd_recovery_op_t *op)
             {
                 // PG is stopped or one of the OSDs is gone, error is harmless
                 printf(
-                    "Recovery operation failed with object %lx:%lx (PG %u/%u)\n",
-                    op->oid.inode, op->oid.stripe, INODE_POOL(op->oid.inode),
-                    map_to_pg(op->oid, st_cli.pool_config.at(INODE_POOL(op->oid.inode)).pg_stripe_size)
+                    "[PG %u/%u] Recovery operation failed with object %lx:%lx\n",
+                    INODE_POOL(op->oid.inode),
+                    map_to_pg(op->oid, st_cli.pool_config.at(INODE_POOL(op->oid.inode)).pg_stripe_size),
+                    op->oid.inode, op->oid.stripe
                 );
             }
             else

src/osd_peering.cpp

@@ -76,7 +76,7 @@ void osd_t::handle_peers()
             peering_state = peering_state & ~OSD_FLUSHING_PGS | OSD_RECOVERING;
         }
     }
-    if ((peering_state & OSD_RECOVERING) && !readonly)
+    if (!(peering_state & OSD_FLUSHING_PGS) && (peering_state & OSD_RECOVERING) && !readonly)
     {
         if (!continue_recovery())
         {

src/osd_peering_pg.cpp

@@ -91,7 +91,7 @@ void pg_obj_state_check_t::walk()
         pg->state |= PG_DEGRADED;
     }
     pg->state |= PG_ACTIVE;
-    if (pg->state == PG_ACTIVE && pg->cur_peers.size() < pg->all_peers.size())
+    if (pg->cur_peers.size() < pg->all_peers.size())
     {
         pg->state |= PG_LEFT_ON_DEAD;
     }

src/osd_primary_sync.cpp

@@ -166,7 +166,7 @@ resume_6:
                 for (int i = 0; i < unstable_osd.len; i++)
                 {
                     // Except those from peered PGs
-                    auto & w = op_data->unstable_writes[i];
+                    auto & w = op_data->unstable_writes[unstable_osd.start + i];
                     pool_pg_num_t wpg = {
                         .pool_id = INODE_POOL(w.oid.inode),
                         .pg_num = map_to_pg(w.oid, st_cli.pool_config.at(INODE_POOL(w.oid.inode)).pg_stripe_size),

src/osd_primary_write.cpp

@@ -12,6 +12,7 @@ bool osd_t::check_write_queue(osd_op_t *cur_op, pg_t & pg)
         .oid = op_data->oid,
         .osd_num = 0,
     });
+    op_data->st = 1;
     if (act_it != pg.flush_actions.end() &&
         act_it->first.oid.inode == op_data->oid.inode &&
         (act_it->first.oid.stripe & ~STRIPE_MASK) == op_data->oid.stripe)
@@ -23,7 +24,6 @@ bool osd_t::check_write_queue(osd_op_t *cur_op, pg_t & pg)
     auto vo_it = pg.write_queue.find(op_data->oid);
     if (vo_it != pg.write_queue.end())
     {
-        op_data->st = 1;
         pg.write_queue.emplace(op_data->oid, cur_op);
         return false;
     }

src/osd_rmw.cpp

@@ -142,11 +142,11 @@ inline bool operator < (const reed_sol_erased_t &a, const reed_sol_erased_t &b)
     for (int i = 0; i < a.size && i < b.size; i++)
     {
         if (a.data[i] < b.data[i])
-            return -1;
+            return true;
         else if (a.data[i] > b.data[i])
-            return 1;
+            return false;
     }
-    return 0;
+    return false;
 }
 
 struct reed_sol_matrix_t
@@ -677,11 +677,11 @@ void* calc_rmw(void *request_buf, osd_rmw_stripe_t *stripes, uint64_t *read_osd_
 static void get_old_new_buffers(osd_rmw_stripe_t & stripe, uint32_t wr_start, uint32_t wr_end, buf_len_t *bufs, int & nbufs)
 {
     uint32_t ns = 0, ne = 0, os = 0, oe = 0;
-    if (stripe.req_end > wr_start &&
-        stripe.req_start < wr_end)
+    if (stripe.write_end > wr_start &&
+        stripe.write_start < wr_end)
     {
-        ns = std::max(stripe.req_start, wr_start);
-        ne = std::min(stripe.req_end, wr_end);
+        ns = std::max(stripe.write_start, wr_start);
+        ne = std::min(stripe.write_end, wr_end);
     }
     if (stripe.read_end > wr_start &&
         stripe.read_start < wr_end)
@@ -692,7 +692,7 @@ static void get_old_new_buffers(osd_rmw_stripe_t & stripe, uint32_t wr_start, ui
     if (ne && (!oe || ns <= os))
     {
         // NEW or NEW->OLD
-        bufs[nbufs++] = { .buf = (uint8_t*)stripe.write_buf + ns - stripe.req_start, .len = ne-ns };
+        bufs[nbufs++] = { .buf = (uint8_t*)stripe.write_buf + ns - stripe.write_start, .len = ne-ns };
         if (os < ne)
             os = ne;
         if (oe > os)
@@ -708,7 +708,7 @@ static void get_old_new_buffers(osd_rmw_stripe_t & stripe, uint32_t wr_start, ui
         {
             // OLD->NEW or OLD->NEW->OLD
             bufs[nbufs++] = { .buf = (uint8_t*)stripe.read_buf + os - stripe.read_start, .len = ns-os };
-            bufs[nbufs++] = { .buf = (uint8_t*)stripe.write_buf + ns - stripe.req_start, .len = ne-ns };
+            bufs[nbufs++] = { .buf = (uint8_t*)stripe.write_buf + ns - stripe.write_start, .len = ne-ns };
             if (oe > ne)
             {
                 // OLD->NEW->OLD

src/osd_rmw_test.cpp

@@ -17,6 +17,7 @@ void test4();
 void test5();
 void test6();
 void test7();
+void test_rmw_4k_degraded_into_lost_to_normal(bool ec);
 void test8();
 void test9();
 void test10();
@@ -39,6 +40,8 @@ int main(int narg, char *args[])
     test6();
     // Test 7
     test7();
+    test_rmw_4k_degraded_into_lost_to_normal(false);
+    test_rmw_4k_degraded_into_lost_to_normal(true);
     // Test 8
     test8();
     // Test 9
@@ -316,6 +319,69 @@ void test7()
 
 /***
 
+7/2. calc_rmw(offset=48K, len=4K, osd_set=[0,2,3], write_set=[1,2,3])
+   = {
+     read: [ [ 0, 128K ], [ 0, 128K ], [ 0, 128K ] ],
+     write: [ [ 48K, 52K ], [ 0, 0 ], [ 48K, 52K ] ],
+     input buffer: [ write0 ],
+     rmw buffer: [ write2, read0, read1, read2 ],
+   }
+   then, after calc_rmw_parity_xor/ec(): {
+     write: [ [ 0, 128K ], [ 0, 0 ], [ 48K, 52K ] ],
+     write0==read0,
+   }
+   + check write0, write2 buffers
+
+***/
+
+void test_rmw_4k_degraded_into_lost_to_normal(bool ec)
+{
+    osd_num_t osd_set[3] = { 0, 2, 3 };
+    osd_num_t write_osd_set[3] = { 1, 2, 3 };
+    osd_rmw_stripe_t stripes[3] = {};
+    // Subtest 1
+    split_stripes(2, 128*1024, 48*1024, 4096, stripes);
+    void *write_buf = malloc(4096);
+    void *rmw_buf = calc_rmw(write_buf, stripes, osd_set, 3, 2, 3, write_osd_set, 128*1024, 0);
+    assert(stripes[0].read_start == 0 && stripes[0].read_end == 128*1024);
+    assert(stripes[1].read_start == 0 && stripes[1].read_end == 128*1024);
+    assert(stripes[2].read_start == 0 && stripes[2].read_end == 128*1024);
+    assert(stripes[0].write_start == 48*1024 && stripes[0].write_end == 52*1024);
+    assert(stripes[1].write_start == 0 && stripes[1].write_end == 0);
+    assert(stripes[2].write_start == 48*1024 && stripes[2].write_end == 52*1024);
+    assert(stripes[0].read_buf == (uint8_t*)rmw_buf+4*1024);
+    assert(stripes[1].read_buf == (uint8_t*)rmw_buf+4*1024+128*1024);
+    assert(stripes[2].read_buf == (uint8_t*)rmw_buf+4*1024+2*128*1024);
+    assert(stripes[0].write_buf == write_buf);
+    assert(stripes[1].write_buf == NULL);
+    assert(stripes[2].write_buf == rmw_buf);
+    // Subtest 2
+    set_pattern(write_buf, 4096, PATTERN2);
+    set_pattern(stripes[1].read_buf, 128*1024, PATTERN1);
+    set_pattern(stripes[2].read_buf, 128*1024, PATTERN0^PATTERN1);
+    if (!ec)
+        calc_rmw_parity_xor(stripes, 3, osd_set, write_osd_set, 128*1024, 0);
+    else
+    {
+        use_ec(3, 2, true);
+        calc_rmw_parity_ec(stripes, 3, 2, osd_set, write_osd_set, 128*1024, 0);
+        use_ec(3, 2, false);
+    }
+    assert(stripes[0].write_start == 0 && stripes[0].write_end == 128*1024);
+    assert(stripes[1].write_start == 0 && stripes[1].write_end == 0);
+    assert(stripes[2].write_start == 48*1024 && stripes[2].write_end == 52*1024);
+    assert(stripes[0].write_buf == stripes[0].read_buf);
+    assert(stripes[1].write_buf == NULL);
+    assert(stripes[2].write_buf == rmw_buf);
+    check_pattern(stripes[0].write_buf, 4096, PATTERN0);
+    check_pattern(stripes[0].write_buf+48*1024, 4096, PATTERN2);
+    check_pattern(stripes[2].write_buf, 4096, PATTERN2^PATTERN1); // new parity
+    free(rmw_buf);
+    free(write_buf);
+}
+
+/***
+
 8. calc_rmw(offset=0, len=128K+4K, osd_set=[0,2,3], write_set=[1,2,3])
    = {
      read: [ [ 0, 0 ], [ 4K, 128K ], [ 0, 0 ] ],

src/vitastor.pc.in

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

tests/common.sh

@@ -64,4 +64,4 @@ echo leak:librbd >> testdata/lsan-suppress.txt
 echo leak:_M_mutate >> testdata/lsan-suppress.txt
 echo leak:_M_assign >> testdata/lsan-suppress.txt
 export LSAN_OPTIONS=report_objects=true:suppressions=`pwd`/testdata/lsan-suppress.txt
-export ASAN_OPTIONS=verify_asan_link_order=false
+export ASAN_OPTIONS=verify_asan_link_order=false:abort_on_error=1

tests/run_3osds.sh

@@ -17,17 +17,17 @@ else
 fi
 
 if [ "$IMMEDIATE_COMMIT" != "" ]; then
-    NO_SAME="--journal_no_same_sector_overwrites true --journal_sector_buffer_count 1024 --disable_data_fsync 1 --immediate_commit all --log_level 1"
+    NO_SAME="--journal_no_same_sector_overwrites true --journal_sector_buffer_count 1024 --disable_data_fsync 1 --immediate_commit all --log_level 10"
     $ETCDCTL put /vitastor/config/global '{"recovery_queue_depth":1,"osd_out_time":1,"immediate_commit":"all"}'
 else
-    NO_SAME="--journal_sector_buffer_count 1024 --log_level 1"
+    NO_SAME="--journal_sector_buffer_count 1024 --log_level 10"
     $ETCDCTL put /vitastor/config/global '{"recovery_queue_depth":1,"osd_out_time":1}'
 fi
 
 start_osd()
 {
     local i=$1
-    build/src/vitastor-osd --osd_num $i --bind_address 127.0.0.1 $NO_SAME $OSD_ARGS --etcd_address $ETCD_URL $(build/src/vitastor-disk simple-offsets --format options ./testdata/test_osd$i.bin 2>/dev/null) &>./testdata/osd$i.log &
+    build/src/vitastor-osd --osd_num $i --bind_address 127.0.0.1 $NO_SAME $OSD_ARGS --etcd_address $ETCD_URL $(build/src/vitastor-disk simple-offsets --format options ./testdata/test_osd$i.bin 2>/dev/null) >>./testdata/osd$i.log 2>&1 &
     eval OSD${i}_PID=$!
 }
 

tests/run_tests.sh

@@ -43,3 +43,6 @@ SCHEME=ec ./test_snapshot.sh
 SCHEME=xor ./test_write.sh
 
 ./test_write_no_same.sh
+
+./test_heal.sh
+SCHEME=ec PG_MINSIZE=2 ./test_heal.sh

tests/test_heal.sh

@@ -43,7 +43,7 @@ kill_osds &
 
 LD_PRELOAD="build/src/libfio_vitastor.so" \
     fio -thread -name=test -ioengine=build/src/libfio_vitastor.so -bs=4k -direct=1 -iodepth=16 -fsync=256 -rw=randwrite \
-        -mirror_file=./testdata/mirror.bin -etcd=$ETCD_URL -image=testimg -loops=10 -runtime=120 2>/dev/null
+        -mirror_file=./testdata/mirror.bin -etcd=$ETCD_URL -image=testimg -loops=10 -runtime=120
 
 qemu-img convert -S 4096 -p \
     -f raw "vitastor:etcd_host=127.0.0.1\:$ETCD_PORT/v3:image=testimg" \

tests/test_move_reappear.sh

@@ -7,7 +7,7 @@ OSD_COUNT=5
 OSD_ARGS=
 for i in $(seq 1 $OSD_COUNT); do
     dd if=/dev/zero of=./testdata/test_osd$i.bin bs=1024 count=1 seek=$((OSD_SIZE*1024-1))
-    build/src/vitastor-osd --osd_num $i --bind_address 127.0.0.1 $OSD_ARGS --etcd_address $ETCD_URL $(build/src/vitastor-disk simple-offsets --format options ./testdata/test_osd$i.bin 2>/dev/null) &>./testdata/osd$i.log &
+    build/src/vitastor-osd --osd_num $i --bind_address 127.0.0.1 $OSD_ARGS --etcd_address $ETCD_URL $(build/src/vitastor-disk simple-offsets --format options ./testdata/test_osd$i.bin 2>/dev/null) >>./testdata/osd$i.log 2>&1 &
     eval OSD${i}_PID=$!
 done
 

tests/test_resize.sh

@@ -53,7 +53,7 @@ for i in $(seq 1 $OSD_COUNT); do
         --data_device ./testdata/test_osd$i.bin \
         --meta_offset 0 \
         --journal_offset $((1024*1024)) \
-        --data_offset $((128*1024*1024)) &>./testdata/osd$i.log &
+        --data_offset $((128*1024*1024)) >>./testdata/osd$i.log 2>&1 &
     eval OSD${i}_PID=$!
 done
 

tests/test_splitbrain.sh

@@ -21,7 +21,8 @@ LD_PRELOAD="build/src/libfio_vitastor.so" \
 # Kill OSD 2, start OSD 1
 
 kill $OSD2_PID
-build/src/vitastor-osd --osd_num 1 --bind_address 127.0.0.1 $OSD_ARGS --etcd_address $ETCD_URL $(build/src/vitastor-disk simple-offsets --format options --device ./testdata/test_osd2.bin 2>/dev/null) >>./testdata/osd2.log 2>&1 &
+build/src/vitastor-osd --osd_num 1 --bind_address 127.0.0.1 $OSD_ARGS --etcd_address $ETCD_URL \
+    $(build/src/vitastor-disk simple-offsets --format options --device ./testdata/test_osd2.bin 2>/dev/null) >>./testdata/osd2.log 2>&1 &
 sleep 2
 
 # Check PG state - it should NOT become active

tests/test_vm_cont.sh

@@ -10,7 +10,7 @@ etcdctl --endpoints=http://127.0.0.1:12379/v3 del --prefix /vitastor/osd/state
 OSD_COUNT=3
 OSD_ARGS=
 for i in $(seq 1 $OSD_COUNT); do
-    build/src/vitastor-osd --osd_num $i --bind_address 127.0.0.1 $OSD_ARGS --etcd_address $ETCD_URL $(build/src/vitastor-disk simple-offsets --format options ./testdata/test_osd$i.bin 2>/dev/null) &>./testdata/osd$i.log &
+    build/src/vitastor-osd --osd_num $i --bind_address 127.0.0.1 $OSD_ARGS --etcd_address $ETCD_URL $(build/src/vitastor-disk simple-offsets --format options ./testdata/test_osd$i.bin 2>/dev/null) >>./testdata/osd$i.log 2>&1 &
     eval OSD${i}_PID=$!
 done