Add no_scrub flag

CMakeLists.txt

@@ -1,4 +1,4 @@
-cmake_minimum_required(VERSION 2.8)
+cmake_minimum_required(VERSION 2.8.12)
 
 project(vitastor)
 

mon/mon.js

@@ -107,6 +107,10 @@ const etcd_tree = {
             slow_log_interval: 10,
             inode_vanish_time: 60,
             osd_memlock: false,
+            scrub_interval: '30d', // 1s/1m/1h/1d
+            scrub_queue_depth: 1,
+            scrub_sleep: 0, // milliseconds
+            scrub_list_limit: 1000, // objects to list on one scrub iteration
             // blockstore - fixed in superblock
             block_size,
             disk_alignment,
@@ -168,6 +172,8 @@ const etcd_tree = {
                 osd_tags?: 'nvme' | [ 'nvme', ... ],
                 // prefer to put primary on OSD with these tags
                 primary_affinity_tags?: 'nvme' | [ 'nvme', ... ],
+                // scrub interval
+                scrub_interval?: '30d',
             },
             ...
         }, */
@@ -262,8 +268,8 @@ const etcd_tree = {
                 <pg_id>: {
                     primary: osd_num_t,
                     state: ("starting"|"peering"|"incomplete"|"active"|"repeering"|"stopping"|"offline"|
-                        "degraded"|"has_incomplete"|"has_degraded"|"has_misplaced"|"has_unclean"|
-                        "has_invalid"|"left_on_dead")[],
+                        "degraded"|"has_corrupted"|"has_incomplete"|"has_degraded"|"has_misplaced"|"has_unclean"|
+                        "has_invalid"|"left_on_dead"|"scrubbing")[],
                 }
             }, */
         },
@@ -285,6 +291,7 @@ const etcd_tree = {
                     osd_sets: osd_num_t[][],
                     all_peers: osd_num_t[],
                     epoch: uint64_t,
+                    scrub_ts: uint64_t,
                 },
             }, */
         },

src/CMakeLists.txt

@@ -1,4 +1,4 @@
-cmake_minimum_required(VERSION 2.8)
+cmake_minimum_required(VERSION 2.8.12)
 
 project(vitastor)
 
@@ -111,7 +111,7 @@ target_compile_options(vitastor_common PUBLIC -fPIC)
 add_executable(vitastor-osd
 	osd_main.cpp osd.cpp osd_secondary.cpp osd_peering.cpp osd_flush.cpp osd_peering_pg.cpp
 	osd_primary.cpp osd_primary_chain.cpp osd_primary_sync.cpp osd_primary_write.cpp osd_primary_subops.cpp
-	osd_cluster.cpp osd_rmw.cpp
+	osd_cluster.cpp osd_rmw.cpp osd_scrub.cpp
 )
 target_link_libraries(vitastor-osd
 	vitastor_common
@@ -299,7 +299,7 @@ add_executable(test_cluster_client
 	EXCLUDE_FROM_ALL
 	test_cluster_client.cpp
 	pg_states.cpp osd_ops.cpp cluster_client.cpp cluster_client_list.cpp msgr_op.cpp mock/messenger.cpp msgr_stop.cpp
-	etcd_state_client.cpp timerfd_manager.cpp ../json11/json11.cpp
+	etcd_state_client.cpp timerfd_manager.cpp str_util.cpp ../json11/json11.cpp
 )
 target_compile_definitions(test_cluster_client PUBLIC -D__MOCK__)
 target_include_directories(test_cluster_client PUBLIC ${CMAKE_SOURCE_DIR}/src/mock)

src/allocator.cpp

@@ -143,34 +143,83 @@ uint64_t allocator::get_free_count()
     return free;
 }
 
+// FIXME: Move to utils?
 void bitmap_set(void *bitmap, uint64_t start, uint64_t len, uint64_t bitmap_granularity)
 {
-    if (start == 0)
+    if (start == 0 && len == 32*bitmap_granularity)
+        *((uint32_t*)bitmap) = UINT32_MAX;
+    else if (start == 0 && len == 64*bitmap_granularity)
+        *((uint64_t*)bitmap) = UINT64_MAX;
+    else
     {
-        if (len == 32*bitmap_granularity)
+        unsigned bit_start = start / bitmap_granularity;
+        unsigned bit_end = ((start + len) + bitmap_granularity - 1) / bitmap_granularity;
+        while (bit_start < bit_end)
         {
-            *((uint32_t*)bitmap) = UINT32_MAX;
-            return;
-        }
-        else if (len == 64*bitmap_granularity)
-        {
-            *((uint64_t*)bitmap) = UINT64_MAX;
-            return;
+            if (!(bit_start & 7) && bit_end >= bit_start+8)
+            {
+                ((uint8_t*)bitmap)[bit_start / 8] = UINT8_MAX;
+                bit_start += 8;
+            }
+            else
+            {
+                ((uint8_t*)bitmap)[bit_start / 8] |= 1 << (bit_start % 8);
+                bit_start++;
+            }
         }
     }
-    unsigned bit_start = start / bitmap_granularity;
-    unsigned bit_end = ((start + len) + bitmap_granularity - 1) / bitmap_granularity;
-    while (bit_start < bit_end)
+}
+
+void bitmap_clear(void *bitmap, uint64_t start, uint64_t len, uint64_t bitmap_granularity)
+{
+    if (start == 0 && len == 32*bitmap_granularity)
+        *((uint32_t*)bitmap) = 0;
+    else if (start == 0 && len == 64*bitmap_granularity)
+        *((uint64_t*)bitmap) = 0;
+    else
     {
-        if (!(bit_start & 7) && bit_end >= bit_start+8)
+        unsigned bit_start = start / bitmap_granularity;
+        unsigned bit_end = ((start + len) + bitmap_granularity - 1) / bitmap_granularity;
+        while (bit_start < bit_end)
         {
-            ((uint8_t*)bitmap)[bit_start / 8] = UINT8_MAX;
-            bit_start += 8;
+            if (!(bit_start & 7) && bit_end >= bit_start+8)
+            {
+                ((uint8_t*)bitmap)[bit_start / 8] = 0;
+                bit_start += 8;
+            }
+            else
+            {
+                ((uint8_t*)bitmap)[bit_start / 8] &= (0xFF ^ (1 << (bit_start % 8)));
+                bit_start++;
+            }
         }
-        else
+    }
+}
+
+bool bitmap_check(void *bitmap, uint64_t start, uint64_t len, uint64_t bitmap_granularity)
+{
+    bool r = false;
+    if (start == 0 && len == 32*bitmap_granularity)
+        r = !!*((uint32_t*)bitmap);
+    else if (start == 0 && len == 64*bitmap_granularity)
+        r = !!*((uint64_t*)bitmap);
+    else
+    {
+        unsigned bit_start = start / bitmap_granularity;
+        unsigned bit_end = ((start + len) + bitmap_granularity - 1) / bitmap_granularity;
+        while (bit_start < bit_end)
         {
-            ((uint8_t*)bitmap)[bit_start / 8] |= 1 << (bit_start % 8);
-            bit_start++;
+            if (!(bit_start & 7) && bit_end >= bit_start+8)
+            {
+                r = r || !!((uint8_t*)bitmap)[bit_start / 8];
+                bit_start += 8;
+            }
+            else
+            {
+                r = r || (((uint8_t*)bitmap)[bit_start / 8] & (1 << (bit_start % 8)));
+                bit_start++;
+            }
         }
     }
+    return r;
 }

src/allocator.h

@@ -23,3 +23,5 @@ public:
 };
 
 void bitmap_set(void *bitmap, uint64_t start, uint64_t len, uint64_t bitmap_granularity);
+void bitmap_clear(void *bitmap, uint64_t start, uint64_t len, uint64_t bitmap_granularity);
+bool bitmap_check(void *bitmap, uint64_t start, uint64_t len, uint64_t bitmap_granularity);

src/blockstore.h

@@ -73,7 +73,8 @@ Input:
   write request is copied into the metadata area bitwise and stored there.
 
 Output:
-- retval = number of bytes actually read/written or negative error number (-EINVAL or -ENOSPC)
+- retval = number of bytes actually read/written or negative error number:
+  -EINVAL, -ENOSPC or -EDOM. -EDOM indicates checksum error.
 - version = the version actually read or written
 
 ## BS_OP_DELETE
@@ -122,11 +123,14 @@ Output:
 Get a list of all objects in this Blockstore.
 
 Input:
-- oid.stripe = PG alignment
-- len = PG count or 0 to list all objects
-- offset = PG number
-- oid.inode = min inode number or 0 to list all inodes
-- version = max inode number or 0 to list all inodes
+- pg_alignment = PG alignment
+- pg_count = PG count or 0 to list all objects
+- pg_number = PG number
+- list_stable_limit = max number of clean objects in the reply
+  it's guaranteed that dirty objects are returned from the same interval,
+  i.e. from (min_oid .. min(max_oid, max(returned stable OIDs)))
+- min_oid = min inode/stripe or 0 to list all objects
+- max_oid = max inode/stripe or 0 to list all objects
 
 Output:
 - retval = total obj_ver_id count
@@ -143,10 +147,27 @@ struct blockstore_op_t
     uint64_t opcode;
     // finish callback
     std::function<void (blockstore_op_t*)> callback;
-    object_id oid;
-    uint64_t version;
-    uint32_t offset;
-    uint32_t len;
+    union
+    {
+        // R/W
+        struct
+        {
+            object_id oid;
+            uint64_t version;
+            uint32_t offset;
+            uint32_t len;
+        };
+        // List
+        struct __attribute__((__packed__))
+        {
+            object_id min_oid;
+            object_id max_oid;
+            uint32_t pg_alignment;
+            uint32_t pg_count;
+            uint32_t pg_number;
+            uint32_t list_stable_limit;
+        };
+    };
     void *buf;
     void *bitmap;
     int retval;

src/blockstore_disk.cpp

@@ -40,10 +40,23 @@ void blockstore_disk_t::parse_config(std::map<std::string, std::string> & config
     data_block_size = parse_size(config["block_size"]);
     journal_device = config["journal_device"];
     journal_offset = parse_size(config["journal_offset"]);
-    disk_alignment = strtoull(config["disk_alignment"].c_str(), NULL, 10);
-    journal_block_size = strtoull(config["journal_block_size"].c_str(), NULL, 10);
-    meta_block_size = strtoull(config["meta_block_size"].c_str(), NULL, 10);
-    bitmap_granularity = strtoull(config["bitmap_granularity"].c_str(), NULL, 10);
+    disk_alignment = parse_size(config["disk_alignment"]);
+    journal_block_size = parse_size(config["journal_block_size"]);
+    meta_block_size = parse_size(config["meta_block_size"]);
+    bitmap_granularity = parse_size(config["bitmap_granularity"]);
+    if (config["data_csum_type"] == "crc32c")
+    {
+        data_csum_type = BLOCKSTORE_CSUM_CRC32C;
+    }
+    else if (config["data_csum_type"] == "" || config["data_csum_type"] == "none")
+    {
+        data_csum_type = BLOCKSTORE_CSUM_NONE;
+    }
+    else
+    {
+        throw std::runtime_error("data_csum_type="+config["data_csum_type"]+" is unsupported, only \"crc32c\" and \"none\" are supported");
+    }
+    csum_block_size = parse_size(config["csum_block_size"]);
     // Validate
     if (!data_block_size)
     {
@@ -91,7 +104,23 @@ void blockstore_disk_t::parse_config(std::map<std::string, std::string> & config
     }
     if (data_block_size % bitmap_granularity)
     {
-        throw std::runtime_error("Block size must be a multiple of sparse write tracking granularity");
+        throw std::runtime_error("Data block size must be a multiple of sparse write tracking granularity");
+    }
+    if (!data_csum_type)
+    {
+        csum_block_size = 0;
+    }
+    else if (!csum_block_size)
+    {
+        csum_block_size = bitmap_granularity;
+    }
+    if (csum_block_size && (csum_block_size % bitmap_granularity))
+    {
+        throw std::runtime_error("Checksum block size must be a multiple of sparse write tracking granularity");
+    }
+    if (csum_block_size && (data_block_size % csum_block_size))
+    {
+        throw std::runtime_error("Checksum block size must be a divisor of data block size");
     }
     if (meta_device == "")
     {
@@ -110,7 +139,9 @@ void blockstore_disk_t::parse_config(std::map<std::string, std::string> & config
         throw std::runtime_error("journal_offset must be a multiple of journal_block_size = "+std::to_string(journal_block_size));
     }
     clean_entry_bitmap_size = data_block_size / bitmap_granularity / 8;
-    clean_entry_size = sizeof(clean_disk_entry) + 2*clean_entry_bitmap_size;
+    clean_dyn_size = clean_entry_bitmap_size*2 + (csum_block_size
+        ? data_block_size/csum_block_size*(data_csum_type & 0xFF) : 0);
+    clean_entry_size = sizeof(clean_disk_entry) + clean_dyn_size + 4 /*entry_csum*/;
 }
 
 void blockstore_disk_t::calc_lengths(bool skip_meta_check)
@@ -160,9 +191,27 @@ void blockstore_disk_t::calc_lengths(bool skip_meta_check)
     // required metadata size
     block_count = data_len / data_block_size;
     meta_len = (1 + (block_count - 1 + meta_block_size / clean_entry_size) / (meta_block_size / clean_entry_size)) * meta_block_size;
+    meta_version = BLOCKSTORE_META_VERSION_V2;
     if (!skip_meta_check && meta_area_size < meta_len)
     {
-        throw std::runtime_error("Metadata area is too small, need at least "+std::to_string(meta_len)+" bytes");
+        if (!data_csum_type && !meta_version)
+        {
+            uint64_t clean_entry_v0_size = sizeof(clean_disk_entry) + 2*clean_entry_bitmap_size;
+            uint64_t meta_v0_len = (1 + (block_count - 1 + meta_block_size / clean_entry_v0_size)
+                / (meta_block_size / clean_entry_v0_size)) * meta_block_size;
+            if (meta_area_size >= meta_v0_len)
+            {
+                // Old metadata fits.
+                printf("Warning: Forcing metadata format without checksums because the new format doesn't fit into provided area\n");
+                clean_entry_size = clean_entry_v0_size;
+                meta_len = meta_v0_len;
+                meta_version = BLOCKSTORE_META_VERSION_V1;
+            }
+        }
+        if (meta_area_size < meta_len)
+        {
+            throw std::runtime_error("Metadata area is too small, need at least "+std::to_string(meta_len)+" bytes");
+        }
     }
     // requested journal size
     if (!skip_meta_check && cfg_journal_size > journal_len)

src/blockstore_disk.h

@@ -8,6 +8,10 @@
 #include <string>
 #include <map>
 
+#define BLOCKSTORE_CSUM_NONE 0
+// Lower byte of checksum type is its length
+#define BLOCKSTORE_CSUM_CRC32C 0x104
+
 struct blockstore_disk_t
 {
     std::string data_device, meta_device, journal_device;
@@ -21,17 +25,21 @@ struct blockstore_disk_t
     uint64_t meta_block_size = 4096;
     // Sparse write tracking granularity. 4 KB is a good choice. Must be a multiple of disk_alignment
     uint64_t bitmap_granularity = 4096;
+    // Data checksum type, BLOCKSTORE_CSUM_NONE or BLOCKSTORE_CSUM_CRC32C
+    uint32_t data_csum_type = BLOCKSTORE_CSUM_NONE;
+    // Checksum block size, must be a multiple of bitmap_granularity
+    uint32_t csum_block_size = 4096;
     // By default, Blockstore locks all opened devices exclusively. This option can be used to disable locking
     bool disable_flock = false;
 
     int meta_fd = -1, data_fd = -1, journal_fd = -1;
-    uint64_t meta_offset, meta_device_sect, meta_device_size, meta_len;
+    uint64_t meta_offset, meta_device_sect, meta_device_size, meta_len, meta_version = 0;
     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;
     uint64_t block_count;
-    uint32_t clean_entry_bitmap_size = 0, clean_entry_size = 0;
+    uint32_t clean_entry_bitmap_size = 0, clean_entry_size = 0, clean_dyn_size = 0;
 
     void parse_config(std::map<std::string, std::string> & config);
     void open_data();
@@ -39,4 +47,13 @@ struct blockstore_disk_t
     void open_journal();
     void calc_lengths(bool skip_meta_check = false);
     void close_all();
+
+    inline uint64_t dirty_dyn_size(uint64_t offset, uint64_t len)
+    {
+        // Checksums may be partial if write is not aligned with csum_block_size
+        return clean_entry_bitmap_size + (csum_block_size
+            ? ((offset+len+csum_block_size-1)/csum_block_size - offset/csum_block_size)
+                * (data_csum_type & 0xFF)
+            : 0);
+    }
 };

src/blockstore_flush.h

@@ -1,10 +1,19 @@
 // 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
+
 struct copy_buffer_t
 {
-    uint64_t offset, len;
-    void *buf;
+    int copy_flags;
+    uint64_t offset, len, disk_offset;
+    uint64_t journal_sector; // only for reads: sector+1 if used and !journal.inmemory, otherwise 0
+    void *buf; // only for writes: new checksum data
+    uint8_t *csum_buf;
 };
 
 struct meta_sector_t
@@ -37,7 +46,7 @@ class journal_flusher_co
 {
     blockstore_impl_t *bs;
     journal_flusher_t *flusher;
-    int wait_state, wait_count;
+    int wait_state, wait_count, wait_journal_count;
     struct io_uring_sqe *sqe;
     struct ring_data_t *data;
 
@@ -46,28 +55,39 @@ class journal_flusher_co
     obj_ver_id cur;
     std::map<obj_ver_id, dirty_entry>::iterator dirty_it, dirty_start, dirty_end;
     std::map<object_id, uint64_t>::iterator repeat_it;
-    std::function<void(ring_data_t*)> simple_callback_r, simple_callback_w;
+    std::function<void(ring_data_t*)> simple_callback_r, simple_callback_rj, simple_callback_w;
 
     bool skip_copy, has_delete, has_writes;
     std::vector<copy_buffer_t> v;
     std::vector<copy_buffer_t>::iterator it;
+    int i;
+    bool fill_incomplete, cleared_incomplete;
+    int read_to_fill_incomplete;
     int copy_count;
-    uint64_t clean_loc, old_clean_loc;
+    uint64_t clean_loc, old_clean_loc, old_clean_ver;
     flusher_meta_write_t meta_old, meta_new;
     bool clean_init_bitmap;
     uint64_t clean_bitmap_offset, clean_bitmap_len;
-    void *new_clean_bitmap;
+    uint8_t *clean_init_dyn_ptr;
+    uint8_t *new_clean_bitmap;
 
     uint64_t new_trim_pos;
 
-    // local: scan_dirty()
-    uint64_t offset, end_offset, submit_offset, submit_len;
-
     friend class journal_flusher_t;
-    bool scan_dirty(int wait_base);
+    void scan_dirty();
+    bool read_dirty(int wait_base);
+    bool modify_meta_do_reads(int wait_base);
+    bool wait_meta_reads(int wait_base);
     bool modify_meta_read(uint64_t meta_loc, flusher_meta_write_t &wr, int wait_base);
+    bool clear_incomplete_csum_block_bits(int wait_base);
+    void calc_block_checksums(uint32_t *new_data_csums, bool skip_overwrites);
+    void update_metadata_entry();
+    bool write_meta_block(flusher_meta_write_t & meta_block, int wait_base);
     void update_clean_db();
+    void free_data_blocks();
     bool fsync_batch(bool fsync_meta, int wait_base);
+    bool trim_journal(int wait_base);
+    void free_buffers();
 public:
     journal_flusher_co();
     bool loop();
@@ -95,9 +115,10 @@ class journal_flusher_t
 
     std::map<uint64_t, meta_sector_t> meta_sectors;
     std::deque<object_id> flush_queue;
-    std::map<object_id, uint64_t> flush_versions;
+    std::map<object_id, uint64_t> flush_versions; // FIXME: consider unordered_map?
 
     bool try_find_older(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
+    bool try_find_other(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
 
 public:
     journal_flusher_t(blockstore_impl_t *bs);

src/blockstore_impl.cpp

@@ -38,8 +38,8 @@ blockstore_impl_t::~blockstore_impl_t()
     dsk.close_all();
     if (metadata_buffer)
         free(metadata_buffer);
-    if (clean_bitmap)
-        free(clean_bitmap);
+    if (clean_dyn_data)
+        free(clean_dyn_data);
 }
 
 bool blockstore_impl_t::is_started()
@@ -445,11 +445,11 @@ void blockstore_impl_t::reshard_clean_db(pool_id_t pool, uint32_t pg_count, uint
 
 void blockstore_impl_t::process_list(blockstore_op_t *op)
 {
-    uint32_t list_pg = op->offset+1;
-    uint32_t pg_count = op->len;
-    uint64_t pg_stripe_size = op->oid.stripe;
-    uint64_t min_inode = op->oid.inode;
-    uint64_t max_inode = op->version;
+    uint32_t list_pg = op->pg_number+1;
+    uint32_t pg_count = op->pg_count;
+    uint64_t pg_stripe_size = op->pg_alignment;
+    uint64_t min_inode = op->min_oid.inode;
+    uint64_t max_inode = op->max_oid.inode;
     // Check PG
     if (pg_count != 0 && (pg_stripe_size < MIN_DATA_BLOCK_SIZE || list_pg > pg_count))
     {
@@ -496,7 +496,13 @@ void blockstore_impl_t::process_list(blockstore_op_t *op)
             stable_alloc += clean_db.size();
         }
     }
-    else
+    if (op->list_stable_limit > 0)
+    {
+        stable_alloc = op->list_stable_limit;
+        if (stable_alloc > 1024*1024)
+            stable_alloc = 1024*1024;
+    }
+    if (stable_alloc < 32768)
     {
         stable_alloc = 32768;
     }
@@ -507,22 +513,21 @@ void blockstore_impl_t::process_list(blockstore_op_t *op)
         FINISH_OP(op);
         return;
     }
+    auto max_oid = op->max_oid;
+    bool limited = false;
     for (auto shard_it = clean_db_shards.lower_bound(first_shard);
         shard_it != clean_db_shards.end() && shard_it->first <= last_shard;
         shard_it++)
     {
         auto & clean_db = shard_it->second;
         auto clean_it = clean_db.begin(), clean_end = clean_db.end();
-        if ((min_inode != 0 || max_inode != 0) && min_inode <= max_inode)
+        if (op->min_oid.inode != 0 || op->min_oid.stripe != 0)
         {
-            clean_it = clean_db.lower_bound({
-                .inode = min_inode,
-                .stripe = 0,
-            });
-            clean_end = clean_db.upper_bound({
-                .inode = max_inode,
-                .stripe = UINT64_MAX,
-            });
+            clean_it = clean_db.lower_bound(op->min_oid);
+        }
+        if ((max_oid.inode != 0 || max_oid.stripe != 0) && !(max_oid < op->min_oid))
+        {
+            clean_end = clean_db.upper_bound(max_oid);
         }
         for (; clean_it != clean_end; clean_it++)
         {
@@ -541,11 +546,24 @@ void blockstore_impl_t::process_list(blockstore_op_t *op)
                 .oid = clean_it->first,
                 .version = clean_it->second.version,
             };
+            if (op->list_stable_limit > 0 && !limited && stable_count >= op->list_stable_limit)
+            {
+                limited = true;
+                break;
+            }
+        }
+        if (op->list_stable_limit > 0 && first_shard != last_shard)
+        {
+            // To maintain the order, we have to include objects in the same range from other shards
+            std::sort(stable, stable+stable_count);
+            if (stable_count > op->list_stable_limit)
+                stable_count = op->list_stable_limit;
+            max_oid = stable[stable_count-1].oid;
         }
     }
-    if (first_shard != last_shard)
+    if (op->list_stable_limit == 0 && first_shard != last_shard)
     {
-        // If that's not a per-PG listing, sort clean entries
+        // If that's not a per-PG listing, sort clean entries (already sorted if list_stable_limit != 0)
         std::sort(stable, stable+stable_count);
     }
     int clean_stable_count = stable_count;
@@ -554,20 +572,17 @@ void blockstore_impl_t::process_list(blockstore_op_t *op)
     obj_ver_id *unstable = NULL;
     {
         auto dirty_it = dirty_db.begin(), dirty_end = dirty_db.end();
-        if ((min_inode != 0 || max_inode != 0) && min_inode <= max_inode)
+        if (op->min_oid.inode != 0 || op->min_oid.stripe != 0)
         {
             dirty_it = dirty_db.lower_bound({
-                .oid = {
-                    .inode = min_inode,
-                    .stripe = 0,
-                },
+                .oid = op->min_oid,
                 .version = 0,
             });
+        }
+        if ((max_oid.inode != 0 || max_oid.stripe != 0) && !(max_oid < op->min_oid))
+        {
             dirty_end = dirty_db.upper_bound({
-                .oid = {
-                    .inode = max_inode,
-                    .stripe = UINT64_MAX,
-                },
+                .oid = max_oid,
                 .version = UINT64_MAX,
             });
         }

src/blockstore_impl.h

@@ -94,10 +94,9 @@
 // "VITAstor"
 #define BLOCKSTORE_META_MAGIC_V1 0x726F747341544956l
 #define BLOCKSTORE_META_VERSION_V1 1
+#define BLOCKSTORE_META_VERSION_V2 2
 
 // metadata header (superblock)
-// FIXME: After adding the OSD superblock, add a key to metadata
-// and journal headers to check if they belong to the same OSD
 struct __attribute__((__packed__)) blockstore_meta_header_v1_t
 {
     uint64_t zero;
@@ -108,14 +107,29 @@ struct __attribute__((__packed__)) blockstore_meta_header_v1_t
     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
-// FIXME: maybe add crc32's to metadata
 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)
@@ -125,7 +139,7 @@ struct __attribute__((__packed__)) clean_entry
     uint64_t location;
 };
 
-// 64 = 24 + 40 bytes per dirty entry in memory (obj_ver_id => dirty_entry)
+// 64 = 24 + 40 bytes per dirty entry in memory (obj_ver_id => dirty_entry). Plus checksums
 struct __attribute__((__packed__)) dirty_entry
 {
     uint32_t state;
@@ -134,7 +148,7 @@ struct __attribute__((__packed__)) dirty_entry
     uint32_t offset;   // data offset within object (stripe)
     uint32_t len;      // data length
     uint64_t journal_sector; // journal sector used for this entry
-    void* bitmap;   // either external bitmap itself when it fits, or a pointer to it when it doesn't
+    void* dyn_data;    // dynamic data: external bitmap and data block checksums. may be a pointer to the in-memory journal
 };
 
 // - Sync must be submitted after previous writes/deletes (not before!)
@@ -161,12 +175,23 @@ struct __attribute__((__packed__)) dirty_entry
 // Suspend operation until the next journal sector buffer is free
 #define WAIT_JOURNAL_BUFFER 4
 
-struct fulfill_read_t
+struct used_clean_obj_t
 {
-    uint64_t offset, len;
-    uint64_t journal_sector; // sector+1 if used and !journal.inmemory, otherwise 0
+    int refs;
+    uint64_t freed_block; // block+1 if freed, otherwise 0
+    uint8_t *meta; // metadata copy
 };
 
+// https://github.com/algorithm-ninja/cpp-btree
+// https://github.com/greg7mdp/sparsepp/ was used previously, but it was TERRIBLY slow after resizing
+// with sparsepp, random reads dropped to ~700 iops very fast with just as much as ~32k objects in the DB
+typedef btree::btree_map<object_id, clean_entry> blockstore_clean_db_t;
+typedef std::map<obj_ver_id, dirty_entry> blockstore_dirty_db_t;
+
+#include "blockstore_init.h"
+
+#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)
 
@@ -179,7 +204,8 @@ struct blockstore_op_private_t
     int op_state;
 
     // Read
-    std::vector<fulfill_read_t> read_vec;
+    uint64_t clean_version_used;
+    std::vector<copy_buffer_t> read_vec;
 
     // Sync, write
     int min_flushed_journal_sector, max_flushed_journal_sector;
@@ -195,16 +221,6 @@ struct blockstore_op_private_t
     int sync_small_checked, sync_big_checked;
 };
 
-// https://github.com/algorithm-ninja/cpp-btree
-// https://github.com/greg7mdp/sparsepp/ was used previously, but it was TERRIBLY slow after resizing
-// with sparsepp, random reads dropped to ~700 iops very fast with just as much as ~32k objects in the DB
-typedef btree::btree_map<object_id, clean_entry> blockstore_clean_db_t;
-typedef std::map<obj_ver_id, dirty_entry> blockstore_dirty_db_t;
-
-#include "blockstore_init.h"
-
-#include "blockstore_flush.h"
-
 typedef uint32_t pool_id_t;
 typedef uint64_t pool_pg_id_t;
 
@@ -246,7 +262,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;
-    uint8_t *clean_bitmap = NULL;
+    uint8_t *clean_dyn_data = NULL;
     blockstore_dirty_db_t dirty_db;
     std::vector<blockstore_op_t*> submit_queue;
     std::vector<obj_ver_id> unsynced_big_writes, unsynced_small_writes;
@@ -260,6 +276,9 @@ class blockstore_impl_t
     journal_flusher_t *flusher;
     int write_iodepth = 0;
 
+    // clean data blocks referenced by read operations
+    std::map<obj_ver_id, used_clean_obj_t> used_clean_objects;
+
     bool live = false, queue_stall = false;
     ring_loop_t *ringloop;
     timerfd_manager_t *tfd;
@@ -302,8 +321,25 @@ class blockstore_impl_t
 
     // Read
     int dequeue_read(blockstore_op_t *read_op);
-    int fulfill_read(blockstore_op_t *read_op, uint64_t &fulfilled, uint32_t item_start, uint32_t item_end,
-        uint32_t item_state, uint64_t item_version, uint64_t item_location, uint64_t journal_sector);
+    void find_holes(std::vector<copy_buffer_t> & read_vec, uint32_t item_start, uint32_t item_end,
+        std::function<int(int, bool, uint32_t, uint32_t)> callback);
+    int fulfill_read(blockstore_op_t *read_op, uint64_t & fulfilled, uint32_t item_start, uint32_t item_end,
+        uint32_t item_state, uint64_t item_version, uint64_t item_location, uint64_t journal_sector, uint8_t *csum);
+    bool fulfill_clean_read(blockstore_op_t *read_op, uint64_t & fulfilled,
+        uint8_t *clean_entry_bitmap, uint64_t clean_loc, uint64_t clean_ver);
+    int fill_partial_checksum_blocks(std::vector<copy_buffer_t> & rv, uint64_t & fulfilled,
+        uint8_t *clean_entry_bitmap, bool from_journal, uint8_t *read_buf, uint64_t read_offset, uint64_t read_end);
+    int pad_journal_read(std::vector<copy_buffer_t> & rv, copy_buffer_t & cp,
+        uint64_t dirty_offset, uint64_t dirty_end, uint64_t dirty_loc, uint8_t *csum_ptr,
+        uint64_t offset, uint64_t submit_len, uint64_t & blk_begin, uint64_t & blk_end, uint8_t* & blk_buf);
+    bool read_range_fulfilled(std::vector<copy_buffer_t> & rv, uint64_t & fulfilled, uint8_t *read_buf,
+        uint8_t *clean_entry_bitmap, uint32_t item_start, uint32_t item_end);
+    bool read_checksum_block(blockstore_op_t *op, int rv_pos, uint64_t &fulfilled, uint64_t clean_loc);
+    bool verify_padded_checksums(uint8_t *clean_entry_bitmap, bool is_journal, uint32_t offset,
+        iovec *iov, int n_iov, std::function<void(uint32_t, uint32_t, uint32_t)> bad_block_cb);
+    bool verify_journal_checksums(uint8_t *csums, uint32_t offset,
+        iovec *iov, int n_iov, std::function<void(uint32_t, uint32_t, uint32_t)> bad_block_cb);
+    bool verify_clean_padded_checksums(blockstore_op_t *op, uint64_t clean_loc, iovec *iov, int n_iov);
     int fulfill_read_push(blockstore_op_t *op, void *buf, uint64_t offset, uint64_t len,
         uint32_t item_state, uint64_t item_version);
     void handle_read_event(ring_data_t *data, blockstore_op_t *op);
@@ -332,6 +368,7 @@ class blockstore_impl_t
     int continue_rollback(blockstore_op_t *op);
     void mark_rolled_back(const obj_ver_id & ov);
     void erase_dirty(blockstore_dirty_db_t::iterator dirty_start, blockstore_dirty_db_t::iterator dirty_end, uint64_t clean_loc);
+    void free_dirty_dyn_data(dirty_entry & e);
 
     // List
     void process_list(blockstore_op_t *op);

src/blockstore_init.cpp

@@ -77,13 +77,17 @@ resume_1:
     if (iszero((uint64_t*)metadata_buffer, bs->dsk.meta_block_size / sizeof(uint64_t)))
     {
         {
-            blockstore_meta_header_v1_t *hdr = (blockstore_meta_header_v1_t *)metadata_buffer;
+            blockstore_meta_header_v2_t *hdr = (blockstore_meta_header_v2_t *)metadata_buffer;
             hdr->zero = 0;
             hdr->magic = BLOCKSTORE_META_MAGIC_V1;
-            hdr->version = BLOCKSTORE_META_VERSION_V1;
+            hdr->version = BLOCKSTORE_META_VERSION_V2;
             hdr->meta_block_size = bs->dsk.meta_block_size;
             hdr->data_block_size = bs->dsk.data_block_size;
             hdr->bitmap_granularity = bs->dsk.bitmap_granularity;
+            hdr->data_csum_type = bs->dsk.data_csum_type;
+            hdr->csum_block_size = bs->dsk.csum_block_size;
+            hdr->header_csum = 0;
+            hdr->header_csum = crc32c(0, hdr, sizeof(*hdr));
         }
         if (bs->readonly)
         {
@@ -109,28 +113,70 @@ resume_1:
     }
     else
     {
-        blockstore_meta_header_v1_t *hdr = (blockstore_meta_header_v1_t *)metadata_buffer;
-        if (hdr->zero != 0 ||
-            hdr->magic != BLOCKSTORE_META_MAGIC_V1 ||
-            hdr->version != BLOCKSTORE_META_VERSION_V1)
+        blockstore_meta_header_v2_t *hdr = (blockstore_meta_header_v2_t *)metadata_buffer;
+        if (hdr->zero != 0 || hdr->magic != BLOCKSTORE_META_MAGIC_V1 || hdr->version < BLOCKSTORE_META_VERSION_V1)
         {
             printf(
-                "Metadata is corrupt or old version.\n"
-                " If this is a new OSD please zero out the metadata area before starting it.\n"
-                " If you need to upgrade from 0.5.x please request it via the issue tracker.\n"
+                "Metadata is corrupt or too old (pre-0.6.x).\n"
+                " If this is a new OSD, please zero out the metadata area before starting it.\n"
+                " If you need to upgrade from 0.5.x, convert metadata with vitastor-disk.\n"
+            );
+            exit(1);
+        }
+        if (bs->dsk.meta_version && bs->dsk.meta_version != hdr->version)
+        {
+            printf(
+                "Metadata format version is %lu on disk, but %lu is currently selected in OSD configuration.\n"
+                " Please upgrade using vitastor-disk.\n", hdr->version, bs->dsk.meta_version
+            );
+            exit(1);
+        }
+        if (hdr->version == BLOCKSTORE_META_VERSION_V2)
+        {
+            uint32_t csum = hdr->header_csum;
+            hdr->header_csum = 0;
+            if (crc32c(0, hdr, sizeof(*hdr)) != csum)
+            {
+                printf("Metadata header is corrupt (checksum mismatch).\n");
+                exit(1);
+            }
+            hdr->header_csum = csum;
+            bs->dsk.meta_version = BLOCKSTORE_META_VERSION_V2;
+        }
+        else if (hdr->version == BLOCKSTORE_META_VERSION_V1)
+        {
+            hdr->data_csum_type = 0;
+            hdr->csum_block_size = 0;
+            hdr->header_csum = 0;
+            // Enable compatibility mode - entries without checksums
+            bs->dsk.clean_entry_size = sizeof(clean_disk_entry) + bs->dsk.clean_entry_bitmap_size*2;
+            bs->dsk.meta_len = (1 + (bs->dsk.block_count - 1 + bs->dsk.meta_block_size / bs->dsk.clean_entry_size)
+                / (bs->dsk.meta_block_size / bs->dsk.clean_entry_size)) * bs->dsk.meta_block_size;
+            bs->dsk.meta_version = BLOCKSTORE_META_VERSION_V1;
+            printf("Warning: Starting with metadata in the old format without checksums, as stored on disk\n");
+        }
+        else if (hdr->version > BLOCKSTORE_META_VERSION_V2)
+        {
+            printf(
+                "Metadata format is too new for me (stored version is %lu, max supported %u).\n",
+                hdr->version, BLOCKSTORE_META_VERSION_V2
             );
             exit(1);
         }
         if (hdr->meta_block_size != bs->dsk.meta_block_size ||
             hdr->data_block_size != bs->dsk.data_block_size ||
-            hdr->bitmap_granularity != bs->dsk.bitmap_granularity)
+            hdr->bitmap_granularity != bs->dsk.bitmap_granularity ||
+            hdr->data_csum_type != bs->dsk.data_csum_type ||
+            hdr->csum_block_size != bs->dsk.csum_block_size)
         {
             printf(
                 "Configuration stored in metadata superblock"
-                " (meta_block_size=%u, data_block_size=%u, bitmap_granularity=%u)"
-                " differs from OSD configuration (%lu/%u/%lu).\n",
+                " (meta_block_size=%u, data_block_size=%u, bitmap_granularity=%u, data_csum_type=%u, csum_block_size=%u)"
+                " differs from OSD configuration (%lu/%u/%lu, %u/%u).\n",
                 hdr->meta_block_size, hdr->data_block_size, hdr->bitmap_granularity,
-                bs->dsk.meta_block_size, bs->dsk.data_block_size, bs->dsk.bitmap_granularity
+                hdr->data_csum_type, hdr->csum_block_size,
+                bs->dsk.meta_block_size, bs->dsk.data_block_size, bs->dsk.bitmap_granularity,
+                bs->dsk.data_csum_type, bs->dsk.csum_block_size
             );
             exit(1);
         }
@@ -279,12 +325,22 @@ bool blockstore_init_meta::handle_meta_block(uint8_t *buf, uint64_t entries_per_
     for (uint64_t i = 0; i < max_i; i++)
     {
         clean_disk_entry *entry = (clean_disk_entry*)(buf + i*bs->dsk.clean_entry_size);
-        if (!bs->inmemory_meta && bs->dsk.clean_entry_bitmap_size)
-        {
-            memcpy(bs->clean_bitmap + (done_cnt+i)*2*bs->dsk.clean_entry_bitmap_size, &entry->bitmap, 2*bs->dsk.clean_entry_bitmap_size);
-        }
         if (entry->oid.inode > 0)
         {
+            if (bs->dsk.meta_version >= BLOCKSTORE_META_VERSION_V2)
+            {
+                // Check entry crc32
+                uint32_t *entry_csum = (uint32_t*)((uint8_t*)entry + bs->dsk.clean_entry_size - 4);
+                if (*entry_csum != crc32c(0, entry, bs->dsk.clean_entry_size - 4))
+                {
+                    printf("Metadata entry %lu is corrupt (checksum mismatch), skipping\n", done_cnt+i);
+                    continue;
+                }
+            }
+            if (!bs->inmemory_meta && bs->dsk.clean_entry_bitmap_size)
+            {
+                memcpy(bs->clean_dyn_data + (done_cnt+i)*bs->dsk.clean_dyn_size, &entry->bitmap, bs->dsk.clean_dyn_size);
+            }
             auto & clean_db = bs->clean_db_shard(entry->oid);
             auto clean_it = clean_db.find(entry->oid);
             if (clean_it == clean_db.end() || clean_it->second.version < entry->version)
@@ -440,7 +496,9 @@ resume_1:
             .size = sizeof(journal_entry_start),
             .reserved = 0,
             .journal_start = bs->journal.block_size,
-            .version = JOURNAL_VERSION,
+            .version = JOURNAL_VERSION_V2,
+            .data_csum_type = bs->dsk.data_csum_type,
+            .csum_block_size = bs->dsk.csum_block_size,
         };
         ((journal_entry_start*)submitted_buf)->crc32 = je_crc32((journal_entry*)submitted_buf);
         if (bs->readonly)
@@ -492,18 +550,36 @@ resume_1:
         if (je_start->magic != JOURNAL_MAGIC ||
             je_start->type != JE_START ||
             je_crc32((journal_entry*)je_start) != je_start->crc32 ||
-            je_start->size != sizeof(journal_entry_start) && je_start->size != JE_START_LEGACY_SIZE)
+            je_start->size != JE_START_V0_SIZE && je_start->size != JE_START_V1_SIZE && je_start->size != JE_START_V2_SIZE)
         {
             // Entry is corrupt
-            fprintf(stderr, "First entry of the journal is corrupt\n");
+            fprintf(stderr, "First entry of the journal is corrupt or unsupported\n");
             exit(1);
         }
-        if (je_start->size == JE_START_LEGACY_SIZE || je_start->version != JOURNAL_VERSION)
+        if (je_start->size == JE_START_V0_SIZE ||
+            (je_start->version != JOURNAL_VERSION_V1 || je_start->size != JE_START_V1_SIZE) &&
+            (je_start->version != JOURNAL_VERSION_V2 || je_start->size != JE_START_V2_SIZE))
         {
             fprintf(
-                stderr, "The code only supports journal version %d, but it is %lu on disk."
-                    " Please use the previous version to flush the journal before upgrading OSD\n",
-                JOURNAL_VERSION, je_start->size == JE_START_LEGACY_SIZE ? 0 : je_start->version
+                stderr, "The code only supports journal versions 2 and 1, but it is %lu on disk."
+                    " Please use vitastor-disk to rewrite the journal\n",
+                je_start->size == JE_START_V0_SIZE ? 0 : je_start->version
+            );
+            exit(1);
+        }
+        if (je_start->version == JOURNAL_VERSION_V1)
+        {
+            je_start->data_csum_type = 0;
+            je_start->csum_block_size = 0;
+        }
+        if (je_start->data_csum_type != bs->dsk.data_csum_type ||
+            je_start->csum_block_size != bs->dsk.csum_block_size)
+        {
+            printf(
+                "Configuration stored in journal superblock (data_csum_type=%u, csum_block_size=%u)"
+                " differs from OSD configuration (%u/%u).\n",
+                je_start->data_csum_type, je_start->csum_block_size,
+                bs->dsk.data_csum_type, bs->dsk.csum_block_size
             );
             exit(1);
         }
@@ -705,11 +781,14 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                     snprintf(err, 1024, "BUG: calculated journal data offset (%08lx) != stored journal data offset (%08lx)", location, je->small_write.data_offset);
                     throw std::runtime_error(err);
                 }
-                uint32_t data_crc32 = 0;
+                small_write_data.clear();
                 if (location >= done_pos && location+je->small_write.len <= done_pos+len)
                 {
                     // data is within this buffer
-                    data_crc32 = crc32c(0, (uint8_t*)buf + location - done_pos, je->small_write.len);
+                    small_write_data.push_back((iovec){
+                        .iov_base = (uint8_t*)buf + location - done_pos,
+                        .iov_len = je->small_write.len,
+                    });
                 }
                 else
                 {
@@ -724,7 +803,10 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                                 ? location+je->small_write.len : done[i].pos+done[i].len);
                             uint64_t part_begin = (location < done[i].pos ? done[i].pos : location);
                             covered += part_end - part_begin;
-                            data_crc32 = crc32c(data_crc32, (uint8_t*)done[i].buf + part_begin - done[i].pos, part_end - part_begin);
+                            small_write_data.push_back((iovec){
+                                .iov_base = (uint8_t*)done[i].buf + part_begin - done[i].pos,
+                                .iov_len = part_end - part_begin,
+                            });
                         }
                     }
                     if (covered < je->small_write.len)
@@ -734,12 +816,68 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                         return 2;
                     }
                 }
-                if (data_crc32 != je->small_write.crc32_data)
+                bool data_csum_valid = true;
+                if (!bs->dsk.csum_block_size)
+                {
+                    uint32_t data_crc32 = 0;
+                    for (auto & sd: small_write_data)
+                    {
+                        data_crc32 = crc32c(data_crc32, sd.iov_base, sd.iov_len);
+                    }
+                    data_csum_valid = data_crc32 == je->small_write.crc32_data;
+                    if (!data_csum_valid)
+                    {
+                        printf("Journal entry data is corrupt (data crc32 %x != %x)\n", data_crc32, je->small_write.crc32_data);
+                    }
+                }
+                else if (je->small_write.len > 0)
+                {
+                    uint32_t *block_csums = (uint32_t*)((uint8_t*)je + sizeof(journal_entry_small_write) + bs->dsk.clean_entry_bitmap_size);
+                    uint32_t start = je->small_write.offset / bs->dsk.csum_block_size;
+                    uint32_t end = (je->small_write.offset+je->small_write.len-1) / bs->dsk.csum_block_size;
+                    int sd_num = 0;
+                    size_t sd_pos = 0;
+                    for (uint32_t pos = start; pos <= end; pos++, block_csums++)
+                    {
+                        size_t block_left = (pos == start
+                            ? (start == end
+                                ? je->small_write.len
+                                : bs->dsk.csum_block_size - je->small_write.offset%bs->dsk.csum_block_size)
+                            : (pos < end
+                                ? bs->dsk.csum_block_size
+                                : (je->small_write.offset + je->small_write.len)%bs->dsk.csum_block_size));
+                        uint32_t block_crc32 = 0;
+                        while (block_left > 0)
+                        {
+                            assert(sd_num < small_write_data.size());
+                            if (small_write_data[sd_num].iov_len >= sd_pos+block_left)
+                            {
+                                block_crc32 = crc32c(block_crc32, (uint8_t*)small_write_data[sd_num].iov_base+sd_pos, block_left);
+                                sd_pos += block_left;
+                                break;
+                            }
+                            else
+                            {
+                                block_crc32 = crc32c(block_crc32, (uint8_t*)small_write_data[sd_num].iov_base+sd_pos, small_write_data[sd_num].iov_len-sd_pos);
+                                block_left -= (small_write_data[sd_num].iov_len-sd_pos);
+                                sd_pos = 0;
+                                sd_num++;
+                            }
+                        }
+                        if (block_crc32 != *block_csums)
+                        {
+                            printf("Journal entry data is corrupt (block %u crc32 %x != %x)\n",
+                                pos, block_crc32, *block_csums);
+                            data_csum_valid = false;
+                            break;
+                        }
+                    }
+                }
+                if (!data_csum_valid)
                 {
                     // journal entry is corrupt, stop here
                     // interesting thing is that we must clear the corrupt entry if we're not readonly,
                     // because we don't write next entries in the same journal block
-                    printf("Journal entry data is corrupt (data crc32 %x != %x)\n", data_crc32, je->small_write.crc32_data);
                     memset((uint8_t*)buf + proc_pos - done_pos + pos, 0, bs->journal.block_size - pos);
                     bs->journal.next_free = prev_free;
                     init_write_buf = (uint8_t*)buf + proc_pos - done_pos;
@@ -755,11 +893,19 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                         .oid = je->small_write.oid,
                         .version = je->small_write.version,
                     };
-                    void *bmp = NULL;
-                    void *bmp_from = (uint8_t*)je + sizeof(journal_entry_small_write);
-                    if (bs->dsk.clean_entry_bitmap_size <= sizeof(void*))
+                    uint64_t dyn_size = bs->dsk.dirty_dyn_size