From fb878707341b992ba7712d9e4f87faa7b002e0d5 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Sat, 2 Aug 2025 17:38:15 +0300 Subject: [PATCH] Support moving objects between blocks --- src/blockstore/blockstore_heap.cpp | 172 ++++++++++++++++++++--------- src/blockstore/blockstore_heap.h | 12 +- src/test/test_heap.cpp | 78 +++++++++++-- 3 files changed, 196 insertions(+), 66 deletions(-) diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index 8ff03d4b..1a6037f1 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -381,7 +381,7 @@ skip_object: { fprintf(stderr, "Warning: Object %jx:%jx in metadata block %u at %u is a newer duplicate (lsn %lu < %lu), overriding\n", obj->inode, obj->stripe, block_num, block_offset, dup_lsn, lsn); - erase_object(dup_block, dup_obj, 0, false); + init_erase(dup_block, dup_obj); } } // Verify checksums @@ -739,7 +739,7 @@ bool blockstore_heap_t::recheck_small_writes(std::functioninode, obj->stripe); - erase_object(block_num, obj, 0, false); + init_erase(block_num, obj); } else { @@ -1047,7 +1047,7 @@ uint32_t blockstore_heap_t::compact_object_to(heap_object_t *obj, uint64_t compa return freed; } -void blockstore_heap_t::compact_block(uint32_t block_num) +void blockstore_heap_t::defragment_block(uint32_t block_num) { auto & inf = block_info[block_num]; assert(inf.data); @@ -1159,6 +1159,30 @@ uint32_t blockstore_heap_t::find_block_run(heap_block_info_t & inf, uint32_t spa return UINT32_MAX; } +uint32_t blockstore_heap_t::block_has_compactable(uint8_t *data) +{ + uint32_t sum = 0; + uint8_t *end = data + dsk->meta_block_size; + while (data < end) + { + uint16_t region_marker = *((uint16_t*)data); + assert(region_marker); + if (!(region_marker & FREE_SPACE_BIT) && + region_marker > sizeof(heap_object_t)) + { + heap_write_t *wr = (heap_write_t*)data; + if (wr->flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) || + wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)) + { + // May be freed in the future + sum += wr->size; + } + } + data += (region_marker & ~FREE_SPACE_BIT); + } + return sum; +} + uint32_t blockstore_heap_t::find_block_space(uint32_t block_num, uint32_t space) { auto & inf = block_info.at(block_num); @@ -1177,10 +1201,20 @@ uint32_t blockstore_heap_t::find_block_space(uint32_t block_num, uint32_t space) return res; } } - compact_block(block_num); + defragment_block(block_num); return find_block_run(inf, space); } +void blockstore_heap_t::allocate_block(heap_block_info_t & inf) +{ + if (!inf.data) + { + inf.data = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk->meta_block_size); + memset(inf.data, 0, dsk->meta_block_size); + *((uint16_t*)inf.data) = FREE_SPACE_BIT | dsk->meta_block_size; + } +} + int blockstore_heap_t::add_object(object_id oid, heap_write_t *wr, uint32_t *modified_block) { // By now, initial small_writes are not allowed @@ -1198,12 +1232,7 @@ int blockstore_heap_t::add_object(object_id oid, heap_write_t *wr, uint32_t *mod return res; } auto & inf = block_info.at(block_num); - if (!inf.data) - { - inf.data = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk->meta_block_size); - memset(inf.data, 0, dsk->meta_block_size); - *((uint16_t*)inf.data) = FREE_SPACE_BIT | dsk->meta_block_size; - } + allocate_block(inf); if (modified_block) { *modified_block = block_num; @@ -1243,6 +1272,20 @@ bool blockstore_heap_t::mvcc_check_tracking(object_id oid) return (mvcc_it->first.oid == oid && mvcc_it->second.entry_copy); } +void blockstore_heap_t::copy_full_object(uint8_t *dst, heap_object_t *obj) +{ + memcpy(dst, obj, sizeof(heap_object_t)); + ((heap_object_t*)dst)->write_pos = obj->size; + dst += obj->size; + for (auto wr = obj->get_writes(); wr; wr = wr->next()) + { + memcpy(dst, wr, wr->size); + if (wr->next_pos) + ((heap_write_t*)dst)->next_pos = wr->size; + dst += wr->size; + } +} + // returns tracking_active, i.e. true if there exists at least one copied MVCC version of the object // it's used for reference tracking because tracking_active=false means that there is 1 implicit reference // for heap_writes of the current version of the object and true means that there isn't @@ -1273,17 +1316,8 @@ bool blockstore_heap_t::mvcc_save_copy(heap_object_t *obj) total_size += wr->size; } heap_object_t *obj_copy = (heap_object_t*)malloc_or_die(total_size); - memcpy(obj_copy, obj, sizeof(heap_object_t)); + copy_full_object((uint8_t*)obj_copy, obj); mvcc_it->second.entry_copy = obj_copy; - total_size = obj->size; - obj_copy->write_pos = obj->size; - for (auto wr = obj->get_writes(); wr; wr = wr->next()) - { - auto new_wr = (heap_write_t*)((uint8_t*)obj_copy + total_size); - memcpy((uint8_t*)new_wr, wr, wr->size); - new_wr->next_pos = wr->next_pos ? wr->size : 0; - total_size += wr->size; - } uint32_t add_ref = 1; bool for_obj = false; // save_copy is performed when the object is modified, so object_mvcc may only @@ -1339,18 +1373,11 @@ void blockstore_heap_t::mark_overwritten(uint64_t over_lsn, uint64_t inode, heap } } -int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, heap_write_t *wr, uint32_t *modified_block) +int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, heap_write_t *wr, uint32_t *modified_block, uint32_t *moved_from_block) { const auto oid = (object_id){ .inode = obj->inode, .stripe = obj->stripe }; - const uint32_t wr_size = wr->get_size(this); - auto & inf = block_info.at(block_num); - assert(inf.data); + // First some validation bool is_overwrite = (wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) || wr->flags == (BS_HEAP_TOMBSTONE|BS_HEAP_STABLE)); - if (dsk->meta_block_size-inf.used_space < wr_size+2) - { - // Something in the block has to be compacted - return ENOSPC; - } auto first_wr = obj->get_writes(); if (first_wr->type() == BS_HEAP_TOMBSTONE && !is_overwrite) { @@ -1379,6 +1406,48 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea // Overwrites with a smaller version are forbidden return EINVAL; } + // Then a free space check + const uint32_t wr_size = wr->get_size(this); + auto *inf = &block_info.at(block_num); + assert(inf->data); + if (inf->used_space+wr_size > dsk->meta_block_size-2) + { + // Something in the block has to be compacted + if (block_has_compactable(inf->data) >= inf->used_space+wr_size-(dsk->meta_block_size-2)) + { + return EAGAIN; + } + // Otherwise, move the object + uint32_t new_block = 0; + int res = get_block_for_new_object(new_block); + if (res == ENOSPC) + { + return ENOSPC; + } + uint32_t full_size = obj->size; + for (auto wr = obj->get_writes(); wr; wr = wr->next()) + { + full_size += wr->size; + } + inf = &block_info.at(new_block); + if (inf->used_space+full_size+wr_size > dsk->meta_block_size-2) + { + return ENOSPC; + } + allocate_block(*inf); + if (moved_from_block) + { + *moved_from_block = block_num; + } + uint32_t new_offset = find_block_space(new_block, full_size); + assert(new_offset != UINT32_MAX); + copy_full_object(inf->data + new_offset, obj); + erase_object(block_num, obj, 0, false); + block_num = new_block; + obj = (heap_object_t*)(inf->data + new_offset); + block_index[get_pg_id(oid.inode, oid.stripe)][oid.inode][oid.stripe] = (uint64_t)new_block*dsk->meta_block_size + new_offset; + add_used_space(new_block, full_size); + } if (modified_block) { *modified_block = block_num; @@ -1397,16 +1466,16 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea mvcc_buffer_refs[wr->location]++; } } - const uint8_t *old_data = inf.data; + const uint8_t *old_data = inf->data; const uint32_t offset = find_block_space(block_num, wr_size); - if (old_data != inf.data) + if (old_data != inf->data) { obj = read_entry(oid, NULL); first_wr = obj->get_writes(); } assert(offset != UINT32_MAX); - memcpy(inf.data + offset, wr, wr_size); - heap_write_t *new_wr = (heap_write_t*)(inf.data + offset); + memcpy(inf->data + offset, wr, wr_size); + heap_write_t *new_wr = (heap_write_t*)(inf->data + offset); new_wr->size = wr_size; new_wr->lsn = ++next_lsn; int32_t used_delta = wr_size; @@ -1441,14 +1510,14 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea } wr->lsn = new_wr->lsn; push_inflight_lsn(oid, new_wr->lsn, new_wr->needs_compact(this) ? HEAP_INFLIGHT_COMPACTABLE : 0); - obj->write_pos = offset - ((uint8_t*)obj - inf.data); + obj->write_pos = offset - ((uint8_t*)obj - inf->data); obj->crc32c = obj->calc_crc32c(); // Change block free space add_used_space(block_num, used_delta); return 0; } -int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block) +int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block, uint32_t *moved_from_block) { uint32_t block_num = 0; heap_object_t *obj = read_entry(oid, &block_num); @@ -1456,16 +1525,16 @@ int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *mod { return add_object(oid, wr, modified_block); } - return update_object(block_num, obj, wr, modified_block); + return update_object(block_num, obj, wr, modified_block, moved_from_block); } -int blockstore_heap_t::post_write(uint32_t & block_num, object_id oid, heap_object_t *obj, heap_write_t *wr) +int blockstore_heap_t::post_write(uint32_t & block_num, object_id oid, heap_object_t *obj, heap_write_t *wr, uint32_t *moved_from_block) { if (!obj) { return add_object(oid, wr, &block_num); } - return update_object(block_num, obj, wr, &block_num); + return update_object(block_num, obj, wr, &block_num, moved_from_block); } int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *new_lsn, uint64_t *new_to_lsn) @@ -1713,6 +1782,7 @@ void blockstore_heap_t::deref_data(uint64_t inode, uint64_t location, bool free_ void blockstore_heap_t::deref_buffer(uint64_t inode, uint64_t location, uint32_t len, bool free_at_0) { + assert(len > 0); auto ref_it = mvcc_buffer_refs.find(location); if (ref_it != mvcc_buffer_refs.end()) { @@ -1777,24 +1847,22 @@ void blockstore_heap_t::erase_block_index(inode_t inode, uint64_t stripe) } } +void blockstore_heap_t::init_erase(uint32_t block_num, heap_object_t *obj) +{ + for (auto wr = obj->get_writes(); wr; wr = wr->next()) + { + if (wr->needs_compact(this)) + mark_lsn_compacted(wr->lsn, true); + } + free_object_space(obj->inode, obj->get_writes(), NULL); + erase_object(block_num, obj, 0, false); +} + void blockstore_heap_t::erase_object(uint32_t block_num, heap_object_t *obj, uint64_t lsn, bool tracking_active) { // Erase object - if (!lsn) - { - for (auto wr = obj->get_writes(); wr; wr = wr->next()) - { - if (wr->needs_compact(this)) - { - mark_lsn_compacted(wr->lsn, true); - } - } - free_object_space(obj->inode, obj->get_writes(), NULL); - } - else - { + if (lsn > 0) mark_overwritten(lsn, obj->inode, obj->get_writes(), NULL, tracking_active); - } erase_block_index(obj->inode, obj->stripe); auto freed = free_writes(obj->get_writes(), NULL); auto obj_size = obj->size; diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index 16afc0d6..3425f404 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -172,15 +172,19 @@ class blockstore_heap_t const uint32_t max_write_entry_size; uint64_t get_pg_id(inode_t inode, uint64_t stripe); - void compact_block(uint32_t block_num); + void defragment_block(uint32_t block_num); uint32_t find_block_run(heap_block_info_t & block, uint32_t space); uint32_t find_block_space(uint32_t block_num, uint32_t space); + uint32_t block_has_compactable(uint8_t *data); uint32_t compact_object_to(heap_object_t *obj, uint64_t lsn, uint8_t *new_csums, bool do_free); + void copy_full_object(uint8_t *dst, heap_object_t *obj); bool mvcc_save_copy(heap_object_t *obj); bool mvcc_check_tracking(object_id oid); + void allocate_block(heap_block_info_t & inf); int add_object(object_id oid, heap_write_t *wr, uint32_t *modified_block); void mark_overwritten(uint64_t over_lsn, uint64_t inode, heap_write_t *wr, heap_write_t *end_wr, bool tracking_active); - int update_object(uint32_t block_num, heap_object_t *obj, heap_write_t *wr, uint32_t *modified_block); + int update_object(uint32_t block_num, heap_object_t *obj, heap_write_t *wr, uint32_t *modified_block, uint32_t *moved_from_block); + void init_erase(uint32_t block_num, heap_object_t *obj); void erase_object(uint32_t block_num, heap_object_t *obj, uint64_t lsn, bool tracking_active); void reindex_block(uint32_t block_num, heap_object_t *from_obj); void erase_block_index(inode_t inode, uint64_t stripe); @@ -232,8 +236,8 @@ public: bool set, std::function bad_block_cb); // auto-compacts the object, then adds a write entry to it and to the compaction queue // return 0 if OK, or maybe ENOSPC - int post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block); - int post_write(uint32_t & block_num, object_id oid, heap_object_t *obj, heap_write_t *wr); + int post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block, uint32_t *moved_from_block); + int post_write(uint32_t & block_num, object_id oid, heap_object_t *obj, heap_write_t *wr, uint32_t *moved_from_block); // stabilize an unstable object version // return 0 if OK, ENOENT if not exists int post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *new_lsn, uint64_t *new_to_lsn); diff --git a/src/test/test_heap.cpp b/src/test/test_heap.cpp index b053ffa2..7177abd8 100644 --- a/src/test/test_heap.cpp +++ b/src/test/test_heap.cpp @@ -83,8 +83,8 @@ int _test_do_big_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64 else memset(wr->get_checksums(&heap), 0xde, dsk.data_block_size/dsk.csum_block_size*4); } - uint32_t mblock; - return heap.post_write(oid, wr, &mblock); + uint32_t mblock, mfblock; + return heap.post_write(oid, wr, &mblock, &mfblock); } void _test_big_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64_t inode, uint64_t stripe, uint64_t version, uint64_t location, @@ -97,7 +97,8 @@ void _test_big_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64_t } int _test_do_small_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64_t inode, uint64_t stripe, uint64_t version, - uint32_t offset, uint32_t len, uint64_t location, bool stable = true, uint32_t *checksums = NULL, bool is_intent = false) + uint32_t offset, uint32_t len, uint64_t location, bool stable = true, uint32_t *checksums = NULL, bool is_intent = false, + uint32_t *mblock = NULL, uint32_t *mfblock = NULL) { object_id oid = { .inode = INODE_WITH_POOL(1, inode), .stripe = stripe }; uint8_t wr_buf[heap.get_max_write_entry_size()]; @@ -122,16 +123,16 @@ int _test_do_small_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint memset(wr->get_checksums(&heap), 0xab, ((offset+len+dsk.csum_block_size-1)/dsk.csum_block_size - offset/dsk.csum_block_size)*4); else *wr->get_checksum(&heap) = 0xabababab; - uint32_t mblock; - return heap.post_write(oid, wr, &mblock); + return heap.post_write(oid, wr, mblock, mfblock); } void _test_small_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64_t inode, uint64_t stripe, uint64_t version, - uint32_t offset, uint32_t len, uint64_t location, bool stable = true, uint32_t *checksums = NULL, bool is_intent = false) + uint32_t offset, uint32_t len, uint64_t location, bool stable = true, uint32_t *checksums = NULL, bool is_intent = false, + uint32_t *mblock = NULL, uint32_t *mfblock = NULL) { if (!is_intent) heap.use_buffer_area(INODE_WITH_POOL(1, inode), location, len); // blocks are allocated before write and outside the heap_t - int res = _test_do_small_write(heap, dsk, inode, stripe, version, offset, len, location, stable, checksums, is_intent); + int res = _test_do_small_write(heap, dsk, inode, stripe, version, offset, len, location, stable, checksums, is_intent, mblock, mfblock); assert(res == 0); if (!is_intent) assert(!heap.is_buffer_area_free(location, len)); @@ -615,13 +616,13 @@ void test_corruption() wr->location = 0; wr->flags = BS_HEAP_TOMBSTONE|BS_HEAP_STABLE; assert(!wr->get_checksums(&heap)); - res = heap.post_write(oid, wr, NULL); + res = heap.post_write(oid, wr, NULL, NULL); assert(res == 0); // try to do a small_write over a tombstone to fail wr->version = 3; wr->flags = BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE; - res = heap.post_write(oid, wr, NULL); + res = heap.post_write(oid, wr, NULL, NULL); assert(res == EINVAL); // persist @@ -1320,7 +1321,7 @@ void test_full_alloc() assert(_test_do_small_write(heap, dsk, 1, 0, 6+i, 0, 4096, epb*4*16384+i*4096) == 0); } assert(dsk.meta_block_size-heap.get_meta_block_used_space(0) < big_write_size); - assert(_test_do_small_write(heap, dsk, 1, 0, 12, 0, 4096, 48*16384+8*4096) == ENOSPC); + assert(_test_do_small_write(heap, dsk, 1, 0, 12, 0, 4096, 48*16384+8*4096) == EAGAIN); // Check that used_alloc_queue doesn't return used blocks { @@ -1558,6 +1559,62 @@ void test_intent_write(bool csum) printf("OK test_intent_write %s\n", csum ? "csum" : "no_csum"); } +void test_move() +{ + blockstore_disk_t dsk; + std::map config; + config["data_csum_type"] = "crc32c"; + dsk.parse_config(config); + dsk.data_device_size = 8*1024*1024; + dsk.meta_device_size = 5*4096; + dsk.journal_device_size = 4*1024*1024; + dsk.data_fd = 0; + dsk.meta_fd = 1; + dsk.journal_fd = 2; + dsk.calc_lengths(); + std::vector buffer_area(dsk.journal_device_size); + + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + assert(heap.get_meta_total_space() == 4*4096); + + uint32_t big_write_size = (sizeof(heap_object_t) + sizeof(heap_write_t) + 2*dsk.clean_entry_bitmap_size + dsk.data_block_size/dsk.csum_block_size*4); + uint32_t small_write_size = (sizeof(heap_write_t) + dsk.clean_entry_bitmap_size + 4); + assert(big_write_size == 197); + assert(small_write_size == 45); + + // Fill block 1 almost completely with unstable small writes + _test_big_write(heap, dsk, 1, 0*0x20000, 1, 0*0x20000); + _test_big_write(heap, dsk, 1, 1*0x20000, 1, 1*0x20000); + int i = 0; + while (i < (dsk.meta_block_size-2*big_write_size-2)/small_write_size/2) + { + _test_small_write(heap, dsk, 1, 0*0x20000, 2+i, (i*4096) % dsk.data_block_size, 4096, 2*i*4096, false); + _test_small_write(heap, dsk, 1, 1*0x20000, 2+i, (i*4096) % dsk.data_block_size, 4096, (2*i+1)*4096, false); + i++; + } + assert(heap.get_meta_block_used_space(0) > 0); + assert(!heap.get_meta_block_used_space(1)); + assert(!heap.get_meta_block_used_space(2)); + assert(!heap.get_meta_block_used_space(3)); + + // Next small_write should auto-move an object + uint32_t mblock = UINT32_MAX, mfblock = UINT32_MAX; + _test_small_write(heap, dsk, 1, 0*0x20000, 2+i, (i*4096) % dsk.data_block_size, 4096, 2*i*4096, false, NULL, false, &mblock, &mfblock); + assert(mblock == 1 && mfblock == 0); + assert(heap.get_meta_block_used_space(0) == big_write_size+i*small_write_size); + assert(heap.get_meta_block_used_space(1) == big_write_size+(i+1)*small_write_size); + + // Check that the object is still readable + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + heap_object_t *obj = heap.read_entry(oid, NULL); + assert(obj); + assert(count_writes(obj) == 2+i); + assert(obj->get_writes()->version == 2+i); + + printf("OK test_move\n"); +} + int main(int narg, char *args[]) { test_mvcc(true); @@ -1594,5 +1651,6 @@ int main(int narg, char *args[]) test_autocompact(false); test_intent_write(true); test_intent_write(false); + test_move(); return 0; }