Release 2.4.4

- Implement `vitastor-kv rescue` command which dumps all readable blocks from the database regardless of the tree structure and allows to recover a corrupted database. - Simplify primary OSD selection - use just PG number instead of the hash. This allows to distribute primary OSDs more evenly and reach better read speeds. - Fix vitastor-kv-stress hang on parallel list & reopen. - Fix CAS writes without immediate_commit. Fix is sufficient for vitastor-kv to operate correctly in non-immediate_commit pools. - Fix NFS ACCESS calls for broken inodes (allow root access to them). - Fix broken vitastor-kv-stress --runtime parameter. - Fix bad vitastor-kv dump output buffering leading to incorrect JSON in dumps. - Fix writing garbage from uninitialized memory to the empty space in VitastorKV DB. - Refactor some code in preparation to the release of the new log-structured metadata store. - Replace assert(done != expected) error message, probably caused by kernel issues with zero-copy TCP send, with a more concise error description.
Fix vitastor-kv-stress hang on parallel list & reopen
2025-12-01 02:22:27 +03:00 · 2025-11-30 16:54:23 +03:00 · 2025-11-30 14:50:50 +03:00 · 2025-11-30 01:49:49 +03:00 · 2025-11-29 19:30:20 +03:00 · 2025-11-29 19:22:01 +03:00
76 changed files with 914 additions and 593 deletions
@@ -414,6 +414,24 @@ jobs:
          echo ""
        done

+  test_degraded:
+    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_degraded.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_rm:
    runs-on: ubuntu-latest
    needs: build
@@ -522,6 +540,42 @@ 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.2")
+set(VITASTOR_VERSION "2.4.4")

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

 all: build push

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

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

 // Config struct fills the parameters of request or user input
@@ -0,0 +1,5 @@
+#!/bin/bash
+# 25.10 Questing quokka
+
+docker build --build-arg DISTRO=ubuntu --build-arg REL=questing -t vitastor-buildenv:questing -f vitastor-buildenv.Dockerfile .
+docker run -it --rm -e REL=questing -v `dirname $0`/../:/root/vitastor vitastor-buildenv:questing /root/vitastor/debian/vitastor-build.sh
@@ -1,4 +1,4 @@
-vitastor (2.4.2-1) unstable; urgency=medium
+vitastor (2.4.4-1) unstable; urgency=medium

  * Bugfixes

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

 all: build push

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

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

 # 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.2`
+   `docker pull vitalif/vitastor:v2.4.4`
 2. Install scripts to the host system: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.2 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.4 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.2`
+   `docker pull vitalif/vitastor:v2.4.4`
 2. Установите скрипты в хост-систему командой: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.2 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.4 install.sh`
 3. Перезагрузите правила udev: \
   `udevadm control --reload-rules`

@@ -1,6 +1,6 @@
 {
  "name": "vitastor-mon",
-  "version": "2.4.2",
+  "version": "2.4.4",
  "description": "Vitastor SDS monitor service",
  "main": "mon-main.js",
  "scripts": {
@@ -9,7 +9,6 @@ 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)
 {
@@ -39,7 +38,7 @@ function pick_primary(pool_id, pg_num, pool_config, osd_set, up_osds, aff_osds)
    {
        return 0;
    }
-    return alive_set[select_murmur3(alive_set.length, osd_num => pool_id+'/'+pg_num+'/'+osd_num)];
+    return alive_set[pg_num % alive_set.length];
 }

 function recheck_primary(state, global_config, up_osds, osd_tree)
@@ -1,6 +1,6 @@
 {
  "name": "vitastor",
-  "version": "2.4.2",
+  "version": "2.4.4",
  "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.2'
+VITASTOR_VERSION = '2.4.4'

 LOG = logging.getLogger(__name__)

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        2.4.2
+Version:        2.4.4
 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.2.el7.tar.gz
+Source0:        vitastor-2.4.4.el7.tar.gz

 BuildRequires:  gperftools-devel
 BuildRequires:  devtoolset-9-gcc-c++
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        2.4.2
+Version:        2.4.4
 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.2.el8.tar.gz
+Source0:        vitastor-2.4.4.el8.tar.gz

 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-toolset-9-gcc-c++
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        2.4.2
+Version:        2.4.4
 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.2.el9.tar.gz
+Source0:        vitastor-2.4.4.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.2")
+add_definitions(-DVITASTOR_VERSION="2.4.4")
 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,87 +3,7 @@

 #include "blockstore_impl.h"

-blockstore_t::blockstore_t(blockstore_config_t & config, ring_loop_t *ringloop, timerfd_manager_t *tfd)
+blockstore_i* blockstore_i::create(blockstore_config_t & config, ring_loop_t *ringloop, timerfd_manager_t *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);
+    return new blockstore_impl_t(config, ringloop, tfd);
 }
@@ -17,22 +17,14 @@
 #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
@@ -46,8 +38,18 @@

 #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
@@ -162,8 +164,8 @@ struct __attribute__ ((visibility("default"))) blockstore_op_t
            uint32_t list_stable_limit;
        };
    };
-    void *buf = NULL;
-    void *bitmap = NULL;
+    uint8_t *buf = NULL;
+    uint8_t *bitmap = NULL;
    int retval = 0;

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

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

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

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

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

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

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

    // 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.
-    bool is_safe_to_stop();
+    virtual bool is_safe_to_stop() = 0;

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

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

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

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

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

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

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

-    uint32_t get_bitmap_granularity();
+    virtual uint64_t get_journal_size() = 0;
+
+    virtual uint32_t get_bitmap_granularity() = 0;
 };
@@ -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 ((block_order = is_power_of_two(data_block_size)) >= 64 || data_block_size < MIN_DATA_BLOCK_SIZE || data_block_size >= MAX_DATA_BLOCK_SIZE)
+    if (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,6 +8,11 @@
 #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
@@ -46,7 +51,6 @@ 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,6 +2,7 @@
 // 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
@@ -542,7 +543,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.block_order, old_entry->oid.inode, old_entry->oid.stripe,
+                        old_clean_loc / bs->dsk.data_block_size, old_entry->oid.inode, old_entry->oid.stripe,
                        old_entry->version, cur.oid.inode, cur.oid.stripe);
                    exit(1);
                }
@@ -645,7 +646,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.block_order, new_entry->oid.inode, new_entry->oid.stripe,
+            clean_loc / bs->dsk.data_block_size, new_entry->oid.inode, new_entry->oid.stripe,
            new_entry->version, cur.oid.inode, cur.oid.stripe, cur.version
        );
        exit(1);
@@ -695,7 +696,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.block_order)*2*bs->dsk.clean_entry_bitmap_size;
+            auto inmem_bmp = (uint8_t*)bs->clean_bitmaps + (clean_loc / bs->dsk.data_block_size)*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)
@@ -848,7 +849,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.block_order, new_entry->oid.inode, new_entry->oid.stripe,
+                    clean_loc / bs->dsk.data_block_size, new_entry->oid.inode, new_entry->oid.stripe,
                    new_entry->version, cur.oid.inode, cur.oid.stripe, cur.version
                );
            }
@@ -864,7 +865,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.block_order)*2*bs->dsk.clean_entry_bitmap_size;
+                auto inmem_bmp = (uint8_t*)bs->clean_bitmaps + (clean_loc / bs->dsk.data_block_size)*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)
@@ -1231,8 +1232,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.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));
+    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));
    if (bs->inmemory_meta)
    {
        wr.buf = (uint8_t*)bs->metadata_buffer + wr.sector;
@@ -1292,14 +1293,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.block_order,
+            old_clean_loc / bs->dsk.data_block_size,
            cur.oid.inode, cur.oid.stripe, cur.version,
-            clean_loc >> bs->dsk.block_order);
+            clean_loc / bs->dsk.data_block_size);
 #endif
        if (used)
            uo_it->second.was_freed = true;
        else
-            bs->data_alloc->set(old_clean_loc >> bs->dsk.block_order, false);
+            bs->data_alloc->set(old_clean_loc / bs->dsk.data_block_size, false);
    }
    if (has_delete)
    {
@@ -1309,13 +1310,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.block_order,
+            old_clean_loc / bs->dsk.data_block_size,
            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.block_order, false);
+            bs->data_alloc->set(old_clean_loc / bs->dsk.data_block_size, false);
    }
 }

@@ -1,14 +1,6 @@
 // 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,6 +2,7 @@
 // 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)
 {
@@ -228,9 +229,9 @@ void blockstore_impl_t::loop()
        for (auto s: journal.submitting_sectors)
        {
            // Mark journal sector writes as submitted
-            journal.sector_info[s].submit_id = 0;
            if (journal.sector_info[s].submit_id)
                journal.sector_info[s].written = true;
+            journal.sector_info[s].submit_id = 0;
        }
        journal.submitting_sectors.clear();
        if ((initial_ring_space - ringloop->space_left()) > 0)
@@ -274,7 +275,7 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
 {
    if (PRIV(op)->wait_for == WAIT_SQE)
    {
-        if (ringloop->sqes_left() < PRIV(op)->wait_detail)
+        if (ringloop->space_left() < PRIV(op)->wait_detail)
        {
            // stop submission if there's still no free space
 #ifdef BLOCKSTORE_DEBUG
@@ -682,7 +683,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 = stable;
+    op->buf = (uint8_t*)stable;
    FINISH_OP(op);
 }

@@ -705,6 +706,11 @@ 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)
@@ -787,3 +793,14 @@ 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,6 +5,7 @@

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

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

 //#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
 {
@@ -170,15 +67,6 @@ 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;
@@ -196,9 +84,6 @@ 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
@@ -224,23 +109,15 @@ 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;
 };

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

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

@@ -276,6 +153,7 @@ class blockstore_impl_t
    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;
@@ -404,6 +282,7 @@ 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
@@ -430,8 +309,8 @@ public:
    // Unstable writes are added here (map of object_id -> version)
    std::unordered_map<object_id, uint64_t> unstable_writes;

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

    // Set per-pool no_inode_stats
    void set_no_inode_stats(const std::vector<uint64_t> & pool_ids);
@@ -439,6 +318,9 @@ 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,6 +2,7 @@
 // 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
@@ -372,7 +373,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.block_order;
+                    uint64_t old_clean_loc = clean_it->second.location / bs->dsk.data_block_size;
                    if (bs->inmemory_meta)
                    {
                        uint64_t sector = (old_clean_loc / entries_per_block) * bs->dsk.meta_block_size;
@@ -390,7 +391,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.block_order);
+                        entries_to_zero.push_back(clean_it->second.location / bs->dsk.data_block_size);
                    }
 #ifdef BLOCKSTORE_DEBUG
                    printf("Free block %ju from %jx:%jx v%ju (new location is %ju)\n",
@@ -412,7 +413,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.block_order,
+                    .location = (done_cnt+i) * bs->dsk.data_block_size,
                };
            }
            else
@@ -1001,7 +1002,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.block_order
+                    je->big_write.oid.inode, je->big_write.oid.stripe, je->big_write.version, je->big_write.location / bs->dsk.data_block_size
                );
 #endif
                auto dirty_it = bs->dirty_db.upper_bound((obj_ver_id){
@@ -1064,7 +1065,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.block_order))
+                    if (bs->data_alloc->get(je->big_write.location / bs->dsk.data_block_size))
                    {
                        // 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.
@@ -1077,11 +1078,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.block_order,
+                            je->big_write.location / bs->dsk.data_block_size,
                            ov.oid.inode, ov.oid.stripe, ov.version
                        );
 #endif
-                        bs->data_alloc->set(je->big_write.location >> bs->dsk.block_order, true);
+                        bs->data_alloc->set(je->big_write.location / bs->dsk.data_block_size, true);
                    }
                    bs->journal.used_sectors[proc_pos]++;
 #ifdef BLOCKSTORE_DEBUG
@@ -0,0 +1,85 @@
+#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,6 +2,7 @@
 // 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,147 +3,7 @@

 #pragma once

-#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);
-}
+class blockstore_impl_t;

 struct journal_sector_info_t
 {
@@ -4,6 +4,11 @@
 #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,6 +3,7 @@

 #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)
@@ -151,7 +152,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.block_order;
+    uint64_t meta_loc = block_loc / dsk.data_block_size;
    if (inmemory_meta)
    {
        uint64_t sector = (meta_loc / (dsk.meta_block_size / dsk.clean_entry_size)) * dsk.meta_block_size;
@@ -691,8 +692,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.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;
+    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;
    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,
@@ -806,7 +807,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.block_order) << dsk.block_order;
+    clean_loc = (clean_loc / dsk.data_block_size) * dsk.data_block_size;
    if (!dyn_data)
    {
        assert(inmemory_meta);
@@ -866,7 +867,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.block_order) << dsk.block_order);
+                        auto & uo = used_clean_objects.at((rv[i].disk_offset / dsk.data_block_size) * dsk.data_block_size);
                        if (!uo.was_changed)
                        {
                            verify_clean_padded_checksums(
@@ -2,6 +2,7 @@
 // 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)
 {
@@ -210,10 +211,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.block_order,
+            printf("Free block %ju from %jx:%jx v%ju\n", dirty_it->second.location / dsk.data_block_size,
                dirty_it->first.oid.inode, dirty_it->first.oid.stripe, dirty_it->first.version);
 #endif
-            data_alloc->set(dirty_it->second.location >> dsk.block_order, false);
+            data_alloc->set(dirty_it->second.location / dsk.data_block_size, false);
        }
        auto used = --journal.used_sectors.at(dirty_it->second.journal_sector);
 #ifdef BLOCKSTORE_DEBUG
@@ -2,6 +2,7 @@
 // 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
@@ -228,17 +229,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 = bad_vers.items;
+        split_stab_op->buf = (uint8_t*)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;
+        uint8_t *orig_buf = op->buf;
        if (good_vers.items)
        {
-            op->buf = good_vers.items;
+            op->buf = (uint8_t*)good_vers.items;
            op->len = good_vers.size;
        }
        // Make a wrapped callback
@@ -2,6 +2,7 @@
 // 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,6 +2,7 @@
 // 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)
 {
@@ -340,7 +341,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.block_order;
+        dirty_it->second.location = loc * dsk.data_block_size;
        dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_SUBMITTED;
 #ifdef BLOCKSTORE_DEBUG
        printf(
@@ -366,7 +367,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.block_order) + op->offset - stripe_offset
+            sqe, dsk.data_fd, PRIV(op)->iov_zerofill, vcnt, dsk.data_offset + (loc * dsk.data_block_size) + 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_t *bs;
+    blockstore_i *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 = new blockstore_t(config, bsd->ringloop, bsd->epmgr->tfd);
+    bsd->bs = blockstore_i::create(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 = io->xfer_buf;
+        op->buf = (uint8_t*)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 = io->xfer_buf;
+        op->buf = (uint8_t*)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 = malloc_or_die(sizeof(obj_ver_id));
+                    stab_op->buf = (uint8_t*)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;
@@ -0,0 +1,186 @@
+// 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,8 +765,13 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
    {
        return;
    }
-    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && enable_writeback && !(op->flags & OP_FLUSH_BUFFER) &&
-        !op->version /* no CAS writeback */)
+    // 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 (wb->writebacks_active >= client_max_writeback_iodepth)
        {
@@ -788,7 +793,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) && !op->version /* no CAS write-repeat */)
+        if (!(op->flags & OP_FLUSH_BUFFER))
        {
            uint64_t flush_id = ++wb->last_flush_id;
            wb->copy_write(op, CACHE_REPEATING, flush_id);
@@ -847,6 +852,72 @@ 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)
@@ -956,13 +1027,6 @@ 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;
@@ -1307,7 +1371,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)
+int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_op_t *op_part)> cb)
 {
    if (!msgr_initialized)
    {
@@ -1367,7 +1431,7 @@ int cluster_client_t::try_send(cluster_op_t *op, int i)
                    ? (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 = [this, part](osd_op_t *op_part)
+                .callback = cb ? cb : [this, part](osd_op_t *op_part)
                {
                    handle_op_part(part);
                },
@@ -175,12 +175,13 @@ 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);
+    int try_send(cluster_op_t *op, int i, std::function<void(osd_op_t *op_part)> cb = nullptr);
    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);
@@ -276,6 +276,7 @@ osd_messenger_t::~osd_messenger_t()
 #ifdef WITH_RDMA
    for (auto rdma_context: rdma_contexts)
    {
+        tfd->set_fd_handler(rdma_context->channel->fd, false, NULL);
        delete rdma_context;
    }
    rdma_contexts.clear();
@@ -323,8 +323,12 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
        }
        if (more)
        {
-            auto expected = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
-            assert(done == expected);
+            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);
+            }
            cl->zc_free_list.push_back(NULL); // end marker
        }
        if (done > 0)
@@ -6,8 +6,20 @@
 #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,11 +44,6 @@
 #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
@@ -267,7 +262,10 @@ struct __attribute__((__packed__)) osd_reply_del_t
    uint32_t left_on_dead_count;
 };

-// sync to the primary OSD
+// 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.
 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.2
+Version: 2.4.4
 Libs: -L${libdir} -lvitastor_client
 Cflags: -I${includedir}

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

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

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

 // vITADisk
 #define VITASTOR_DISK_MAGIC 0x6b73694441544976
@@ -1,7 +1,10 @@
 // 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,6 +5,7 @@
 #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,6 +4,7 @@
 #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,6 +7,7 @@
 #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,11 +2,13 @@
 // 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,6 +97,7 @@ 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"
@@ -298,6 +299,50 @@ 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)
@@ -316,6 +361,7 @@ struct kv_cli_list_t
            if (res > 0)
                done += res;
        }
+        buf.clear();
    }
 };

@@ -624,44 +670,18 @@ 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)
        {
-            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);
-            }
+            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);
        });
    }
    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);
+    int parse(uint64_t offset, uint8_t *data, int size, bool allow_empty = false);
    bool serialize(uint8_t *data, int size);
    void apply_change();
    void cancel_change();
@@ -139,7 +139,6 @@ 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;
@@ -173,6 +172,7 @@ 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)
+int kv_block_t::parse(uint64_t offset, uint8_t *data, int size, bool allow_empty)
 {
    kv_stored_block_t *blk = (kv_stored_block_t *)data;
    if (blk->magic == 0 || blk->type == KV_EMPTY)
    {
        // empty block
-        if (offset != 0)
+        if (!allow_empty)
            fprintf(stderr, "K/V: Block %ju is %s\n", offset, blk->magic == 0 ? "empty" : "cleared");
        return -ENOTBLK;
    }
@@ -389,6 +389,10 @@ 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;
 }

@@ -524,7 +528,6 @@ 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;
@@ -542,6 +545,127 @@ 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();
@@ -563,7 +687,6 @@ 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();
@@ -939,7 +1062,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);
+            int err = blk->parse(op->offset, (uint8_t*)op->iov.buf[0].iov_base, op->len, op->offset == 0);
            if (err == 0)
            {
                blk->level = cur_level;
@@ -1002,6 +1125,8 @@ kv_op_t::~kv_op_t()
    {
        done = true;
        db->active_ops--;
+        if (!db->active_ops && db->closing)
+            db->close(db->on_close);
    }
 }

@@ -1272,22 +1397,7 @@ static void write_block(kv_db_t *db, kv_block_t *blk, std::function<void(int)> c
            }
        }
        delete op;
-        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);
-        }
+        cb(res);
    };
    db->cli->execute(op);
 }
@@ -1662,6 +1772,7 @@ 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);
@@ -1794,6 +1905,7 @@ 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);
@@ -1981,6 +2093,11 @@ 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,6 +82,7 @@ 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;
@@ -299,9 +300,11 @@ 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)
+    while (!finished || in_progress > 0)
    {
        ringloop->loop();
        if (!finished)
@@ -309,6 +312,8 @@ 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,6 +28,7 @@ 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,7 +166,20 @@ 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)
        {
-            *reply = (ACCESS3res){ .status = vitastor_nfs_map_err(-res) };
+            // 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) };
+            }
        }
        else
        {
@@ -276,6 +276,19 @@ void nfs_proxy_t::run(json11::Json cfg)
    }
    // Check default pool
    check_default_pool();
+    // Daemonize before initializing messenger and RDMA because otherwise RDMA doesn't survive fork()
+    bool bg = cfg["foreground"].is_null() && cfg["cmd"].is_null();
+    int notifyfd[2] = { -1, -1 };
+    if (bg)
+    {
+        if (socketpair(AF_UNIX, SOCK_STREAM, 0, notifyfd) < 0)
+        {
+            perror("socketpair");
+            exit(1);
+        }
+        daemonize_fork(notifyfd);
+        close(notifyfd[0]);
+    }
    // Init VitastorFS after starting client because it depends on loaded inode configuration
    if (fsname != "")
    {
@@ -294,6 +307,13 @@ void nfs_proxy_t::run(json11::Json cfg)
    {
        kvfs->upgrade_db([this](int res) { finished = true; });
    }
+    if (bg)
+    {
+        daemonize_reopen_stdio();
+        int ok = 0;
+        (void)write(notifyfd[1], &ok, sizeof(ok));
+        close(notifyfd[1]);
+    }
    while (!finished)
    {
        ringloop->loop();
@@ -409,10 +429,6 @@ void nfs_proxy_t::run_server(json11::Json cfg)
    {
        mount_fs();
    }
-    if (cfg["foreground"].is_null())
-    {
-        daemonize();
-    }
    if (pidfile != "")
    {
        write_pid();
@@ -699,7 +715,8 @@ void nfs_client_t::handle_read(int result)
        return;
    if (result <= 0 && result != -EAGAIN && result != -EINTR && result != -ECANCELED)
    {
-        printf("Failed read from client %d: %d (%s)\n", nfs_fd, result, strerror(-result));
+        if (result != 0)
+            printf("Failed read from client %d: %d (%s)\n", nfs_fd, result, strerror(-result));
        stop();
        return;
    }
@@ -1224,26 +1241,31 @@ void nfs_client_t::free_or_rdma(rpc_op_t *rop, void *buf)
 #endif
 }

-void nfs_proxy_t::daemonize()
+void nfs_proxy_t::daemonize_fork(int *notifyfd)
 {
-    // Stop all clients because client I/O sometimes breaks during daemonize
-    // I.e. the new process stops receiving events on the old FD
-    // It doesn't happen if we call sleep(1) here, but we don't want to call sleep(1)...
-    for (auto & cli: rpc_clients)
-        cli->stop();
    if (fork())
-        exit(0);
+    {
+        // Parent - check status
+        close(notifyfd[1]);
+        int child_errno = 1;
+        (void)read(notifyfd[0], &child_errno, sizeof(child_errno));
+        exit(child_errno);
+    }
    setsid();
    if (fork())
        exit(0);
-    if (chdir("/") != 0)
-        fprintf(stderr, "Warning: Failed to chdir into /\n");
+}
+
+void nfs_proxy_t::daemonize_reopen_stdio()
+{
    close(0);
    close(1);
    close(2);
    open("/dev/null", O_RDONLY);
    open(logfile.c_str(), O_WRONLY|O_APPEND|O_CREAT, 0666);
    open(logfile.c_str(), O_WRONLY|O_APPEND|O_CREAT, 0666);
+    if (chdir("/") != 0)
+        fprintf(stderr, "Warning: Failed to chdir into /\n");
 }

 void nfs_proxy_t::write_pid()
@@ -83,7 +83,8 @@ public:
    void check_default_pool();
    nfs_client_t* create_client();
    void do_accept(int listen_fd);
-    void daemonize();
+    void daemonize_fork(int *notifyfd);
+    void daemonize_reopen_stdio();
    void write_pid();
    void mount_fs();
    void check_already_mounted();
@@ -9,7 +9,6 @@
 #include <arpa/inet.h>

 #include "addr_util.h"
-#include "blockstore_impl.h"
 #include "osd_primary.h"
 #include "osd.h"
 #include "http_client.h"
@@ -35,7 +34,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 = new blockstore_t(bs_cfg, ringloop, tfd);
+        this->bs = blockstore_i::create(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())
@@ -650,11 +649,11 @@ void osd_t::print_slow()
                    op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
                {
                    cur_slow_op_secondary++;
-                    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)
+                    if (op->bs_op)
                    {
-                        bufprintf(" wait=%d (detail=%ju)", wait_for, PRIV(op->bs_op)->wait_detail);
+                        auto diag = bs->get_op_diag(op->bs_op);
+                        if (diag != "")
+                            bufprintf(" %s", diag.c_str());
                    }
                }
                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_t *bs = NULL;
+    blockstore_i *bs = NULL;
    void *zero_buffer = NULL;
    uint64_t zero_buffer_size = 0;
    uint32_t bs_block_size, bs_bitmap_granularity, clean_entry_bitmap_size;
@@ -410,7 +410,11 @@ void osd_t::report_statistics()

 void osd_t::on_change_osd_state_hook(osd_num_t peer_osd)
 {
-    if (msgr.wanted_peers.find(peer_osd) != msgr.wanted_peers.end())
+    if (st_cli.peer_states[peer_osd].is_null())
+    {
+        repeer_pgs(peer_osd);
+    }
+    else if (msgr.wanted_peers.find(peer_osd) != msgr.wanted_peers.end())
    {
        msgr.connect_peer(peer_osd, st_cli.peer_states[peer_osd]);
    }
@@ -5,18 +5,6 @@

 #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;

@@ -4,6 +4,7 @@
 #include <map>
 #include <vector>
 #include <algorithm>
+#include <stdint.h>

 #include "cpp-btree/btree_map.h"

@@ -165,7 +165,6 @@ 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,11 +8,6 @@
 #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,6 +16,7 @@

 #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_t *bs = new blockstore_t(config, ringloop, epmgr->tfd);
+    blockstore_i *bs = blockstore_i::create(config, ringloop, epmgr->tfd);

    blockstore_op_t op;
    int main_state = 0;
@@ -6,6 +6,11 @@
 #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;
 }

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

 ./test_move_reappear.sh

+./test_degraded.sh
+
 ./test_rm.sh

 ./test_rm_degraded.sh
@@ -46,6 +48,9 @@ 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
@@ -0,0 +1,28 @@
+#!/bin/bash -ex
+
+OSD_COUNT=0
+PG_COUNT=32
+PG_SIZE=3
+PG_MINSIZE=2
+POOLCFG='"failure_domain":"host",'
+. `dirname $0`/run_3osds.sh
+
+$ETCDCTL put /vitastor/config/node_placement '{"1":{"parent":"host1"},"2":{"parent":"host2"},"3":{"parent":"host3"},"4":{"parent":"host1"},"5":{"parent":"host2"},"6":{"parent":"host3"},"host1":{"level":"host"},"host2":{"level":"host"}}'
+
+OSD_COUNT=6
+for i in $(seq 1 $OSD_COUNT); do
+    start_osd $i
+done
+wait_up 120
+
+LD_PRELOAD="build/src/client/libfio_vitastor.so" \
+fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so -bs=4k -direct=1 -iodepth=4 -rw=randwrite -loops=1000 \
+    -etcd=$ETCD_URL -pool=1 -inode=2 -size=256M -cluster_log_level=10 -runtime=5 &
+FIO_PID=$!
+
+sleep 15
+kill -9 $OSD1_PID $OSD4_PID
+
+wait $FIO_PID
+
+format_green OK
@@ -0,0 +1,8 @@
+#!/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,5 +1,7 @@
 #!/bin/bash -ex

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

@@ -177,7 +179,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
-rm ./testdata/nfs/settings.jsonLGNmGn
+sudo 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

@@ -187,7 +189,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
-rm ./testdata/nfs/settings.jsonLGNmGn
+sudo 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
Author	SHA1	Message	Date
Vitaliy Filippov	4c11e3ad3d	Release 2.4.4 - Implement `vitastor-kv rescue` command which dumps all readable blocks from the database regardless of the tree structure and allows to recover a corrupted database. - Simplify primary OSD selection - use just PG number instead of the hash. This allows to distribute primary OSDs more evenly and reach better read speeds. - Fix vitastor-kv-stress hang on parallel list & reopen. - Fix CAS writes without immediate_commit. Fix is sufficient for vitastor-kv to operate correctly in non-immediate_commit pools. - Fix NFS ACCESS calls for broken inodes (allow root access to them). - Fix broken vitastor-kv-stress --runtime parameter. - Fix bad vitastor-kv dump output buffering leading to incorrect JSON in dumps. - Fix writing garbage from uninitialized memory to the empty space in VitastorKV DB. - Refactor some code in preparation to the release of the new log-structured metadata store. - Replace assert(done != expected) error message, probably caused by kernel issues with zero-copy TCP send, with a more concise error description.	2025-12-01 02:22:27 +03:00
Vitaliy Filippov	d88b49872b	Fix vitastor-kv-stress hang on parallel list & reopen	2025-11-30 16:54:23 +03:00
Vitaliy Filippov	ca27b91919	Handle CAS writes with built-in SYNC and without retries in the client	2025-11-30 14:50:50 +03:00
Vitaliy Filippov	94f31b96b8	Cancel block changes on write failures/CAS retries (should fix #115 )	2025-11-30 01:49:49 +03:00
Vitaliy Filippov	76c7c26d32	Fix broken inode removal again (ACCESS will now allow root access to such inodes)	2025-11-29 19:30:20 +03:00
Vitaliy Filippov	8aa2c49202	Use immediate_commit in test_nfs	2025-11-29 19:22:01 +03:00
Vitaliy Filippov	0f330b10f1	Add kv-stress test to CI	2025-11-29 16:48:38 +03:00
Vitaliy Filippov	477b54a0d8	Fix broken vitastor-kv-stress --runtime parameter	2025-11-29 16:47:13 +03:00
Vitaliy Filippov	5823a7de66	Disable K/V DB operation on non-immediate_commit pools (unsafe because of broken CAS)	2025-11-29 16:43:46 +03:00
Vitaliy Filippov	eb0deaa3f5	Implement vitastor-kv rescue command	2025-11-29 16:34:34 +03:00
Vitaliy Filippov	59e6527303	Fix dump buffering, do not write uninitialized memory to the end K/V blocks on serialize	2025-11-29 16:34:34 +03:00
Vitaliy Filippov	67ba9f9b7c	Simplify primary OSD selection - use just PG number instead of the hash	2025-11-28 16:38:55 +03:00
Vitaliy Filippov	3ad83e8d13	Use only space_left, not sqes_left	2025-11-23 15:17:57 +03:00
Vitaliy Filippov	5d3f3f47a7	Move all #defines to internal.h	2025-11-23 14:53:46 +03:00
Vitaliy Filippov	8ee7058ec8	Use an interface instead of explicit blockstore_t wrapper	2025-11-23 01:48:44 +03:00
Vitaliy Filippov	1badc6ad13	Remove include of blockstore_impl.h from disk_tool and osd	2025-11-23 01:48:44 +03:00
Vitaliy Filippov	be1858848e	Remove block_order and replace << >> by / * data_block_size	2025-11-22 18:10:32 +03:00
Vitaliy Filippov	d75b1cb2d2	Refactor some defines	2025-11-22 18:07:44 +03:00
Vitaliy Filippov	a1c17d90a3	Add done != expected send error message	2025-11-18 01:23:52 +03:00
Vitaliy Filippov	f0112050ce	Release 2.4.3 - Daemonize before forking in NFS proxy to fix OSD RDMA support (#107) - Fix possible PG_INCOMPLETE on node outage when allow_net_split is false - Fix build under Ubuntu 25.10 - Fix journal_no_same_sector_overwrites mode after optimizing WA (#109)	2025-11-09 02:18:35 +03:00
Vitaliy Filippov	d6b8d921d6	Daemonize before forking in NFS proxy to fix OSD RDMA support (fix #107 )	2025-11-09 01:07:11 +03:00
Vitaliy Filippov	65872f5d0e	Remove RDMA context FD handlers during msgr destroy	2025-11-08 14:11:03 +03:00
Vitaliy Filippov	15eef27d44	Fix PG_INCOMPLETE on node outage when allow_net_split is false	2025-11-07 21:56:56 +03:00
Vitaliy Filippov	aa1e51de5f	Add include stdint (fix ubuntu 25.10 build)	2025-11-07 21:56:56 +03:00
Flynn049andGitHub	c164adb43c	Fix: journal sector written flag not set when submitted (#109 )	2025-11-07 00:59:04 +03:00