From 3cb7ec69bc3bf4b8f4f85afc28665cd0a2b8a0a2 Mon Sep 17 00:00:00 2001
From: Vitaliy Filippov <vitalif@yourcmc.ru>
Date: Sun, 29 Jun 2025 00:46:07 +0300
Subject: [PATCH] Remove sync_to_repeat map and use simpler repeating

---
 src/blockstore/blockstore_flush.cpp | 35 +++++++++++------------------
 src/blockstore/blockstore_flush.h   |  3 ++-
 2 files changed, 15 insertions(+), 23 deletions(-)

diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp
index 245cdd4c..205dd8d3 100644
--- a/src/blockstore/blockstore_flush.cpp
+++ b/src/blockstore/blockstore_flush.cpp
@@ -23,6 +23,7 @@ journal_flusher_t::journal_flusher_t(blockstore_impl_t *bs)
     co = new journal_flusher_co[max_flusher_count];
     for (int i = 0; i < max_flusher_count; i++)
     {
+        co[i].co_id = i;
         co[i].bs = bs;
         co[i].flusher = this;
         if (bs->dsk.csum_block_size)
@@ -161,10 +162,21 @@ resume_0:
     res = bs->heap->get_next_compact(cur_oid);
     if (res == ENOENT)
     {
+        cur_oid = {};
         wait_state = 0;
         return true;
     }
+    for (int i = 0; i < flusher->cur_flusher_count; i++)
+    {
+        if (i != co_id && flusher->co[i].cur_oid == cur_oid)
+        {
+            // Already flushing it
+            flusher->co[i].should_repeat = true;
+            goto resume_0;
+        }
+    }
 resume_1:
+    should_repeat = false;
     cur_obj = bs->heap->lock_and_read_entry(cur_oid, cur_lsn);
     if (!cur_obj)
     {
@@ -183,24 +195,6 @@ resume_1:
     compact_lsn = begin_wr->lsn;
     assert(!end_wr->next() && end_wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE));
     clean_loc = end_wr->location;
-    // "Lock" object for flushing
-    repeat_it = flusher->sync_to_repeat.find(cur_oid);
-    if (repeat_it != flusher->sync_to_repeat.end())
-    {
-#ifdef BLOCKSTORE_DEBUG
-        printf("Postpone %jx:%jx v%ju\n", cur_oid.inode, cur_oid.stripe, cur_version);
-#endif
-        // We don't flush different parts of history of the same object in parallel
-        // So we check if someone is already flushing this object
-        // In that case we set sync_to_repeat and pick another object
-        // Another coroutine will see it and re-queue the object after it finishes
-        if (repeat_it->second < cur_version)
-            repeat_it->second = cur_version;
-        bs->heap->unlock_entry(cur_oid, cur_lsn);
-        goto resume_0;
-    }
-    else
-        flusher->sync_to_repeat[cur_oid] = 0;
 #ifdef BLOCKSTORE_DEBUG
     printf("Flushing %jx:%jx v%ju .. v%ju\n", cur_oid.inode, cur_oid.stripe, end_wr->version, begin_wr->version);
 #endif
@@ -304,11 +298,8 @@ resume_17:
             return false;
     }
 release_oid:
-    repeat_it = flusher->sync_to_repeat.find(cur_oid);
-    do_repeat = (repeat_it != flusher->sync_to_repeat.end() && repeat_it->second > cur_version);
-    flusher->sync_to_repeat.erase(repeat_it);
     flusher->active_flushers--;
-    if (do_repeat)
+    if (should_repeat)
     {
         // Flush the same object again
         goto resume_1;
diff --git a/src/blockstore/blockstore_flush.h b/src/blockstore/blockstore_flush.h
index 7a605e19..aec9245a 100644
--- a/src/blockstore/blockstore_flush.h
+++ b/src/blockstore/blockstore_flush.h
@@ -39,6 +39,7 @@ class journal_flusher_co
 {
     blockstore_impl_t *bs;
     journal_flusher_t *flusher;
+    int co_id;
     int wait_state, wait_count;
     struct io_uring_sqe *sqe;
     struct ring_data_t *data;
@@ -55,6 +56,7 @@ class journal_flusher_co
     heap_object_t *cur_obj;
     heap_write_t *begin_wr, *end_wr;
     uint32_t modified_block;
+    bool should_repeat;
 
     std::vector<copy_buffer_t> read_vec;
     uint32_t overwrite_start, overwrite_end;
@@ -100,7 +102,6 @@ class journal_flusher_t
     int active_flushers = 0;
     int syncing_flushers = 0;
     std::list<flusher_sync_t> syncs;
-    std::map<object_id, uint64_t> sync_to_repeat;
 
 public:
     journal_flusher_t(blockstore_impl_t *bs);