From ec3bf4ae6ce80103f619be11e7773da62f07e361 Mon Sep 17 00:00:00 2001
From: Vitaliy Filippov <vitalif@yourcmc.ru>
Date: Tue, 8 Jul 2025 02:20:08 +0300
Subject: [PATCH] Correctly track fsynced_lsn when fsyncs are enabled

---
 src/blockstore/blockstore_disk.cpp   | 11 ++++
 src/blockstore/blockstore_disk.h     |  2 +
 src/blockstore/blockstore_flush.cpp  | 80 +++++++++++++++++++++++++---
 src/blockstore/blockstore_flush.h    |  3 ++
 src/blockstore/blockstore_heap.cpp   | 18 +++++++
 src/blockstore/blockstore_heap.h     |  3 ++
 src/blockstore/blockstore_impl.h     |  4 +-
 src/blockstore/blockstore_init.cpp   |  2 +-
 src/blockstore/blockstore_open.cpp   | 24 +--------
 src/blockstore/blockstore_stable.cpp |  2 +-
 src/blockstore/blockstore_sync.cpp   | 29 ++++++----
 src/blockstore/blockstore_write.cpp  | 11 ++--
 12 files changed, 140 insertions(+), 49 deletions(-)

diff --git a/src/blockstore/blockstore_disk.cpp b/src/blockstore/blockstore_disk.cpp
index 57dea67e..a4b6c1c4 100644
--- a/src/blockstore/blockstore_disk.cpp
+++ b/src/blockstore/blockstore_disk.cpp
@@ -97,6 +97,9 @@ void blockstore_disk_t::parse_config(std::map<std::string, std::string> & config
         config["inmemory_metadata"] != "no";
     inmemory_journal = config["inmemory_journal"] != "false" && config["inmemory_journal"] != "0" &&
         config["inmemory_journal"] != "no";
+    disable_data_fsync = config["disable_data_fsync"] == "true" || config["disable_data_fsync"] == "1" || config["disable_data_fsync"] == "yes";
+    disable_meta_fsync = config["disable_meta_fsync"] == "true" || config["disable_meta_fsync"] == "1" || config["disable_meta_fsync"] == "yes";
+    disable_journal_fsync = config["disable_journal_fsync"] == "true" || config["disable_journal_fsync"] == "1" || config["disable_journal_fsync"] == "yes";
     // Validate
     if (!data_block_size)
     {
@@ -194,6 +197,14 @@ 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));
     }
+    if (meta_device == data_device)
+    {
+        disable_meta_fsync = disable_data_fsync;
+    }
+    if (journal_device == meta_device)
+    {
+        disable_journal_fsync = disable_meta_fsync;
+    }
 }
 
 void blockstore_disk_t::calc_lengths(bool skip_meta_check)
diff --git a/src/blockstore/blockstore_disk.h b/src/blockstore/blockstore_disk.h
index dbfc74de..b10a63e0 100644
--- a/src/blockstore/blockstore_disk.h
+++ b/src/blockstore/blockstore_disk.h
@@ -43,6 +43,8 @@ struct blockstore_disk_t
     // I/O modes for data, metadata and journal: direct or "" = O_DIRECT, cached = O_SYNC, directsync = O_DIRECT|O_SYNC
     // O_SYNC without O_DIRECT = use Linux page cache for reads and writes
     std::string data_io, meta_io, journal_io;
+    // It is safe to disable fsync() if drive write cache is writethrough
+    bool disable_data_fsync = false, disable_meta_fsync = false, disable_journal_fsync = false;
     // Keep journal (buffered data) in memory?
     bool inmemory_meta = true;
     // Keep metadata in memory?
diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp
index 548589e1..5496671f 100644
--- a/src/blockstore/blockstore_flush.cpp
+++ b/src/blockstore/blockstore_flush.cpp
@@ -68,6 +68,11 @@ int journal_flusher_t::get_active()
     return active_flushers;
 }
 
+int journal_flusher_t::get_syncing_buffer()
+{
+    return syncing_buffer;
+}
+
 uint64_t journal_flusher_t::get_compact_counter()
 {
     return compact_counter;
@@ -75,7 +80,7 @@ uint64_t journal_flusher_t::get_compact_counter()
 
 bool journal_flusher_t::is_active()
 {
-    return active_flushers > 0 || force_start > 0 || bs->heap->get_compact_queue_size() > bs->flusher_start_threshold;
+    return active_flushers > 0;
 }
 
 void journal_flusher_t::request_trim()
@@ -119,7 +124,7 @@ void journal_flusher_t::loop()
         }
     }
     int prev_active = active_flushers;
-    for (int i = 0; is_active() && i < cur_flusher_count; i++)
+    for (int i = 0; (active_flushers > 0 || force_start > 0 || bs->heap->get_compact_queue_size() > bs->flusher_start_threshold) && i < cur_flusher_count; i++)
         co[i].loop();
     if (prev_active && !active_flushers && force_start > 0)
         bs->ringloop->wakeup();
@@ -161,9 +166,26 @@ bool journal_flusher_co::loop()
     else if (wait_state == 18) goto resume_18;
     else if (wait_state == 19) goto resume_19;
     else if (wait_state == 20) goto resume_20;
+    else if (wait_state == 21) goto resume_21;
+    else if (wait_state == 22) goto resume_22;
+    else if (wait_state == 23) goto resume_23;
 resume_0:
+    wait_state = 0;
     cur_oid = {};
     res = bs->heap->get_next_compact(cur_oid);
+    if (res == ENOENT && flusher->force_start > 0 && co_id == 0 &&
+        (!bs->dsk.disable_journal_fsync || !bs->dsk.disable_meta_fsync))
+    {
+resume_21:
+resume_22:
+resume_23:
+        res = fsync_buffer(21);
+        if (!res)
+        {
+            return false;
+        }
+        res = (res == 2 ? bs->heap->get_next_compact(cur_oid) : ENOENT);
+    }
     if (res == ENOENT)
     {
         if (co_id == 0 && flusher->force_start > 0)
@@ -323,7 +345,6 @@ release_oid:
         goto resume_1;
     }
     // All done
-    wait_state = 0;
     goto resume_0;
 }
 
@@ -619,7 +640,7 @@ bool journal_flusher_co::fsync_batch(bool fsync_meta, int wait_base)
     if (wait_state == wait_base)        goto resume_0;
     else if (wait_state == wait_base+1) goto resume_1;
     else if (wait_state == wait_base+2) goto resume_2;
-    if (!(fsync_meta ? bs->disable_meta_fsync : bs->disable_data_fsync))
+    if (!(fsync_meta ? bs->dsk.disable_meta_fsync : bs->dsk.disable_data_fsync))
     {
         cur_sync = flusher->syncs.end();
         while (cur_sync != flusher->syncs.begin())
@@ -677,6 +698,53 @@ bool journal_flusher_co::fsync_batch(bool fsync_meta, int wait_base)
     return true;
 }
 
+int journal_flusher_co::fsync_buffer(int wait_base)
+{
+    if (wait_state == wait_base)        goto resume_0;
+    else if (wait_state == wait_base+1) goto resume_1;
+    else if (wait_state == wait_base+2) goto resume_2;
+    if (!bs->unsynced_big_write_count && !bs->unsynced_small_write_count)
+    {
+        return 1;
+    }
+    if (flusher->syncing_buffer)
+    {
+        return 0;
+    }
+    compact_lsn = bs->heap->get_completed_lsn();
+    flusher->active_flushers++;
+    flusher->syncing_buffer++;
+    assert(!wait_count);
+    if (!bs->dsk.disable_meta_fsync)
+    {
+        bs->unsynced_big_write_count = 0;
+        await_sqe(0);
+        data->iov = { 0 };
+        data->callback = simple_callback_w;
+        io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC);
+        wait_count++;
+    }
+    if (bs->unsynced_small_write_count > 0 && !bs->dsk.disable_journal_fsync && bs->dsk.journal_fd != bs->dsk.meta_fd)
+    {
+        bs->unsynced_small_write_count = 0;
+        await_sqe(1);
+        data->iov = { 0 };
+        data->callback = simple_callback_w;
+        io_uring_prep_fsync(sqe, bs->dsk.journal_fd, IORING_FSYNC_DATASYNC);
+        wait_count++;
+    }
+resume_2:
+    if (wait_count > 0)
+    {
+        wait_state = wait_base+2;
+        return 0;
+    }
+    bs->heap->mark_lsn_fsynced(compact_lsn);
+    flusher->active_flushers--;
+    flusher->syncing_buffer--;
+    return 2;
+}
+
 bool journal_flusher_co::trim_lsn(int wait_base)
 {
     if (wait_state == wait_base)        goto resume_0;
@@ -691,7 +759,7 @@ bool journal_flusher_co::trim_lsn(int wait_base)
     }
     flusher->active_flushers++;
     assert(!wait_count);
-    if (!bs->disable_meta_fsync)
+    if (!bs->dsk.disable_meta_fsync)
     {
         await_sqe(0);
         data->iov = { 0 };
@@ -699,7 +767,7 @@ bool journal_flusher_co::trim_lsn(int wait_base)
         io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC);
         wait_count++;
     }
-    if (!bs->disable_data_fsync && bs->dsk.data_fd != bs->dsk.meta_fd)
+    if (!bs->dsk.disable_data_fsync && bs->dsk.data_fd != bs->dsk.meta_fd)
     {
         await_sqe(1);
         data->iov = { 0 };
diff --git a/src/blockstore/blockstore_flush.h b/src/blockstore/blockstore_flush.h
index cee23b5d..5e0fbe7c 100644
--- a/src/blockstore/blockstore_flush.h
+++ b/src/blockstore/blockstore_flush.h
@@ -79,6 +79,7 @@ class journal_flusher_co
     bool write_meta_block(int wait_base);
     bool read_buffered(int wait_base);
     bool fsync_batch(bool fsync_meta, int wait_base);
+    int fsync_buffer(int wait_base);
     bool trim_lsn(int wait_base);
 public:
     journal_flusher_co();
@@ -100,6 +101,7 @@ class journal_flusher_t
 
     int active_flushers = 0;
     int syncing_flushers = 0;
+    int syncing_buffer = 0;
     std::list<flusher_sync_t> syncs;
 
 public:
@@ -107,6 +109,7 @@ public:
     ~journal_flusher_t();
     void loop();
     int get_active();
+    int get_syncing_buffer();
     uint64_t get_compact_counter();
     bool is_active();
     void request_trim();
diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp
index 51c96b05..80f74140 100644
--- a/src/blockstore/blockstore_heap.cpp
+++ b/src/blockstore/blockstore_heap.cpp
@@ -1506,6 +1506,10 @@ int blockstore_heap_t::get_next_compact(object_id & oid)
     }
     while (next_compact_lsn-first_inflight_lsn < inflight_lsn.size())
     {
+        if (next_compact_lsn > (dsk->disable_meta_fsync && dsk->disable_journal_fsync ? completed_lsn : fsynced_lsn))
+        {
+            return ENOENT;
+        }
         auto & item = inflight_lsn[next_compact_lsn-first_inflight_lsn];
         if (!(item.flags & HEAP_INFLIGHT_COMPACTABLE))
         {
@@ -1895,6 +1899,15 @@ void blockstore_heap_t::mark_lsn_completed(uint64_t lsn)
     }
 }
 
+void blockstore_heap_t::mark_lsn_fsynced(uint64_t lsn)
+{
+    if (lsn > fsynced_lsn)
+    {
+        assert(lsn >= first_inflight_lsn && lsn <= completed_lsn);
+        fsynced_lsn = lsn;
+    }
+}
+
 void blockstore_heap_t::mark_lsn_compacted(uint64_t lsn)
 {
     assert(lsn >= first_inflight_lsn && lsn < first_inflight_lsn+inflight_lsn.size());
@@ -1943,3 +1956,8 @@ uint64_t blockstore_heap_t::get_completed_lsn()
 {
     return completed_lsn;
 }
+
+uint64_t blockstore_heap_t::get_fsynced_lsn()
+{
+    return dsk->disable_meta_fsync && dsk->disable_journal_fsync ? completed_lsn : fsynced_lsn;
+}
diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h
index eb874d92..cc8113b3 100644
--- a/src/blockstore/blockstore_heap.h
+++ b/src/blockstore/blockstore_heap.h
@@ -147,6 +147,7 @@ class blockstore_heap_t
     uint32_t to_compact_count = 0;
     uint64_t first_inflight_lsn = 0;
     uint64_t completed_lsn = 0;
+    uint64_t fsynced_lsn = 0;
     uint64_t compacted_lsn = 0;
     uint64_t next_compact_lsn = 0;
 
@@ -236,10 +237,12 @@ public:
 
     // inflight write tracking
     void mark_lsn_completed(uint64_t lsn);
+    void mark_lsn_fsynced(uint64_t lsn);
     void mark_lsn_compacted(uint64_t lsn);
     void mark_object_compacted(heap_object_t *obj, uint64_t max_lsn);
     void mark_lsn_trimmed(uint64_t lsn);
     uint64_t get_completed_lsn();
+    uint64_t get_fsynced_lsn();
 
     // data device block allocator functions
     uint64_t find_free_data();
diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h
index 3092cff0..f54a6aa4 100644
--- a/src/blockstore/blockstore_impl.h
+++ b/src/blockstore/blockstore_impl.h
@@ -53,7 +53,7 @@ struct blockstore_op_private_t
     int pending_ops;
     int op_state;
 
-    // Read, write, stabilize
+    // Read, write, sync, stabilize
     uint64_t lsn;
 
     // Read
@@ -80,8 +80,6 @@ class blockstore_impl_t: public blockstore_i
 
     /******* OPTIONS *******/
     bool readonly = false;
-    // It is safe to disable fsync() if drive write cache is writethrough
-    bool disable_data_fsync = false, disable_meta_fsync = false, disable_journal_fsync = false;
     // Enable if you want every operation to be executed with an "implicit fsync"
     // Suitable only for server SSDs with capacitors, requires disabled data and journal fsyncs
     int immediate_commit = IMMEDIATE_NONE;
diff --git a/src/blockstore/blockstore_init.cpp b/src/blockstore/blockstore_init.cpp
index 2c9dadf7..2bc2787e 100644
--- a/src/blockstore/blockstore_init.cpp
+++ b/src/blockstore/blockstore_init.cpp
@@ -235,7 +235,7 @@ resume_4:
     // metadata read finished
     bs->heap->finish_load();
     printf("Metadata entries loaded: %ju, used blocks: %ju / %ju\n", entries_loaded, bs->heap->get_data_used_space() / bs->dsk.data_block_size, bs->dsk.block_count);
-    if (zero_on_init && !bs->disable_meta_fsync)
+    if (zero_on_init && !bs->dsk.disable_meta_fsync)
     {
         GET_SQE();
         io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC);
diff --git a/src/blockstore/blockstore_open.cpp b/src/blockstore/blockstore_open.cpp
index 1fc8dae5..97031732 100644
--- a/src/blockstore/blockstore_open.cpp
+++ b/src/blockstore/blockstore_open.cpp
@@ -78,18 +78,6 @@ void blockstore_impl_t::parse_config(blockstore_config_t & config, bool init)
     {
         readonly = true;
     }
-    if (config["disable_data_fsync"] == "true" || config["disable_data_fsync"] == "1" || config["disable_data_fsync"] == "yes")
-    {
-        disable_data_fsync = true;
-    }
-    if (config["disable_meta_fsync"] == "true" || config["disable_meta_fsync"] == "1" || config["disable_meta_fsync"] == "yes")
-    {
-        disable_meta_fsync = true;
-    }
-    if (config["disable_journal_fsync"] == "true" || config["disable_journal_fsync"] == "1" || config["disable_journal_fsync"] == "yes")
-    {
-        disable_journal_fsync = true;
-    }
     if (config["immediate_commit"] == "all")
     {
         immediate_commit = IMMEDIATE_ALL;
@@ -110,19 +98,11 @@ void blockstore_impl_t::parse_config(blockstore_config_t & config, bool init)
     {
         meta_write_recheck_parallelism = 16;
     }
-    if (dsk.meta_device == dsk.data_device)
-    {
-        disable_meta_fsync = disable_data_fsync;
-    }
-    if (dsk.journal_device == dsk.meta_device)
-    {
-        disable_journal_fsync = disable_meta_fsync;
-    }
-    if (immediate_commit != IMMEDIATE_NONE && !disable_journal_fsync)
+    if (immediate_commit != IMMEDIATE_NONE && !dsk.disable_journal_fsync)
     {
         throw std::runtime_error("immediate_commit requires disable_journal_fsync");
     }
-    if (immediate_commit == IMMEDIATE_ALL && !disable_data_fsync)
+    if (immediate_commit == IMMEDIATE_ALL && !dsk.disable_data_fsync)
     {
         throw std::runtime_error("immediate_commit=all requires disable_journal_fsync and disable_data_fsync");
     }
diff --git a/src/blockstore/blockstore_stable.cpp b/src/blockstore/blockstore_stable.cpp
index 0d506b6b..eb4d1e5b 100644
--- a/src/blockstore/blockstore_stable.cpp
+++ b/src/blockstore/blockstore_stable.cpp
@@ -67,7 +67,7 @@ resume_1:
         return 0;
     }
 resume_2:
-    if (!disable_meta_fsync)
+    if (!dsk.disable_meta_fsync)
     {
         BS_SUBMIT_GET_SQE(sqe, data);
         io_uring_prep_fsync(sqe, dsk.meta_fd, IORING_FSYNC_DATASYNC);
diff --git a/src/blockstore/blockstore_sync.cpp b/src/blockstore/blockstore_sync.cpp
index ee57eb45..e293c0e2 100644
--- a/src/blockstore/blockstore_sync.cpp
+++ b/src/blockstore/blockstore_sync.cpp
@@ -10,7 +10,12 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
     if (op_state == 1) goto resume_1;
     if (op_state == 2) goto resume_2;
     assert(!op_state);
-    if (immediate_commit == IMMEDIATE_ALL || !unsynced_big_write_count && !unsynced_small_write_count)
+    if (flusher->get_syncing_buffer())
+    {
+        // Wait for flusher-initiated sync
+        return 0;
+    }
+    if (dsk.disable_journal_fsync && dsk.disable_meta_fsync || !unsynced_big_write_count && !unsynced_small_write_count)
     {
         // We can return immediately because sync is only dequeued after all previous writes
         unsynced_big_write_count = unsynced_small_write_count = 0;
@@ -18,17 +23,9 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
         FINISH_OP(op);
         return 2;
     }
+    PRIV(op)->lsn = heap->get_completed_lsn();
     stop_sync_submitted = false;
-    if (unsynced_small_write_count > 0 && !disable_journal_fsync)
-    {
-        // fsync buffer
-        BS_SUBMIT_GET_SQE(sqe, data);
-        io_uring_prep_fsync(sqe, dsk.journal_fd, IORING_FSYNC_DATASYNC);
-        data->iov = { 0 };
-        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
-        PRIV(op)->pending_ops++;
-    }
-    if (!disable_meta_fsync && dsk.meta_fd != dsk.journal_fd)
+    if (!dsk.disable_meta_fsync)
     {
         // fsync meta
         BS_SUBMIT_GET_SQE(sqe, data);
@@ -37,6 +34,15 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
         data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
         PRIV(op)->pending_ops++;
     }
+    if (unsynced_small_write_count > 0 && !dsk.disable_journal_fsync && dsk.meta_fd != dsk.journal_fd)
+    {
+        // fsync buffer
+        BS_SUBMIT_GET_SQE(sqe, data);
+        io_uring_prep_fsync(sqe, dsk.journal_fd, IORING_FSYNC_DATASYNC);
+        data->iov = { 0 };
+        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
+        PRIV(op)->pending_ops++;
+    }
     unsynced_big_write_count = 0;
     unsynced_small_write_count = 0;
 resume_1:
@@ -46,6 +52,7 @@ resume_1:
         return 1;
     }
 resume_2:
+    heap->mark_lsn_fsynced(PRIV(op)->lsn);
     op->retval = 0;
     FINISH_OP(op);
     return 2;
diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp
index 737d156c..3fa3f6b4 100644
--- a/src/blockstore/blockstore_write.cpp
+++ b/src/blockstore/blockstore_write.cpp
@@ -142,14 +142,13 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             sqe, dsk.data_fd, PRIV(op)->iov_zerofill, vcnt, dsk.data_offset + loc + op->offset - stripe_offset
         );
         PRIV(op)->pending_ops++;
-        unsynced_big_write_count++;
         PRIV(op)->op_state = 1;
         write_iodepth++;
         inflight_big++;
     }
     // Only one INTENT_WRITE is allowed at a time, but in fact,
     // parallel writes to the same object are forbidden anyway
-    else if (disable_data_fsync &&
+    else if (dsk.disable_data_fsync &&
         op->opcode == BS_OP_WRITE_STABLE &&
         op->len > 0 && op->len <= dsk.bitmap_granularity /* FIXME atomic_write_size */ &&
         (obj->get_writes()->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) ||
@@ -193,7 +192,6 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         assert(res == 0);
         PRIV(op)->lsn = wr->lsn;
         prepare_meta_block_write(op, modified_block);
-        unsynced_small_write_count++;
         PRIV(op)->op_state = 9;
         write_iodepth++;
     }
@@ -256,7 +254,6 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         {
             // Zero-length overwrite. Allowed to bump object version in EC placement groups without actually writing data
         }
-        unsynced_small_write_count++;
         assert(PRIV(op)->pending_ops);
         PRIV(op)->op_state = 5;
         write_iodepth++;
@@ -291,7 +288,7 @@ resume_2:
     // It's OK for all HDDs and for server SSDs, but slightly worse for desktop SSDs
     // The other way is to add another type of MVCC to blockstore_heap: "forward" MVCC :)
     inflight_big--;
-    if (!disable_data_fsync)
+    if (!dsk.disable_data_fsync)
     {
         // fsync data in a batch
 resume_11:
@@ -388,6 +385,10 @@ resume_8:
 #endif
     op->retval = op->len;
     heap->mark_lsn_completed(PRIV(op)->lsn);
+    if (PRIV(op)->is_big)
+        unsynced_big_write_count++;
+    else
+        unsynced_small_write_count++;
     write_iodepth--;
     FINISH_OP(op);
     return 2;