Fix enospc detection logic and test_enospc

src/blockstore/blockstore_flush.cpp

@@ -80,11 +80,6 @@ int journal_flusher_t::get_syncing_buffer()
     return syncing_buffer;
 }
 
-uint64_t journal_flusher_t::get_compact_counter()
-{
-    return compact_counter;
-}
-
 bool journal_flusher_t::is_active()
 {
     return active_flushers > 0;
@@ -104,7 +99,7 @@ void journal_flusher_t::release_trim()
 void journal_flusher_t::dump_diagnostics()
 {
     printf(
-        "Compaction queue: %u/%u items, data: %ju/%ju blocks used, buffer: %ju/%ju bytes used, meta: %ju/%ju bytes used, %u/%ju blocks nearfull\n",
+        "Compaction queue: %u items + %u future, data: %ju/%ju blocks used, buffer: %ju/%ju bytes used, meta: %ju/%ju bytes used, %u/%ju blocks nearfull\n",
         bs->heap->get_compact_queue_size(), bs->heap->get_to_compact_count(),
         bs->heap->get_data_used_space()/bs->dsk.data_block_size, bs->dsk.block_count,
         bs->heap->get_buffer_area_used_space(), bs->dsk.journal_len,
@@ -133,7 +128,7 @@ 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 ||
+        bs->heap->get_compact_queue_size() > bs->flusher_start_threshold ||
         i == 0 && bs->intent_write_counter >= bs->journal_trim_interval) && i < cur_flusher_count; i++)
     {
         co[i].loop();
@@ -424,7 +419,6 @@ resume_13:
     {
         printf("Compacted %jx:%jx l%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, compact_info.compact_lsn, copy_count);
     }
-    flusher->compact_counter++;
     flusher->active_flushers--;
     if (should_repeat)
     {
@@ -727,7 +721,7 @@ bool journal_flusher_co::fsync_buffer(int wait_base)
     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 ||
-        !bs->unsynced_big_write_count && !bs->unsynced_small_write_count && !bs->unsynced_meta_write_count)
+        !bs->unsynced_data_write_count && !bs->unsynced_small_write_count && !bs->unsynced_meta_write_count)
     {
         return true;
     }
@@ -784,7 +778,6 @@ resume_1:
         wait_state = wait_base+1;
         return false;
     }
-    flusher->compact_counter++;
     flusher->active_flushers--;
     return true;
 }

src/blockstore/blockstore_flush.h

@@ -87,8 +87,6 @@ class journal_flusher_t
     blockstore_impl_t *bs;
     friend class journal_flusher_co;
 
-    uint64_t compact_counter = 0;
-
     robin_hood::unordered_flat_set<object_id> flushing;
     int active_flushers = 0;
     int wanting_meta_fsync = 0;
@@ -100,7 +98,6 @@ public:
     ~journal_flusher_t();
     void loop();
     int get_syncing_buffer();
-    uint64_t get_compact_counter();
     bool is_active();
     void request_trim();
     void release_trim();

src/blockstore/blockstore_heap.cpp

@@ -1920,15 +1920,15 @@ uint32_t blockstore_heap_t::get_to_compact_count()
     return to_compact_count;
 }
 
-uint32_t blockstore_heap_t::get_inflight_queue_size()
+uint64_t blockstore_heap_t::get_compacted_count()
 {
-    return inflight_lsn.size();
+    return compacted_count;
 }
 
 void blockstore_heap_t::push_inflight_lsn(uint64_t lsn, heap_entry_t *wr, uint64_t flags)
 {
     uint64_t next_inf = first_inflight_lsn + inflight_lsn.size();
-    if (flags & HEAP_INFLIGHT_COMPACTABLE)
+    if (flags & (HEAP_INFLIGHT_COMPACTABLE|HEAP_INFLIGHT_COMPACTED))
     {
         to_compact_count++;
     }
@@ -1999,6 +1999,8 @@ void blockstore_heap_t::apply_inflight(heap_inflight_lsn_t & inflight)
     {
         // Mark previous entries as garbage, sequentially
         mark_garbage_up_to(wr);
+        to_compact_count--;
+        compacted_count++;
     }
     else if (inflight.flags & HEAP_INFLIGHT_COMPACTABLE)
     {

src/blockstore/blockstore_heap.h

@@ -176,6 +176,8 @@ class blockstore_heap_t
     // LSN queue: inflight (writing) -> completed [-> fsynced]
     std::deque<heap_inflight_lsn_t> inflight_lsn;
     uint32_t to_compact_count = 0;
+    uint64_t compacted_count = 0;
+    uint32_t inflight_overwrite_count = 0;
     uint64_t first_inflight_lsn = 0;
     uint64_t completed_lsn = 0;
     uint64_t fsynced_lsn = 0;
@@ -313,9 +315,9 @@ public:
     uint64_t get_meta_total_space();
     uint64_t get_meta_used_space();
     uint32_t get_meta_nearfull_blocks();
-    uint32_t get_inflight_queue_size();
     uint32_t get_compact_queue_size();
     uint32_t get_to_compact_count();
+    uint64_t get_compacted_count();
 
     uint64_t entry_pos(uint32_t block_num, uint32_t offset);
     heap_entry_t *entry_from_pos(uint64_t entry_pos, bool allow_unallocated = false);

src/blockstore/blockstore_impl.cpp

@@ -222,7 +222,7 @@ bool blockstore_impl_t::is_safe_to_stop()
     {
         return false;
     }
-    if (unsynced_big_write_count > 0 || unsynced_small_write_count > 0)
+    if (unsynced_data_write_count > 0 || unsynced_small_write_count > 0 || unsynced_meta_write_count > 0)
     {
         if (!readonly && !stop_sync_submitted)
         {
@@ -258,7 +258,7 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
     }
     else if (PRIV(op)->wait_for == WAIT_COMPACTION)
     {
-        if (flusher->get_compact_counter() <= PRIV(op)->wait_detail)
+        if (heap->get_compacted_count() <= PRIV(op)->wait_detail)
         {
             // do not submit
 #ifdef BLOCKSTORE_DEBUG

src/blockstore/blockstore_impl.h

@@ -49,7 +49,7 @@ struct blockstore_op_private_t
 
     // Write
     uint64_t location;
-    bool is_big;
+    uint32_t write_type;
 
     // Stabilize, rollback
     int stab_pos;
@@ -106,7 +106,7 @@ public:
     uint8_t* meta_superblock = NULL;
     uint8_t *buffer_area = NULL;
     std::vector<blockstore_op_t*> submit_queue;
-    int unsynced_big_write_count = 0, unsynced_small_write_count = 0, unsynced_meta_write_count = 0;
+    int unsynced_data_write_count = 0, unsynced_small_write_count = 0, unsynced_meta_write_count = 0;
     int unsynced_queued_ops = 0;
     uint8_t *zero_object = NULL;
 

src/blockstore/blockstore_stable.cpp

@@ -38,7 +38,7 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
             }
             if (res == ENOSPC)
             {
-                if (!heap->get_inflight_queue_size())
+                if (!heap->get_to_compact_count())
                 {
                     // no space
                     op->retval = -ENOSPC;
@@ -51,7 +51,7 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
                     goto resume_1;
                 }
                 priv->wait_for = WAIT_COMPACTION;
-                priv->wait_detail = flusher->get_compact_counter();
+                priv->wait_detail = heap->get_compacted_count();
                 flusher->request_trim();
                 return 0;
             }

src/blockstore/blockstore_sync.cpp

@@ -20,9 +20,9 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
 
 bool blockstore_impl_t::submit_fsyncs(int & wait_count)
 {
-    int n = ((unsynced_small_write_count > 0 || unsynced_big_write_count > 0 || unsynced_meta_write_count > 0) && !dsk.disable_meta_fsync) +
+    int n = ((unsynced_small_write_count > 0 || unsynced_data_write_count > 0 || 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_big_write_count > 0 && !dsk.disable_data_fsync && dsk.data_fd != dsk.meta_fd && dsk.data_fd != dsk.journal_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)
     {
         return false;
@@ -40,7 +40,7 @@ bool blockstore_impl_t::submit_fsyncs(int & wait_count)
         if (!wait_count)
             ringloop->wakeup();
     };
-    if ((unsynced_small_write_count > 0 || unsynced_big_write_count > 0 || unsynced_meta_write_count > 0) && !dsk.disable_meta_fsync)
+    if ((unsynced_small_write_count > 0 || unsynced_data_write_count > 0 || unsynced_meta_write_count > 0) && !dsk.disable_meta_fsync)
     {
         // fsync meta
         io_uring_sqe *sqe = get_sqe();
@@ -62,7 +62,7 @@ bool blockstore_impl_t::submit_fsyncs(int & wait_count)
         data->callback = cb;
         wait_count++;
     }
-    if (unsynced_big_write_count > 0 && !dsk.disable_data_fsync && dsk.data_fd != dsk.meta_fd && dsk.data_fd != dsk.journal_fd)
+    if (unsynced_data_write_count > 0 && !dsk.disable_data_fsync && dsk.data_fd != dsk.meta_fd && dsk.data_fd != dsk.journal_fd)
     {
         // fsync data
         io_uring_sqe *sqe = get_sqe();
@@ -73,7 +73,7 @@ bool blockstore_impl_t::submit_fsyncs(int & wait_count)
         data->callback = cb;
         wait_count++;
     }
-    unsynced_big_write_count = 0;
+    unsynced_data_write_count = 0;
     unsynced_small_write_count = 0;
     unsynced_meta_write_count = 0;
     return true;
@@ -91,10 +91,10 @@ int blockstore_impl_t::do_sync(blockstore_op_t *op, int base_state)
         return 0;
     }
     if (dsk.disable_journal_fsync && dsk.disable_meta_fsync && dsk.disable_data_fsync ||
-        !unsynced_big_write_count && !unsynced_small_write_count && !unsynced_meta_write_count)
+        !unsynced_data_write_count && !unsynced_small_write_count && !unsynced_meta_write_count)
     {
         // We can return immediately because sync only syncs previous writes
-        unsynced_big_write_count = unsynced_small_write_count = unsynced_meta_write_count = 0;
+        unsynced_data_write_count = unsynced_small_write_count = unsynced_meta_write_count = 0;
         return 2;
     }
     PRIV(op)->modified_block = heap->get_completed_lsn();

src/blockstore/blockstore_write.cpp

@@ -116,7 +116,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         return 0;
     }
     PRIV(op)->modified_block = UINT32_MAX;
-    PRIV(op)->is_big = false;
+    PRIV(op)->write_type = 0;
     heap_entry_t *obj = heap->read_entry(op->oid);
     if (op->opcode == BS_OP_DELETE)
     {
@@ -128,8 +128,11 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             FINISH_OP(op);
             return 2;
         }
+        PRIV(op)->write_type = BS_HEAP_DELETE;
         BS_SUBMIT_CHECK_SQES(1);
         int res = heap->add_delete(obj, &PRIV(op)->modified_block);
+        if (res == ENOSPC)
+            goto enospc;
         assert(res == 0);
         prepare_meta_block_write(PRIV(op)->modified_block);
         PRIV(op)->pending_ops++;
@@ -142,12 +145,13 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     else if (!obj || obj->type() == BS_HEAP_DELETE || op->offset == 0 && op->len == dsk.data_block_size)
     {
         // Big (redirect) write
+        PRIV(op)->write_type = BS_HEAP_BIG_WRITE;
         BS_SUBMIT_CHECK_SQES(1);
-        PRIV(op)->is_big = true;
+        PRIV(op)->location = heap->find_free_data();
-        uint64_t loc = heap->find_free_data();
+        if (PRIV(op)->location == UINT64_MAX)
-        if (loc == UINT64_MAX)
         {
-            if (!heap->get_inflight_queue_size())
+enospc:
+            if (!heap->get_to_compact_count())
             {
                 // no space
                 op->retval = -ENOSPC;
@@ -155,11 +159,11 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
                 return 2;
             }
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_compact_counter();
+            PRIV(op)->wait_detail = heap->get_compacted_count();
             flusher->request_trim();
             return 0;
         }
-        PRIV(op)->location = loc;
+        uint64_t loc = PRIV(op)->location;
 #ifdef BLOCKSTORE_DEBUG
         printf(
             "Allocate offset %ju for %jx:%jx v%ju\n",
@@ -204,12 +208,14 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         {
             // Even more simplified BIG_INTENT writes
             // FIXME: Support RMW mode for csum_block_size > bitmap_granularity
+            PRIV(op)->write_type = BS_HEAP_BIG_INTENT;
             PRIV(op)->location = obj->big_location(heap);
             res = heap->add_big_intent(op->oid, obj, op->version, op->offset, op->len, op->bitmap,
                 (uint8_t*)op->buf, NULL, &PRIV(op)->modified_block);
         }
         else
         {
+            PRIV(op)->write_type = BS_HEAP_INTENT_WRITE;
             auto wr = obj;
             while (wr && (wr->type() == BS_HEAP_INTENT_WRITE || wr->type() == BS_HEAP_COMMIT || wr->type() == BS_HEAP_ROLLBACK))
             {
@@ -220,21 +226,8 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             res = heap->add_small_write(op->oid, obj, (BS_HEAP_INTENT_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)),
                 op->version, op->offset, op->len, 0, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
         }
-        if (res == EAGAIN)
+        if (res == ENOSPC)
-        {
+            goto enospc;
-            assert(heap->get_inflight_queue_size());
-            PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_compact_counter();
-            flusher->request_trim();
-            return 0;
-        }
-        else if (res == ENOSPC)
-        {
-            // no space
-            op->retval = -ENOSPC;
-            FINISH_OP(op);
-            return 2;
-        }
         assert(res == 0);
         prepare_meta_block_write(PRIV(op)->modified_block);
         intent_write_counter++;
@@ -246,11 +239,12 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     {
         // Small (buffered) overwrite
         // First check if there is free buffer space
+        PRIV(op)->write_type = BS_HEAP_SMALL_WRITE;
         uint64_t loc = !op->len ? 0 : heap->find_free_buffer_area(op->len);
         if (loc == UINT64_MAX)
         {
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_compact_counter();
+            PRIV(op)->wait_detail = heap->get_compacted_count();
             flusher->request_trim();
             return 0;
         }
@@ -258,21 +252,8 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0));
         int res = heap->add_small_write(op->oid, obj, (BS_HEAP_SMALL_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)),
             op->version, op->offset, op->len, loc, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
-        if (res == EAGAIN)
+        if (res == ENOSPC)
-        {
+            goto enospc;
-            assert(heap->get_inflight_queue_size());
-            PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_compact_counter();
-            flusher->request_trim();
-            return 0;
-        }
-        else if (res == ENOSPC)
-        {
-            // no space
-            op->retval = -ENOSPC;
-            FINISH_OP(op);
-            return 2;
-        }
         assert(res == 0);
         if (op->len)
             heap->use_buffer_area(op->oid.inode, loc, op->len);
@@ -376,21 +357,21 @@ resume_4:
         auto obj = heap->read_entry(op->oid);
         int res = heap->add_big_write(op->oid, obj, (op->opcode == BS_OP_WRITE_STABLE), op->version,
             op->offset, op->len, PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
-        if (res == EAGAIN)
+        if (res == ENOSPC)
         {
-            assert(heap->get_inflight_queue_size());
+            if (!heap->get_to_compact_count())
+            {
+                // no space
+                heap->free_data(op->oid.inode, PRIV(op)->location);
+                write_iodepth--;
+                op->retval = -ENOSPC;
+                FINISH_OP(op);
+                return 2;
+            }
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_compact_counter();
+            PRIV(op)->wait_detail = heap->get_compacted_count();
             flusher->request_trim();
-            return 1;
+            return 0;
-        }
-        else if (res == ENOSPC)
-        {
-            heap->free_data(op->oid.inode, PRIV(op)->location);
-            write_iodepth--;
-            op->retval = -ENOSPC;
-            FINISH_OP(op);
-            return 2;
         }
         assert(res == 0);
         prepare_meta_block_write(PRIV(op)->modified_block);
@@ -399,8 +380,8 @@ resume_4:
         return 1;
     }
 resume_6:
-    // Apply throttling to not fill the journal too fast for the SSD+HDD case
+    // Apply throttling to not fill the journal too quickly for the SSD+HDD case
-    if (!PRIV(op)->is_big && throttle_small_writes)
+    if (PRIV(op)->write_type == BS_HEAP_SMALL_WRITE && throttle_small_writes)
     {
         // Apply throttling
         timespec tv_end;
@@ -436,8 +417,12 @@ resume_8:
     printf("Ack write %jx:%jx v%ju\n", op->oid.inode, op->oid.stripe, op->version);
 #endif
     op->retval = op->len;
-    if (PRIV(op)->is_big)
+    if (PRIV(op)->write_type == BS_HEAP_BIG_WRITE ||
-        unsynced_big_write_count++;
+        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)
+        unsynced_meta_write_count++;
     else
         unsynced_small_write_count++;
     write_iodepth--;