thebestvibedev/tromcho.net
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sort items on load to fix O(N^2) startup

Browse Source
This commit is contained in:
Vitaliy Filippov
2026-02-17 01:12:26 +03:00
parent 3a057ed5af
commit 4fa442a5de
5 changed files with 82 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/blockstore/blockstore_heap.cpp
+24 -6
View File
@@ -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<void(bool is_data, ui
}
if (!recheck_queue_filled)
{
finish_load();
fill_recheck_queue();
recheck_queue_filled = true;
}
@@ -803,7 +821,7 @@ std::vector<uint32_t> 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)
{
src/blockstore/blockstore_heap.h
+4 -2
View File
@@ -200,6 +200,7 @@ class blockstore_heap_t
bool marked_used_blocks = false;
bool recheck_queue_filled = false;
std::vector<heap_list_item_t*> loaded_list_items;
std::set<uint32_t> recheck_modified_blocks;
std::deque<heap_entry_t*> recheck_queue;
int recheck_in_progress = 0;
@@ -237,13 +238,14 @@ public:
std::function<void(uint32_t, uint32_t, uint8_t*)> 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<uint32_t> get_recheck_modified_blocks();
// recheck small write data after reading the database from disk
bool recheck_small_writes(std::function<void(bool is_data, uint64_t offset, uint64_t len, uint8_t* buf, std::function<void()>)> 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);
src/blockstore/blockstore_init.cpp
+2 -1
View File
@@ -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);
}
src/disk_tool/disk_tool_meta.cpp
+1
View File
@@ -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);
src/test/test_heap.cpp
+51 -41
View File
@@ -131,7 +131,7 @@ void test_mvcc(bool csum)
_test_init(dsk, csum);
std::vector<uint8_t> 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<uint8_t> 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<uint8_t> 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<uint8_t> 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<uint8_t> 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<void()> 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<void()> 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<void()> 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<void()> 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<void()> 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<void()> 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<uint8_t> 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<void()> 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<uint8_t> 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<void()> 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<uint8_t> 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<void()> 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<void()> 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<uint8_t> 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());