First just recheck version without actually re-reading block in vitastor-kv

src/cluster_client.cpp

@@ -673,7 +673,7 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
         return false;
     }
     // Check alignment
-    if (!op->len && (op->opcode == OSD_OP_READ || op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP || op->opcode == OSD_OP_WRITE) ||
+    if (!op->len && (op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP || op->opcode == OSD_OP_WRITE) ||
         op->offset % pool_it->second.bitmap_granularity || op->len % pool_it->second.bitmap_granularity)
     {
         op->retval = -EINVAL;

src/kv_db.cpp

@@ -197,6 +197,7 @@ struct kv_op_t
 
     void exec();
     void next(); // for list
+    ~kv_op_t();
 protected:
     int recheck_policy = KV_RECHECK_LEAF;
     bool started = false;
@@ -887,13 +888,22 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
     op->opcode = OSD_OP_READ;
     op->inode = db->inode_id;
     op->offset = offset;
+    if (b_it != db->block_cache.end() && !b_it->second.invalidated && !b_it->second.updating)
+    {
+        // just recheck version - it's cheaper than re-reading the block
+        op->len = 0;
+    }
+    else
+    {
         op->len = db->kv_block_size;
         op->iov.push_back(malloc_or_die(op->len), op->len);
+    }
     op->callback = [=](cluster_op_t *op)
     {
         if (op->retval != op->len)
         {
             // error
+            if (op->len)
                 free(op->iov.buf[0].iov_base);
             cb(op->retval >= 0 ? -EIO : op->retval, BLK_NOCHANGE);
             delete op;
@@ -909,6 +919,7 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
             if (blk->updating > 0 && recheck_policy == KV_RECHECK_WAIT)
             {
                 // Wait until block update stops
+                if (op->len)
                     free(op->iov.buf[0].iov_base);
                 delete op;
                 db->continue_update.emplace(blk->offset, [=, blk_offset = blk->offset]()
@@ -923,6 +934,13 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
         }
         else
         {
+            if (!op->len)
+            {
+                // Version check failed, re-read block
+                delete op;
+                get_block(db, offset, cur_level, recheck_policy, cb);
+                return;
+            }
             auto blk = &db->block_cache[op->offset];
             if (blk_it != db->block_cache.end())
             {
@@ -944,6 +962,7 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
             }
             try_evict(db);
         }
+        if (op->len)
             free(op->iov.buf[0].iov_base);
         delete op;
     };
@@ -985,14 +1004,24 @@ void kv_op_t::exec()
         finish(-ENOSYS);
 }
 
+kv_op_t::~kv_op_t()
+{
+    if (started && !done)
+    {
+        done = true;
+        db->active_ops--;
+    }
+}
+
 void kv_op_t::finish(int res)
 {
+    auto db = this->db;
     this->res = res;
     this->done = true;
     db->active_ops--;
+    (std::function<void(kv_op_t *)>(callback))(this);
     if (!db->active_ops && db->closing)
         db->close(db->on_close);
-    (std::function<void(kv_op_t *)>(callback))(this);
 }
 
 void kv_op_t::get()

src/kv_stress.cpp

@@ -147,6 +147,8 @@ json11::Json::object kv_test_t::parse_args(int narg, const char *args[])
                 "    Fraction of key delete operations\n"
                 "  --list_prob 300\n"
                 "    Fraction of listing operations\n"
+                "  --reopen_prob 1\n"
+                "    Fraction of database reopens\n"
                 "  --min_key_len 10\n"
                 "    Minimum key size in bytes\n"
                 "  --max_key_len 70\n"
@@ -607,11 +609,9 @@ void kv_test_t::add_stat(kv_test_lat_t & stat, timespec tv_begin)
     int64_t usec = (tv_end.tv_sec - tv_begin.tv_sec)*1000000 +
         (tv_end.tv_nsec - tv_begin.tv_nsec)/1000;
     if (usec > 0)
-    {
         stat.usec += usec;
     stat.count++;
 }
-}
 
 void kv_test_t::print_stats(kv_test_stat_t & prev_stat, timespec & prev_stat_time)
 {