From 4e0b203552cc10197cb0045fe95afea5a29050fa Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Sun, 2 Nov 2025 21:03:24 +0300 Subject: [PATCH] Fix iterate_compaction for commit with non-latest version --- src/blockstore/blockstore_flush.cpp | 12 +++++--- src/blockstore/blockstore_heap.cpp | 12 +++++--- src/blockstore/blockstore_sync.cpp | 3 +- src/test/test_heap.cpp | 47 +++++++++++++++++++++++++++++ 4 files changed, 65 insertions(+), 9 deletions(-) diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index 866abe19..02fc3f3f 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -104,9 +104,10 @@ void journal_flusher_t::release_trim() void journal_flusher_t::dump_diagnostics() { printf( - "Compaction queue: %u items, data: %ju/%ju blocks used, meta: %ju/%ju bytes used, %u/%ju blocks nearfull\n", - bs->heap->get_to_compact_count(), + "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", + 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, bs->heap->get_meta_used_space(), bs->heap->get_meta_total_space(), bs->heap->get_meta_nearfull_blocks(), bs->dsk.meta_area_size/bs->dsk.meta_block_size-1 ); @@ -220,6 +221,7 @@ resume_16: goto resume_0; } } + assert(false); } flusher->flushing.insert(cur_oid); resume_1: @@ -286,7 +288,9 @@ resume_1: flusher->active_flushers++; if (bs->log_level > 10) { - printf("Compacting %jx:%jx l%ju .. l%ju\n", cur_oid.inode, cur_oid.stripe, compact_info.clean_wr->lsn, compact_info.compact_lsn); + printf("Compacting %jx:%jx v%ju..v%ju / l%ju..l%ju\n", cur_oid.inode, cur_oid.stripe, + compact_info.clean_wr->version, compact_info.compact_version, + compact_info.clean_wr->lsn, compact_info.compact_lsn); } overwrite_start = overwrite_end = 0; if (read_vec.size() > 0) @@ -723,7 +727,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_big_write_count && !bs->unsynced_small_write_count && !bs->unsynced_meta_write_count) { return true; } diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index 68c2a34a..c72cacc6 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -1347,7 +1347,8 @@ int blockstore_heap_t::add_commit(heap_entry_t *obj, uint64_t version, uint32_t } if (wr->type() == BS_HEAP_COMMIT) { - commit_version = wr->version; + if (commit_version < wr->version) + commit_version = wr->version; wr = prev(wr); continue; } @@ -1397,7 +1398,8 @@ int blockstore_heap_t::add_rollback(heap_entry_t *obj, uint64_t version, uint32_ } if (wr->type() == BS_HEAP_COMMIT) { - commit_version = wr->version; + if (commit_version < wr->version) + commit_version = wr->version; wr = prev(wr); continue; } @@ -1581,7 +1583,8 @@ void blockstore_heap_t::iterate_with_stable(heap_entry_t *obj, uint64_t max_lsn, } else if (old_wr->type() == BS_HEAP_COMMIT) { - commit_version = old_wr->version; + if (commit_version < old_wr->version) + commit_version = old_wr->version; } else { @@ -1639,7 +1642,8 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t res.compact_version = wr->version; } res.do_delete = false; - commit_version = wr->version; + if (commit_version < wr->version) + commit_version = wr->version; continue; } bool rolled_back = (wr->version > rollback_version); diff --git a/src/blockstore/blockstore_sync.cpp b/src/blockstore/blockstore_sync.cpp index 99a9894d..cde97d3d 100644 --- a/src/blockstore/blockstore_sync.cpp +++ b/src/blockstore/blockstore_sync.cpp @@ -90,7 +90,8 @@ 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 || !unsynced_big_write_count && !unsynced_small_write_count) + 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) { // We can return immediately because sync only syncs previous writes unsynced_big_write_count = unsynced_small_write_count = unsynced_meta_write_count = 0; diff --git a/src/test/test_heap.cpp b/src/test/test_heap.cpp index f053f70e..e66757ca 100644 --- a/src/test/test_heap.cpp +++ b/src/test/test_heap.cpp @@ -470,6 +470,52 @@ void test_compact(bool csum, bool stable) printf("OK test_compact %s %s\n", stable ? "stable" : "unstable", csum ? "csum" : "no_csum"); } +void test_iterate_compaction() +{ + int res; + blockstore_disk_t dsk; + _test_init(dsk, false); + std::vector buffer_area(dsk.journal_device_size); + + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + // Case: BIG_STABLE(v1 l1) -> SMALL(v2 l2) -> SMALL(v3 l3) -> SMALL(v4 l4) -> COMMIT(v2 l5) -> SMALL(v5 l6) -> COMMIT(v5 l7) + uint32_t mblock = 0; + _test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 4096, buffer_area.data()); + _test_small_write(heap, dsk, 1, 0, 2, 0, 4096, 0, false, buffer_area.data(), false); + _test_small_write(heap, dsk, 1, 0, 3, 4*1024, 4096, 4096, false, buffer_area.data(), false); + _test_small_write(heap, dsk, 1, 0, 4, 8*1024, 4096, 8*1024, false, buffer_area.data(), false); + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + auto obj = heap.read_entry(oid); + res = heap.add_commit(obj, 2, &mblock); + assert(res == 0); + heap.start_block_write(mblock); + heap.complete_block_write(mblock); + _test_small_write(heap, dsk, 1, 0, 5, 12*1024, 4096, 12*1024, false, buffer_area.data(), false); + obj = heap.read_entry(oid); + res = heap.add_commit(obj, 5, &mblock); + assert(res == 0); + heap.start_block_write(mblock); + heap.complete_block_write(mblock); + + assert(heap.get_fsynced_lsn() == 7); + int small_writes = 0; + obj = heap.read_entry(oid); + auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr) + { + small_writes++; + }); + assert(compact_info.compact_lsn == 7); + assert(compact_info.compact_version == 5); + assert(compact_info.clean_wr->lsn == 1); + assert(small_writes == 4); + } + + printf("OK test_iterate_compaction\n"); +} + void test_modify_bitmap() { blockstore_disk_t dsk; @@ -1167,6 +1213,7 @@ int main(int narg, char *args[]) test_compact(true, false); test_compact(false, true); test_compact(false, false); + test_iterate_compaction(); test_modify_bitmap(); test_recheck(false, true, false); test_recheck(false, false, false);