WIP refactor dequeue_write/continue_write into a single "coroutine"

CMakeLists.txt

@@ -2,7 +2,7 @@ cmake_minimum_required(VERSION 2.8.12)
 
 project(vitastor)
 
-set(VITASTOR_VERSION "3.0.3")
+set(VITASTOR_VERSION "3.0.2")
 
 include(CTest)
 

csi/Makefile

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v3.0.3
+VITASTOR_VERSION ?= v3.0.2
 
 all: build push
 

csi/deploy/004-csi-nodeplugin.yaml

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

csi/deploy/007-csi-provisioner.yaml

@@ -121,7 +121,7 @@ spec:
             privileged: true
             capabilities:
               add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v3.0.3
+          image: vitalif/vitastor-csi:v3.0.2
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/src/config.go

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

debian/changelog

@@ -1,4 +1,4 @@
-vitastor (3.0.3-1) unstable; urgency=medium
+vitastor (3.0.2-1) unstable; urgency=medium
 
   * Bugfixes
 

docker/Makefile

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v3.0.3
+VITASTOR_VERSION ?= v3.0.2
 
 all: build push
 

docker/etc/vitastor/docker.conf

@@ -4,7 +4,7 @@
 #
 
 # Desired Vitastor version
-VITASTOR_VERSION=v3.0.3
+VITASTOR_VERSION=v3.0.2
 
 # Additional arguments for all containers
 # For example, you may want to specify a custom logging driver here

docs/installation/docker.en.md

@@ -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:v3.0.3`
+   `docker pull vitalif/vitastor:v3.0.2`
 2. Install scripts to the host system: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.0.3 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.0.2 install.sh`
 3. Reload udev rules: \
    `udevadm control --reload-rules`
 

docs/installation/docker.ru.md

@@ -25,9 +25,9 @@ Vitastor можно установить в Docker/Podman. При этом etcd,
 Инструкция по установке максимально простая.
 
 1. Скачайте Docker-образ желаемой версии: \
-   `docker pull vitalif/vitastor:v3.0.3`
+   `docker pull vitalif/vitastor:v3.0.2`
 2. Установите скрипты в хост-систему командой: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.0.3 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.0.2 install.sh`
 3. Перезагрузите правила udev: \
    `udevadm control --reload-rules`
 

mon/lp_optimizer/lp_optimizer.js

@@ -10,19 +10,16 @@ const NO_OSD = 'Z';
 async function lp_solve(text)
 {
     const cp = child_process.spawn('lp_solve');
-    let stdout = '', stderr = '', finish_cb, finished = 0;
+    let stdout = '', stderr = '', finish_cb;
     cp.stdout.on('data', buf => stdout += buf.toString());
     cp.stderr.on('data', buf => stderr += buf.toString());
-    cp.stdout.on('end', () => finish_cb());
+    cp.on('exit', () => finish_cb && finish_cb());
-    cp.stderr.on('end', () => finish_cb());
     cp.stdin.write(text);
     cp.stdin.end();
-    await new Promise(ok => (finish_cb = () =>
+    if (cp.exitCode == null)
     {
-        finished++;
+        await new Promise(ok => finish_cb = ok);
-        if (finished == 2)
+    }
-            ok();
-    }));
     if (!stdout.trim())
     {
         return null;

mon/package.json

@@ -1,6 +1,6 @@
 {
   "name": "vitastor-mon",
-  "version": "3.0.3",
+  "version": "3.0.2",
   "description": "Vitastor SDS monitor service",
   "main": "mon-main.js",
   "scripts": {

mon/pg_gen.js

@@ -195,7 +195,7 @@ async function generate_pool_pgs(state, global_config, pool_id, osd_tree, levels
     const folded_tree = make_hier_tree(global_config, folded.nodes.reduce((a, c) => { a[c.id] = c; return a; }, {}));
     const old_pg_count = prev_pgs.length;
     const optimize_cfg = {
-        osd_weights: folded.nodes.reduce((a, c) => { if (/^\d+$/.exec(c.id) && c.size != null) { a[c.id] = c.size||0; } return a; }, {}),
+        osd_weights: folded.nodes.reduce((a, c) => { if (Number(c.id)) { a[c.id] = c.size; } return a; }, {}),
         combinator: use_rules
             // new algorithm:
             ? new RuleCombinator(folded_tree, rules, pool_cfg.max_osd_combinations)

node-binding/package.json

@@ -1,6 +1,6 @@
 {
   "name": "vitastor",
-  "version": "3.0.3",
+  "version": "3.0.2",
   "description": "Low-level native bindings to Vitastor client library",
   "main": "index.js",
   "keywords": [

patches/cinder-vitastor.py

@@ -50,7 +50,7 @@ from cinder.volume import configuration
 from cinder.volume import driver
 from cinder.volume import volume_utils
 
-VITASTOR_VERSION = '3.0.3'
+VITASTOR_VERSION = '3.0.2'
 
 LOG = logging.getLogger(__name__)
 

rpm/vitastor-el7.spec

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

rpm/vitastor-el8.spec

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

rpm/vitastor-el9.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.0.3
+Version:        3.0.2
 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-3.0.3.el9.tar.gz
+Source0:        vitastor-3.0.2.el9.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-c++

src/CMakeLists.txt

@@ -21,7 +21,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 endif()
 set(ENABLE_COVERAGE false CACHE BOOL "Enable code coverage")
 
-add_definitions(-DVITASTOR_VERSION="3.0.3")
+add_definitions(-DVITASTOR_VERSION="3.0.2")
 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})

src/blockstore/blockstore.h

@@ -187,6 +187,7 @@ public:
     // MUST be called only when nobody makes any modifications to the DB for this pool
     virtual void* reshard_start(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size, uint64_t chunk_limit) = 0;
     virtual bool reshard_continue(void *reshard_state, uint64_t chunk_limit) = 0;
+    virtual void reshard_abort(void *reshard_state) = 0;
 
     // Event loop
     virtual void loop() = 0;

src/blockstore/blockstore_heap.cpp

@@ -458,22 +458,19 @@ bool blockstore_heap_t::validate_object(heap_entry_t *obj)
         next_wr = wr;
         if (wr->type() == BS_HEAP_ROLLBACK)
         {
+            if (commit_wr && wr->version > commit_wr->version)
+            {
+                // rollback may not come before commit with a smaller version
+                fprintf(stderr, "Error: rollback entry %jx:%jx v%ju l%ju comes before a commit entry v%ju l%ju\n",
+                    wr->inode, wr->stripe, wr->version, wr->lsn, commit_wr->version, commit_wr->lsn);
+                return false;
+            }
             rollback_wr = wr;
             continue;
         }
         if (wr->type() == BS_HEAP_COMMIT)
         {
-            if (commit_wr && wr->version > commit_wr->version)
+            commit_wr = wr;
-            {
-                // commit may not come before commit with a smaller version
-                fprintf(stderr, "Error: commit entry %jx:%jx v%ju l%ju comes before a commit entry v%ju l%ju\n",
-                    wr->inode, wr->stripe, wr->version, wr->lsn, commit_wr->version, commit_wr->lsn);
-                return false;
-            }
-            if (!commit_wr)
-            {
-                commit_wr = wr;
-            }
             continue;
         }
         if (wr->entry_type & BS_HEAP_STABLE)
@@ -840,6 +837,7 @@ bool blockstore_heap_t::calc_checksums(heap_entry_t *wr, uint8_t *data, bool set
         {
             return true;
         }
+        uint32_t len = 0;
         if (wr->type() == BS_HEAP_SMALL_WRITE || wr->type() == BS_HEAP_INTENT_WRITE)
             len = wr->small().len;
         else if (wr->type() == BS_HEAP_BIG_INTENT)
@@ -856,14 +854,13 @@ bool blockstore_heap_t::calc_checksums(heap_entry_t *wr, uint8_t *data, bool set
     }
     if (wr->type() == BS_HEAP_BIG_WRITE)
     {
-        assert(offset != UINT32_MAX && len != UINT32_MAX);
         return calc_block_checksums((uint32_t*)(wr->get_checksums(this) + offset/dsk->csum_block_size * (dsk->data_csum_type & 0xFF)),
             data, wr->get_int_bitmap(this), offset, offset+len, set, NULL);
     }
     if (wr->type() == BS_HEAP_BIG_INTENT)
     {
         auto & bi = wr->big_intent();
-        return calc_block_checksums((uint32_t*)(wr->get_checksums(this) + bi.offset/dsk->csum_block_size * (dsk->data_csum_type & 0xFF)),
+        return calc_block_checksums((uint32_t*)(wr->get_checksums(this) + offset/dsk->csum_block_size * (dsk->data_csum_type & 0xFF)),
             data, wr->get_int_bitmap(this), bi.offset, bi.offset+bi.len, set, NULL);
     }
     assert(wr->type() == BS_HEAP_SMALL_WRITE || wr->type() == BS_HEAP_INTENT_WRITE);
@@ -1080,6 +1077,16 @@ bool blockstore_heap_t::reshard_check(pool_id_t pool, uint32_t pg_count, uint32_
         set_it->second.pg_stripe_size == pg_stripe_size);
 }
 
+void blockstore_heap_t::reshard_abort(void* reshard_state)
+{
+    heap_reshard_state_t *st = (heap_reshard_state_t*)reshard_state;
+    for (auto sh_it = st->old_shards.begin(); sh_it != st->old_shards.end(); sh_it++)
+    {
+        block_index[sh_it->first] = std::move(sh_it->second);
+    }
+    delete st;
+}
+
 heap_entry_t *blockstore_heap_t::lock_and_read_entry(object_id oid)
 {
     auto obj = read_entry(oid);
@@ -1092,6 +1099,27 @@ heap_entry_t *blockstore_heap_t::lock_and_read_entry(object_id oid)
     return obj;
 }
 
+heap_entry_t *blockstore_heap_t::read_locked_entry(object_id oid, uint64_t lsn)
+{
+    auto obj = read_entry(oid);
+    assert(obj);
+    for (auto wr = obj; wr; wr = prev(wr))
+    {
+        if (wr->is_overwrite())
+        {
+            if (lsn == wr->lsn)
+            {
+                return obj;
+            }
+            else
+            {
+                obj = prev(wr);
+            }
+        }
+    }
+    return NULL;
+}
+
 bool blockstore_heap_t::unlock_entry(object_id oid)
 {
     auto mvcc_it = object_mvcc.find(oid);
@@ -1319,6 +1347,7 @@ int blockstore_heap_t::add_small_write(object_id oid, heap_entry_t **obj_ptr, ui
     // Small writes are written in parallel with buffered data so they require explicit_complete
     return add_entry(wr_size, modified_block, false, true, [&](heap_entry_t *wr)
     {
+        printf("add_small_write t%u %lx:%lx l%lu v%lu %u +%u loc:%lx\n", type, oid.inode, oid.stripe, wr->lsn, version, offset, len, location);
         wr->entry_type = type;
         wr->inode = oid.inode;
         wr->stripe = oid.stripe;
@@ -1348,6 +1377,7 @@ int blockstore_heap_t::add_big_write(object_id oid, heap_entry_t *old_head, bool
     // Big writes are written after writing data so they don't require explicit_complete
     return add_entry(wr_size, modified_block, false, false, [&](heap_entry_t *wr)
     {
+        printf("add_big_write %lx:%lx l%lu v%lu loc:%lx\n", oid.inode, oid.stripe, wr->lsn, version, location);
         wr->entry_type = BS_HEAP_BIG_WRITE | (stable ? BS_HEAP_STABLE : 0);
         wr->inode = oid.inode;
         wr->stripe = oid.stripe;
@@ -1374,6 +1404,7 @@ int blockstore_heap_t::add_redirect_intent(object_id oid, heap_entry_t **obj_ptr
     // Big-redirect intents, just like regular big writes, are written after writing data so they don't require explicit_complete
     return add_entry(wr_size, modified_block, false, false, [&](heap_entry_t *wr)
     {
+        printf("add_redir_intent %lx:%lx l%lu v%lu %u +%u loc:%lx\n", oid.inode, oid.stripe, wr->lsn, version, offset, len, location);
         wr->entry_type = BS_HEAP_BIG_INTENT|BS_HEAP_STABLE;
         wr->inode = oid.inode;
         wr->stripe = oid.stripe;
@@ -1390,7 +1421,7 @@ int blockstore_heap_t::add_redirect_intent(object_id oid, heap_entry_t **obj_ptr
         bitmap_set(wr->get_int_bitmap(this), offset, len, dsk->bitmap_granularity);
         if (dsk->data_csum_type)
             memset(wr->get_checksums(this), 0, get_csum_size(wr));
-        calc_checksums(wr, (uint8_t*)data, true);
+        calc_checksums(wr, (uint8_t*)data, true, offset, len);
         *obj_ptr = wr;
     });
 }
@@ -1410,6 +1441,7 @@ int blockstore_heap_t::add_big_intent(object_id oid, heap_entry_t **obj_ptr, uin
     // Big intents are written before writing data so they require explicit_complete
     return add_entry(wr_size, modified_block, false, true, [&](heap_entry_t *wr)
     {
+        printf("add_big_intent %lx:%lx l%lu v%lu %u +%u loc:%lx\n", oid.inode, oid.stripe, wr->lsn, version, offset, len, obj->big_location(this));
         wr->entry_type = BS_HEAP_BIG_INTENT | BS_HEAP_STABLE;
         wr->inode = oid.inode;
         wr->stripe = oid.stripe;
@@ -1429,11 +1461,11 @@ int blockstore_heap_t::add_big_intent(object_id oid, heap_entry_t **obj_ptr, uin
         if (dsk->data_csum_type)
         {
             if (checksums)
-                memcpy(wr->get_checksums(this), checksums, get_csum_size(wr));
+                memcpy(wr->get_checksums(this), checksums, dsk->clean_entry_bitmap_size);
             else
             {
-                memcpy(wr->get_checksums(this), obj->get_checksums(this), get_csum_size(wr));
+                memcpy(wr->get_checksums(this), obj->get_checksums(this), dsk->clean_entry_bitmap_size);
-                calc_checksums(wr, (uint8_t*)data, true);
+                calc_checksums(wr, (uint8_t*)data, true, offset, len);
             }
         }
         else
@@ -1465,6 +1497,7 @@ int blockstore_heap_t::add_compact(heap_entry_t *obj, uint64_t compact_version,
     // Compaction entry is added after copying data so it doesn't require explicit_complete
     return add_entry(wr_size, modified_block, true, false, [&](heap_entry_t *new_wr)
     {
+        printf("add_compact %lx:%lx l%lu v%lu loc:%lx\n", obj->inode, obj->stripe, compact_lsn, compact_version, compact_location);
         new_wr->entry_type = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE;
         new_wr->inode = obj->inode;
         new_wr->stripe = obj->stripe;
@@ -1554,7 +1587,7 @@ int blockstore_heap_t::add_commit(heap_entry_t *obj, uint64_t version, uint32_t
     }
     if (!uncommitted)
     {
-        return 0;
+        return EBUSY;
     }
     return add_simple(obj, version, modified_block, BS_HEAP_COMMIT);
 }
@@ -1564,32 +1597,23 @@ int blockstore_heap_t::add_rollback(heap_entry_t *obj, uint64_t version, uint32_
     heap_entry_t *wr = obj;
     bool found_uncommitted = false;
     uint64_t commit_version = 0;
-    uint64_t rollback_version = UINT64_MAX;
+    while (wr && !wr->is_overwrite())
-    while (wr)
     {
         if (wr->type() == BS_HEAP_ROLLBACK)
         {
-            if (wr->version <= version)
+            auto rollback_version = wr->version;
-            {
-                // All previous writes are already rolled back, stop
-                break;
-            }
-            rollback_version = wr->version;
             wr = prev(wr);
+            while (wr->version > rollback_version)
+            {
+                assert(!(wr->entry_type & BS_HEAP_STABLE));
+                wr = prev(wr);
+            }
             continue;
         }
         if (wr->type() == BS_HEAP_COMMIT)
         {
             if (commit_version < wr->version)
-            {
                 commit_version = wr->version;
-            }
-            wr = prev(wr);
-            continue;
-        }
-        if (wr->version > rollback_version)
-        {
-            // Already rolled back, skip
             wr = prev(wr);
             continue;
         }
@@ -1600,10 +1624,14 @@ int blockstore_heap_t::add_rollback(heap_entry_t *obj, uint64_t version, uint32_
             {
                 return EBUSY;
             }
-            else
+            else if (wr->version == version)
             {
                 break;
             }
+            else if (wr->version < version)
+            {
+                return ENOENT;
+            }
         }
         else if (wr->version > version)
         {
@@ -1784,9 +1812,9 @@ void blockstore_heap_t::iterate_with_stable(heap_entry_t *obj, uint64_t max_lsn,
         }
         else
         {
-            // 1) 1 2 3 ROLLBACK(2) COMMIT(3) -> 3 is unstable
+            // 1) 1 2 3 ROLLBACK(2) COMMIT(3) -> impossible
             // 2) 1 2 3 4 ROLLBACK(3) COMMIT(2) -> OK
-            // 3) 1 2 3 ROLLBACK(2) 3 COMMIT(3) -> first 3 is unstable
+            // 3) 1 2 3 ROLLBACK(2) 3 COMMIT(3) -> first 3 shouldn't be treated as stable
             // 4) 1 2 3 COMMIT(3) ROLLBACK(2) -> impossible
             //    I.e. a rollback always has version >= previous commit
             // 5) 1 2 3 4 5 ROLLBACK(4) 5 ROLLBACK(3)
@@ -2022,13 +2050,7 @@ void blockstore_heap_t::free_data(inode_t inode, uint64_t location)
         inode = (INODE_POOL(inode) << POOL_ID_BITS);
     assert(data_alloc->get(location / dsk->data_block_size));
     data_alloc->set(location / dsk->data_block_size, false);
-    auto sp_it = inode_space_stats.find(inode);
+    inode_space_stats[inode] -= dsk->data_block_size;
-    if (sp_it != inode_space_stats.end())
-    {
-        sp_it->second -= dsk->data_block_size;
-        if (sp_it->second == 0)
-            inode_space_stats.erase(sp_it);
-    }
     data_used_space -= dsk->data_block_size;
 }
 
@@ -2311,7 +2333,7 @@ void blockstore_heap_t::set_no_inode_stats(const std::vector<uint64_t> & pool_id
     {
         // Recalculate if changed
         if (ps.second.no_inode_stats == 2 || ps.second.no_inode_stats == 1)
-            recalc_inode_space_stats(ps.first, ps.second.no_inode_stats == 2);
+            recalc_inode_space_stats(ps.first, ps.second.no_inode_stats == 1);
         ps.second.no_inode_stats &= 1;
     }
 }
@@ -2322,8 +2344,8 @@ void blockstore_heap_t::recalc_inode_space_stats(uint64_t pool_id, bool per_inod
     auto sp_begin = inode_space_stats.lower_bound((pool_id << (64-POOL_ID_BITS)));
     auto sp_end = inode_space_stats.lower_bound(((pool_id+1) << (64-POOL_ID_BITS)));
     inode_space_stats.erase(sp_begin, sp_end);
-    uint32_t pg_count = ps.pg_count;
+    uint32_t pg_count = ps.pg_count ? ps.pg_count : 1;
-    for (uint32_t pg_num = pg_count ? 1 : 0; pg_num <= pg_count; pg_num++)
+    for (uint32_t pg_num = 1; pg_num <= pg_count; pg_num++)
     {
         auto & pg_idx = block_index[(pool_id << (64-POOL_ID_BITS)) | pg_num];
         for (auto & ip: pg_idx)

src/blockstore/blockstore_heap.h

@@ -248,17 +248,20 @@ public:
     void* reshard_start(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size, uint64_t chunk_limit);
     bool reshard_continue(void* reshard_state, uint64_t chunk_limit);
     bool reshard_check(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size);
+    void reshard_abort(void* reshard_state);
     void set_no_inode_stats(const std::vector<uint64_t> & pool_ids);
     void recalc_inode_space_stats(uint64_t pool_id, bool per_inode);
     // read an object entry and lock it against removal
     // in the future, may become asynchronous
     heap_entry_t *lock_and_read_entry(object_id oid);
+    // re-read a locked object entry with the given lsn (pointer may be invalidated)
+    heap_entry_t *read_locked_entry(object_id oid, uint64_t lsn);
     // read an object entry without locking it
     heap_entry_t *read_entry(object_id oid);
     // unlock an entry
     bool unlock_entry(object_id oid);
     // set or verify checksums in a write request
-    bool calc_checksums(heap_entry_t *wr, uint8_t *data, bool set, uint32_t offset = UINT32_MAX, uint32_t len = UINT32_MAX);
+    bool calc_checksums(heap_entry_t *wr, uint8_t *data, bool set, uint32_t offset = 0, uint32_t len = 0);
     // set or verify raw block checksums
     bool calc_block_checksums(uint32_t *block_csums, uint8_t *data, uint8_t *bitmap, uint32_t start, uint32_t end,
         bool set, std::function<void(uint32_t, uint32_t, uint32_t)> bad_block_cb);

src/blockstore/blockstore_impl.cpp

@@ -406,3 +406,8 @@ bool blockstore_impl_t::reshard_continue(void *reshard_state, uint64_t chunk_lim
 {
     return heap->reshard_continue(reshard_state, chunk_limit);
 }
+
+void blockstore_impl_t::reshard_abort(void *reshard_state)
+{
+    return heap->reshard_abort(reshard_state);
+}

src/blockstore/blockstore_impl.h

@@ -191,6 +191,7 @@ public:
 
     void* reshard_start(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size, uint64_t chunk_limit);
     bool reshard_continue(void *reshard_state, uint64_t chunk_limit);
+    void reshard_abort(void *reshard_state);
 
     // Event loop
     void loop();

src/blockstore/blockstore_stable.cpp

@@ -57,9 +57,9 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
             }
             assert(res == 0);
         }
-resume_1:
         if (priv->modified_block != UINT32_MAX && priv->modified_block2 != priv->modified_block)
         {
+resume_1:
             BS_SUBMIT_CHECK_SQES(1);
             prepare_meta_block_write(priv->modified_block);
 resume_2:

src/blockstore/blockstore_write.cpp

@@ -13,10 +13,13 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
     return true;
 }
 
-void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
+bool blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
 {
-    if (modified_blocks.find(modified_block) != modified_blocks.end())
+    auto mod_it = modified_blocks.find(modified_block);
-        return;
+    if (mod_it != modified_blocks.end())
+    {
+        return !mod_it->second.sent;
+    }
     io_uring_sqe *sqe = get_sqe();
     assert(sqe != NULL);
     ring_data_t *data = ((ring_data_t*)sqe->user_data);
@@ -41,6 +44,7 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
     unsynced_meta_write_count++;
     pending_modified_blocks.push_back(modified_block);
     modified_blocks[modified_block] = { .sent = false, .buf = buf };
+    return true;
 }
 
 bool blockstore_impl_t::meta_block_is_pending(uint32_t modified_block)
@@ -121,6 +125,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     heap_entry_t *obj = heap->read_entry(op->oid);
     if (op->opcode == BS_OP_DELETE)
     {
+        return continue_delete(op, 0);
         // Delete
         if (!obj || obj->type() == BS_HEAP_DELETE)
         {
@@ -133,17 +138,38 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         BS_SUBMIT_CHECK_SQES(1);
         int res = heap->add_delete(obj, &PRIV(op)->modified_block);
         if (res == ENOSPC)
+        {
             goto enospc;
+        }
         assert(res == 0);
-        prepare_meta_block_write(PRIV(op)->modified_block);
-        PRIV(op)->pending_ops++;
-        PRIV(op)->op_state = 5;
         write_iodepth++;
+resume_1:
+        while (!prepare_meta_block_write(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 1;
+            return 1;
+        }
+rseume_2:
+        while (meta_block_is_pending(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 2;
+            return 1;
+        }
+resume_3:
+resume_4:
+        if (!throttle_write(op, 3))
+        {
+            return 1;
+        }
+        write_iodepth--;
+        ack_write(op);
+        return 2;
     }
     // FIXME: Allow to do initial writes as buffered, not redirected
     // FIXME: Allow to do direct writes over holes
     else if (!obj || obj->type() == BS_HEAP_DELETE || op->offset == 0 && op->len == dsk.data_block_size)
     {
+        return continue_big_write(op, 10);
         // Big (redirect) write
         PRIV(op)->write_type = dsk.disable_data_fsync || op->opcode != BS_OP_WRITE_STABLE ? BS_HEAP_BIG_WRITE : _REDIRECT_INTENT;
         BS_SUBMIT_CHECK_SQES(1);
@@ -163,6 +189,7 @@ enospc:
             flusher->request_trim();
             return 0;
         }
+        write_iodepth++;
         uint64_t loc = PRIV(op)->location;
 #ifdef BLOCKSTORE_DEBUG
         printf(
@@ -176,18 +203,72 @@ enospc:
         data->iov = (struct iovec){ op->buf, op->len };
         data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
         io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + loc + op->offset);
+        if (PRIV(op)->write_type == BS_HEAP_BIG_WRITE)
+            inflight_big++;
         PRIV(op)->pending_ops++;
-        write_iodepth++;
+resume_10:
+        if (PRIV(op)->pending_ops > 0)
+        {
+            PRIV(op)->op_state = 10;
+            return 1;
+        }
         if (PRIV(op)->write_type == BS_HEAP_BIG_WRITE)
         {
-            PRIV(op)->op_state = 1;
+            inflight_big--;
-            inflight_big++;
+resume_11:
+resume_12:
+resume_13:
+            if (!fsync_big_write(op, 11))
+                return 1;
+        }
+        heap_entry_t *obj = heap->read_entry(op->oid);
+        int res = 0;
+        if (PRIV(op)->write_type == _REDIRECT_INTENT)
+        {
+            res = heap->add_redirect_intent(op->oid, &obj, op->version, op->offset, op->len,
+                PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
         }
         else
-            PRIV(op)->op_state = 3;
+        {
+            res = heap->add_big_write(op->oid, obj, op->opcode == BS_OP_WRITE_STABLE,
+                op->version, op->offset, op->len, PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
+        }
+        if (res == ENOSPC)
+        {
+            if (!heap->get_to_compact_count())
+            {
+                // no space
+                heap->free_data(op->oid.inode, PRIV(op)->location);
+                write_iodepth--;
+                op->retval = -ENOSPC;
+                FINISH_OP(op);
+                return 2;
+            }
+            PRIV(op)->wait_for = WAIT_COMPACTION;
+            PRIV(op)->wait_detail = heap->get_compacted_count();
+            flusher->request_trim();
+            return 0;
+        }
+        assert(res == 0);
+resume_14:
+        while (!prepare_meta_block_write(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 14;
+            return 1;
+        }
+resume_15:
+        while (meta_block_is_pending(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 15;
+            return 1;
+        }
+        write_iodepth--;
+        ack_write(op);
+        return 2;
     }
     else if (intent_write_allowed(op, obj))
     {
+        return continue_intent_write(op, 20);
         // Direct intent-write
         BS_SUBMIT_CHECK_SQES(1);
         int res = 0;
@@ -223,13 +304,41 @@ enospc:
             assert(res == 0);
             PRIV(op)->lsn = obj->lsn;
         }
-        prepare_meta_block_write(PRIV(op)->modified_block);
-        PRIV(op)->pending_ops++;
-        PRIV(op)->op_state = 9;
         write_iodepth++;
+resume_20:
+        while (!prepare_meta_block_write(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 20;
+            return 1;
+        }
+resume_21:
+        while (meta_block_is_pending(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 21;
+            return 1;
+        }
+        // Direct intent-write
+        // LSN is not marked as completed so big_write won't be freed
+        BS_SUBMIT_GET_SQE(sqe, data);
+        data->iov = (struct iovec){ op->buf, op->len };
+        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
+        io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + PRIV(op)->location + op->offset);
+        if (dsk.use_atomic_flag)
+            sqe->rw_flags = RWF_ATOMIC;
+        PRIV(op)->pending_ops++;
+resume_22:
+        if (PRIV(op)->pending_ops > 0)
+        {
+            PRIV(op)->op_state = 22;
+            return 1;
+        }
+        write_iodepth--;
+        ack_write(op);
+        return 2;
     }
     else
     {
+        return continue_small_write(op, 30);
         // Small (buffered) overwrite
         // First check if there is free buffer space
         PRIV(op)->write_type = BS_HEAP_SMALL_WRITE;
@@ -242,7 +351,7 @@ enospc:
             return 0;
         }
         // There is sufficient space. Check SQE(s)
-        BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0));
+        BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0)); ---> refactor too
         int res = heap->add_small_write(op->oid, &obj, (BS_HEAP_SMALL_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)),
             op->version, op->offset, op->len, loc, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
         if (res == ENOSPC)
@@ -311,27 +420,46 @@ again:
         goto again;
     }
 resume_2:
+
+
+resume_4:
+resume_6:
+
+resume_8:
+    ack
+    return 2;
+resume_10:
+    return 1;
+}
+
+bool blockstore_impl_t::fsync_big_write(blockstore_op_t *op, int base_state)
+{
+    if (PRIV(op)->state == base_state)
+        goto resume_0;
+    else if (PRIV(op)->state == base_state+1)
+        goto resume_1;
+    else if (PRIV(op)->state == base_state+2)
+        goto resume_2;
     // We must fsync all big writes to avoid complex write workflows
     // It's OK for all HDDs and for server SSDs, but slightly worse for desktop SSDs
-    inflight_big--;
     if (!dsk.disable_data_fsync)
     {
         // fsync data in a batch
-resume_11:
+resume_0:
         if (inflight_big > 0)
         {
-            PRIV(op)->op_state = 11;
+            PRIV(op)->op_state = base_state;
-            return 1;
+            return false;
         }
         if (fsyncing_data)
         {
-resume_12:
+resume_1:
             if (fsyncing_data)
             {
-                PRIV(op)->op_state = 12;
+                PRIV(op)->op_state = base_state+1;
-                return 1;
+                return false;
             }
-            goto resume_4;
+            return true;
         }
         fsyncing_data = true;
         BS_SUBMIT_GET_SQE(sqe, data);
@@ -343,47 +471,23 @@ resume_12:
             handle_write_event(data, op);
         };
         PRIV(op)->pending_ops++;
-        PRIV(op)->op_state = 3;
+resume_2:
-        return 1;
+        if (PRIV(op)->pending_ops > 0)
+        {
+            PRIV(op)->op_state = base_state+2;
+            return false;
+        }
     }
-resume_4:
+    return true;
-    {
+}
-        auto obj = heap->read_entry(op->oid);
+
-        int res = 0;
+bool blockstore_impl_t::throttle_write(blockstore_op_t *op, int base_state)
-        if (PRIV(op)->write_type == _REDIRECT_INTENT)
+{
-        {
-            res = heap->add_redirect_intent(op->oid, &obj, op->version, op->offset, op->len,
-                PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
-        }
-        else
-        {
-            res = heap->add_big_write(op->oid, obj, op->opcode == BS_OP_WRITE_STABLE,
-                op->version, op->offset, op->len, PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
-        }
-        if (res == ENOSPC)
-        {
-            if (!heap->get_to_compact_count())
-            {
-                // no space
-                heap->free_data(op->oid.inode, PRIV(op)->location);
-                write_iodepth--;
-                op->retval = -ENOSPC;
-                FINISH_OP(op);
-                return 2;
-            }
-            PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = heap->get_compacted_count();
-            flusher->request_trim();
-            return 0;
-        }
-        assert(res == 0);
-        prepare_meta_block_write(PRIV(op)->modified_block);
-        PRIV(op)->pending_ops++;
-        PRIV(op)->op_state = 5;
-        return 1;
-    }
-resume_6:
     // Apply throttling to not fill the journal too quickly for the SSD+HDD case
+    if (PRIV(op)->op_state >= base_state+1)
+    {
+        return true;
+    }
     if (PRIV(op)->write_type == BS_HEAP_SMALL_WRITE && throttle_small_writes)
     {
         // Apply throttling
@@ -404,17 +508,21 @@ resume_6:
         if (ref_us > exec_us + throttle_threshold_us)
         {
             // Pause reply
-            PRIV(op)->op_state = 7;
+            PRIV(op)->op_state = base_state;
             // Remember that the timer can in theory be called right here
             tfd->set_timer_us(ref_us-exec_us, false, [this, op](int timer_id)
             {
-                PRIV(op)->op_state = 8;
+                PRIV(op)->op_state++;
                 ringloop->wakeup();
             });
-            return 1;
+            return false;
         }
     }
-resume_8:
+    return true;
+}
+
+void blockstore_impl_t::ack_write(blockstore_op_t *op)
+{
     // Acknowledge write
 #ifdef BLOCKSTORE_DEBUG
     printf("Ack write %jx:%jx v%ju\n", op->oid.inode, op->oid.stripe, op->version);
@@ -441,21 +549,7 @@ resume_8:
         unsynced_data_write_count++;
         intent_write_counter++;
     }
-    write_iodepth--;
     FINISH_OP(op);
-    return 2;
-resume_10:
-    // Direct intent-write
-    // LSN is not marked as completed so big_write won't be freed
-    BS_SUBMIT_GET_SQE(sqe, data);
-    data->iov = (struct iovec){ op->buf, op->len };
-    data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
-    io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + PRIV(op)->location + op->offset);
-    if (dsk.use_atomic_flag)
-        sqe->rw_flags = RWF_ATOMIC;
-    PRIV(op)->pending_ops++;
-    PRIV(op)->op_state = 7;
-    return 1;
 }
 
 void blockstore_impl_t::handle_write_event(ring_data_t *data, blockstore_op_t *op)

src/blockstore/v1/impl.cpp

@@ -480,6 +480,17 @@ resume_1:
     return true;
 }
 
+void blockstore_impl_t::reshard_abort(void *reshard_state)
+{
+    bs_reshard_state_t *st = (bs_reshard_state_t*)reshard_state;
+    for (auto sh_it = st->old_shards.begin(); sh_it != st->old_shards.end(); sh_it++)
+    {
+        auto & to = clean_db_shards[sh_it->first];
+        to.swap(sh_it->second);
+    }
+    delete st;
+}
+
 void blockstore_impl_t::process_list(blockstore_op_t *op)
 {
     uint32_t list_pg = op->pg_number+1;

src/blockstore/v1/impl.h

@@ -290,6 +290,7 @@ public:
     // Reshard database for a pool
     void* reshard_start(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size, uint64_t chunk_limit);
     bool reshard_continue(void *reshard_state, uint64_t chunk_limit);
+    void reshard_abort(void *reshard_state);
 
     // Event loop
     void loop();

src/client/vitastor.pc.in

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

src/osd/osd_cluster.cpp

@@ -809,17 +809,21 @@ again:
         goto again;
     }
     auto & pool_cfg = pool_it->second;
-    bool done = bs->reshard_continue(pool_cfg.reshard_state, pg_reshard_chunk_size);
+    bool done = false;
-    if (done &&
+    if (pool_cfg.real_pg_count != pool_cfg.applied_pg_count ||
-        (pool_cfg.real_pg_count != pool_cfg.applied_pg_count ||
+        pool_cfg.pg_stripe_size != pool_cfg.applied_pg_stripe_size)
-        pool_cfg.pg_stripe_size != pool_cfg.applied_pg_stripe_size))
     {
         // PG count changed again, reshard again
+        bs->reshard_abort(pool_cfg.reshard_state);
         pool_cfg.applied_pg_count = pool_cfg.real_pg_count;
         pool_cfg.applied_pg_stripe_size = pool_cfg.pg_stripe_size;
         pool_cfg.reshard_state = bs->reshard_start(pool_id, pool_cfg.real_pg_count, pool_cfg.pg_stripe_size, pg_reshard_chunk_size);
         done = !pool_cfg.reshard_state;
     }
+    else
+    {
+        done = bs->reshard_continue(pool_cfg.reshard_state, pg_reshard_chunk_size);
+    }
     if (done)
     {
         // Pool is resharded