From 4fa442a5deeed618343107a4d5deda1c1b578356 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Sun, 15 Feb 2026 21:45:29 +0300 Subject: [PATCH] Sort items on load to fix O(N^2) startup --- src/blockstore/blockstore_heap.cpp | 30 ++++++++-- src/blockstore/blockstore_heap.h | 6 +- src/blockstore/blockstore_init.cpp | 3 +- src/disk_tool/disk_tool_meta.cpp | 1 + src/test/test_heap.cpp | 92 +++++++++++++++++------------- 5 files changed, 82 insertions(+), 50 deletions(-) diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index 604cb2c3..4f9f94eb 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -360,7 +360,7 @@ corrupted_object: { // Small writes require accessing offset & len to calculate correct length, // so require at least sizeof(heap_small_write_t) for them - fprintf(stderr, "Error: entry %jx:%jx v%ju has invalid size in metadata block %u at %u (%u < min %zu bytes). Metadata is corrupted, aborting\n", + fprintf(stderr, "Error: entry %jx:%jx v%ju has invalid size in metadata block %u at %u (%u < min %zu bytes)\n", wr->inode, wr->stripe, wr->version, block_num, block_offset, wr->size, sizeof(heap_small_write_t)); goto corrupted_object; } @@ -374,7 +374,7 @@ corrupted_object: uint32_t expected_crc32c = wr->calc_crc32c(); if (wr->crc32c != expected_crc32c) { - fprintf(stderr, "Error: entry %jx:%jx v%ju in metadata block %u at %u is corrupt (crc32c mismatch: expected %08x, got %08x). Metadata is corrupted, aborting\n", + fprintf(stderr, "Error: entry %jx:%jx v%ju in metadata block %u at %u is corrupt (crc32c mismatch: expected %08x, got %08x). ", wr->inode, wr->stripe, wr->version, block_num, block_offset, expected_crc32c, wr->crc32c); goto corrupted_object; @@ -385,7 +385,7 @@ corrupted_object: wr->small().offset % dsk->bitmap_granularity || wr->small().len % dsk->bitmap_granularity)) { - fprintf(stderr, "Error: %s entry %jx:%jx v%ju has invalid offset/length: %u/%u. Metadata is incompatible with current parameters, aborting\n", + fprintf(stderr, "Error: %s entry %jx:%jx v%ju has invalid offset/length: %u/%u. Metadata is incompatible with current parameters. ", wr->type() == BS_HEAP_SMALL_WRITE ? "small_write" : "intent_write", wr->inode, wr->stripe, wr->version, wr->small().offset, wr->small().len); goto corrupted_object; @@ -395,7 +395,7 @@ corrupted_object: wr->big_intent().offset % dsk->bitmap_granularity || wr->big_intent().len % dsk->bitmap_granularity)) { - fprintf(stderr, "Error: big_intent entry %jx:%jx v%ju has invalid offset/length: %u/%u. Metadata is incompatible with current parameters, aborting\n", + fprintf(stderr, "Error: big_intent entry %jx:%jx v%ju has invalid offset/length: %u/%u. Metadata is incompatible with current parameters. ", wr->inode, wr->stripe, wr->version, wr->big_intent().offset, wr->big_intent().len); goto corrupted_object; } @@ -422,7 +422,7 @@ int blockstore_heap_t::load_blocks(uint64_t disk_offset, uint64_t size, uint8_t next_lsn = wr->lsn; } entries_loaded++; - insert_list_item(li); + loaded_list_items.push_back(li); modify_alloc(block_num, [&](heap_block_info_t & inf) { if (!inf.entries.size()) @@ -526,6 +526,23 @@ bool blockstore_heap_t::validate_object(heap_entry_t *obj) return true; } +void blockstore_heap_t::finish_load() +{ + if (loaded_list_items.size()) + { + // Sort everything and load in correct order + std::sort(loaded_list_items.begin(), loaded_list_items.end(), [this](const heap_list_item_t* a, const heap_list_item_t* b) + { + return a->entry.lsn < b->entry.lsn; + }); + for (auto & li: loaded_list_items) + { + insert_list_item(li); + } + loaded_list_items.clear(); + } +} + void blockstore_heap_t::fill_recheck_queue() { for (auto & pgp: block_index) @@ -712,6 +729,7 @@ bool blockstore_heap_t::recheck_small_writes(std::function blockstore_heap_t::get_recheck_modified_blocks() return modified; } -int blockstore_heap_t::finish_load(bool allow_corrupted) +int blockstore_heap_t::finish_recheck() { if (!marked_used_blocks) { diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index 30c208ae..676c13aa 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -200,6 +200,7 @@ class blockstore_heap_t bool marked_used_blocks = false; bool recheck_queue_filled = false; + std::vector loaded_list_items; std::set recheck_modified_blocks; std::deque recheck_queue; int recheck_in_progress = 0; @@ -237,13 +238,14 @@ public: std::function handle_block); int load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf, bool allow_corrupted, uint64_t &entries_loaded); - // finish loading - int finish_load(bool allow_corrupted = false); + // finish loading - should be called after load_blocks + void finish_load(); // get blocks which are modified during loading and should be written to the disk // before finishing initialization if not R/O std::vector get_recheck_modified_blocks(); // recheck small write data after reading the database from disk bool recheck_small_writes(std::function)> read_buffer, int queue_depth); + int finish_recheck(); // reshard database according to the pool's PG count void* reshard_start(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size, uint64_t chunk_limit); bool reshard_continue(void* reshard_state, uint64_t chunk_limit); diff --git a/src/blockstore/blockstore_init.cpp b/src/blockstore/blockstore_init.cpp index e596c698..db99d519 100644 --- a/src/blockstore/blockstore_init.cpp +++ b/src/blockstore/blockstore_init.cpp @@ -239,6 +239,7 @@ resume_4: return 1; } // metadata read finished + bs->heap->finish_load(); printf("Metadata entries loaded: %ju, used blocks: %ju / %ju\n", entries_loaded, bs->heap->get_data_used_space() / bs->dsk.data_block_size, bs->dsk.block_count); if (zero_on_init && !bs->dsk.disable_meta_fsync) { @@ -284,7 +285,7 @@ resume_6: }, bs->meta_write_recheck_parallelism); return 1; resume_7: - if (bs->heap->finish_load() != 0) + if (bs->heap->finish_recheck() != 0) { exit(1); } diff --git a/src/disk_tool/disk_tool_meta.cpp b/src/disk_tool/disk_tool_meta.cpp index 60edfa5f..b9815d1c 100644 --- a/src/disk_tool/disk_tool_meta.cpp +++ b/src/disk_tool/disk_tool_meta.cpp @@ -109,6 +109,7 @@ close_error: r = heap->load_blocks(meta_pos-dsk.meta_block_size, read_len, data, true, entries_loaded); meta_pos += read_len; } + heap->finish_load(); heap->iterate_objects([&](heap_entry_t* obj, uint32_t meta_block_num) { obj_fn(heap, obj, meta_block_num); diff --git a/src/test/test_heap.cpp b/src/test/test_heap.cpp index 7ef02a97..e43359d3 100644 --- a/src/test/test_heap.cpp +++ b/src/test/test_heap.cpp @@ -131,7 +131,7 @@ void test_mvcc(bool csum) _test_init(dsk, csum); std::vector buffer_area(dsk.journal_device_size); blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // write, read, modify, check basic mvcc { @@ -186,7 +186,7 @@ void test_update(bool csum) _test_init(dsk, csum); std::vector buffer_area(dsk.journal_device_size); blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); { _test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 0, buffer_area.data()); @@ -206,7 +206,7 @@ void test_delete(bool csum) _test_init(dsk, csum); std::vector buffer_area(dsk.journal_device_size); blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); { // Add 1:0 and 1:20000 @@ -302,7 +302,7 @@ void test_defrag_block() std::vector buffer_area(dsk.journal_device_size); dsk.meta_area_size = 4096*3; blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); uint32_t big_write_size = heap.get_big_entry_size(); uint32_t small_write_size = heap.get_small_entry_size(0, 4096); @@ -377,7 +377,7 @@ void test_compact(bool csum, bool stable) std::vector buffer_area(dsk.journal_device_size); blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); memset(buffer_area.data(), 0x19, 4096); _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 4096, buffer_area.data()); @@ -508,7 +508,7 @@ void test_iterate_compaction() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // 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; @@ -551,9 +551,10 @@ void test_iterate_compaction() blockstore_heap_t heap(&dsk, buffer_area.data()); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); bool done = heap.recheck_small_writes([&](bool, uint64_t, uint64_t, uint8_t*, std::function cb) {}, 1); assert(done); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 0); @@ -572,7 +573,7 @@ void test_iterate_compaction() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // Case: BIG_STABLE(v1 l1) SMALL(v2 l2) COMMIT(v2 l3) SMALL(v3 l4) COMMIT(v3 l5) unfinished uint32_t mblock = 0; @@ -613,9 +614,10 @@ void test_iterate_compaction() blockstore_heap_t heap(&dsk, buffer_area.data()); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); bool done = heap.recheck_small_writes([&](bool, uint64_t, uint64_t, uint8_t*, std::function cb) {}, 1); assert(done); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 0); @@ -634,7 +636,7 @@ void test_iterate_compaction() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // Case: BIG_STABLE(v1 l1) SMALL(v2 l2) SMALL(v3 l3) SMALL(v4 l4) ROLLBACK(v3 l5) COMMIT(v2 l6) // -> compact by adding BIG_STABLE(v2 l2) @@ -685,9 +687,10 @@ void test_iterate_compaction() blockstore_heap_t heap(&dsk, buffer_area.data()); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); bool done = heap.recheck_small_writes([&](bool, uint64_t, uint64_t, uint8_t*, std::function cb) {}, 1); assert(done); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 0); @@ -708,7 +711,7 @@ void test_iterate_compaction() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // Case: BIG_STABLE(v1 l1) DELETE(l2) BIG_UNSTABLE(v1 l3) ROLLBACK(v0 l4) // -> compact by adding DELETE(l4) @@ -760,9 +763,10 @@ void test_iterate_compaction() blockstore_heap_t heap(&dsk, buffer_area.data()); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); bool done = heap.recheck_small_writes([&](bool, uint64_t, uint64_t, uint8_t*, std::function cb) {}, 1); assert(done); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 0); @@ -782,7 +786,7 @@ void test_iterate_compaction() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // Case: BIG_STABLE(v1 l1) SMALL(v2 l2) SMALL(v3 l3) ROLLBACK(v2 l4) SMALL(v3 l5) COMMIT(v3 l6) // -> compact by adding BIG_STABLE(v3 l6) and skip l3 @@ -836,9 +840,10 @@ void test_iterate_compaction() blockstore_heap_t heap(&dsk, buffer_area.data()); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); bool done = heap.recheck_small_writes([&](bool, uint64_t, uint64_t, uint8_t*, std::function cb) {}, 1); assert(done); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 0); @@ -859,7 +864,7 @@ void test_iterate_compaction() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // Case: BIG_STABLE(v1 l1) SMALL_STABLE(v2 l2) BIG_UNSTABLE(v3 l3) // -> skip compaction of l2 into l1 if not under pressure @@ -902,9 +907,10 @@ void test_iterate_compaction() blockstore_heap_t heap(&dsk, buffer_area.data()); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); bool done = heap.recheck_small_writes([&](bool, uint64_t, uint64_t, uint8_t*, std::function cb) {}, 1); assert(done); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 0); @@ -933,7 +939,7 @@ void test_modify_bitmap() std::vector buffer_area(dsk.journal_device_size); blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); memset(buffer_area.data(), 0x19, 8192); _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 8192, buffer_area.data()); @@ -981,7 +987,7 @@ void test_recheck(bool async, bool csum, bool intent) // write { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // object 1 _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 8192, buffer_area.data()); @@ -1007,6 +1013,7 @@ void test_recheck(bool async, bool csum, bool intent) blockstore_heap_t heap(&dsk, async ? NULL : buffer_area.data(), 10); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); int calls = 0; bool done = heap.recheck_small_writes([&](bool is_data, uint64_t offset, uint64_t len, uint8_t *buf, std::function cb) @@ -1033,7 +1040,7 @@ void test_recheck(bool async, bool csum, bool intent) assert(done); assert(calls == (async || intent ? 3 : 1)); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 1); @@ -1075,7 +1082,7 @@ void test_corruption() // write { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // big_write _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 0, buffer_area.data()); @@ -1128,7 +1135,7 @@ void test_full_overwrite(bool stable) // write { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // big_write _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 0, buffer_area.data()); @@ -1233,7 +1240,7 @@ void test_reshard_list() // write { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 0, buffer_area.data()); _test_big_write(heap, dsk, 1, 0x20000, 1, 0x40000, true, 0, 0, buffer_area.data()); @@ -1312,7 +1319,7 @@ void test_reshard_chunked() // write { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); for (int i = 0; i < 30; i++) _test_big_write(heap, dsk, 1, i*0x20000, 1, i*0x20000, true, 0, 0, buffer_area.data()); @@ -1380,7 +1387,7 @@ void test_destructor_mvcc() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // some writes _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 0, buffer_area.data()); @@ -1406,7 +1413,7 @@ void test_rollback() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // some writes _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 0, buffer_area.data()); @@ -1495,7 +1502,7 @@ void test_rollback() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // Remove a big write at all object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0x20000 }; @@ -1534,7 +1541,7 @@ void test_rollback() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // v1 unstable -> v2 unstable -> v3 unstable -> rollback v2 -> rollback v1 _test_big_write(heap, dsk, 1, 0, 1, 0x20000, false, 0, 0, buffer_area.data()); @@ -1561,7 +1568,7 @@ void test_rollback() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // v1 unstable -> v2 unstable -> v3 unstable -> rollback v2 -> rollback v1 _test_big_write(heap, dsk, 1, 0, 1, 0x20000, false, 0, 0, buffer_area.data()); @@ -1588,7 +1595,7 @@ void test_rollback() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // v1 unstable -> v2 unstable -> v3 unstable -> commit v1 -> commit v2 -> rollback v1 _test_big_write(heap, dsk, 1, 0, 1, 0x20000, false, 0, 0, buffer_area.data()); @@ -1641,7 +1648,7 @@ void test_rollback() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); // v1 stable -> rollback v2 _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 0, buffer_area.data()); @@ -1676,7 +1683,7 @@ void test_alloc_buffer() } blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); uint64_t pos; @@ -1739,7 +1746,7 @@ void test_full_alloc() std::vector buffer_area(dsk.journal_device_size); blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); assert(heap.get_meta_total_space() == 4*4096); uint32_t big_write_size = heap.get_big_entry_size(); @@ -1784,7 +1791,7 @@ void test_intent_write(bool csum) { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 4096, buffer_area.data()); @@ -1817,7 +1824,7 @@ void test_big_intent_csums() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 4096, buffer_area.data()); @@ -1865,6 +1872,7 @@ void test_big_intent_csums() blockstore_heap_t heap(&dsk, buffer_area.data(), 10); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); int calls = 0; bool done = heap.recheck_small_writes([&](bool is_data, uint64_t offset, uint64_t len, uint8_t *buf, std::function cb) @@ -1882,7 +1890,7 @@ void test_big_intent_csums() assert(done); assert(calls == 2); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 0); @@ -1907,7 +1915,7 @@ void test_recalc_stats() _test_init(dsk, false); std::vector buffer_area(dsk.journal_device_size); blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); { _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 0, buffer_area.data()); @@ -1959,7 +1967,7 @@ void test_redirect_intent_csums() { blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 4096, buffer_area.data()); @@ -1985,6 +1993,7 @@ void test_redirect_intent_csums() blockstore_heap_t heap(&dsk, buffer_area.data(), 10); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); int calls = 0; bool done = heap.recheck_small_writes([&](bool is_data, uint64_t offset, uint64_t len, uint8_t *buf, std::function cb) @@ -2002,7 +2011,7 @@ void test_redirect_intent_csums() assert(done); assert(calls == 2); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 0); @@ -2039,6 +2048,7 @@ void test_redirect_intent_csums() blockstore_heap_t heap(&dsk, buffer_area.data(), 10); uint64_t entries_loaded; heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded); + heap.finish_load(); int calls = 0; bool done = heap.recheck_small_writes([&](bool is_data, uint64_t offset, uint64_t len, uint8_t *buf, std::function cb) @@ -2054,7 +2064,7 @@ void test_redirect_intent_csums() assert(done); assert(calls == 2); - heap.finish_load(); + heap.finish_recheck(); auto mod = heap.get_recheck_modified_blocks(); assert(mod.size() == 1); @@ -2081,7 +2091,7 @@ void test_explicit_complete() dsk.disable_journal_fsync = dsk.disable_meta_fsync = false; std::vector buffer_area(dsk.journal_device_size); blockstore_heap_t heap(&dsk, buffer_area.data()); - heap.finish_load(); + heap.finish_recheck(); { _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 0, buffer_area.data());