Write recheck_modified_blocks to disk

src/blockstore/blockstore_heap.cpp

@@ -595,7 +595,6 @@ void blockstore_heap_t::recheck_buffer(heap_entry_t *cwr, uint8_t *buf)
                 }
             }
         });
-        // FIXME: Write recheck_modified_blocks to disk
         recheck_modified_blocks.insert(block_num);
     };
     if (cwr->is_garbage())
@@ -729,6 +728,13 @@ bool blockstore_heap_t::recheck_small_writes(std::function<void(bool is_data, ui
     return false;
 }
 
+std::vector<uint32_t> blockstore_heap_t::get_recheck_modified_blocks()
+{
+    std::vector<uint32_t> modified(recheck_modified_blocks.begin(), recheck_modified_blocks.end());
+    recheck_modified_blocks.clear();
+    return modified;
+}
+
 int blockstore_heap_t::finish_load(bool allow_corrupted)
 {
     if (!marked_used_blocks)

src/blockstore/blockstore_heap.h

@@ -221,6 +221,9 @@ public:
         bool allow_corrupted, uint64_t &entries_loaded);
     // finish loading
     int finish_load(bool allow_corrupted = false);
+    // get blocks which are modified during loading and should be written to the disk
+    // before finishing initialization if not R/O
+    std::vector<uint32_t> get_recheck_modified_blocks();
     // recheck small write data after reading the database from disk
     bool recheck_small_writes(std::function<void(bool is_data, uint64_t offset, uint64_t len, uint8_t* buf, std::function<void()>)> read_buffer, int queue_depth);
     // reshard database according to the pool's PG count

src/blockstore/blockstore_init.cpp

@@ -51,6 +51,8 @@ int blockstore_init_meta::loop()
     else if (wait_state == 5) goto resume_5;
     else if (wait_state == 6) goto resume_6;
     else if (wait_state == 7) goto resume_7;
+    else if (wait_state == 8) goto resume_8;
+    else if (wait_state == 9) goto resume_9;
     metadata_buffer = memalign(MEM_ALIGNMENT, 2*bs->metadata_buf_size);
     if (!metadata_buffer)
         throw std::runtime_error("Failed to allocate metadata read buffer");
@@ -283,6 +285,46 @@ resume_6:
     }, bs->meta_write_recheck_parallelism);
     return 1;
 resume_7:
+    recheck_mod = bs->heap->get_recheck_modified_blocks();
+    if (bs->readonly)
+    {
+        recheck_mod.clear();
+    }
+    for (i = 0; i < recheck_mod.size(); i++)
+    {
+resume_8:
+        if (wait_count >= bs->meta_write_recheck_parallelism || !(sqe = bs->get_sqe()))
+        {
+            bs->ringloop->submit();
+            wait_state = 8;
+            return 1;
+        }
+        data = ((ring_data_t*)sqe->user_data);
+        uint8_t *buf = (uint8_t*)malloc_or_die(bs->dsk.meta_block_size);
+        bs->heap->get_meta_block(recheck_mod[i], buf);
+        data->iov = { buf, bs->dsk.meta_block_size };
+        data->callback = [this, buf, block_num = i](ring_data_t *data)
+        {
+            wait_count--;
+            free(buf);
+            if (data->res != bs->dsk.meta_block_size)
+            {
+                throw std::runtime_error(
+                    "write metadata failed at offset " + std::to_string(bs->dsk.meta_offset + (block_num+1)*bs->dsk.meta_block_size) +
+                    ": " + strerror(-data->res)
+                );
+            }
+        };
+        io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset + (i+1)*bs->dsk.meta_block_size);
+        wait_count++;
+    }
+resume_9:
+    if (wait_count > 0)
+    {
+        bs->ringloop->submit();
+        wait_state = 9;
+        return 1;
+    }
     if (bs->heap->finish_load() != 0)
     {
         exit(1);

src/blockstore/blockstore_init.h

@@ -15,6 +15,7 @@ class blockstore_init_meta
 {
     blockstore_impl_t *bs;
     int wait_state = 0;
+    int wait_count = 0;
     bool zero_on_init = false;
     void *metadata_buffer = NULL;
     blockstore_init_meta_buf bufs[2] = {};
@@ -25,6 +26,7 @@ class blockstore_init_meta
     uint64_t next_offset = 0;
     uint64_t last_read_offset = 0;
     uint64_t entries_loaded = 0;
+    std::vector<uint32_t> recheck_mod;
     int i = 0, j = 0;
     bool handle_meta_block(uint8_t *buf, uint64_t count, uint64_t done_cnt);
     void handle_event(ring_data_t *data, int buf_num);

src/test/test_heap.cpp

@@ -579,6 +579,10 @@ void test_recheck(bool async, bool csum, bool intent)
 
         heap.finish_load();
 
+        auto mod = heap.get_recheck_modified_blocks();
+        assert(mod.size() == 1);
+        assert(mod[0] == 0);
+
         // read object 1 - big_write should be there but small_write should be rechecked and removed
         object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
         heap_entry_t *obj = heap.read_entry(oid);