Save completed_lsn in the superblock

src/blockstore/blockstore_flush.cpp

@@ -18,7 +18,6 @@ journal_flusher_t::journal_flusher_t(blockstore_impl_t *bs)
     this->cur_flusher_count = bs->min_flusher_count;
     this->target_flusher_count = bs->min_flusher_count;
     active_flushers = 0;
-    advance_lsn_counter = 0;
     co = new journal_flusher_co[max_flusher_count];
     for (int i = 0; i < max_flusher_count; i++)
     {
@@ -132,8 +131,12 @@ void journal_flusher_t::loop()
         }
     }
     int prev_active = active_flushers;
-    for (int i = 0; (active_flushers > 0 || force_start > 0 || bs->heap->get_to_compact_count() > bs->flusher_start_threshold) && i < cur_flusher_count; i++)
+    for (int i = 0; (active_flushers > 0 || force_start > 0 ||
+        bs->heap->get_to_compact_count() > bs->flusher_start_threshold ||
+        i == 0 && bs->intent_write_counter >= bs->journal_trim_interval) && i < cur_flusher_count; i++)
+    {
         co[i].loop();
+    }
     if (prev_active && !active_flushers && force_start > 0)
         bs->ringloop->wakeup();
 }
@@ -170,11 +173,22 @@ bool journal_flusher_co::loop()
     else if (wait_state == 14) goto resume_14;
     else if (wait_state == 15) goto resume_15;
     else if (wait_state == 16) goto resume_16;
+    else if (wait_state == 17) goto resume_17;
+    else if (wait_state == 18) goto resume_18;
 resume_0:
     wait_state = 0;
     wait_count = 0;
     cur_oid = {};
     res = bs->heap->get_next_compact(cur_oid);
+    // Advance fsynced_lsn every <journal_trim_interval> intent writes
+    if ((bs->intent_write_counter >= bs->journal_trim_interval) && co_id == 0)
+    {
+        bs->intent_write_counter = 0;
+resume_17:
+resume_18:
+        if (!trim_lsn(17))
+            return false;
+    }
     if (res == ENOENT && flusher->force_start > 0 && co_id == 0 &&
         (!bs->dsk.disable_journal_fsync || !bs->dsk.disable_meta_fsync))
     {
@@ -740,3 +754,33 @@ resume_2:
     flusher->syncing_buffer--;
     return true;
 }
+
+bool journal_flusher_co::trim_lsn(int wait_base)
+{
+    if (wait_state == wait_base)        goto resume_0;
+    else if (wait_state == wait_base+1) goto resume_1;
+    fsynced_lsn = bs->heap->get_fsynced_lsn();
+    if (((blockstore_meta_header_v3_t*)bs->meta_superblock)->completed_lsn == fsynced_lsn)
+    {
+        return true;
+    }
+    flusher->active_flushers++;
+    ((blockstore_meta_header_v3_t*)bs->meta_superblock)->completed_lsn = fsynced_lsn;
+    ((blockstore_meta_header_v3_t*)bs->meta_superblock)->set_crc32c();
+    await_sqe(0);
+    data->iov = (struct iovec){ bs->meta_superblock, (size_t)bs->dsk.meta_block_size };
+    data->callback = simple_callback_w;
+    io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset);
+    // Update superblock with datasync
+    sqe->rw_flags = RWF_DSYNC;
+    wait_count++;
+resume_1:
+    if (wait_count > 0)
+    {
+        wait_state = wait_base+1;
+        return false;
+    }
+    flusher->compact_counter++;
+    flusher->active_flushers--;
+    return true;
+}

src/blockstore/blockstore_flush.h

@@ -71,6 +71,7 @@ class journal_flusher_co
     bool read_buffered(int wait_base);
     bool fsync_meta(int wait_base);
     bool fsync_buffer(int wait_base);
+    bool trim_lsn(int wait_base);
 public:
     journal_flusher_co();
     ~journal_flusher_co();
@@ -86,7 +87,6 @@ class journal_flusher_t
     blockstore_impl_t *bs;
     friend class journal_flusher_co;
 
-    int advance_lsn_counter = 0;
     uint64_t compact_counter = 0;
 
     robin_hood::unordered_flat_set<object_id> flushing;

src/blockstore/blockstore_heap.cpp

@@ -255,6 +255,11 @@ blockstore_heap_t::~blockstore_heap_t()
     delete buffer_alloc;
 }
 
+void blockstore_heap_t::start_load(uint64_t completed_lsn)
+{
+    this->completed_lsn = completed_lsn;
+}
+
 int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uint8_t *buf,
     std::function<void(uint32_t block_num, heap_entry_t* wr)> handle_write, std::function<void(uint32_t, uint32_t, uint8_t*)> handle_block)
 {
@@ -385,8 +390,11 @@ void blockstore_heap_t::fill_recheck_queue()
                 if (obj->type() == BS_HEAP_INTENT_WRITE || obj->type() == BS_HEAP_BIG_INTENT)
                 {
                     // Recheck only the latest intent_write
-                    // FIXME Save checked_lsn in the superblock
-                    recheck_queue.push_back(obj);
+                    if (obj->lsn > completed_lsn)
+                    {
+                        // Do not recheck if it's already marked as completed in the superblock
+                        recheck_queue.push_back(obj);
+                    }
                 }
                 else
                 {
@@ -1344,7 +1352,6 @@ void blockstore_heap_t::start_block_write(uint32_t block_num)
         assert(!inf.is_writing);
         inf.is_writing = true;
     });
-    // FIXME also get_meta_block
 }
 
 void blockstore_heap_t::complete_block_write(uint32_t block_num)

src/blockstore/blockstore_heap.h

@@ -211,6 +211,7 @@ class blockstore_heap_t
 public:
     blockstore_heap_t(blockstore_disk_t *dsk, uint8_t *buffer_area, int log_level = 0);
     ~blockstore_heap_t();
+    void start_load(uint64_t completed_lsn);
     // load data from the disk, returns EDOM on corruption
     int read_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf,
         std::function<void(uint32_t block_num, heap_entry_t* wr)> handle_write,

src/blockstore/blockstore_impl.h

@@ -29,18 +29,6 @@ class blockstore_impl_t;
 
 //#define BLOCKSTORE_DEBUG
 
-// - Sync must be submitted after previous writes/deletes (not before!)
-// - Reads may be submitted in parallel with writes/deletes because we use MVCC
-// - Writes may be submitted in any order, because they don't overlap. Each write
-//   goes into a new location - either on the journal device or on the data device
-// - Stable (stabilize) must be submitted after sync of that object is completed
-//   It's even OK to return an error to the caller if that object is not synced yet
-// - compacted_lsn should be moved forward only after all versions are moved to the main storage
-// - If an operation can not be submitted because the ring is full
-//   we should stop submission of other operations. Otherwise some "scatter" reads
-//   may end up blocked for a long time.
-// Otherwise, the submission order is free.
-
 #include "blockstore_init.h"
 
 #include "blockstore_flush.h"
@@ -128,6 +116,7 @@ public:
     journal_flusher_t *flusher;
     int write_iodepth = 0;
     int inflight_big = 0;
+    int intent_write_counter = 0;
     bool fsyncing_data = false;
 
     bool live = false, queue_stall = false;

src/blockstore/blockstore_init.cpp

@@ -155,6 +155,7 @@ resume_1:
         bs->dsk.check_lengths();
     }
     bs->init();
+    bs->heap->start_load(((blockstore_meta_header_v3_t *)bs->meta_superblock)->completed_lsn);
     if (bs->dsk.inmemory_journal)
     {
         // Read buffer area

src/blockstore/blockstore_open.cpp

@@ -42,7 +42,7 @@ void blockstore_impl_t::parse_config(blockstore_config_t & config, bool init)
     }
     if (!journal_trim_interval)
     {
-        journal_trim_interval = 1024;
+        journal_trim_interval = 4096;
     }
     if (!flusher_start_threshold)
     {

src/blockstore/blockstore_write.cpp

@@ -237,6 +237,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         }
         assert(res == 0);
         prepare_meta_block_write(PRIV(op)->modified_block);
+        intent_write_counter++;
         PRIV(op)->pending_ops++;
         PRIV(op)->op_state = 9;
         write_iodepth++;

src/blockstore/ondisk_formats.h

@@ -185,7 +185,7 @@ struct __attribute__((__packed__)) blockstore_meta_header_v3_t
     uint32_t data_csum_type;
     uint32_t csum_block_size;
     uint32_t header_csum;
-    uint64_t compacted_lsn;
+    uint64_t completed_lsn;
 
     void set_crc32c();
 };

src/disk_tool/disk_tool_meta.cpp

@@ -275,7 +275,7 @@ int disk_tool_t::dump_meta()
             if (dump_as_old)
             {
                 hdr->version = BLOCKSTORE_META_FORMAT_V2;
-                hdr->compacted_lsn = 0;
+                hdr->completed_lsn = 0;
                 hdr->header_csum = 0;
                 hdr->header_csum = crc32c(0, hdr, sizeof(blockstore_meta_header_v2_t));
             }

src/disk_tool/disk_tool_resize.cpp

@@ -554,7 +554,7 @@ int disk_tool_t::resize_rebuild_meta()
             new_meta_hdr->bitmap_granularity = dsk.bitmap_granularity ? dsk.bitmap_granularity : 4096;
             new_meta_hdr->data_csum_type = dsk.data_csum_type;
             new_meta_hdr->csum_block_size = dsk.csum_block_size;
-            new_meta_hdr->compacted_lsn = hdr->compacted_lsn;
+            new_meta_hdr->completed_lsn = hdr->completed_lsn;
             new_meta_hdr->header_csum = 0;
             new_meta_hdr->header_csum = crc32c(0, new_meta_hdr, new_meta_hdr->version == BLOCKSTORE_META_FORMAT_HEAP
                 ? sizeof(blockstore_meta_header_v3_t) : sizeof(blockstore_meta_header_v2_t));