Remove peer_fd from the event loop only after clearing cl->refs

2025-11-15 15:27:06 +03:00
72 changed files with 611 additions and 843 deletions
@@ -540,42 +540,6 @@ jobs:
          echo ""
        done

-  test_kv_stress:
-    runs-on: ubuntu-latest
-    needs: build
-    container: ${{env.TEST_IMAGE}}:${{github.sha}}
-    steps:
-    - name: Run test
-      id: test
-      timeout-minutes: 3
-      run: /root/vitastor/tests/test_kv_stress.sh
-    - name: Print logs
-      if: always() && steps.test.outcome == 'failure'
-      run: |
-        for i in /root/vitastor/testdata/*.log /root/vitastor/testdata/*.txt; do
-          echo "-------- $i --------"
-          cat $i
-          echo ""
-        done
-
-  test_kv_stress_imm:
-    runs-on: ubuntu-latest
-    needs: build
-    container: ${{env.TEST_IMAGE}}:${{github.sha}}
-    steps:
-    - name: Run test
-      id: test
-      timeout-minutes: 3
-      run: IMMEDIATE_COMMIT=1 /root/vitastor/tests/test_kv_stress.sh
-    - name: Print logs
-      if: always() && steps.test.outcome == 'failure'
-      run: |
-        for i in /root/vitastor/testdata/*.log /root/vitastor/testdata/*.txt; do
-          echo "-------- $i --------"
-          cat $i
-          echo ""
-        done
-
  test_splitbrain:
    runs-on: ubuntu-latest
    needs: build
@@ -2,6 +2,6 @@ cmake_minimum_required(VERSION 2.8.12)

 project(vitastor)

-set(VITASTOR_VERSION "2.4.4")
+set(VITASTOR_VERSION "2.4.3")

 add_subdirectory(src)
@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v2.4.4
+VITASTOR_VERSION ?= v2.4.3

 all: build push

@@ -49,7 +49,7 @@ spec:
            capabilities:
              add: ["SYS_ADMIN"]
            allowPrivilegeEscalation: true
-          image: vitalif/vitastor-csi:v2.4.4
+          image: vitalif/vitastor-csi:v2.4.3
          args:
            - "--node=$(NODE_ID)"
            - "--endpoint=$(CSI_ENDPOINT)"
@@ -121,7 +121,7 @@ spec:
            privileged: true
            capabilities:
              add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v2.4.4
+          image: vitalif/vitastor-csi:v2.4.3
          args:
            - "--node=$(NODE_ID)"
            - "--endpoint=$(CSI_ENDPOINT)"
@@ -5,7 +5,7 @@ package vitastor

 const (
    vitastorCSIDriverName    = "csi.vitastor.io"
-    vitastorCSIDriverVersion = "2.4.4"
+    vitastorCSIDriverVersion = "2.4.3"
 )

 // Config struct fills the parameters of request or user input
@@ -1,4 +1,4 @@
-vitastor (2.4.4-1) unstable; urgency=medium
+vitastor (2.4.3-1) unstable; urgency=medium

  * Bugfixes

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v2.4.4
+VITASTOR_VERSION ?= v2.4.3

 all: build push

@@ -4,7 +4,7 @@
 #

 # Desired Vitastor version
-VITASTOR_VERSION=v2.4.4
+VITASTOR_VERSION=v2.4.3

 # Additional arguments for all containers
 # For example, you may want to specify a custom logging driver here
@@ -26,9 +26,9 @@ at Vitastor Kubernetes operator: https://github.com/Antilles7227/vitastor-operat
 The instruction is very simple.

 1. Download a Docker image of the desired version: \
-   `docker pull vitalif/vitastor:v2.4.4`
+   `docker pull vitalif/vitastor:v2.4.3`
 2. Install scripts to the host system: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.4 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.3 install.sh`
 3. Reload udev rules: \
   `udevadm control --reload-rules`

@@ -25,9 +25,9 @@ Vitastor можно установить в Docker/Podman. При этом etcd,
 Инструкция по установке максимально простая.

 1. Скачайте Docker-образ желаемой версии: \
-   `docker pull vitalif/vitastor:v2.4.4`
+   `docker pull vitalif/vitastor:v2.4.3`
 2. Установите скрипты в хост-систему командой: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.4 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.3 install.sh`
 3. Перезагрузите правила udev: \
   `udevadm control --reload-rules`

@@ -1,6 +1,6 @@
 {
  "name": "vitastor-mon",
-  "version": "2.4.4",
+  "version": "2.4.3",
  "description": "Vitastor SDS monitor service",
  "main": "mon-main.js",
  "scripts": {
@@ -9,6 +9,7 @@ const LPOptimizer = require('./lp_optimizer/lp_optimizer.js');
 const { scale_pg_count } = require('./pg_utils.js');
 const { make_hier_tree, filter_osds_by_root_node,
    filter_osds_by_tags, filter_osds_by_block_layout, get_affinity_osds } = require('./osd_tree.js');
+const { select_murmur3 } = require('./lp_optimizer/murmur3.js');

 function pick_primary(pool_id, pg_num, pool_config, osd_set, up_osds, aff_osds)
 {
@@ -38,7 +39,7 @@ function pick_primary(pool_id, pg_num, pool_config, osd_set, up_osds, aff_osds)
    {
        return 0;
    }
-    return alive_set[pg_num % alive_set.length];
+    return alive_set[select_murmur3(alive_set.length, osd_num => pool_id+'/'+pg_num+'/'+osd_num)];
 }

 function recheck_primary(state, global_config, up_osds, osd_tree)
@@ -1,6 +1,6 @@
 {
  "name": "vitastor",
-  "version": "2.4.4",
+  "version": "2.4.3",
  "description": "Low-level native bindings to Vitastor client library",
  "main": "index.js",
  "keywords": [
@@ -50,7 +50,7 @@ from cinder.volume import configuration
 from cinder.volume import driver
 from cinder.volume import volume_utils

-VITASTOR_VERSION = '2.4.4'
+VITASTOR_VERSION = '2.4.3'

 LOG = logging.getLogger(__name__)

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        2.4.4
+Version:        2.4.3
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage

 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-2.4.4.el7.tar.gz
+Source0:        vitastor-2.4.3.el7.tar.gz

 BuildRequires:  gperftools-devel
 BuildRequires:  devtoolset-9-gcc-c++
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        2.4.4
+Version:        2.4.3
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage

 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-2.4.4.el8.tar.gz
+Source0:        vitastor-2.4.3.el8.tar.gz

 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-toolset-9-gcc-c++
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        2.4.4
+Version:        2.4.3
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage

 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-2.4.4.el9.tar.gz
+Source0:        vitastor-2.4.3.el9.tar.gz

 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-c++
@@ -20,7 +20,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 	set(CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/${CMAKE_INSTALL_LIBDIR}")
 endif()

-add_definitions(-DVITASTOR_VERSION="2.4.4")
+add_definitions(-DVITASTOR_VERSION="2.4.3")
 add_definitions(-D_GNU_SOURCE -D_LARGEFILE64_SOURCE -D_FILE_OFFSET_BITS=64 -Wall -Wno-sign-compare -Wno-comment -Wno-parentheses -Wno-pointer-arith -fdiagnostics-color=always -fno-omit-frame-pointer -fvisibility=hidden -I ${CMAKE_SOURCE_DIR}/src)
 add_link_options(-fno-omit-frame-pointer)
 if (${WITH_ASAN})
@@ -3,7 +3,87 @@

 #include "blockstore_impl.h"

-blockstore_i* blockstore_i::create(blockstore_config_t & config, ring_loop_t *ringloop, timerfd_manager_t *tfd)
+blockstore_t::blockstore_t(blockstore_config_t & config, ring_loop_t *ringloop, timerfd_manager_t *tfd)
 {
-    return new blockstore_impl_t(config, ringloop, tfd);
+    impl = new blockstore_impl_t(config, ringloop, tfd);
+}
+
+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();
+}
+
+bool blockstore_t::is_started()
+{
+    return impl->is_started();
+}
+
+bool blockstore_t::is_stalled()
+{
+    return impl->is_stalled();
+}
+
+bool blockstore_t::is_safe_to_stop()
+{
+    return impl->is_safe_to_stop();
+}
+
+void blockstore_t::enqueue_op(blockstore_op_t *op)
+{
+    impl->enqueue_op(op);
+}
+
+int blockstore_t::read_bitmap(object_id oid, uint64_t target_version, void *bitmap, uint64_t *result_version)
+{
+    return impl->read_bitmap(oid, target_version, bitmap, result_version);
+}
+
+std::map<uint64_t, uint64_t> & blockstore_t::get_inode_space_stats()
+{
+    return impl->inode_space_stats;
+}
+
+void blockstore_t::dump_diagnostics()
+{
+    return impl->dump_diagnostics();
+}
+
+uint32_t blockstore_t::get_block_size()
+{
+    return impl->get_block_size();
+}
+
+uint64_t blockstore_t::get_block_count()
+{
+    return impl->get_block_count();
+}
+
+uint64_t blockstore_t::get_free_block_count()
+{
+    return impl->get_free_block_count();
+}
+
+uint64_t blockstore_t::get_journal_size()
+{
+    return impl->get_journal_size();
+}
+
+uint32_t blockstore_t::get_bitmap_granularity()
+{
+    return impl->get_bitmap_granularity();
+}
+
+void blockstore_t::set_no_inode_stats(const std::vector<uint64_t> & pool_ids)
+{
+    impl->set_no_inode_stats(pool_ids);
 }
@@ -17,14 +17,22 @@
 #include "ringloop.h"
 #include "timerfd_manager.h"

+// Memory alignment for direct I/O (usually 512 bytes)
+#ifndef DIRECT_IO_ALIGNMENT
+#define DIRECT_IO_ALIGNMENT 512
+#endif
+
+// Memory allocation alignment (page size is usually optimal)
+#ifndef MEM_ALIGNMENT
+#define MEM_ALIGNMENT 4096
+#endif
+
 // Default block size is 128 KB, current allowed range is 4K - 128M
 #define DEFAULT_DATA_BLOCK_ORDER 17
 #define MIN_DATA_BLOCK_SIZE 4*1024
 #define MAX_DATA_BLOCK_SIZE 128*1024*1024
 #define DEFAULT_BITMAP_GRANULARITY 4096

-#define MIN_JOURNAL_SIZE 1024*1024
-
 #define BS_OP_MIN 1
 #define BS_OP_READ 1
 #define BS_OP_WRITE 2
@@ -38,18 +46,8 @@

 #define BS_OP_PRIVATE_DATA_SIZE 256

-#define IMMEDIATE_NONE 0
-#define IMMEDIATE_SMALL 1
-#define IMMEDIATE_ALL 2
-
 /*

-All operations may be submitted in any order, because reads only see completed writes,
-syncs only sync completed writes and writes don't depend on each other.
-
-The only restriction is that the external code MUST NOT submit multiple writes for one
-object in parallel. This is a natural restriction because `version` numbers are used though.
-
 Blockstore opcode documentation:

 ## BS_OP_READ / BS_OP_WRITE / BS_OP_WRITE_STABLE
@@ -164,8 +162,8 @@ struct __attribute__ ((visibility("default"))) blockstore_op_t
            uint32_t list_stable_limit;
        };
    };
-    uint8_t *buf = NULL;
-    uint8_t *bitmap = NULL;
+    void *buf = NULL;
+    void *bitmap = NULL;
    int retval = 0;

    uint8_t private_data[BS_OP_PRIVATE_DATA_SIZE];
@@ -173,54 +171,53 @@ struct __attribute__ ((visibility("default"))) blockstore_op_t

 typedef std::map<std::string, std::string> blockstore_config_t;

-class __attribute__((visibility("default"))) blockstore_i
-{
-public:
-    static blockstore_i* create(blockstore_config_t & config, ring_loop_t *ringloop, timerfd_manager_t *tfd);
+class blockstore_impl_t;

-    virtual ~blockstore_i() = default;
+class __attribute__((visibility("default"))) blockstore_t
+{
+    blockstore_impl_t *impl;
+public:
+    blockstore_t(blockstore_config_t & config, ring_loop_t *ringloop, timerfd_manager_t *tfd);
+    ~blockstore_t();

    // Update configuration
-    virtual void parse_config(blockstore_config_t & config) = 0;
+    void parse_config(blockstore_config_t & config);

    // Event loop
-    virtual void loop() = 0;
+    void loop();

    // Returns true when blockstore is ready to process operations
    // (Although you're free to enqueue them before that)
-    virtual bool is_started() = 0;
+    bool is_started();

    // Returns true when blockstore is stalled
-    virtual bool is_stalled() = 0;
+    bool is_stalled();

    // Returns true when it's safe to destroy the instance. If destroying the instance
    // requires to purge some queues, starts that process. Should be called in the event
    // loop until it returns true.
-    virtual bool is_safe_to_stop() = 0;
+    bool is_safe_to_stop();

    // Submission
-    virtual void enqueue_op(blockstore_op_t *op) = 0;
+    void enqueue_op(blockstore_op_t *op);

    // Simplified synchronous operation: get object bitmap & current version
-    virtual int read_bitmap(object_id oid, uint64_t target_version, void *bitmap, uint64_t *result_version = NULL) = 0;
+    int read_bitmap(object_id oid, uint64_t target_version, void *bitmap, uint64_t *result_version = NULL);

    // Get per-inode space usage statistics
-    virtual const std::map<uint64_t, uint64_t> & get_inode_space_stats() = 0;
+    std::map<uint64_t, uint64_t> & get_inode_space_stats();

    // Set per-pool no_inode_stats
-    virtual void set_no_inode_stats(const std::vector<uint64_t> & pool_ids) = 0;
+    void set_no_inode_stats(const std::vector<uint64_t> & pool_ids);

    // Print diagnostics to stdout
-    virtual void dump_diagnostics() = 0;
+    void dump_diagnostics();

-    // Get diagnostic string for an operation
-    virtual std::string get_op_diag(blockstore_op_t *op) = 0;
+    uint32_t get_block_size();
+    uint64_t get_block_count();
+    uint64_t get_free_block_count();

-    virtual uint32_t get_block_size() = 0;
-    virtual uint64_t get_block_count() = 0;
-    virtual uint64_t get_free_block_count() = 0;
+    uint64_t get_journal_size();

-    virtual uint64_t get_journal_size() = 0;
-
-    virtual uint32_t get_bitmap_granularity() = 0;
+    uint32_t get_bitmap_granularity();
 };
@@ -95,7 +95,7 @@ void blockstore_disk_t::parse_config(std::map<std::string, std::string> & config
    {
        data_block_size = (1 << DEFAULT_DATA_BLOCK_ORDER);
    }
-    if (is_power_of_two(data_block_size) >= 64 || data_block_size < MIN_DATA_BLOCK_SIZE || data_block_size >= MAX_DATA_BLOCK_SIZE)
+    if ((block_order = is_power_of_two(data_block_size)) >= 64 || data_block_size < MIN_DATA_BLOCK_SIZE || data_block_size >= MAX_DATA_BLOCK_SIZE)
    {
        throw std::runtime_error("Bad block size");
    }
@@ -8,11 +8,6 @@
 #include <string>
 #include <map>

-// Memory alignment for direct I/O (usually 512 bytes)
-#ifndef DIRECT_IO_ALIGNMENT
-#define DIRECT_IO_ALIGNMENT 512
-#endif
-
 #define BLOCKSTORE_CSUM_NONE 0
 // Lower byte of checksum type is its length
 #define BLOCKSTORE_CSUM_CRC32C 0x104
@@ -51,6 +46,7 @@ struct blockstore_disk_t
    uint64_t data_offset, data_device_sect, data_device_size, data_len;
    uint64_t journal_offset, journal_device_sect, journal_device_size, journal_len;

+    uint32_t block_order = 0;
    uint64_t block_count = 0;
    uint32_t clean_entry_bitmap_size = 0, clean_entry_size = 0, clean_dyn_size = 0;

@@ -2,7 +2,6 @@
 // License: VNPL-1.1 (see README.md for details)

 #include "blockstore_impl.h"
-#include "blockstore_internal.h"

 #define META_BLOCK_UNREAD 0
 #define META_BLOCK_READ 1
@@ -543,7 +542,7 @@ resume_2:
                if (old_entry->oid.inode != 0 && old_entry->oid != cur.oid)
                {
                    printf("Fatal error (metadata corruption or bug): tried to wipe metadata entry %ju (%jx:%jx v%ju) as old location of %jx:%jx\n",
-                        old_clean_loc / bs->dsk.data_block_size, old_entry->oid.inode, old_entry->oid.stripe,
+                        old_clean_loc >> bs->dsk.block_order, old_entry->oid.inode, old_entry->oid.stripe,
                        old_entry->version, cur.oid.inode, cur.oid.stripe);
                    exit(1);
                }
@@ -646,7 +645,7 @@ void journal_flusher_co::update_metadata_entry()
            has_delete
                ? "Fatal error (metadata corruption or bug): tried to delete metadata entry %ju (%jx:%jx v%ju) while deleting %jx:%jx v%ju\n"
                : "Fatal error (metadata corruption or bug): tried to overwrite non-zero metadata entry %ju (%jx:%jx v%ju) with %jx:%jx v%ju\n",
-            clean_loc / bs->dsk.data_block_size, new_entry->oid.inode, new_entry->oid.stripe,
+            clean_loc >> bs->dsk.block_order, new_entry->oid.inode, new_entry->oid.stripe,
            new_entry->version, cur.oid.inode, cur.oid.stripe, cur.version
        );
        exit(1);
@@ -696,7 +695,7 @@ void journal_flusher_co::update_metadata_entry()
        new_entry->version = cur.version;
        if (!bs->inmemory_meta)
        {
-            auto inmem_bmp = (uint8_t*)bs->clean_bitmaps + (clean_loc / bs->dsk.data_block_size)*2*bs->dsk.clean_entry_bitmap_size;
+            auto inmem_bmp = (uint8_t*)bs->clean_bitmaps + (clean_loc >> bs->dsk.block_order)*2*bs->dsk.clean_entry_bitmap_size;
            memcpy(inmem_bmp, new_clean_bitmap, 2*bs->dsk.clean_entry_bitmap_size);
        }
        if (bs->dsk.meta_format >= BLOCKSTORE_META_FORMAT_V2)
@@ -849,7 +848,7 @@ bool journal_flusher_co::clear_incomplete_csum_block_bits(int wait_base)
            {
                printf(
                    "Fatal error (metadata corruption or bug): tried to make holes in %ju (%jx:%jx v%ju) with %jx:%jx v%ju\n",
-                    clean_loc / bs->dsk.data_block_size, new_entry->oid.inode, new_entry->oid.stripe,
+                    clean_loc >> bs->dsk.block_order, new_entry->oid.inode, new_entry->oid.stripe,
                    new_entry->version, cur.oid.inode, cur.oid.stripe, cur.version
                );
            }
@@ -865,7 +864,7 @@ bool journal_flusher_co::clear_incomplete_csum_block_bits(int wait_base)
            calc_block_checksums(new_data_csums, true);
            if (!bs->inmemory_meta)
            {
-                auto inmem_bmp = (uint8_t*)bs->clean_bitmaps + (clean_loc / bs->dsk.data_block_size)*2*bs->dsk.clean_entry_bitmap_size;
+                auto inmem_bmp = (uint8_t*)bs->clean_bitmaps + (clean_loc >> bs->dsk.block_order)*2*bs->dsk.clean_entry_bitmap_size;
                memcpy(inmem_bmp, new_clean_bitmap, 2*bs->dsk.clean_entry_bitmap_size);
            }
            if (bs->dsk.meta_format >= BLOCKSTORE_META_FORMAT_V2)
@@ -1232,8 +1231,8 @@ bool journal_flusher_co::modify_meta_read(uint64_t meta_loc, flusher_meta_write_
    // And yet another option is to use LSM trees for metadata, but it sophisticates everything a lot,
    // so I'll avoid it as long as I can.
    wr.submitted = false;
-    wr.sector = ((meta_loc / bs->dsk.data_block_size) / (bs->dsk.meta_block_size / bs->dsk.clean_entry_size)) * bs->dsk.meta_block_size;
-    wr.pos = ((meta_loc / bs->dsk.data_block_size) % (bs->dsk.meta_block_size / bs->dsk.clean_entry_size));
+    wr.sector = ((meta_loc >> bs->dsk.block_order) / (bs->dsk.meta_block_size / bs->dsk.clean_entry_size)) * bs->dsk.meta_block_size;
+    wr.pos = ((meta_loc >> bs->dsk.block_order) % (bs->dsk.meta_block_size / bs->dsk.clean_entry_size));
    if (bs->inmemory_meta)
    {
        wr.buf = (uint8_t*)bs->metadata_buffer + wr.sector;
@@ -1293,14 +1292,14 @@ void journal_flusher_co::free_data_blocks()
 #ifdef BLOCKSTORE_DEBUG
        printf("%s block %ju from %jx:%jx v%ju (new location is %ju)\n",
            used ? "Postpone free" : "Free",
-            old_clean_loc / bs->dsk.data_block_size,
+            old_clean_loc >> bs->dsk.block_order,
            cur.oid.inode, cur.oid.stripe, cur.version,
-            clean_loc / bs->dsk.data_block_size);
+            clean_loc >> bs->dsk.block_order);
 #endif
        if (used)
            uo_it->second.was_freed = true;
        else
-            bs->data_alloc->set(old_clean_loc / bs->dsk.data_block_size, false);
+            bs->data_alloc->set(old_clean_loc >> bs->dsk.block_order, false);
    }
    if (has_delete)
    {
@@ -1310,13 +1309,13 @@ void journal_flusher_co::free_data_blocks()
 #ifdef BLOCKSTORE_DEBUG
        printf("%s block %ju from %jx:%jx v%ju (delete)\n",
            used ? "Postpone free" : "Free",
-            old_clean_loc / bs->dsk.data_block_size,
+            old_clean_loc >> bs->dsk.block_order,
            cur.oid.inode, cur.oid.stripe, cur.version);
 #endif
        if (used)
            uo_it->second.was_freed = true;
        else
-            bs->data_alloc->set(old_clean_loc / bs->dsk.data_block_size, false);
+            bs->data_alloc->set(old_clean_loc >> bs->dsk.block_order, false);
    }
 }

@@ -1,6 +1,14 @@
 // Copyright (c) Vitaliy Filippov, 2019+
 // License: VNPL-1.1 (see README.md for details)

+#define COPY_BUF_JOURNAL 1
+#define COPY_BUF_DATA 2
+#define COPY_BUF_ZERO 4
+#define COPY_BUF_CSUM_FILL 8
+#define COPY_BUF_COALESCED 16
+#define COPY_BUF_META_BLOCK 32
+#define COPY_BUF_JOURNALED_BIG 64
+
 struct copy_buffer_t
 {
    int copy_flags;
@@ -2,7 +2,6 @@
 // License: VNPL-1.1 (see README.md for details)

 #include "blockstore_impl.h"
-#include "blockstore_internal.h"

 blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_t *ringloop, timerfd_manager_t *tfd)
 {
@@ -275,7 +274,7 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
 {
    if (PRIV(op)->wait_for == WAIT_SQE)
    {
-        if (ringloop->space_left() < PRIV(op)->wait_detail)
+        if (ringloop->sqes_left() < PRIV(op)->wait_detail)
        {
            // stop submission if there's still no free space
 #ifdef BLOCKSTORE_DEBUG
@@ -683,7 +682,7 @@ void blockstore_impl_t::process_list(blockstore_op_t *op)
    free(unstable);
    op->version = stable_count;
    op->retval = stable_count+unstable_count;
-    op->buf = (uint8_t*)stable;
+    op->buf = stable;
    FINISH_OP(op);
 }

@@ -706,11 +705,6 @@ void blockstore_impl_t::disk_error_abort(const char *op, int retval, int expecte
    exit(1);
 }

-const std::map<uint64_t, uint64_t> & blockstore_impl_t::get_inode_space_stats()
-{
-    return inode_space_stats;
-}
-
 void blockstore_impl_t::set_no_inode_stats(const std::vector<uint64_t> & pool_ids)
 {
    for (auto & np: no_inode_stats)
@@ -793,14 +787,3 @@ void blockstore_impl_t::recalc_inode_space_stats(uint64_t pool_id, bool per_inod
        dirty_it++;
    }
 }
-
-std::string blockstore_impl_t::get_op_diag(blockstore_op_t *op)
-{
-    char buf[256];
-    auto priv = PRIV(op);
-    if (priv->wait_for)
-        snprintf(buf, sizeof(buf), "state=%d wait=%d (detail=%ju)", priv->op_state, priv->wait_for, priv->wait_detail);
-    else
-        snprintf(buf, sizeof(buf), "state=%d", priv->op_state);
-    return std::string(buf);
-}
@@ -5,7 +5,6 @@

 #include "blockstore.h"
 #include "blockstore_disk.h"
-#include "ondisk_formats.h"

 #include <sys/types.h>
 #include <sys/ioctl.h>
@@ -29,8 +28,112 @@

 //#define BLOCKSTORE_DEBUG

+// States are not stored on disk. Instead, they're deduced from the journal
+
+#define BS_ST_SMALL_WRITE 0x01
+#define BS_ST_BIG_WRITE 0x02
+#define BS_ST_DELETE 0x03
+
+#define BS_ST_WAIT_DEL 0x10
+#define BS_ST_WAIT_BIG 0x20
+#define BS_ST_IN_FLIGHT 0x30
+#define BS_ST_SUBMITTED 0x40
+#define BS_ST_WRITTEN 0x50
+#define BS_ST_SYNCED 0x60
+#define BS_ST_STABLE 0x70
+
+#define BS_ST_INSTANT 0x100
+
+#define IMMEDIATE_NONE 0
+#define IMMEDIATE_SMALL 1
+#define IMMEDIATE_ALL 2
+
+#define BS_ST_TYPE_MASK 0x0F
+#define BS_ST_WORKFLOW_MASK 0xF0
+#define IS_IN_FLIGHT(st) (((st) & 0xF0) <= BS_ST_SUBMITTED)
+#define IS_STABLE(st) (((st) & 0xF0) == BS_ST_STABLE)
+#define IS_SYNCED(st) (((st) & 0xF0) >= BS_ST_SYNCED)
+#define IS_JOURNAL(st) (((st) & 0x0F) == BS_ST_SMALL_WRITE)
+#define IS_BIG_WRITE(st) (((st) & 0x0F) == BS_ST_BIG_WRITE)
+#define IS_DELETE(st) (((st) & 0x0F) == BS_ST_DELETE)
+#define IS_INSTANT(st) (((st) & BS_ST_TYPE_MASK) == BS_ST_DELETE || ((st) & BS_ST_INSTANT))
+
+#define BS_SUBMIT_CHECK_SQES(n) \
+    if (ringloop->sqes_left() < (n))\
+    {\
+        /* Pause until there are more requests available */\
+        PRIV(op)->wait_detail = (n);\
+        PRIV(op)->wait_for = WAIT_SQE;\
+        return 0;\
+    }
+
+#define BS_SUBMIT_GET_SQE(sqe, data) \
+    BS_SUBMIT_GET_ONLY_SQE(sqe); \
+    struct ring_data_t *data = ((ring_data_t*)sqe->user_data)
+
+#define BS_SUBMIT_GET_ONLY_SQE(sqe) \
+    struct io_uring_sqe *sqe = get_sqe();\
+    if (!sqe)\
+    {\
+        /* Pause until there are more requests available */\
+        PRIV(op)->wait_detail = 1;\
+        PRIV(op)->wait_for = WAIT_SQE;\
+        return 0;\
+    }
+
+#define BS_SUBMIT_GET_SQE_DECL(sqe) \
+    sqe = get_sqe();\
+    if (!sqe)\
+    {\
+        /* Pause until there are more requests available */\
+        PRIV(op)->wait_detail = 1;\
+        PRIV(op)->wait_for = WAIT_SQE;\
+        return 0;\
+    }
+
 #include "blockstore_journal.h"

+// "VITAstor"
+#define BLOCKSTORE_META_MAGIC_V1 0x726F747341544956l
+#define BLOCKSTORE_META_FORMAT_V1 1
+#define BLOCKSTORE_META_FORMAT_V2 2
+
+// metadata header (superblock)
+struct __attribute__((__packed__)) blockstore_meta_header_v1_t
+{
+    uint64_t zero;
+    uint64_t magic;
+    uint64_t version;
+    uint32_t meta_block_size;
+    uint32_t data_block_size;
+    uint32_t bitmap_granularity;
+};
+
+struct __attribute__((__packed__)) blockstore_meta_header_v2_t
+{
+    uint64_t zero;
+    uint64_t magic;
+    uint64_t version;
+    uint32_t meta_block_size;
+    uint32_t data_block_size;
+    uint32_t bitmap_granularity;
+    uint32_t data_csum_type;
+    uint32_t csum_block_size;
+    uint32_t header_csum;
+};
+
+// 32 bytes = 24 bytes + block bitmap (4 bytes by default) + external attributes (also bitmap, 4 bytes by default)
+// per "clean" entry on disk with fixed metadata tables
+struct __attribute__((__packed__)) clean_disk_entry
+{
+    object_id oid;
+    uint64_t version;
+    uint8_t bitmap[];
+    // Two more fields come after bitmap in metadata version 2:
+    // uint32_t data_csum[];
+    // uint32_t entry_csum;
+};
+
 // 32 = 16 + 16 bytes per "clean" entry in memory (object_id => clean_entry)
 struct __attribute__((__packed__)) clean_entry
 {
@@ -67,6 +170,15 @@ struct __attribute__((__packed__)) dirty_entry
 // Otherwise, the submit order is free, that is all operations may be submitted immediately
 // In fact, adding a write operation must immediately result in dirty_db being populated

+// Suspend operation until there are more free SQEs
+#define WAIT_SQE 1
+// Suspend operation until there are <wait_detail> bytes of free space in the journal on disk
+#define WAIT_JOURNAL 3
+// Suspend operation until the next journal sector buffer is free
+#define WAIT_JOURNAL_BUFFER 4
+// Suspend operation until there is some free space on the data device
+#define WAIT_FREE 5
+
 struct used_clean_obj_t
 {
    int refs;
@@ -84,6 +196,9 @@ typedef std::map<obj_ver_id, dirty_entry> blockstore_dirty_db_t;

 #include "blockstore_flush.h"

+#define PRIV(op) ((blockstore_op_private_t*)(op)->private_data)
+#define FINISH_OP(op) PRIV(op)->~blockstore_op_private_t(); std::function<void (blockstore_op_t*)>(op->callback)(op)
+
 struct blockstore_op_private_t
 {
    // Wait status
@@ -109,15 +224,23 @@ struct blockstore_op_private_t
    std::vector<obj_ver_id> sync_big_writes, sync_small_writes;
 };

+typedef uint32_t pool_id_t;
+typedef uint64_t pool_pg_id_t;
+
+#define POOL_ID_BITS 16
+
 struct pool_shard_settings_t
 {
    uint32_t pg_count;
    uint32_t pg_stripe_size;
 };

-typedef uint64_t pool_pg_id_t;
+#define STAB_SPLIT_DONE 1
+#define STAB_SPLIT_WAIT 2
+#define STAB_SPLIT_SYNC 3
+#define STAB_SPLIT_TODO 4

-class blockstore_impl_t: public blockstore_i
+class blockstore_impl_t
 {
    blockstore_disk_t dsk;

@@ -153,7 +276,6 @@ class blockstore_impl_t: public blockstore_i
    std::map<pool_id_t, pool_shard_settings_t> clean_db_settings;
    std::map<pool_pg_id_t, blockstore_clean_db_t> clean_db_shards;
    std::map<uint64_t, int> no_inode_stats;
-    std::map<uint64_t, uint64_t> inode_space_stats;
    uint8_t *clean_bitmaps = NULL;
    blockstore_dirty_db_t dirty_db;
    std::vector<blockstore_op_t*> submit_queue;
@@ -282,7 +404,6 @@ 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);
    void parse_config(blockstore_config_t & config, bool init);

    // Event loop
@@ -309,8 +430,8 @@ public:
    // Unstable writes are added here (map of object_id -> version)
    std::unordered_map<object_id, uint64_t> unstable_writes;

-    // Get space usage statistics
-    const std::map<uint64_t, uint64_t> & get_inode_space_stats();
+    // Space usage statistics
+    std::map<uint64_t, uint64_t> inode_space_stats;

    // Set per-pool no_inode_stats
    void set_no_inode_stats(const std::vector<uint64_t> & pool_ids);
@@ -318,9 +439,6 @@ public:
    // Print diagnostics to stdout
    void dump_diagnostics();

-    // Get diagnostic string for an operation
-    std::string get_op_diag(blockstore_op_t *op);
-
    inline uint32_t get_block_size() { return dsk.data_block_size; }
    inline uint64_t get_block_count() { return dsk.block_count; }
    inline uint64_t get_free_block_count() { return dsk.block_count - used_blocks; }
@@ -2,7 +2,6 @@
 // License: VNPL-1.1 (see README.md for details)

 #include "blockstore_impl.h"
-#include "blockstore_internal.h"

 #define INIT_META_EMPTY 0
 #define INIT_META_READING 1
@@ -373,7 +372,7 @@ bool blockstore_init_meta::handle_meta_block(uint8_t *buf, uint64_t entries_per_
                    // free the previous block
                    // here we have to zero out the previous entry because otherwise we'll hit
                    // "tried to overwrite non-zero metadata entry" later
-                    uint64_t old_clean_loc = clean_it->second.location / bs->dsk.data_block_size;
+                    uint64_t old_clean_loc = clean_it->second.location >> bs->dsk.block_order;
                    if (bs->inmemory_meta)
                    {
                        uint64_t sector = (old_clean_loc / entries_per_block) * bs->dsk.meta_block_size;
@@ -391,7 +390,7 @@ bool blockstore_init_meta::handle_meta_block(uint8_t *buf, uint64_t entries_per_
                    }
                    else
                    {
-                        entries_to_zero.push_back(clean_it->second.location / bs->dsk.data_block_size);
+                        entries_to_zero.push_back(clean_it->second.location >> bs->dsk.block_order);
                    }
 #ifdef BLOCKSTORE_DEBUG
                    printf("Free block %ju from %jx:%jx v%ju (new location is %ju)\n",
@@ -413,7 +412,7 @@ bool blockstore_init_meta::handle_meta_block(uint8_t *buf, uint64_t entries_per_
                bs->data_alloc->set(done_cnt+i, true);
                clean_db[entry->oid] = (struct clean_entry){
                    .version = entry->version,
-                    .location = (done_cnt+i) * bs->dsk.data_block_size,
+                    .location = (done_cnt+i) << bs->dsk.block_order,
                };
            }
            else
@@ -1002,7 +1001,7 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                printf(
                    "je_big_write%s oid=%jx:%jx ver=%ju loc=%ju\n",
                    je->type == JE_BIG_WRITE_INSTANT ? "_instant" : "",
-                    je->big_write.oid.inode, je->big_write.oid.stripe, je->big_write.version, je->big_write.location / bs->dsk.data_block_size
+                    je->big_write.oid.inode, je->big_write.oid.stripe, je->big_write.version, je->big_write.location >> bs->dsk.block_order
                );
 #endif
                auto dirty_it = bs->dirty_db.upper_bound((obj_ver_id){
@@ -1065,7 +1064,7 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                        .journal_sector = proc_pos,
                        .dyn_data = dyn,
                    }).first;
-                    if (bs->data_alloc->get(je->big_write.location / bs->dsk.data_block_size))
+                    if (bs->data_alloc->get(je->big_write.location >> bs->dsk.block_order))
                    {
                        // This is probably a big_write that's already flushed and freed, but it may
                        // also indicate a bug. So we remember such entries and recheck them afterwards.
@@ -1078,11 +1077,11 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
 #ifdef BLOCKSTORE_DEBUG
                        printf(
                            "Allocate block (journal) %ju: %jx:%jx v%ju\n",
-                            je->big_write.location / bs->dsk.data_block_size,
+                            je->big_write.location >> bs->dsk.block_order,
                            ov.oid.inode, ov.oid.stripe, ov.version
                        );
 #endif
-                        bs->data_alloc->set(je->big_write.location / bs->dsk.data_block_size, true);
+                        bs->data_alloc->set(je->big_write.location >> bs->dsk.block_order, true);
                    }
                    bs->journal.used_sectors[proc_pos]++;
 #ifdef BLOCKSTORE_DEBUG
@@ -1,85 +0,0 @@
-#pragma once
-
-// States are not stored on disk. Instead, they're deduced from the journal
-
-#define BS_ST_SMALL_WRITE 0x01
-#define BS_ST_BIG_WRITE 0x02
-#define BS_ST_DELETE 0x03
-
-#define BS_ST_WAIT_DEL 0x10
-#define BS_ST_WAIT_BIG 0x20
-#define BS_ST_IN_FLIGHT 0x30
-#define BS_ST_SUBMITTED 0x40
-#define BS_ST_WRITTEN 0x50
-#define BS_ST_SYNCED 0x60
-#define BS_ST_STABLE 0x70
-
-#define BS_ST_INSTANT 0x100
-
-#define BS_ST_TYPE_MASK 0x0F
-#define BS_ST_WORKFLOW_MASK 0xF0
-#define IS_IN_FLIGHT(st) (((st) & 0xF0) <= BS_ST_SUBMITTED)
-#define IS_STABLE(st) (((st) & 0xF0) == BS_ST_STABLE)
-#define IS_SYNCED(st) (((st) & 0xF0) >= BS_ST_SYNCED)
-#define IS_JOURNAL(st) (((st) & 0x0F) == BS_ST_SMALL_WRITE)
-#define IS_BIG_WRITE(st) (((st) & 0x0F) == BS_ST_BIG_WRITE)
-#define IS_DELETE(st) (((st) & 0x0F) == BS_ST_DELETE)
-#define IS_INSTANT(st) (((st) & BS_ST_TYPE_MASK) == BS_ST_DELETE || ((st) & BS_ST_INSTANT))
-
-#define BS_SUBMIT_CHECK_SQES(n) \
-    if (ringloop->space_left() < (n))\
-    {\
-        /* Pause until there are more requests available */\
-        PRIV(op)->wait_detail = (n);\
-        PRIV(op)->wait_for = WAIT_SQE;\
-        return 0;\
-    }
-
-#define BS_SUBMIT_GET_SQE(sqe, data) \
-    BS_SUBMIT_GET_ONLY_SQE(sqe); \
-    struct ring_data_t *data = ((ring_data_t*)sqe->user_data)
-
-#define BS_SUBMIT_GET_ONLY_SQE(sqe) \
-    struct io_uring_sqe *sqe = get_sqe();\
-    if (!sqe)\
-    {\
-        /* Pause until there are more requests available */\
-        PRIV(op)->wait_detail = 1;\
-        PRIV(op)->wait_for = WAIT_SQE;\
-        return 0;\
-    }
-
-#define BS_SUBMIT_GET_SQE_DECL(sqe) \
-    sqe = get_sqe();\
-    if (!sqe)\
-    {\
-        /* Pause until there are more requests available */\
-        PRIV(op)->wait_detail = 1;\
-        PRIV(op)->wait_for = WAIT_SQE;\
-        return 0;\
-    }
-
-#define PRIV(op) ((blockstore_op_private_t*)(op)->private_data)
-#define FINISH_OP(op) PRIV(op)->~blockstore_op_private_t(); std::function<void (blockstore_op_t*)>(op->callback)(op)
-
-// Suspend operation until there are more free SQEs
-#define WAIT_SQE 1
-// Suspend operation until there are <wait_detail> bytes of free space in the journal on disk
-#define WAIT_JOURNAL 3
-// Suspend operation until the next journal sector buffer is free
-#define WAIT_JOURNAL_BUFFER 4
-// Suspend operation until there is some free space on the data device
-#define WAIT_FREE 5
-
-#define COPY_BUF_JOURNAL 1
-#define COPY_BUF_DATA 2
-#define COPY_BUF_ZERO 4
-#define COPY_BUF_CSUM_FILL 8
-#define COPY_BUF_COALESCED 16
-#define COPY_BUF_META_BLOCK 32
-#define COPY_BUF_JOURNALED_BIG 64
-
-#define STAB_SPLIT_DONE 1
-#define STAB_SPLIT_WAIT 2
-#define STAB_SPLIT_SYNC 3
-#define STAB_SPLIT_TODO 4
@@ -2,7 +2,6 @@
 // License: VNPL-1.1 (see README.md for details)

 #include "blockstore_impl.h"
-#include "blockstore_internal.h"

 blockstore_journal_check_t::blockstore_journal_check_t(blockstore_impl_t *bs)
 {
@@ -3,7 +3,147 @@

 #pragma once

-class blockstore_impl_t;
+#include "crc32c.h"
+#include <set>
+
+#define MIN_JOURNAL_SIZE 4*1024*1024
+#define JOURNAL_MAGIC 0x4A33
+#define JOURNAL_VERSION_V1 1
+#define JOURNAL_VERSION_V2 2
+#define JOURNAL_BUFFER_SIZE 4*1024*1024
+#define JOURNAL_ENTRY_HEADER_SIZE 16
+
+// Journal entries
+// Journal entries are linked to each other by their crc32 value
+// The journal is almost a blockchain, because object versions constantly increase
+#define JE_MIN         0x01
+#define JE_START       0x01
+#define JE_SMALL_WRITE 0x02
+#define JE_BIG_WRITE   0x03
+#define JE_STABLE      0x04
+#define JE_DELETE      0x05
+#define JE_ROLLBACK    0x06
+#define JE_SMALL_WRITE_INSTANT 0x07
+#define JE_BIG_WRITE_INSTANT   0x08
+#define JE_MAX         0x08
+
+// crc32c comes first to ease calculation
+struct __attribute__((__packed__)) journal_entry_start
+{
+    uint32_t crc32;
+    uint16_t magic;
+    uint16_t type;
+    uint32_t size;
+    uint32_t reserved;
+    uint64_t journal_start;
+    uint64_t version;
+    uint32_t data_csum_type;
+    uint32_t csum_block_size;
+};
+#define JE_START_V0_SIZE 24
+#define JE_START_V1_SIZE 32
+#define JE_START_V2_SIZE 40
+
+struct __attribute__((__packed__)) journal_entry_small_write
+{
+    uint32_t crc32;
+    uint16_t magic;
+    uint16_t type;
+    uint32_t size;
+    uint32_t crc32_prev;
+    object_id oid;
+    uint64_t version;
+    uint32_t offset;
+    uint32_t len;
+    // small_write entries contain <len> bytes of data which is stored in next sectors
+    // data_offset is its offset within journal
+    uint64_t data_offset;
+    uint32_t crc32_data; // zero when data_csum_type != 0
+    // small_write and big_write entries are followed by the "external" bitmap
+    // its size is dynamic and included in journal entry's <size> field
+    uint8_t bitmap[];
+    // and then data checksums if data_csum_type != 0
+    // uint32_t data_crc32c[];
+};
+
+struct __attribute__((__packed__)) journal_entry_big_write
+{
+    uint32_t crc32;
+    uint16_t magic;
+    uint16_t type;
+    uint32_t size;
+    uint32_t crc32_prev;
+    object_id oid;
+    uint64_t version;
+    uint32_t offset;
+    uint32_t len;
+    uint64_t location;
+    // small_write and big_write entries are followed by the "external" bitmap
+    // its size is dynamic and included in journal entry's <size> field
+    uint8_t bitmap[];
+    // and then data checksums if data_csum_type != 0
+    // uint32_t data_crc32c[];
+};
+
+struct __attribute__((__packed__)) journal_entry_stable
+{
+    uint32_t crc32;
+    uint16_t magic;
+    uint16_t type;
+    uint32_t size;
+    uint32_t crc32_prev;
+    object_id oid;
+    uint64_t version;
+};
+
+struct __attribute__((__packed__)) journal_entry_rollback
+{
+    uint32_t crc32;
+    uint16_t magic;
+    uint16_t type;
+    uint32_t size;
+    uint32_t crc32_prev;
+    object_id oid;
+    uint64_t version;
+};
+
+struct __attribute__((__packed__)) journal_entry_del
+{
+    uint32_t crc32;
+    uint16_t magic;
+    uint16_t type;
+    uint32_t size;
+    uint32_t crc32_prev;
+    object_id oid;
+    uint64_t version;
+};
+
+struct __attribute__((__packed__)) journal_entry
+{
+    union
+    {
+        struct __attribute__((__packed__))
+        {
+            uint32_t crc32;
+            uint16_t magic;
+            uint16_t type;
+            uint32_t size;
+            uint32_t crc32_prev;
+        };
+        journal_entry_start start;
+        journal_entry_small_write small_write;
+        journal_entry_big_write big_write;
+        journal_entry_stable stable;
+        journal_entry_rollback rollback;
+        journal_entry_del del;
+    };
+};
+
+inline uint32_t je_crc32(journal_entry *je)
+{
+    // 0x48674bc7 = crc32(4 zero bytes)
+    return crc32c(0x48674bc7, ((uint8_t*)je)+4, je->size-4);
+}

 struct journal_sector_info_t
 {
@@ -4,11 +4,6 @@
 #include <sys/file.h>
 #include "blockstore_impl.h"

-void blockstore_impl_t::parse_config(blockstore_config_t & config)
-{
-    return parse_config(config, false);
-}
-
 void blockstore_impl_t::parse_config(blockstore_config_t & config, bool init)
 {
    // Online-configurable options:
@@ -3,7 +3,6 @@

 #include <limits.h>
 #include "blockstore_impl.h"
-#include "blockstore_internal.h"

 int blockstore_impl_t::fulfill_read_push(blockstore_op_t *op, void *buf, uint64_t offset, uint64_t len,
    uint32_t item_state, uint64_t item_version)
@@ -152,7 +151,7 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *read_op,
 uint8_t* blockstore_impl_t::get_clean_entry_bitmap(uint64_t block_loc, int offset)
 {
    uint8_t *clean_entry_bitmap;
-    uint64_t meta_loc = block_loc / dsk.data_block_size;
+    uint64_t meta_loc = block_loc >> dsk.block_order;
    if (inmemory_meta)
    {
        uint64_t sector = (meta_loc / (dsk.meta_block_size / dsk.clean_entry_size)) * dsk.meta_block_size;
@@ -692,8 +691,8 @@ bool blockstore_impl_t::fulfill_clean_read(blockstore_op_t *read_op, uint64_t &
 uint8_t* blockstore_impl_t::read_clean_meta_block(blockstore_op_t *op, uint64_t clean_loc, int rv_pos)
 {
    auto & rv = PRIV(op)->read_vec;
-    auto sector = ((clean_loc / dsk.data_block_size) / (dsk.meta_block_size / dsk.clean_entry_size)) * dsk.meta_block_size;
-    auto pos = ((clean_loc / dsk.data_block_size) % (dsk.meta_block_size / dsk.clean_entry_size)) * dsk.clean_entry_size;
+    auto sector = ((clean_loc >> dsk.block_order) / (dsk.meta_block_size / dsk.clean_entry_size)) * dsk.meta_block_size;
+    auto pos = ((clean_loc >> dsk.block_order) % (dsk.meta_block_size / dsk.clean_entry_size)) * dsk.clean_entry_size;
    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size);
    rv.insert(rv.begin()+rv_pos, (copy_buffer_t){
        .copy_flags = COPY_BUF_META_BLOCK|COPY_BUF_CSUM_FILL,
@@ -807,7 +806,7 @@ bool blockstore_impl_t::verify_clean_padded_checksums(blockstore_op_t *op, uint6
    uint32_t offset = clean_loc % dsk.data_block_size;
    if (from_journal)
        return verify_padded_checksums(dyn_data, dyn_data + dsk.clean_entry_bitmap_size, offset, iov, n_iov, bad_block_cb);
-    clean_loc = (clean_loc / dsk.data_block_size) * dsk.data_block_size;
+    clean_loc = (clean_loc >> dsk.block_order) << dsk.block_order;
    if (!dyn_data)
    {
        assert(inmemory_meta);
@@ -867,7 +866,7 @@ void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op
                    {
                        // BIG_WRITE from journal or clean data
                        // Do not verify checksums if the data location is/was mutated by flushers
-                        auto & uo = used_clean_objects.at((rv[i].disk_offset / dsk.data_block_size) * dsk.data_block_size);
+                        auto & uo = used_clean_objects.at((rv[i].disk_offset >> dsk.block_order) << dsk.block_order);
                        if (!uo.was_changed)
                        {
                            verify_clean_padded_checksums(
@@ -2,7 +2,6 @@
 // License: VNPL-1.1 (see README.md for details)

 #include "blockstore_impl.h"
-#include "blockstore_internal.h"

 int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
 {
@@ -211,10 +210,10 @@ void blockstore_impl_t::erase_dirty(blockstore_dirty_db_t::iterator dirty_start,
            dirty_it->second.location != UINT64_MAX)
        {
 #ifdef BLOCKSTORE_DEBUG
-            printf("Free block %ju from %jx:%jx v%ju\n", dirty_it->second.location / dsk.data_block_size,
+            printf("Free block %ju from %jx:%jx v%ju\n", dirty_it->second.location >> dsk.block_order,
                dirty_it->first.oid.inode, dirty_it->first.oid.stripe, dirty_it->first.version);
 #endif
-            data_alloc->set(dirty_it->second.location / dsk.data_block_size, false);
+            data_alloc->set(dirty_it->second.location >> dsk.block_order, false);
        }
        auto used = --journal.used_sectors.at(dirty_it->second.journal_sector);
 #ifdef BLOCKSTORE_DEBUG
@@ -2,7 +2,6 @@
 // License: VNPL-1.1 (see README.md for details)

 #include "blockstore_impl.h"
-#include "blockstore_internal.h"

 // Stabilize small write:
 // 1) Copy data from the journal to the data device
@@ -229,17 +228,17 @@ int blockstore_impl_t::split_stab_op(blockstore_op_t *op, std::function<int(obj_
        // 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 = (uint8_t*)bad_vers.items;
+        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)
    {
-        uint8_t *orig_buf = op->buf;
+        void *orig_buf = op->buf;
        if (good_vers.items)
        {
-            op->buf = (uint8_t*)good_vers.items;
+            op->buf = good_vers.items;
            op->len = good_vers.size;
        }
        // Make a wrapped callback
@@ -2,7 +2,6 @@
 // License: VNPL-1.1 (see README.md for details)

 #include "blockstore_impl.h"
-#include "blockstore_internal.h"

 #define SYNC_HAS_SMALL 1
 #define SYNC_HAS_BIG 2
@@ -2,7 +2,6 @@
 // License: VNPL-1.1 (see README.md for details)

 #include "blockstore_impl.h"
-#include "blockstore_internal.h"

 bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
 {
@@ -341,7 +340,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
        }
        BS_SUBMIT_GET_SQE(sqe, data);
        write_iodepth++;
-        dirty_it->second.location = loc * dsk.data_block_size;
+        dirty_it->second.location = loc << dsk.block_order;
        dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_SUBMITTED;
 #ifdef BLOCKSTORE_DEBUG
        printf(
@@ -367,7 +366,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
        data->iov.iov_len = op->len + stripe_offset + stripe_end; // to check it in the callback
        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
        io_uring_prep_writev(
-            sqe, dsk.data_fd, PRIV(op)->iov_zerofill, vcnt, dsk.data_offset + (loc * dsk.data_block_size) + op->offset - stripe_offset
+            sqe, dsk.data_fd, PRIV(op)->iov_zerofill, vcnt, dsk.data_offset + (loc << dsk.block_order) + op->offset - stripe_offset
        );
        PRIV(op)->pending_ops = 1;
        if (!(dirty_it->second.state & BS_ST_INSTANT))
@@ -32,7 +32,7 @@

 struct bs_data
 {
-    blockstore_i *bs;
+    blockstore_t *bs;
    epoll_manager_t *epmgr;
    ring_loop_t *ringloop;
    /* The list of completed io_u structs. */
@@ -162,7 +162,7 @@ static int bs_init(struct thread_data *td)
    }
    bsd->ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
    bsd->epmgr = new epoll_manager_t(bsd->ringloop);
-    bsd->bs = blockstore_i::create(config, bsd->ringloop, bsd->epmgr->tfd);
+    bsd->bs = new blockstore_t(config, bsd->ringloop, bsd->epmgr->tfd);
    bsd->imm = config.find("immediate_commit") == config.end() ||
        config["immediate_commit"] == "all";
    while (1)
@@ -200,7 +200,7 @@ static enum fio_q_status bs_queue(struct thread_data *td, struct io_u *io)
    {
    case DDIR_READ:
        op->opcode = BS_OP_READ;
-        op->buf = (uint8_t*)io->xfer_buf;
+        op->buf = io->xfer_buf;
        op->oid = {
            .inode = 1,
            .stripe = io->offset / bsd->bs->get_block_size(),
@@ -221,7 +221,7 @@ static enum fio_q_status bs_queue(struct thread_data *td, struct io_u *io)
        break;
    case DDIR_WRITE:
        op->opcode = bsd->ec ? BS_OP_WRITE : BS_OP_WRITE_STABLE;
-        op->buf = (uint8_t*)io->xfer_buf;
+        op->buf = io->xfer_buf;
        op->oid = {
            .inode = 1,
            .stripe = io->offset / bsd->bs->get_block_size(),
@@ -247,7 +247,7 @@ static enum fio_q_status bs_queue(struct thread_data *td, struct io_u *io)
                {
                    auto stab_op = new blockstore_op_t;
                    stab_op->opcode = BS_OP_STABLE;
-                    stab_op->buf = (uint8_t*)malloc_or_die(sizeof(obj_ver_id));
+                    stab_op->buf = malloc_or_die(sizeof(obj_ver_id));
                    obj_ver_id *ver = (obj_ver_id *)stab_op->buf;
                    ver[0].oid = op->oid;
                    ver[0].version = op->version;
@@ -1,186 +0,0 @@
-// Metadata on-disk structures
-// Copyright (c) Vitaliy Filippov, 2019+
-// License: VNPL-1.1 (see README.md for details)
-
-#pragma once
-
-#include "crc32c.h"
-
-#define JOURNAL_MAGIC 0x4A33
-#define JOURNAL_VERSION_V1 1
-#define JOURNAL_VERSION_V2 2
-#define JOURNAL_BUFFER_SIZE 4*1024*1024
-#define JOURNAL_ENTRY_HEADER_SIZE 16
-
-// Journal entries
-// Journal entries are linked to each other by their crc32 value
-// The journal is almost a blockchain, because object versions constantly increase
-#define JE_MIN         0x01
-#define JE_START       0x01
-#define JE_SMALL_WRITE 0x02
-#define JE_BIG_WRITE   0x03
-#define JE_STABLE      0x04
-#define JE_DELETE      0x05
-#define JE_ROLLBACK    0x06
-#define JE_SMALL_WRITE_INSTANT 0x07
-#define JE_BIG_WRITE_INSTANT   0x08
-#define JE_MAX         0x08
-
-// crc32c comes first to ease calculation
-struct __attribute__((__packed__)) journal_entry_start
-{
-    uint32_t crc32;
-    uint16_t magic;
-    uint16_t type;
-    uint32_t size;
-    uint32_t reserved;
-    uint64_t journal_start;
-    uint64_t version;
-    uint32_t data_csum_type;
-    uint32_t csum_block_size;
-};
-#define JE_START_V0_SIZE 24
-#define JE_START_V1_SIZE 32
-#define JE_START_V2_SIZE 40
-
-struct __attribute__((__packed__)) journal_entry_small_write
-{
-    uint32_t crc32;
-    uint16_t magic;
-    uint16_t type;
-    uint32_t size;
-    uint32_t crc32_prev;
-    object_id oid;
-    uint64_t version;
-    uint32_t offset;
-    uint32_t len;
-    // small_write entries contain <len> bytes of data which is stored in next sectors
-    // data_offset is its offset within journal
-    uint64_t data_offset;
-    uint32_t crc32_data; // zero when data_csum_type != 0
-    // small_write and big_write entries are followed by the "external" bitmap
-    // its size is dynamic and included in journal entry's <size> field
-    uint8_t bitmap[];
-    // and then data checksums if data_csum_type != 0
-    // uint32_t data_crc32c[];
-};
-
-struct __attribute__((__packed__)) journal_entry_big_write
-{
-    uint32_t crc32;
-    uint16_t magic;
-    uint16_t type;
-    uint32_t size;
-    uint32_t crc32_prev;
-    object_id oid;
-    uint64_t version;
-    uint32_t offset;
-    uint32_t len;
-    uint64_t location;
-    // small_write and big_write entries are followed by the "external" bitmap
-    // its size is dynamic and included in journal entry's <size> field
-    uint8_t bitmap[];
-    // and then data checksums if data_csum_type != 0
-    // uint32_t data_crc32c[];
-};
-
-struct __attribute__((__packed__)) journal_entry_stable
-{
-    uint32_t crc32;
-    uint16_t magic;
-    uint16_t type;
-    uint32_t size;
-    uint32_t crc32_prev;
-    object_id oid;
-    uint64_t version;
-};
-
-struct __attribute__((__packed__)) journal_entry_rollback
-{
-    uint32_t crc32;
-    uint16_t magic;
-    uint16_t type;
-    uint32_t size;
-    uint32_t crc32_prev;
-    object_id oid;
-    uint64_t version;
-};
-
-struct __attribute__((__packed__)) journal_entry_del
-{
-    uint32_t crc32;
-    uint16_t magic;
-    uint16_t type;
-    uint32_t size;
-    uint32_t crc32_prev;
-    object_id oid;
-    uint64_t version;
-};
-
-struct __attribute__((__packed__)) journal_entry
-{
-    union
-    {
-        struct __attribute__((__packed__))
-        {
-            uint32_t crc32;
-            uint16_t magic;
-            uint16_t type;
-            uint32_t size;
-            uint32_t crc32_prev;
-        };
-        journal_entry_start start;
-        journal_entry_small_write small_write;
-        journal_entry_big_write big_write;
-        journal_entry_stable stable;
-        journal_entry_rollback rollback;
-        journal_entry_del del;
-    };
-};
-
-inline uint32_t je_crc32(journal_entry *je)
-{
-    // 0x48674bc7 = crc32(4 zero bytes)
-    return crc32c(0x48674bc7, ((uint8_t*)je)+4, je->size-4);
-}
-
-// "VITAstor"
-#define BLOCKSTORE_META_MAGIC_V1 0x726F747341544956l
-#define BLOCKSTORE_META_FORMAT_V1 1
-#define BLOCKSTORE_META_FORMAT_V2 2
-
-// metadata header (superblock)
-struct __attribute__((__packed__)) blockstore_meta_header_v1_t
-{
-    uint64_t zero;
-    uint64_t magic;
-    uint64_t version;
-    uint32_t meta_block_size;
-    uint32_t data_block_size;
-    uint32_t bitmap_granularity;
-};
-
-struct __attribute__((__packed__)) blockstore_meta_header_v2_t
-{
-    uint64_t zero;
-    uint64_t magic;
-    uint64_t version;
-    uint32_t meta_block_size;
-    uint32_t data_block_size;
-    uint32_t bitmap_granularity;
-    uint32_t data_csum_type;
-    uint32_t csum_block_size;
-    uint32_t header_csum;
-};
-
-// 32 bytes = 24 bytes + block bitmap (4 bytes by default) + external attributes (also bitmap, 4 bytes by default)
-// per "clean" entry on disk with fixed metadata tables
-struct __attribute__((__packed__)) clean_disk_entry
-{
-    object_id oid;
-    uint64_t version;
-    uint8_t bitmap[];
-    // Two more fields come after bitmap in metadata version 2:
-    // uint32_t data_csum[];
-    // uint32_t entry_csum;
-};
@@ -765,13 +765,8 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
    {
        return;
    }
-    // CAS writes are simplified: they're not cached, not resliced, not retried, and not part of the regular write queue at all
-    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version)
-    {
-        execute_cas(op);
-        return;
-    }
-    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && enable_writeback && !(op->flags & OP_FLUSH_BUFFER))
+    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && enable_writeback && !(op->flags & OP_FLUSH_BUFFER) &&
+        !op->version /* no CAS writeback */)
    {
        if (wb->writebacks_active >= client_max_writeback_iodepth)
        {
@@ -793,7 +788,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
    }
    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && !(op->flags & OP_IMMEDIATE_COMMIT))
    {
-        if (!(op->flags & OP_FLUSH_BUFFER))
+        if (!(op->flags & OP_FLUSH_BUFFER) && !op->version /* no CAS write-repeat */)
        {
            uint64_t flush_id = ++wb->last_flush_id;
            wb->copy_write(op, CACHE_REPEATING, flush_id);
@@ -852,72 +847,6 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
    }
 }

-void cluster_client_t::execute_cas(cluster_op_t *op)
-{
-    slice_rw(op);
-    op->needs_reslice = false;
-    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1)
-    {
-        // Atomic writes to multiple stripes are unsupported
-        op->retval = -EINVAL;
-        auto cb = std::move(op->callback);
-        cb(op);
-        return;
-    }
-    int res = try_send(op, 0, [this, op](osd_op_t *part)
-    {
-        int expected = part->req.hdr.opcode == OSD_OP_DELETE ? 0 : part->req.rw.len;
-        op->retval = part->reply.hdr.retval;
-        op->retval = op->retval == expected ? 0 : (op->retval >= 0 ? -EIO : op->retval);
-        op->retval = op->retval == -EPIPE ? -EINTR : op->retval;
-        auto peer_it = msgr.osd_peer_fds.find(op->parts[0].osd_num);
-        if (op->retval != 0 || (op->flags & OP_IMMEDIATE_COMMIT))
-        {
-            auto cb = std::move(op->callback);
-            cb(op);
-        }
-        else if (peer_it == msgr.osd_peer_fds.end())
-        {
-            // Care must be taken to make sure that the client doesn't reconnect to the OSD
-            // before executing the previously completed operation callback (!)
-            op->retval = -EINTR;
-            auto cb = std::move(op->callback);
-            cb(op);
-        }
-        else
-        {
-            // CAS writes have a built-in sync
-            auto peer_fd = peer_it->second;
-            *part = (osd_op_t){
-                .op_type = OSD_OP_OUT,
-                .peer_fd = peer_fd,
-                .req = {
-                    .hdr = {
-                        .magic = SECONDARY_OSD_OP_MAGIC,
-                        .opcode = OSD_OP_SYNC,
-                    },
-                },
-                .callback = [this, op](osd_op_t *part)
-                {
-                    op->retval = part->reply.hdr.retval;
-                    op->retval = op->retval == -EPIPE ? -EINTR : op->retval;
-                    auto cb = std::move(op->callback);
-                    cb(op);
-                },
-            };
-            msgr.outbox_push(part);
-        }
-    });
-    if (res == TRY_SEND_CONNECTING || res == TRY_SEND_OFFLINE)
-    {
-        // In theory, CAS writes could wait for the PG to come up, but it's easier to just fail it
-        op->retval = -EINTR;
-        auto cb = std::move(op->callback);
-        cb(op);
-        return;
-    }
-}
-
 bool cluster_client_t::check_rw(cluster_op_t *op)
 {
    if (op->opcode == OSD_OP_SYNC)
@@ -1027,6 +956,13 @@ resume_0:
    // Slice the operation into parts
    slice_rw(op);
    op->needs_reslice = false;
+    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1)
+    {
+        // Atomic writes to multiple stripes are unsupported
+        op->retval = -EINVAL;
+        erase_op(op);
+        return 1;
+    }
 resume_1:
    // Send unsent parts, if they're not subject to change
    op->state = 2;
@@ -1371,7 +1307,7 @@ bool cluster_client_t::affects_osd(uint64_t inode, uint64_t offset, uint64_t len
    return false;
 }

-int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_op_t *op_part)> cb)
+int cluster_client_t::try_send(cluster_op_t *op, int i)
 {
    if (!msgr_initialized)
    {
@@ -1431,7 +1367,7 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
                    ? (uint8_t*)op->part_bitmaps + pg_bitmap_size*i : NULL),
                .bitmap_len = (unsigned)(op->opcode == OSD_OP_READ || op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP
                    ? pg_bitmap_size : 0),
-                .callback = cb ? cb : [this, part](osd_op_t *op_part)
+                .callback = [this, part](osd_op_t *op_part)
                {
                    handle_op_part(part);
                },
@@ -175,13 +175,12 @@ protected:
    void on_change_node_placement_hook();

    void execute_internal(cluster_op_t *op);
-    void execute_cas(cluster_op_t *op);
    void unshift_op(cluster_op_t *op);
    int continue_rw(cluster_op_t *op);
    bool check_rw(cluster_op_t *op);
    void slice_rw(cluster_op_t *op);
    void reset_retry_timer(int new_duration);
-    int try_send(cluster_op_t *op, int i, std::function<void(osd_op_t *op_part)> cb = nullptr);
+    int try_send(cluster_op_t *op, int i);
    int continue_sync(cluster_op_t *op);
    void send_sync(cluster_op_t *op, cluster_op_part_t *part);
    void handle_op_part(cluster_op_part_t *part);
@@ -234,6 +234,7 @@ public:
    void parse_config(const json11::Json & config);
    void connect_peer(uint64_t osd_num, json11::Json peer_state);
    void stop_client(int peer_fd, bool force = false, bool force_delete = false);
+    void destroy_client(osd_client_t *cl);
    void outbox_push(osd_op_t *cur_op);
    std::function<void(osd_op_t*)> exec_op;
    std::function<void(osd_num_t)> repeer_pgs;
@@ -78,7 +78,7 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl)
    {
        if (cl->refs <= 0)
        {
-            delete cl;
+            destroy_client(cl);
        }
        return false;
    }
@@ -274,7 +274,7 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
    {
        if (cl->refs <= 0)
        {
-            delete cl;
+            destroy_client(cl);
        }
        return;
    }
@@ -323,12 +323,8 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
        }
        if (more)
        {
-            int expected = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
-            if (done != expected)
-            {
-                fprintf(stderr, "BUG (maybe kernel): Expected to send %d iovecs with MSG_WAITALL but sent %d\n", expected, done);
-                exit(1);
-            }
+            auto expected = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
+            assert(done == expected);
            cl->zc_free_list.push_back(NULL); // end marker
        }
        if (done > 0)
@@ -85,30 +85,13 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
            osd_peer_fds.erase(osd_it);
        }
    }
+    // Do not remove socket from the event loop as it may have refs > 0 and we want to clear them
 #ifndef __MOCK__
-    // Then remove FD from the eventloop so we don't accidentally read something
-    tfd->set_fd_handler(peer_fd, false, NULL);
    if (cl->connect_timeout_id >= 0)
    {
        tfd->clear_timer(cl->connect_timeout_id);
        cl->connect_timeout_id = -1;
    }
-    for (auto rit = read_ready_clients.begin(); rit != read_ready_clients.end(); rit++)
-    {
-        if (*rit == peer_fd)
-        {
-            read_ready_clients.erase(rit);
-            break;
-        }
-    }
-    for (auto wit = write_ready_clients.begin(); wit != write_ready_clients.end(); wit++)
-    {
-        if (*wit == peer_fd)
-        {
-            write_ready_clients.erase(wit);
-            break;
-        }
-    }
 #endif
    if (cl->in_osd_num && break_pg_locks)
    {
@@ -123,19 +106,42 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
        // so do not repeer on it.
        repeer_pgs(cl->osd_num);
    }
-    // Find the item again because it can be invalidated at this point
-    it = clients.find(peer_fd);
-    if (it != clients.end())
-    {
-        clients.erase(it);
-    }
    cl->refs--;
    if (cl->refs <= 0 || force_delete)
    {
-        delete cl;
+        destroy_client(cl);
    }
 }

+void osd_messenger_t::destroy_client(osd_client_t *cl)
+{
+#ifndef __MOCK__
+    if (cl->peer_fd >= 0)
+    {
+        // Remove FD from the eventloop
+        tfd->set_fd_handler(cl->peer_fd, false, NULL);
+        for (auto rit = read_ready_clients.begin(); rit != read_ready_clients.end(); rit++)
+        {
+            if (*rit == cl->peer_fd)
+            {
+                read_ready_clients.erase(rit);
+                break;
+            }
+        }
+        for (auto wit = write_ready_clients.begin(); wit != write_ready_clients.end(); wit++)
+        {
+            if (*wit == cl->peer_fd)
+            {
+                write_ready_clients.erase(wit);
+                break;
+            }
+        }
+    }
+#endif
+    clients.erase(cl->peer_fd);
+    delete cl;
+}
+
 osd_client_t::~osd_client_t()
 {
    free(in_buf);
@@ -6,20 +6,8 @@
 #include <stdint.h>
 #include <functional>

-#define POOL_SCHEME_REPLICATED 1
-#define POOL_SCHEME_XOR 2
-#define POOL_SCHEME_EC 3
-#define POOL_ID_MAX 0x10000
-#define POOL_ID_BITS 16
-#define INODE_POOL(inode) (pool_id_t)((inode) >> (64 - POOL_ID_BITS))
-#define INODE_NO_POOL(inode) (inode_t)((inode) & (((uint64_t)1 << (64-POOL_ID_BITS)) - 1))
-#define INODE_WITH_POOL(pool_id, inode) (((inode_t)(pool_id) << (64-POOL_ID_BITS)) | INODE_NO_POOL(inode))
-
 typedef uint64_t inode_t;

-// Pool ID is 16 bits long
-typedef uint32_t pool_id_t;
-
 // 16 bytes per object/stripe id
 // stripe = (start of the parity stripe + peer role)
 // i.e. for example (256KB + one of 0,1,2)
@@ -44,6 +44,11 @@
 #define DIRECT_IO_ALIGNMENT 512
 #endif

+// Memory allocation alignment (page size is usually optimal)
+#ifndef MEM_ALIGNMENT
+#define MEM_ALIGNMENT 4096
+#endif
+
 // Constants for osd_reply_describe_item_t.loc_bad
 #define LOC_OUTDATED 1
 #define LOC_CORRUPTED 2
@@ -262,10 +267,7 @@ struct __attribute__((__packed__)) osd_reply_del_t
    uint32_t left_on_dead_count;
 };

-// sync to the primary OSD. semantics:
-// 1) any non-synced write may disappear on OSD restart. even if it's a CAS write.
-// 2) sync only guarantees to commit completed writes, not in-progress ones.
-// 3) sync is a no-op when immediate_commit is active.
+// sync to the primary OSD
 struct __attribute__((__packed__)) osd_op_sync_t
 {
    osd_op_header_t header;
@@ -6,7 +6,7 @@ includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@

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

@@ -8,17 +8,14 @@
 #endif

 #include <map>
-#include <set>
 #include <vector>
 #include <string>
 #include <functional>

 #include "json11/json11.hpp"
 #include "blockstore_disk.h"
-#include "blockstore.h"
-#include "ondisk_formats.h"
+#include "blockstore_impl.h"
 #include "crc32c.h"
-#include "allocator.h"

 // vITADisk
 #define VITASTOR_DISK_MAGIC 0x6b73694441544976
@@ -1,10 +1,7 @@
 // Copyright (c) Vitaliy Filippov, 2019+
 // License: VNPL-1.1 (see README.md for details)

-#include <unistd.h>
-
 #include "disk_tool.h"
-#include "malloc_or_die.h"

 int disk_tool_t::dump_journal()
 {
@@ -5,7 +5,6 @@
 #include "rw_blocking.h"
 #include "osd_id.h"
 #include "json_util.h"
-#include "malloc_or_die.h"

 int disk_tool_t::process_meta(std::function<void(blockstore_meta_header_v2_t *)> hdr_fn,
    std::function<void(uint64_t, clean_disk_entry*, uint8_t*)> record_fn, bool do_open)
@@ -4,7 +4,6 @@
 #include "disk_tool.h"
 #include "rw_blocking.h"
 #include "str_util.h"
-#include "malloc_or_die.h"

 #define DM_ST_EMPTY 0
 #define DM_ST_TO_READ 1
@@ -7,7 +7,6 @@
 #include "rw_blocking.h"
 #include "str_util.h"
 #include "json_util.h"
-#include "malloc_or_die.h"

 struct __attribute__((__packed__)) vitastor_disk_superblock_t
 {
@@ -2,13 +2,11 @@
 // License: VNPL-1.1 (see README.md for details)

 #include <sys/wait.h>
-#include <sys/ioctl.h>
 #include <dirent.h>

 #include "disk_tool.h"
 #include "rw_blocking.h"
 #include "str_util.h"
-#include "malloc_or_die.h"

 uint64_t sscanf_json(const char *fmt, const json11::Json & str)
 {
@@ -97,7 +97,6 @@ void kv_cli_t::parse_args(int narg, const char *args[])
                "  dump [<start> [end]]\n"
                "  dumpjson [<start> [end]]\n"
                "  loadjson\n"
-                "  rescue\n"
                "\n"
                "<IMAGE> should be the name of Vitastor image with the DB.\n"
                "Without <COMMAND>, you get an interactive DB shell.\n"
@@ -299,50 +298,6 @@ struct kv_cli_list_t
    int n = 0;
    std::function<void(int)> cb;

-    void handle_key(int res, const std::string & key, const std::string & value)
-    {
-        if (res < 0)
-        {
-            if (res != -ENOENT)
-                fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
-            if (format == 2)
-                write("\n}\n");
-            if (handle)
-                db->list_close(handle);
-            flush();
-            cb(res == -ENOENT ? 0 : res);
-            delete this;
-        }
-        else
-        {
-            if (format == 2)
-            {
-                write(n ? ",\n  " : "{\n  ");
-                write(addslashes(key));
-                write(": ");
-                write(addslashes(value));
-            }
-            else if (format == 1)
-            {
-                write("set ");
-                write(auto_addslashes(key));
-                write(" ");
-                write(value);
-                write("\n");
-            }
-            else
-            {
-                write(key);
-                write(" = ");
-                write(value);
-                write("\n");
-            }
-            n++;
-            if (handle)
-                db->list_next(handle, NULL);
-        }
-    }
-
    void write(const std::string & str)
    {
        if (buf.capacity() < KV_LIST_BUF_SIZE)
@@ -361,7 +316,6 @@ struct kv_cli_list_t
            if (res > 0)
                done += res;
        }
-        buf.clear();
    }
 };

@@ -670,18 +624,44 @@ void kv_cli_t::handle_cmd(const std::vector<std::string> & cmd, std::function<vo
        lst->cb = std::move(cb);
        db->list_next(lst->handle, [lst](int res, const std::string & key, const std::string & value)
        {
-            lst->handle_key(res, key, value);
-        });
-    }
-    else if (opname == "rescue")
-    {
-        kv_cli_list_t *lst = new kv_cli_list_t;
-        lst->db = db;
-        lst->format = 2;
-        lst->cb = std::move(cb);
-        db->rescue([lst](int res, const std::string & key, const std::string & value)
-        {
-            lst->handle_key(res, key, value);
+            if (res < 0)
+            {
+                if (res != -ENOENT)
+                    fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
+                if (lst->format == 2)
+                    lst->write("\n}\n");
+                lst->flush();
+                lst->db->list_close(lst->handle);
+                lst->cb(res == -ENOENT ? 0 : res);
+                delete lst;
+            }
+            else
+            {
+                if (lst->format == 2)
+                {
+                    lst->write(lst->n ? ",\n  " : "{\n  ");
+                    lst->write(addslashes(key));
+                    lst->write(": ");
+                    lst->write(addslashes(value));
+                }
+                else if (lst->format == 1)
+                {
+                    lst->write("set ");
+                    lst->write(auto_addslashes(key));
+                    lst->write(" ");
+                    lst->write(value);
+                    lst->write("\n");
+                }
+                else
+                {
+                    lst->write(key);
+                    lst->write(" = ");
+                    lst->write(value);
+                    lst->write("\n");
+                }
+                lst->n++;
+                lst->db->list_next(lst->handle, NULL);
+            }
        });
    }
    else if (opname == "loadjson")
@@ -96,7 +96,7 @@ struct kv_block_t

    void set_data_size();
    static int kv_size(const std::string & key, const std::string & value);
-    int parse(uint64_t offset, uint8_t *data, int size, bool allow_empty = false);
+    int parse(uint64_t offset, uint8_t *data, int size);
    bool serialize(uint8_t *data, int size);
    void apply_change();
    void cancel_change();
@@ -139,6 +139,7 @@ struct kv_db_t
    uint64_t next_free = 0;
    uint32_t kv_block_size = 0;
    uint32_t ino_block_size = 0;
+    bool immediate_commit = false;
    uint64_t memory_limit = 128*1024*1024;
    uint64_t evict_unused_age = 1000;
    uint64_t evict_max_misses = 10;
@@ -172,7 +173,6 @@ struct kv_db_t
    void open(inode_t inode_id, json11::Json cfg, std::function<void(int)> cb);
    void set_config(json11::Json cfg);
    void close(std::function<void()> cb);
-    void rescue(std::function<void(int res, const std::string & key, const std::string & value)> cb);

    void find_size(uint64_t min, uint64_t max, int phase, std::function<void(int, uint64_t)> cb);
    void run_continue_update(uint64_t offset);
@@ -243,13 +243,13 @@ static std::string read_string(uint8_t *data, int size, int *pos)
    return key;
 }

-int kv_block_t::parse(uint64_t offset, uint8_t *data, int size, bool allow_empty)
+int kv_block_t::parse(uint64_t offset, uint8_t *data, int size)
 {
    kv_stored_block_t *blk = (kv_stored_block_t *)data;
    if (blk->magic == 0 || blk->type == KV_EMPTY)
    {
        // empty block
-        if (!allow_empty)
+        if (offset != 0)
            fprintf(stderr, "K/V: Block %ju is %s\n", offset, blk->magic == 0 ? "empty" : "cleared");
        return -ENOTBLK;
    }
@@ -389,10 +389,6 @@ bool kv_block_t::serialize(uint8_t *buf, int size)
            return false;
        blk->items++;
    }
-    if (pos < size)
-    {
-        memset(buf+pos, 0, size-pos);
-    }
    return true;
 }

@@ -528,6 +524,7 @@ void kv_db_t::open(inode_t inode_id, json11::Json cfg, std::function<void(int)>
        return;
    }
    this->inode_id = inode_id;
+    this->immediate_commit = cli->get_immediate_commit(inode_id);
    this->ino_block_size = pool_cfg.data_block_size * pg_data_size;
    this->kv_block_size = kv_block_size;
    this->next_free = 0;
@@ -545,127 +542,6 @@ void kv_db_t::open(inode_t inode_id, json11::Json cfg, std::function<void(int)>
    });
 }

-struct kv_rescue_t
-{
-    kv_db_t *db = NULL;
-    uint64_t size = 0;
-    uint64_t pos = 0;
-    uint64_t cur_size = 0;
-    uint64_t cur_offset = 0;
-    int state = 0;
-    kv_block_t blk;
-    std::vector<uint8_t> buf;
-    std::function<void(int res, const std::string & key, const std::string & value)> cb;
-
-    void finish(int retval)
-    {
-        auto cb = std::move(this->cb);
-        cb(retval, "", "");
-        delete this;
-    }
-
-    void send_read()
-    {
-        if (pos >= size)
-        {
-            finish(-ENOENT);
-            return;
-        }
-        if (!buf.size())
-        {
-            buf.resize(1048576);
-        }
-        cluster_op_t *op = new cluster_op_t;
-        op->opcode = OSD_OP_READ;
-        op->inode = db->inode_id;
-        op->offset = pos;
-        cur_size = op->len = pos+buf.size() < size ? buf.size() : size-pos;
-        op->iov.push_back(buf.data(), cur_size);
-        op->callback = [=](cluster_op_t *op)
-        {
-            if (op->retval != op->len)
-            {
-                // error
-                finish(op->retval >= 0 ? -EIO : op->retval);
-                return;
-            }
-            state = 2;
-            cur_offset = 0;
-            run();
-            delete op;
-        };
-        db->cli->execute(op);
-        state = 1;
-    }
-
-    void parse_block()
-    {
-        if (cur_offset < cur_size)
-        {
-            blk = {};
-            int err = blk.parse(pos+cur_offset, buf.data()+cur_offset, db->kv_block_size, true);
-            if (err != 0)
-            {
-            }
-            else if (blk.type == KV_LEAF || blk.type == KV_LEAF_SPLIT)
-            {
-                for (auto it = blk.data.begin(); it != blk.data.end(); it++)
-                {
-                    cb(0, it->first, it->second);
-                }
-            }
-            cur_offset += db->kv_block_size;
-        }
-        else
-        {
-            pos += cur_size;
-            state = 0;
-        }
-    }
-
-    void run()
-    {
-        while (true)
-        {
-            if (state == 0)
-            {
-                send_read();
-                return;
-            }
-            else if (state == 1)
-            {
-                return;
-            }
-            else if (state == 2)
-            {
-                parse_block();
-            }
-        }
-    }
-};
-
-void kv_db_t::rescue(std::function<void(int res, const std::string & key, const std::string & value)> cb)
-{
-    if (!inode_id || closing)
-    {
-        cb(-EINVAL, "", "");
-        return;
-    }
-    find_size(0, 0, 1, [=](int res, uint64_t size)
-    {
-        if (res < 0)
-        {
-            cb(res, "", "");
-            return;
-        }
-        kv_rescue_t *st = new kv_rescue_t();
-        st->db = this;
-        st->size = size;
-        st->cb = cb;
-        st->run();
-    });
-}
-
 void kv_db_t::set_config(json11::Json cfg)
 {
    this->memory_limit = cfg["kv_memory_limit"].is_null() ? 128*1024*1024 : cfg["kv_memory_limit"].uint64_value();
@@ -687,6 +563,7 @@ void kv_db_t::close(std::function<void()> cb)
        next_free = 0;
        kv_block_size = 0;
        ino_block_size = 0;
+        immediate_commit = false;
        block_cache.clear();
        known_versions.clear();
        cb();
@@ -1062,7 +939,7 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
                del_block_level(db, blk);
                *blk = {};
            }
-            int err = blk->parse(op->offset, (uint8_t*)op->iov.buf[0].iov_base, op->len, op->offset == 0);
+            int err = blk->parse(op->offset, (uint8_t*)op->iov.buf[0].iov_base, op->len);
            if (err == 0)
            {
                blk->level = cur_level;
@@ -1125,8 +1002,6 @@ kv_op_t::~kv_op_t()
    {
        done = true;
        db->active_ops--;
-        if (!db->active_ops && db->closing)
-            db->close(db->on_close);
    }
 }

@@ -1397,7 +1272,22 @@ static void write_block(kv_db_t *db, kv_block_t *blk, std::function<void(int)> c
            }
        }
        delete op;
-        cb(res);
+        if (res < 0 || db->immediate_commit)
+        {
+            cb(res);
+        }
+        else
+        {
+            op = new cluster_op_t;
+            op->opcode = OSD_OP_SYNC;
+            op->callback = [cb](cluster_op_t *op)
+            {
+                auto res = op->retval;
+                delete op;
+                cb(res);
+            };
+            db->cli->execute(op);
+        }
    };
    db->cli->execute(op);
 }
@@ -1772,7 +1662,6 @@ void kv_op_t::update_block(int path_pos, bool is_delete, const std::string & key
        {
            if (res < 0)
            {
-                blk->cancel_change();
                auto blk_offset = blk->offset;
                del_block_level(db, blk);
                db->block_cache.erase(blk_offset);
@@ -1905,7 +1794,6 @@ void kv_op_t::update_block(int path_pos, bool is_delete, const std::string & key
            {
                if (write_res < 0)
                {
-                    blk->cancel_change();
                    auto blk_offset = blk->offset;
                    del_block_level(db, blk);
                    db->block_cache.erase(blk_offset);
@@ -2093,11 +1981,6 @@ void vitastorkv_dbw_t::close(std::function<void()> cb)
    db->close(cb);
 }

-void vitastorkv_dbw_t::rescue(std::function<void(int res, const std::string & key, const std::string & value)> cb)
-{
-    db->rescue(cb);
-}
-
 void vitastorkv_dbw_t::get(const std::string & key, std::function<void(int res, const std::string & value)> cb, bool cached)
 {
    auto *op = new kv_op_t;
@@ -82,7 +82,6 @@ public:
    uint64_t total_prob = 0;
    uint64_t ops_sent = 0, ops_done = 0;
    int stat_timer_id = -1;
-    int run_timer_id = -1;
    int in_progress = 0;
    bool reopening = false;
    std::set<kv_test_listing_t*> listings;
@@ -300,11 +299,9 @@ void kv_test_t::run(json11::Json cfg)
    ringloop->register_consumer(&consumer);
    if (print_stats_interval)
        stat_timer_id = epmgr->tfd->set_timer(print_stats_interval*1000, true, [this](int) { print_stats(prev_stat, prev_stat_time); });
-    if (runtime_sec)
-        run_timer_id = epmgr->tfd->set_timer(runtime_sec*1000, false, [this](int) { run_timer_id = -1; op_count = 0; });
    clock_gettime(CLOCK_REALTIME, &start_stat_time);
    prev_stat_time = start_stat_time;
-    while (!finished || in_progress > 0)
+    while (!finished)
    {
        ringloop->loop();
        if (!finished)
@@ -312,8 +309,6 @@ void kv_test_t::run(json11::Json cfg)
    }
    if (stat_timer_id >= 0)
        epmgr->tfd->clear_timer(stat_timer_id);
-    if (run_timer_id >= 0)
-        epmgr->tfd->clear_timer(run_timer_id);
    ringloop->unregister_consumer(&consumer);
    // Print total stats
    print_total_stats();
@@ -28,7 +28,6 @@ struct __attribute__((visibility("default"))) vitastorkv_dbw_t
    void open(uint64_t inode_id, std::map<std::string, std::string> cfg, std::function<void(int)> cb);
    void set_config(std::map<std::string, std::string> cfg);
    void close(std::function<void()> cb);
-    void rescue(std::function<void(int res, const std::string & key, const std::string & value)> cb);

    uint64_t get_size();

@@ -166,20 +166,7 @@ int kv_nfs3_access_proc(void *opaque, rpc_op_t *rop)
            fprintf(stderr, "[%d] ACCESS %ju -> %s\n", self->nfs_fd, ino, value.c_str());
        if (res < 0)
        {
-            // Broken inode (non-existing), allow only root access
-            if (!rop->auth_sys.uid)
-            {
-                *reply = (ACCESS3res){
-                    .status = NFS3_OK,
-                    .resok = (ACCESS3resok){
-                        .access = args->access,
-                    },
-                };
-            }
-            else
-            {
-                *reply = (ACCESS3res){ .status = vitastor_nfs_map_err(-res) };
-            }
+            *reply = (ACCESS3res){ .status = vitastor_nfs_map_err(-res) };
        }
        else
        {
@@ -9,6 +9,7 @@
 #include <arpa/inet.h>

 #include "addr_util.h"
+#include "blockstore_impl.h"
 #include "osd_primary.h"
 #include "osd.h"
 #include "http_client.h"
@@ -34,7 +35,7 @@ osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop)
    if (!json_is_true(this->config["disable_blockstore"]))
    {
        auto bs_cfg = json_to_string_map(this->config);
-        this->bs = blockstore_i::create(bs_cfg, ringloop, tfd);
+        this->bs = new blockstore_t(bs_cfg, ringloop, tfd);
        // Wait for blockstore initialisation before actually starting OSD logic
        // to prevent peering timeouts during restart with filled databases
        while (!bs->is_started())
@@ -649,11 +650,11 @@ void osd_t::print_slow()
                    op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
                {
                    cur_slow_op_secondary++;
-                    if (op->bs_op)
+                    bufprintf(" state=%d", op->bs_op ? PRIV(op->bs_op)->op_state : -1);
+                    int wait_for = op->bs_op ? PRIV(op->bs_op)->wait_for : 0;
+                    if (wait_for)
                    {
-                        auto diag = bs->get_op_diag(op->bs_op);
-                        if (diag != "")
-                            bufprintf(" %s", diag.c_str());
+                        bufprintf(" wait=%d (detail=%ju)", wait_for, PRIV(op->bs_op)->wait_detail);
                    }
                }
                else if (op->req.hdr.opcode == OSD_OP_READ || op->req.hdr.opcode == OSD_OP_WRITE ||
@@ -202,7 +202,7 @@ class osd_t

    bool stopping = false;
    int inflight_ops = 0;
-    blockstore_i *bs = NULL;
+    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;
@@ -5,6 +5,18 @@

 #include "object_id.h"

+#define POOL_SCHEME_REPLICATED 1
+#define POOL_SCHEME_XOR 2
+#define POOL_SCHEME_EC 3
+#define POOL_ID_MAX 0x10000
+#define POOL_ID_BITS 16
+#define INODE_POOL(inode) (pool_id_t)((inode) >> (64 - POOL_ID_BITS))
+#define INODE_NO_POOL(inode) (inode_t)((inode) & (((uint64_t)1 << (64-POOL_ID_BITS)) - 1))
+#define INODE_WITH_POOL(pool_id, inode) (((inode_t)(pool_id) << (64-POOL_ID_BITS)) | INODE_NO_POOL(inode))
+
+// Pool ID is 16 bits long
+typedef uint32_t pool_id_t;
+
 typedef uint64_t osd_num_t;
 typedef uint32_t pg_num_t;

@@ -165,6 +165,7 @@ resume_3:
        return;
    }
    // Check CAS version
+    // FIXME: Handle CAS writes as "immediate" in non-immediate_commit pools, otherwise CAS doesn't make sense
    if (cur_op->req.rw.version && op_data->fact_ver != (cur_op->req.rw.version-1))
    {
        deref_object_state(pg, &op_data->object_state, true);
@@ -8,6 +8,11 @@
 #include "object_id.h"
 #include "osd_id.h"

+// Memory allocation alignment (page size is usually optimal)
+#ifndef MEM_ALIGNMENT
+#define MEM_ALIGNMENT 4096
+#endif
+
 struct buf_len_t
 {
    void *buf;
@@ -16,7 +16,6 @@

 #include <stdexcept>

-#include "malloc_or_die.h"
 #include "addr_util.h"
 #include "osd_ops.h"
 #include "rw_blocking.h"
@@ -13,7 +13,7 @@ int main(int narg, char *args[])
    config["data_device"] = "./test_data.bin";
    ring_loop_t *ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
    epoll_manager_t *epmgr = new epoll_manager_t(ringloop);
-    blockstore_i *bs = blockstore_i::create(config, ringloop, epmgr->tfd);
+    blockstore_t *bs = new blockstore_t(config, ringloop, epmgr->tfd);

    blockstore_op_t op;
    int main_state = 0;
@@ -6,11 +6,6 @@
 #include <malloc.h>
 #include <stdlib.h>

-// Memory allocation alignment (page size is usually optimal)
-#ifndef MEM_ALIGNMENT
-#define MEM_ALIGNMENT 4096
-#endif
-
 #pragma GCC visibility push(default)

 inline void* memalign_or_die(size_t alignment, size_t size)
@@ -192,7 +192,7 @@ void ring_loop_t::restore(unsigned sqe_tail)
    ring.sq.sqe_tail = sqe_tail;
 }

-unsigned ring_loop_t::space_left()
+int ring_loop_t::sqes_left()
 {
    struct io_uring_sq *sq = &ring.sq;
    unsigned int head = io_uring_smp_load_acquire(sq->khead);
@@ -68,7 +68,11 @@ public:
        struct io_uring_cqe *cqe;
        return io_uring_wait_cqe(&ring, &cqe);
    }
-    unsigned space_left();
+    int sqes_left();
+    inline unsigned space_left()
+    {
+        return free_ring_data_ptr;
+    }
    inline bool has_work()
    {
        return loop_again;
@@ -48,9 +48,6 @@ SCHEME=ec ./test_snapshot_chain.sh
 ./test_snapshot_down.sh
 SCHEME=ec ./test_snapshot_down.sh

-./test_kv_stress.sh
-IMMEDIATE_COMMIT=1 ./test_kv_stress.sh
-
 ./test_splitbrain.sh

 ./test_rebalance_verify.sh
@@ -1,8 +0,0 @@
-#!/bin/bash -ex
-
-PG_COUNT=16
-. `dirname $0`/run_3osds.sh
-
-build/src/kv/vitastor-kv-stress --etcd_address $ETCD_URL --pool_id 1 --inode_id 1 --runtime 30
-
-format_green OK
@@ -1,7 +1,5 @@
 #!/bin/bash -ex

-GLOBAL_CONFIG=',"client_enable_writeback":false'
-IMMEDIATE_COMMIT=1
 PG_COUNT=16
 . `dirname $0`/run_3osds.sh

@@ -179,7 +177,7 @@ build/src/kv/vitastor-kv --etcd_address $ETCD_URL fsmeta set d11/settings.jsonLG
 sudo mount localhost:/ ./testdata/nfs -o port=2050,mountport=2050,nfsvers=3,soft,nolock,tcp
 ls -l ./testdata/nfs
 ls -l ./testdata/nfs/settings.jsonLGNmGn
-sudo rm ./testdata/nfs/settings.jsonLGNmGn
+rm ./testdata/nfs/settings.jsonLGNmGn
 build/src/kv/vitastor-kv --etcd_address $ETCD_URL fsmeta get d11/settings.jsonLGNmGn 2>&1 | grep '(code -2)'
 ls -l ./testdata/nfs

@@ -189,7 +187,7 @@ build/src/kv/vitastor-kv --etcd_address $ETCD_URL fsmeta set d11/settings.jsonLG
 sudo mount localhost:/ ./testdata/nfs -o port=2050,mountport=2050,nfsvers=3,soft,nolock,tcp
 ls -l ./testdata/nfs
 ls -l ./testdata/nfs/settings.jsonLGNmGn
-sudo rm ./testdata/nfs/settings.jsonLGNmGn
+rm ./testdata/nfs/settings.jsonLGNmGn
 build/src/kv/vitastor-kv --etcd_address $ETCD_URL fsmeta get d11/settings.jsonLGNmGn 2>&1 | grep '(code -2)'
 ls -l ./testdata/nfs