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 :)
Fix a small memory leak caused by bad indexing of EC recovery matrices
2023-04-11 02:11:57 +03:00 · 2023-04-11 00:30:36 +03:00 · 2023-04-09 02:10:42 +03:00 · 2023-04-09 02:06:10 +03:00 · 2023-04-09 02:06:10 +03:00 · 2023-04-09 02:06:10 +03:00
61 changed files with 888 additions and 285 deletions
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -1,7 +1,7 @@
-cmake_minimum_required(VERSION 2.8)
+cmake_minimum_required(VERSION 2.8.12)

 project(vitastor)

-set(VERSION "0.8.6")
+set(VERSION "0.8.7")

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

 all: build push

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

 const (
    vitastorCSIDriverName    = "csi.vitastor.io"
-    vitastorCSIDriverVersion = "0.8.6"
+    vitastorCSIDriverVersion = "0.8.7"
 )

 // Config struct fills the parameters of request or user input
--- a/csi/src/controllerserver.go
+++ b/csi/src/controllerserver.go
@@ -6,6 +6,7 @@ package vitastor
 import (
    "context"
    "encoding/json"
+    "fmt"
    "strings"
    "bytes"
    "strconv"
@@ -178,7 +179,7 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
    }

    // Create image using vitastor-cli
-    _, err := invokeCLI(ctxVars, []string{ "create", volName, "-s", string(volSize), "--pool", string(poolId) })
+    _, err := invokeCLI(ctxVars, []string{ "create", volName, "-s", fmt.Sprintf("%v", volSize), "--pool", fmt.Sprintf("%v", poolId) })
    if (err != nil)
    {
        if (strings.Index(err.Error(), "already exists") > 0)
--- a/debian/changelog
+++ b/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
--- a/debian/vitastor.Dockerfile
+++ b/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; \
--- a/docs/performance/theoretical.en.md
+++ b/docs/performance/theoretical.en.md
@@ -35,15 +35,24 @@ Write amplification for 4 KB blocks is usually 3-5 in Vitastor:
 If you manage to get an SSD which handles 512 byte blocks well (Optane?) you may
 lower 1, 3 and 4 to 512 bytes (1/8 of data size) and get WA as low as 2.375.

+Implemented NVDIMM support can basically eliminate WA at all - all extra writes will
+go to DRAM memory. But this requires a test cluster with NVDIMM - please contact me
+if you want to provide me with such cluster for tests.
+
 Lazy fsync also reduces WA for parallel workloads because journal blocks are only
 written when they fill up or fsync is requested.

 ## In Practice

-In practice, using tests from [Understanding Performance](understanding.en.md)
-and good server-grade SSD/NVMe drives, you should head for:
+In practice, using tests from [Understanding Performance](understanding.en.md), decent TCP network,
+good server-grade SSD/NVMe drives and disabled CPU power saving, you should head for:
 - At least 5000 T1Q1 replicated read and write iops (maximum 0.2ms latency)
+- At least 5000 T1Q1 EC read IOPS and at least 2200 EC write IOPS (maximum 0.45ms latency)
 - At least ~80k parallel read iops or ~30k write iops per 1 core (1 OSD)
 - Disk-speed or wire-speed linear reads and writes, whichever is the bottleneck in your case

 Lower results may mean that you have bad drives, bad network or some kind of misconfiguration.
+
+Current latency records:
+- 9668 T1Q1 replicated write iops (0.103 ms latency) with TCP and NVMe
+- 9143 T1Q1 replicated read iops (0.109 ms latency) with TCP and NVMe
--- a/docs/performance/theoretical.ru.md
+++ b/docs/performance/theoretical.ru.md
@@ -36,6 +36,25 @@ WA (мультипликатор записи) для 4 КБ блоков в Vit
 Если вы найдёте SSD, хорошо работающий с 512-байтными блоками данных (Optane?),
 то 1, 3 и 4 можно снизить до 512 байт (1/8 от размера данных) и получить WA всего 2.375.

+Если реализовать поддержку NVDIMM, то WA можно, условно говоря, ликвидировать вообще - все
+дополнительные операции записи смогут обслуживаться DRAM памятью. Но для этого необходим
+тестовый кластер с NVDIMM - пишите, если готовы предоставить такой для тестов.
+
 Кроме того, WA снижается при использовании отложенного/ленивого сброса при параллельной
 нагрузке, т.к. блоки журнала записываются на диск только когда они заполняются или явным
 образом запрашивается fsync.
+
+## На практике
+
+На практике, используя тесты fio со страницы [Понимание сути производительности систем хранения](understanding.ru.md),
+нормальную TCP-сеть, хорошие серверные SSD/NVMe, при отключённом энергосбережении процессоров вы можете рассчитывать на:
+- От 5000 IOPS в 1 поток (T1Q1) и на чтение, и на запись при использовании репликации (задержка до 0.2мс)
+- От 5000 IOPS в 1 поток (T1Q1) на чтение и 2200 IOPS в 1 поток на запись при использовании EC (задержка до 0.45мс)
+- От 80000 IOPS на чтение в параллельном режиме на 1 ядро, от 30000 IOPS на запись на 1 ядро (на 1 OSD)
+- Скорость параллельного линейного чтения и записи, равная меньшему значению из скорости дисков или сети
+
+Худшие результаты означают, что у вас либо медленные диски, либо медленная сеть, либо что-то неправильно настроено.
+
+Зафиксированный на данный момент рекорд задержки:
+- 9668 IOPS (0.103 мс задержка) в 1 поток (T1Q1) на запись с TCP и NVMe при использовании репликации
+- 9143 IOPS (0.109 мс задержка) в 1 поток (T1Q1) на чтение с TCP и NVMe при использовании репликации
--- a/mon/mon.js
+++ b/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,
--- a/patches/cinder-vitastor.py
+++ b/patches/cinder-vitastor.py
@@ -50,7 +50,7 @@ from cinder.volume import configuration
 from cinder.volume import driver
 from cinder.volume import volume_utils

-VERSION = '0.8.6'
+VERSION = '0.8.7'

 LOG = logging.getLogger(__name__)

--- a/rpm/build-tarball.sh
+++ b/rpm/build-tarball.sh
@@ -25,4 +25,4 @@ rm fio
 mv fio-copy fio
 FIO=`rpm -qi fio | perl -e 'while(<>) { /^Epoch[\s:]+(\S+)/ && print "$1:"; /^Version[\s:]+(\S+)/ && print $1; /^Release[\s:]+(\S+)/ && print "-$1"; }'`
 perl -i -pe 's/(Requires:\s*fio)([^\n]+)?/$1 = '$FIO'/' $VITASTOR/rpm/vitastor-el$EL.spec
-tar --transform 's#^#vitastor-0.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 *
--- a/rpm/vitastor-el7.Dockerfile
+++ b/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; \
--- a/rpm/vitastor-el7.spec
+++ b/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
--- a/rpm/vitastor-el8.Dockerfile
+++ b/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; \
--- a/rpm/vitastor-el8.spec
+++ b/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
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -1,4 +1,4 @@
-cmake_minimum_required(VERSION 2.8)
+cmake_minimum_required(VERSION 2.8.12)

 project(vitastor)

@@ -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)
--- a/src/blockstore.cpp
+++ b/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();
--- a/src/blockstore.h
+++ b/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();

--- a/src/blockstore_impl.cpp
+++ b/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)
--- a/src/blockstore_impl.h
+++ b/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();

--- a/src/blockstore_open.cpp
+++ b/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;
--- a/src/blockstore_rollback.cpp
+++ b/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)
--- a/src/blockstore_stable.cpp
+++ b/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)
        {
--- a/src/blockstore_sync.cpp
+++ b/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;
--- a/src/cli_rm_osd.cpp
+++ b/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;
--- a/src/cli_status.cpp
+++ b/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
--- a/src/cluster_client.cpp
+++ b/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);
+        }
    }
 }

--- a/src/cluster_client.h
+++ b/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();
--- a/src/disk_tool_journal.cpp
+++ b/src/disk_tool_journal.cpp
@@ -281,7 +281,7 @@ void disk_tool_t::dump_journal_entry(int num, journal_entry *je, bool json)
        if (je->big_write.size > sizeof(journal_entry_big_write))
        {
            printf(json ? ",\"bitmap\":\"" : " (bitmap: ");
-            for (int i = sizeof(journal_entry_big_write); i < je->small_write.size; i++)
+            for (int i = sizeof(journal_entry_big_write); i < je->big_write.size; i++)
            {
                printf("%02x", ((uint8_t*)je)[i]);
            }
--- a/src/disk_tool_meta.cpp
+++ b/src/disk_tool_meta.cpp
@@ -26,7 +26,7 @@ int disk_tool_t::process_meta(std::function<void(blockstore_meta_header_v1_t *)>
        buf_size = dsk.meta_len;
    void *data = memalign_or_die(MEM_ALIGNMENT, buf_size);
    lseek64(dsk.meta_fd, dsk.meta_offset, 0);
-    read_blocking(dsk.meta_fd, data, buf_size);
+    read_blocking(dsk.meta_fd, data, dsk.meta_block_size);
    // Check superblock
    blockstore_meta_header_v1_t *hdr = (blockstore_meta_header_v1_t *)data;
    if (hdr->zero == 0 &&
@@ -41,8 +41,11 @@ int disk_tool_t::process_meta(std::function<void(blockstore_meta_header_v1_t *)>
            if (buf_size % dsk.meta_block_size)
            {
                buf_size = 8*dsk.meta_block_size;
+                void *new_data = memalign_or_die(MEM_ALIGNMENT, buf_size);
+                memcpy(new_data, data, dsk.meta_block_size);
                free(data);
-                data = memalign_or_die(MEM_ALIGNMENT, buf_size);
+                data = new_data;
+                hdr = (blockstore_meta_header_v1_t *)data;
            }
        }
        dsk.bitmap_granularity = hdr->bitmap_granularity;
--- a/src/etcd_state_client.cpp
+++ b/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)
--- a/src/etcd_state_client.h
+++ b/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);
--- a/src/messenger.cpp
+++ b/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;
+}
--- a/src/messenger.h
+++ b/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();
--- a/src/mock/messenger.cpp
+++ b/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;
 }
--- a/src/object_id.h
+++ b/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;
--- a/src/osd.cpp
+++ b/src/osd.cpp
@@ -35,18 +35,18 @@ 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
    this->tfd = epmgr->tfd;

-    auto bs_cfg = json_to_bs(this->config);
-    this->bs = new blockstore_t(bs_cfg, ringloop, tfd);
+    if (!json_is_true(this->config["disable_blockstore"]))
    {
+        auto bs_cfg = json_to_bs(this->config);
+        this->bs = new blockstore_t(bs_cfg, ringloop, tfd);
        // Autosync based on the number of unstable writes to prevent stalls due to insufficient journal space
        uint64_t max_autosync = bs->get_journal_size() / bs->get_block_size() / 2;
        if (autosync_writes > max_autosync)
@@ -67,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();
    });
@@ -91,18 +91,42 @@ 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;
-    delete bs;
+    if (bs)
+        delete bs;
    close(listen_fd);
    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();
@@ -124,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
@@ -169,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()
@@ -475,7 +533,7 @@ void osd_t::print_slow()
            }
        }
    }
-    if (has_slow)
+    if (has_slow && bs)
    {
        bs->dump_diagnostics();
    }
--- a/src/osd.h
+++ b/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

@@ -152,7 +153,7 @@ class osd_t

    bool stopping = false;
    int inflight_ops = 0;
-    blockstore_t *bs;
+    blockstore_t *bs = NULL;
    void *zero_buffer = NULL;
    uint64_t zero_buffer_size = 0;
    uint32_t bs_block_size, bs_bitmap_granularity, clean_entry_bitmap_size;
@@ -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);
--- a/src/osd_cluster.cpp
+++ b/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();
        });
@@ -182,10 +182,10 @@ json11::Json osd_t::get_statistics()
    char time_str[50] = { 0 };
    sprintf(time_str, "%ld.%03ld", ts.tv_sec, ts.tv_nsec/1000000);
    st["time"] = time_str;
-    st["blockstore_ready"] = bs->is_started();
-    st["data_block_size"] = (uint64_t)bs->get_block_size();
    if (bs)
    {
+        st["blockstore_ready"] = bs->is_started();
+        st["data_block_size"] = (uint64_t)bs->get_block_size();
        st["size"] = bs->get_block_count() * bs->get_block_size();
        st["free"] = bs->get_free_block_count() * bs->get_block_size();
    }
@@ -233,7 +233,8 @@ void osd_t::report_statistics()
    json11::Json::object inode_space;
    json11::Json::object last_stat;
    pool_id_t last_pool = 0;
-    auto & bs_inode_space = bs->get_inode_space_stats();
+    std::map<uint64_t, uint64_t> bs_empty_space;
+    auto & bs_inode_space = bs ? bs->get_inode_space_stats() : bs_empty_space;
    for (auto kv: bs_inode_space)
    {
        pool_id_t pool_id = INODE_POOL(kv.first);
@@ -374,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();
@@ -384,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();
 }
--- a/src/osd_flush.cpp
+++ b/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
--- a/src/osd_peering.cpp
+++ b/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())
        {
--- a/src/osd_peering_pg.cpp
+++ b/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;
    }
--- a/src/osd_primary_subops.cpp
+++ b/src/osd_primary_subops.cpp
@@ -53,7 +53,10 @@ void osd_t::finish_op(osd_op_t *cur_op, int retval)
        inode_stats[cur_op->req.rw.inode].op_count[inode_st_op]++;
        inode_stats[cur_op->req.rw.inode].op_sum[inode_st_op] += usec;
        if (cur_op->req.hdr.opcode == OSD_OP_DELETE)
-            inode_stats[cur_op->req.rw.inode].op_bytes[inode_st_op] += cur_op->op_data->pg_data_size * bs_block_size;
+        {
+            if (cur_op->op_data)
+                inode_stats[cur_op->req.rw.inode].op_bytes[inode_st_op] += cur_op->op_data->pg_data_size * bs_block_size;
+        }
        else
            inode_stats[cur_op->req.rw.inode].op_bytes[inode_st_op] += cur_op->req.rw.len;
    }
--- a/src/osd_primary_sync.cpp
+++ b/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),
--- a/src/osd_primary_write.cpp
+++ b/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;
    }
--- a/src/osd_rmw.cpp
+++ b/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
--- a/src/osd_rmw_test.cpp
+++ b/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 ] ],
--- a/src/osd_test.cpp
+++ b/src/osd_test.cpp
@@ -150,6 +150,7 @@ int connect_osd(const char *osd_address, int osd_port)
    if (connect(connect_fd, (sockaddr*)&addr, sizeof(addr)) < 0)
    {
        perror("connect");
+        close(connect_fd);
        return -1;
    }
    int one = 1;
--- a/src/rw_blocking.cpp
+++ b/src/rw_blocking.cpp
@@ -15,7 +15,7 @@ int read_blocking(int fd, void *read_buf, size_t remaining)
    size_t done = 0;
    while (done < remaining)
    {
-        size_t r = read(fd, read_buf, remaining-done);
+        ssize_t r = read(fd, read_buf, remaining-done);
        if (r <= 0)
        {
            if (!errno)
@@ -41,7 +41,7 @@ int write_blocking(int fd, void *write_buf, size_t remaining)
    size_t done = 0;
    while (done < remaining)
    {
-        size_t r = write(fd, write_buf, remaining-done);
+        ssize_t r = write(fd, write_buf, remaining-done);
        if (r < 0)
        {
            if (errno != EINTR && errno != EAGAIN && errno != EPIPE)
--- a/src/stub_bench.cpp
+++ b/src/stub_bench.cpp
@@ -83,6 +83,7 @@ int connect_stub(const char *server_address, int server_port)
    if (connect(connect_fd, (sockaddr*)&addr, sizeof(addr)) < 0)
    {
        perror("connect");
+        close(connect_fd);
        return -1;
    }
    int one = 1;
--- a/src/vitastor.pc.in
+++ b/src/vitastor.pc.in
@@ -6,7 +6,7 @@ includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@

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

--- a/tests/common.sh
+++ b/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
--- a/tests/run_3osds.sh
+++ b/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=$!
 }

--- a/tests/run_tests.sh
+++ b/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
--- a/tests/test_heal.sh
+++ b/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" \
--- a/tests/test_move_reappear.sh
+++ b/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

--- a/tests/test_resize.sh
+++ b/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

--- a/tests/test_splitbrain.sh
+++ b/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
--- a/tests/test_vm_cont.sh
+++ b/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
Author	SHA1	Message	Date
Vitaliy Filippov	7e958afeda	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 :)	2023-04-11 02:11:57 +03:00
Vitaliy Filippov	2f5e769a29	Fix a small memory leak caused by bad indexing of EC recovery matrices	2023-04-11 00:30:36 +03:00
Vitaliy Filippov	28d5e53c6c	Add test_heal to run_tests	2023-04-09 02:10:42 +03:00
Vitaliy Filippov	d9f55f11d8	More logs (log_level 10), append to log instead of overwriting on restart in tests	2023-04-09 02:06:10 +03:00
Vitaliy Filippov	3237014608	Fix incorrect parity read-modify-write calculation when writing into a lost chunk	2023-04-09 02:06:10 +03:00
Vitaliy Filippov	baaf8f6f44	Fix write operations not continued correctly after flush	2023-04-09 02:06:10 +03:00
Vitaliy Filippov	1d83fdcd17	Add debug logs to osd_flush	2023-04-09 02:06:10 +03:00
Vitaliy Filippov	0ddd787c38	Fix non-working "pagination" during PG dirty object flushing	2023-04-08 02:44:02 +03:00
Vitaliy Filippov	6eff3a60a5	Do not lose left_on_dead PG state of non-clean PGs	2023-04-08 02:44:02 +03:00
Vitaliy Filippov	888a6975ab	Fix a rare use-after-free in cluster_client caused by a reenterability issue	2023-04-08 02:44:02 +03:00
Vitaliy Filippov	cd1e890bd4	Fix "opcode=5 retval=-2" errors sometimes possible with EC	2023-04-08 02:44:02 +03:00
Vitaliy Filippov	0fbf4c6a08	Selectively sync nonsynced objects on STABILIZE/ROLLBACK (fix for github issue #51 )	2023-04-08 02:44:02 +03:00
Vitaliy Filippov	d06ed2b0e7	Implement online config update	2023-03-26 19:21:50 +03:00
Vitaliy Filippov	3bbc46543d	Fix vitastor-cli create syntax	2023-03-17 11:12:58 +03:00
Vitaliy Filippov	2fb0c85618	Allow to start OSDs without local store (only for tests)	2023-03-15 01:13:59 +03:00
Vitaliy Filippov	d81a6c04fc	Update cmake min version so it does not complain about deprecation	2023-03-15 01:08:23 +03:00
Vitaliy Filippov	7b35801647	Fix possible bad realloc in disk_tool_meta for non-standard metadata block sizes	2023-03-15 01:08:23 +03:00
Vitaliy Filippov	f3228d5c07	Fix typo (did not affect execution though)	2023-03-15 01:08:23 +03:00
Vitaliy Filippov	18366f5055	Fix read/write return type in rw_blocking	2023-03-15 01:08:14 +03:00
Vitaliy Filippov	851507c147	Add missing close() in test stubs	2023-03-15 00:23:56 +03:00
Vitaliy Filippov	9aaad28488	Fix "null pointer exception" for unhandled OSD_OP_DELETEs (when pool is not loaded yet)	2023-03-02 11:16:39 +03:00
Vitaliy Filippov	dd57d086fe	Add a missing part of the "theoretical performance" to the Russian version	2023-03-01 00:24:54 +03:00