From 08f21edaf7c4bf20d65a96d539071fe76db6f364 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Mon, 3 Nov 2025 02:30:51 +0300 Subject: [PATCH] Fix enospc detection logic and test_enospc --- src/blockstore/blockstore_flush.cpp | 13 +--- src/blockstore/blockstore_flush.h | 3 - src/blockstore/blockstore_heap.cpp | 8 ++- src/blockstore/blockstore_heap.h | 4 +- src/blockstore/blockstore_impl.cpp | 4 +- src/blockstore/blockstore_impl.h | 4 +- src/blockstore/blockstore_stable.cpp | 4 +- src/blockstore/blockstore_sync.cpp | 14 ++--- src/blockstore/blockstore_write.cpp | 93 ++++++++++++---------------- 9 files changed, 63 insertions(+), 84 deletions(-) diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index 02fc3f3f..73fcc10d 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/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; } diff --git a/src/blockstore/blockstore_flush.h b/src/blockstore/blockstore_flush.h index f8fdcbc6..a9a24ef9 100644 --- a/src/blockstore/blockstore_flush.h +++ b/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 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(); diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index cc73588e..21c5e3cb 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/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) { diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index c2387144..feda9306 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -176,6 +176,8 @@ class blockstore_heap_t // LSN queue: inflight (writing) -> completed [-> fsynced] std::deque 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); diff --git a/src/blockstore/blockstore_impl.cpp b/src/blockstore/blockstore_impl.cpp index 0d5cea00..24638cff 100644 --- a/src/blockstore/blockstore_impl.cpp +++ b/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 diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index 46d1d1a1..93cdccfa 100644 --- a/src/blockstore/blockstore_impl.h +++ b/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 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; diff --git a/src/blockstore/blockstore_stable.cpp b/src/blockstore/blockstore_stable.cpp index f3253ec4..33ec076c 100644 --- a/src/blockstore/blockstore_stable.cpp +++ b/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; } diff --git a/src/blockstore/blockstore_sync.cpp b/src/blockstore/blockstore_sync.cpp index cde97d3d..bd21605c 100644 --- a/src/blockstore/blockstore_sync.cpp +++ b/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(); diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index 02d88bde..b4b86285 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/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; - uint64_t loc = heap->find_free_data(); - if (loc == UINT64_MAX) + PRIV(op)->location = heap->find_free_data(); + if (PRIV(op)->location == 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) - { - 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; - } + if (res == ENOSPC) + goto enospc; 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) - { - 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; - } + if (res == ENOSPC) + goto enospc; 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; - } - else if (res == ENOSPC) - { - heap->free_data(op->oid.inode, PRIV(op)->location); - write_iodepth--; - op->retval = -ENOSPC; - FINISH_OP(op); - return 2; + return 0; } 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 - if (!PRIV(op)->is_big && throttle_small_writes) + // Apply throttling to not fill the journal too quickly for the SSD+HDD case + 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) - unsynced_big_write_count++; + if (PRIV(op)->write_type == BS_HEAP_BIG_WRITE || + 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--;