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

@@ -887,13 +887,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 +918,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 +933,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 +961,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;
     };
@@ -987,12 +1005,13 @@ void kv_op_t::exec()
 
 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()
@@ -2039,5 +2058,10 @@ void kv_dbw_t::list_next(void *handle, std::function<void(int res, const std::st
 void kv_dbw_t::list_close(void *handle)
 {
     kv_op_t *op = (kv_op_t*)handle;
+    if (op->started && !op->done)
+    {
+        op->done = true;
+        op->db->active_ops--;
+    }
     delete op;
 }

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,10 +609,8 @@ 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)