Fix submitting forced fsyncs in flusher

src/blockstore/blockstore_flush.cpp

@@ -186,7 +186,7 @@ resume_18:
             return false;
     }
     if (res == ENOENT && flusher->force_start > 0 && co_id == 0 &&
-        (!bs->dsk.disable_journal_fsync || !bs->dsk.disable_meta_fsync))
+        (!bs->dsk.disable_journal_fsync || !bs->dsk.disable_meta_fsync || !bs->dsk.disable_data_fsync))
     {
         flusher->active_flushers++;
 resume_14:
@@ -723,8 +723,7 @@ bool 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->dsk.disable_journal_fsync && bs->dsk.disable_meta_fsync && bs->dsk.disable_data_fsync ||
+    if (!bs->has_unsynced())
-        !bs->unsynced_data_write_count && !bs->unsynced_small_write_count && !bs->unsynced_meta_write_count)
     {
         return true;
     }

src/blockstore/blockstore_impl.cpp

@@ -222,7 +222,7 @@ bool blockstore_impl_t::is_safe_to_stop()
     {
         return false;
     }
-    if (unsynced_data_write_count > 0 || unsynced_small_write_count > 0 || unsynced_meta_write_count > 0)
+    if (has_unsynced())
     {
         if (!readonly && !stop_sync_submitted)
         {

src/blockstore/blockstore_impl.h

@@ -106,7 +106,7 @@ public:
     uint8_t* meta_superblock = NULL;
     uint8_t *buffer_area = NULL;
     std::vector<blockstore_op_t*> submit_queue;
-    int unsynced_data_write_count = 0, unsynced_small_write_count = 0, unsynced_meta_write_count = 0;
+    int unsynced_data_write_count = 0, unsynced_buffer_write_count = 0, unsynced_meta_write_count = 0;
     int unsynced_queued_ops = 0;
     uint8_t *zero_object = NULL;
 
@@ -173,6 +173,7 @@ public:
     int continue_sync(blockstore_op_t *op);
     bool submit_fsyncs(int & wait_count);
     int do_sync(blockstore_op_t *op, int base_state);
+    bool has_unsynced();
 
     // Stabilize
     int dequeue_stable(blockstore_op_t *op);

src/blockstore/blockstore_stable.cpp

@@ -76,7 +76,6 @@ resume_2:
             goto resume_1;
         }
     }
-    unsynced_meta_write_count++;
     // Fsync, just because our semantics imply that commit (stabilize) is immediately fsynced
     priv->op_state = 3;
 resume_3:

src/blockstore/blockstore_sync.cpp

@@ -18,10 +18,18 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
     return res;
 }
 
+bool blockstore_impl_t::has_unsynced()
+{
+    bool data = (!dsk.disable_data_fsync && unsynced_data_write_count);
+    bool buffer = (!dsk.disable_journal_fsync && unsynced_buffer_write_count);
+    bool meta = (!dsk.disable_meta_fsync && unsynced_meta_write_count);
+    return data || buffer || meta;
+}
+
 bool blockstore_impl_t::submit_fsyncs(int & wait_count)
 {
-    int n = ((unsynced_small_write_count > 0 || unsynced_data_write_count > 0 || unsynced_meta_write_count > 0) && !dsk.disable_meta_fsync) +
+    int n = (unsynced_meta_write_count > 0 && !dsk.disable_meta_fsync) +
-        (unsynced_small_write_count > 0 && !dsk.disable_journal_fsync && dsk.journal_fd != dsk.meta_fd) +
+        (unsynced_buffer_write_count > 0 && !dsk.disable_journal_fsync && dsk.journal_fd != dsk.meta_fd) +
         (unsynced_data_write_count > 0 && !dsk.disable_data_fsync && dsk.data_fd != dsk.meta_fd && dsk.data_fd != dsk.journal_fd);
     if (ringloop->space_left() < n)
     {
@@ -40,7 +48,7 @@ bool blockstore_impl_t::submit_fsyncs(int & wait_count)
         if (!wait_count)
             ringloop->wakeup();
     };
-    if ((unsynced_small_write_count > 0 || unsynced_data_write_count > 0 || unsynced_meta_write_count > 0) && !dsk.disable_meta_fsync)
+    if (unsynced_meta_write_count > 0 && !dsk.disable_meta_fsync)
     {
         // fsync meta
         io_uring_sqe *sqe = get_sqe();
@@ -51,7 +59,7 @@ bool blockstore_impl_t::submit_fsyncs(int & wait_count)
         data->callback = cb;
         wait_count++;
     }
-    if (unsynced_small_write_count > 0 && !dsk.disable_journal_fsync && dsk.meta_fd != dsk.journal_fd)
+    if (unsynced_buffer_write_count > 0 && !dsk.disable_journal_fsync && dsk.meta_fd != dsk.journal_fd)
     {
         // fsync buffer
         io_uring_sqe *sqe = get_sqe();
@@ -74,7 +82,7 @@ bool blockstore_impl_t::submit_fsyncs(int & wait_count)
         wait_count++;
     }
     unsynced_data_write_count = 0;
-    unsynced_small_write_count = 0;
+    unsynced_buffer_write_count = 0;
     unsynced_meta_write_count = 0;
     return true;
 }
@@ -90,11 +98,10 @@ int blockstore_impl_t::do_sync(blockstore_op_t *op, int base_state)
         // Wait for flusher-initiated sync
         return 0;
     }
-    if (dsk.disable_journal_fsync && dsk.disable_meta_fsync && dsk.disable_data_fsync ||
+    if (!has_unsynced())
-        !unsynced_data_write_count && !unsynced_small_write_count && !unsynced_meta_write_count)
     {
         // We can return immediately because sync only syncs previous writes
-        unsynced_data_write_count = unsynced_small_write_count = unsynced_meta_write_count = 0;
+        unsynced_data_write_count = unsynced_buffer_write_count = unsynced_meta_write_count = 0;
         return 2;
     }
     PRIV(op)->modified_block = heap->get_completed_lsn();

src/blockstore/blockstore_write.cpp

@@ -36,6 +36,7 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
     io_uring_prep_writev(
         sqe, dsk.meta_fd, &data->iov, 1, dsk.meta_offset + (modified_block+1)*dsk.meta_block_size
     );
+    unsynced_meta_write_count++;
     pending_modified_blocks.push_back(modified_block);
     modified_blocks[modified_block] = { .sent = false, .buf = buf };
 }
@@ -421,10 +422,8 @@ resume_8:
         PRIV(op)->write_type == BS_HEAP_BIG_INTENT ||
         PRIV(op)->write_type == BS_HEAP_INTENT_WRITE)
         unsynced_data_write_count++;
-    else if (PRIV(op)->write_type == BS_HEAP_DELETE)
+    else if (PRIV(op)->write_type != BS_HEAP_DELETE)
-        unsynced_meta_write_count++;
+        unsynced_buffer_write_count++;
-    else
-        unsynced_small_write_count++;
     write_iodepth--;
     FINISH_OP(op);
     return 2;