diff --git a/CMakeLists.txt b/CMakeLists.txt index 19f67698..71a77794 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -4,4 +4,6 @@ project(vitastor) set(VITASTOR_VERSION "2.4.4") +include(CTest) + add_subdirectory(src) diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 3fc35647..ee8219e2 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -84,6 +84,8 @@ add_custom_target(test echo leak:tcmalloc > ${CMAKE_CURRENT_BINARY_DIR}/lsan-suppress.txt && env LSAN_OPTIONS=suppressions=${CMAKE_CURRENT_BINARY_DIR}/lsan-suppress.txt ${CMAKE_CTEST_COMMAND} ) +# make -j16 -C ../../build test_heap && ../../build/src/test/test_heap +# make -j16 -C ../../build test_heap && rm -f $(find ../../build -name '*.gcda') && ctest -V -T test -T coverage -R heap --test-dir ../../build && (cd ../../build; gcovr -f ../src --html --html-nested -o coverage/index.html; cd ../src/test) add_dependencies(test build_tests) include_directories( diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp new file mode 100644 index 00000000..16f4c4e1 --- /dev/null +++ b/src/blockstore/blockstore_heap.cpp @@ -0,0 +1,1716 @@ +// Metadata storage version 3 ("heap") +// Copyright (c) Vitaliy Filippov, 2025+ +// License: VNPL-1.1 (see README.md for details) + +#include "blockstore_heap.h" + +#include "../util/allocator.h" +#include "../util/crc32c.h" +#include "../util/malloc_or_die.h" + +#define BS_HEAP_FREE_MVCC 1 +#define BS_HEAP_FREE_MAIN 2 + +heap_write_t *heap_write_t::next(blockstore_heap_t *heap) +{ + return (heap_write_t*)((uint8_t*)this + get_size(heap)); +} + +uint32_t heap_write_t::get_size(blockstore_heap_t *heap) +{ + return (sizeof(heap_write_t) + + heap->dsk->clean_entry_bitmap_size + + ((flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE + ? heap->dsk->clean_entry_bitmap_size + : 0) + + get_csum_size(heap)); +} + +uint32_t heap_write_t::get_csum_size(blockstore_heap_t *heap) +{ + if (!heap->dsk->csum_block_size) + { + return ((flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE ? 4 : 0); + } + return ((offset+len+heap->dsk->csum_block_size-1)/heap->dsk->csum_block_size - offset/heap->dsk->csum_block_size) + * (heap->dsk->data_csum_type & 0xFF); +} + +bool heap_write_t::needs_recheck(blockstore_heap_t *heap) +{ + return len > 0 && lsn >= heap->compacted_lsn && (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) || flags == BS_HEAP_SMALL_WRITE); +} + +bool heap_write_t::needs_compact(uint64_t compacted_lsn) +{ + return lsn > compacted_lsn && flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); +} + +bool heap_write_t::is_compacted(uint64_t compacted_lsn) +{ + return lsn <= compacted_lsn && flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); +} + +bool heap_write_t::can_be_collapsed(blockstore_heap_t *heap) +{ + return !heap->dsk->csum_block_size || heap->dsk->csum_block_size == heap->dsk->bitmap_granularity || + !(offset % heap->dsk->csum_block_size) && !(len % heap->dsk->csum_block_size); +} + +bool heap_write_t::is_allowed_before_compacted(uint64_t compacted_lsn, bool is_last_entry) +{ + return lsn <= compacted_lsn && flags == ((is_last_entry ? BS_HEAP_BIG_WRITE : BS_HEAP_SMALL_WRITE) | BS_HEAP_STABLE); +} + +uint8_t *heap_write_t::get_ext_bitmap(blockstore_heap_t *heap) +{ + return ((uint8_t*)this + sizeof(heap_write_t)); +} + +uint8_t *heap_write_t::get_int_bitmap(blockstore_heap_t *heap) +{ + if ((flags & BS_HEAP_TYPE) != BS_HEAP_BIG_WRITE || !len) + return NULL; + return ((uint8_t*)this + sizeof(heap_write_t) + heap->dsk->clean_entry_bitmap_size); +} + +uint8_t *heap_write_t::get_checksums(blockstore_heap_t *heap) +{ + if (!heap->dsk->csum_block_size || !len) + return NULL; + if ((flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE) + return ((uint8_t*)this + sizeof(heap_write_t) + heap->dsk->clean_entry_bitmap_size); + if ((flags & BS_HEAP_TYPE) != BS_HEAP_BIG_WRITE) + return NULL; + return ((uint8_t*)this + sizeof(heap_write_t) + 2*heap->dsk->clean_entry_bitmap_size); +} + +uint32_t *heap_write_t::get_checksum(blockstore_heap_t *heap) +{ + if (heap->dsk->csum_block_size || (flags & BS_HEAP_TYPE) != BS_HEAP_SMALL_WRITE || !len) + return NULL; + return (uint32_t*)((uint8_t*)this + sizeof(heap_write_t) + heap->dsk->clean_entry_bitmap_size); +} + +heap_object_t *heap_object_t::next() +{ + return (heap_object_t*)((uint8_t*)this + size); +} + +heap_write_t *heap_object_t::get_writes() +{ + return (heap_write_t*)((uint8_t*)this + sizeof(heap_object_t)); +} + +uint32_t heap_object_t::calc_crc32c() +{ + return ::crc32c(0, (uint8_t*)&inode, size - ((uint8_t*)(&inode) - (uint8_t*)this)); +} + +uint64_t blockstore_heap_t::get_pg_id(inode_t inode, uint64_t stripe) +{ + uint64_t pg_num = 0; + uint64_t pool_id = (inode >> (64-POOL_ID_BITS)); + auto sh_it = pool_shard_settings.find(pool_id); + if (sh_it != pool_shard_settings.end()) + { + // like map_to_pg() + pg_num = (stripe / sh_it->second.pg_stripe_size) % sh_it->second.pg_count + 1; + } + return ((pool_id << (64-POOL_ID_BITS)) | pg_num); +} + +blockstore_heap_t::blockstore_heap_t(blockstore_disk_t *dsk, uint8_t *buffer_area, int log_level): + dsk(dsk), + buffer_area(buffer_area), + log_level(log_level), + meta_block_count(dsk->meta_area_size/dsk->meta_block_size-1), // first block is the superblock + target_block_free_space(dsk->meta_block_target_free_space), + max_write_entry_size(sizeof(heap_write_t) + 2*dsk->clean_entry_bitmap_size + + (dsk->csum_block_size ? dsk->data_block_size/dsk->csum_block_size*(dsk->data_csum_type & 0xFF) : 4 /*sizeof crc32c*/)) +{ + assert(target_block_free_space < dsk->meta_block_size); + meta_alloc = new allocator_t(meta_block_count); + block_info.resize(meta_block_count); + buffer_by_end.insert((heap_extent_t){ .start = 0, .end = dsk->journal_len }); + buffer_by_size.insert((heap_extent_t){ .start = 0, .end = dsk->journal_len }); + data_alloc = new allocator_t(dsk->block_count); + if (!target_block_free_space) + target_block_free_space = 800; +} + +blockstore_heap_t::~blockstore_heap_t() +{ + for (auto & inf: block_info) + { + if (inf.data) + { + free(inf.data); + } + } + block_info.clear(); + for (auto & mvcc: object_mvcc) + { + if (mvcc.second.entry_copy) + { + free(mvcc.second.entry_copy); + } + } + object_mvcc.clear(); + if (meta_alloc) + { + delete meta_alloc; + } + if (data_alloc) + { + delete data_alloc; + } +} + +// set initially compacted lsn - should be done before loading +void blockstore_heap_t::set_compacted_lsn(uint64_t compacted_lsn) +{ + assert(!next_lsn || next_lsn >= compacted_lsn); + this->compacted_lsn = compacted_lsn; +} + +uint64_t blockstore_heap_t::get_compacted_lsn() +{ + return compacted_lsn; +} + +// EASY PEASY LEMON SQUEEZIE +uint64_t blockstore_heap_t::load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf) +{ + uint64_t entries_loaded = 0; + for (uint64_t buf_offset = 0; buf_offset < size; buf_offset += dsk->meta_block_size) + { + uint32_t block_num = (disk_offset + buf_offset) / dsk->meta_block_size; + assert(block_num < block_info.size()); + uint32_t virtual_free_space = 0; + uint32_t block_offset = 0; + uint32_t block_end = dsk->meta_block_size - sizeof(heap_object_t); + uint32_t src_offset = 0; + while (block_offset < block_end) + { + heap_object_t *obj = (heap_object_t *)(buf + buf_offset + block_offset); + if (!obj->size) + { + break; + } + if (obj->size < sizeof(heap_object_t)) + { + fprintf(stderr, "Warning: Object is too small in metadata block %u at %u (%u bytes), skipping the rest of block\n", + block_num, src_offset, obj->size); +skip_block: + if (fail_on_warn) + abort(); + if (block_offset > 0) + memset((void*)obj, 0, dsk->meta_block_size-block_offset); + break; + } + if (obj->size > dsk->meta_block_size-block_offset) + { + fprintf(stderr, "Warning: Object is too large in metadata block %u at %u (%u bytes), skipping the rest of block\n", + block_num, src_offset, obj->size); + goto skip_block; + } + uint32_t expected_crc32c = obj->calc_crc32c(); + if (obj->crc32c != expected_crc32c) + { + fprintf(stderr, "Warning: Object is corrupt in metadata block %u at %u (crc32c mismatch: expected %08x, got %08x), skipping\n", + block_num, src_offset, expected_crc32c, obj->crc32c); +skip_object: + uint32_t obj_size = obj->size; + src_offset += obj_size; + uint32_t to_copy = dsk->meta_block_size-block_offset-obj_size; + memmove(obj, (uint8_t*)obj + obj_size, to_copy); + continue; + } + if (!obj->write_count) + { + fprintf(stderr, "Warning: Object in metadata block %u at %u does not contain writes, skipping\n", block_num, src_offset); + if (fail_on_warn) + abort(); + goto skip_object; + } + uint64_t to_compact = 0; + bool to_recheck = false; + heap_write_t *wr = obj->get_writes(); + uint32_t calc_obj_size = sizeof(heap_object_t); + uint32_t remove_entry_bytes = 0; + uint32_t remove_entry_count = 0; + uint64_t remove_lsn = 0; + for (uint16_t wr_i = 0; wr_i < obj->write_count; wr_i++) + { + auto sz = wr->get_size(this); + calc_obj_size += sz; + if (calc_obj_size > obj->size) + { + fprintf(stderr, "Warning: Object write entries exceed object size in metadata block %u at %u, skipping object\n", + block_num, src_offset); + if (fail_on_warn) + abort(); + goto skip_object; + } + if (wr->needs_recheck(this)) + { + if (!buffer_area) + { + to_recheck = true; + } + // recheck small write data immediately + else if (!calc_checksums(wr, buffer_area + wr->location, false)) + { + // entry is invalid (not fully written before OSD crash) - remove it and all newer (previous) entries too + remove_entry_bytes = calc_obj_size - sizeof(heap_object_t); + remove_entry_count = wr_i+1; + remove_lsn = wr->lsn; + } + } + if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE) + { + use_buffer_area(obj->inode, wr->location, wr->len); + } + else if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) + { + // Mark data block as used + use_data(obj->inode, wr->location); + } + if (wr->needs_compact(this->compacted_lsn)) + { + to_compact = to_compact ? to_compact : wr->lsn; + } + else if (wr->is_compacted(this->compacted_lsn)) + { + if (wr->can_be_collapsed(this)) + { + virtual_free_space += sz; + } + else + { + // We can't just collapse the object entry when csum_block_size is larger + // than bitmap_granularity, so we add the object into the compact queue + to_compact = to_compact ? to_compact : wr->lsn; + } + } + wr = (heap_write_t*)((uint8_t*)wr + sz); + } + if (obj->write_count == remove_entry_count) + { + // Skip the whole object + goto skip_object; + } + uint64_t lsn = obj->get_writes()->lsn; + auto oid = (object_id){ .inode = obj->inode, .stripe = obj->stripe }; + uint32_t dup_block; + heap_object_t *dup_obj = read_entry(oid, &dup_block); + if (dup_obj != NULL) + { + if (dup_obj->get_writes()->lsn >= lsn) + { + // Object is duplicated on disk + fprintf(stderr, "Warning: Object in metadata block %u at %u is an older duplicate, skipping\n", + block_num, src_offset); + free_object_space(obj->inode, obj->get_writes(), (heap_write_t*)obj->next()); + goto skip_object; + } + else + { + fprintf(stderr, "Warning: Object in metadata block %u at %u is a newer duplicate, overriding\n", + block_num, src_offset); + free_object_space(dup_obj->inode, dup_obj->get_writes(), (heap_write_t*)dup_obj->next()); + compact_block(dup_block, oid); + } + } + if (remove_entry_count) + { + if (log_level > 3) + { + fprintf(stderr, "Notice: %u unfinished writes to %jx:%jx v%jx since lsn %ju, rolling back\n", + remove_entry_count, obj->inode, obj->stripe, obj->get_writes()->version, remove_lsn); + } + uint32_t to_copy = dsk->meta_block_size-block_offset-obj->size + obj->size-sizeof(heap_object_t)-remove_entry_bytes; + if (to_copy < remove_entry_bytes) + { + memset((uint8_t*)obj + sizeof(heap_object_t), 0, remove_entry_bytes); + } + memmove((uint8_t*)obj + sizeof(heap_object_t), (uint8_t*)obj + sizeof(heap_object_t) + remove_entry_bytes, to_copy); + obj->size -= remove_entry_bytes; + obj->write_count -= remove_entry_count; + obj->crc32c = obj->calc_crc32c(); + src_offset += remove_entry_bytes; + } + if (lsn > next_lsn) + { + next_lsn = lsn; + } + if (to_compact) + { + if (compact_queue_lsn.find(oid) == compact_queue_lsn.end()) + { + compact_queue.push_back(oid); + } + compact_queue_lsn[oid] = to_compact; + } + if (to_recheck) + { + recheck_queue.push_back(oid); + } + // btree_map anyway stores std::pair's of 16 bytes size + // so we can store block_offset in it too + block_index[get_pg_id(obj->inode, obj->stripe)][obj->inode][obj->stripe] = (uint32_t)block_num*dsk->meta_block_size + block_offset; + entries_loaded += obj->write_count; + src_offset += obj->size; + block_offset += obj->size; + } + uint8_t *copy = NULL; + if (block_offset > 0) + { + // Do not store free blocks in memory + copy = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk->meta_block_size); + memcpy(copy, buf+buf_offset, block_offset); + memset(copy+block_offset, 0, dsk->meta_block_size-block_offset); + } + block_info[block_num] = { + .used_space = block_offset, + .virtual_free_space = virtual_free_space, + .data = copy, + }; + if (block_offset > 0) + { + mark_allocated_block(block_num); + } + } + return entries_loaded; +} + +void blockstore_heap_t::finish_load() +{ + std::sort(compact_queue.begin(), compact_queue.end(), [this](const object_id & a, const object_id & b) + { + return compact_queue_lsn[a] < compact_queue_lsn[b]; + }); +} + +bool blockstore_heap_t::calc_checksums(heap_write_t *wr, uint8_t *data, bool set) +{ + if (!dsk->csum_block_size) + { + if ((wr->flags & BS_HEAP_TYPE) != BS_HEAP_SMALL_WRITE) + { + return true; + } + // Single checksum + uint32_t *wr_csum = wr->get_checksum(this); + uint32_t real_csum = crc32c(0, data, wr->len); + if (set) + { + *wr_csum = real_csum; + return true; + } + return ((*wr_csum) == real_csum); + } + return calc_block_checksums((uint32_t*)wr->get_checksums(this), data, wr->get_int_bitmap(this), + wr->offset, wr->offset+wr->len, set, NULL); +} + +bool blockstore_heap_t::calc_block_checksums(uint32_t *block_csums, uint8_t *data, uint8_t *bitmap, uint32_t start, uint32_t end, + bool set, std::function bad_block_cb) +{ + bool res = true; + uint32_t pos = start; + uint32_t block_end = (start/dsk->csum_block_size + 1)*dsk->csum_block_size; + uint32_t block_crc = 0; + while (pos < end) + { + if (bitmap) + { + while (pos < end && pos < block_end) + { + if (!bitmap[pos/dsk->bitmap_granularity/8] & (1 << ((pos/dsk->bitmap_granularity) % 8))) + block_crc = crc32c_pad(block_crc, NULL, 0, dsk->bitmap_granularity, 0); + else + block_crc = crc32c(block_crc, data, dsk->bitmap_granularity); + data += dsk->bitmap_granularity; + pos += dsk->bitmap_granularity; + } + } + else + { + block_crc = crc32c(block_crc, data, (end > block_end ? block_end : end) - pos); + pos = (end > block_end ? block_end : end); + } + if (set) + { + *block_csums = block_crc; + } + else if (block_crc != *block_csums) + { + if (bad_block_cb) + { + bad_block_cb(pos-start, *block_csums, block_crc); + res = false; + } + else + return false; + } + block_csums++; + } + return res; +} + +bool blockstore_heap_t::recheck_small_writes(std::function)> read_buffer, int queue_depth) +{ + if (buffer_area) + { + // Already checked + return true; + } + if (in_recheck) + { + // Recheck already entered + return false; + } + if (read_buffer) + { + recheck_cb = read_buffer; + recheck_queue_depth = queue_depth; + } + in_recheck = true; + while (recheck_queue.size() > 0 && recheck_in_progress < recheck_queue_depth) + { + object_id oid = recheck_queue.front(); + recheck_queue.pop_front(); + heap_object_t *obj = read_entry(oid, NULL); + assert(obj); + for (heap_write_t *wr = obj->get_writes(); wr < (heap_write_t*)obj->next(); wr = wr->next(this)) + { + if (wr->needs_recheck(this)) + { + recheck_in_progress++; + uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, wr->len); + recheck_cb(wr->location, wr->len, buf, [this, oid, lsn = wr->lsn, buf]() + { + uint32_t block_num = 0; + heap_object_t *obj = read_entry(oid, &block_num); + if (obj) + { + heap_write_t *wr = obj->get_writes(), *end = (heap_write_t*)obj->next(); + uint32_t wr_i = 0; + while (wr < end && wr->lsn != lsn) + { + wr = wr->next(this); + wr_i++; + } + if (wr < end && !calc_checksums(wr, buf, false)) + { + // Erase all writes to the object from this one to the newest + if (log_level > 3) + { + fprintf(stderr, "Notice: %u unfinished writes to %jx:%jx v%jx since lsn %ju, rolling back\n", + wr_i+1, obj->inode, obj->stripe, obj->get_writes()->version, wr->lsn); + } + auto & inf = block_info.at(block_num); + unmark_allocated_block(block_num); + uint32_t to_move, to_erase; + obj->write_count -= wr_i+1; + if (!obj->write_count) + { + to_erase = obj->size; + to_move = inf.used_space - ((uint8_t*)obj->next() - inf.data); + memmove((uint8_t*)obj, wr->next(this), to_move); + } + else + { + to_erase = (uint8_t*)wr->next(this) - (uint8_t*)obj->get_writes(); + to_move = inf.used_space - ((uint8_t*)wr->next(this) - inf.data); + memmove((uint8_t*)obj->get_writes(), wr->next(this), to_move); + obj->size -= to_erase; + obj->crc32c = obj->calc_crc32c(); + } + memset(inf.data+inf.used_space-to_erase, 0, to_erase); + inf.used_space -= to_erase; + mark_allocated_block(block_num); + reindex_block(block_num, obj); + } + } + free(buf); + recheck_in_progress--; + recheck_small_writes(NULL, 0); + }); + } + } + } + in_recheck = false; + if (!recheck_queue.size() && !recheck_in_progress) + { + auto cb = std::move(recheck_cb); + recheck_queue_depth = 0; + if (cb) + cb(0, 0, NULL, NULL); + return true; + } + return false; +} + +void blockstore_heap_t::reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size) +{ + auto & pool_settings = pool_shard_settings[pool]; + if (pool_settings.pg_count == pg_count && pool_settings.pg_stripe_size == pg_stripe_size) + { + return; + } + uint64_t pool_id = (uint64_t)pool; + std::map>> new_shards; + auto sh_it = block_index.lower_bound((pool_id << (64-POOL_ID_BITS))); + while (sh_it != block_index.end() && (sh_it->first >> (64-POOL_ID_BITS)) == pool_id) + { + for (auto & inode_pair: sh_it->second) + { + inode_t inode = inode_pair.first; + for (auto & pair: inode_pair.second) + { + // like map_to_pg() + uint64_t pg_num = (pair.first / pg_stripe_size) % pg_count + 1; + uint64_t shard_id = (pool_id << (64-POOL_ID_BITS)) | pg_num; + new_shards[shard_id][inode][pair.first] = std::move(pair.second); + } + } + block_index.erase(sh_it++); + } + for (sh_it = new_shards.begin(); sh_it != new_shards.end(); sh_it++) + { + auto & to = block_index[sh_it->first]; + to.swap(sh_it->second); + } + pool_settings = (pool_shard_settings_t){ + .pg_count = pg_count, + .pg_stripe_size = pg_stripe_size, + }; +} + +heap_object_t *blockstore_heap_t::lock_and_read_entry(object_id oid, uint64_t & lsn) +{ + auto obj = read_entry(oid, NULL); + if (!obj) + { + return NULL; + } + lsn = obj->get_writes()->lsn; + auto & mvcc = object_mvcc[(heap_object_lsn_t){ .oid = oid, .lsn = lsn }]; + mvcc.readers++; + if (mvcc.entry_copy) + { + return mvcc.entry_copy; + } + return obj; +} + +heap_object_t *blockstore_heap_t::read_locked_entry(object_id oid, uint64_t lsn) +{ + auto mvcc_it = object_mvcc.find((heap_object_lsn_t){ .oid = oid, .lsn = lsn }); + if (mvcc_it == object_mvcc.end()) + { + return NULL; + } + if (mvcc_it->second.entry_copy) + { + return mvcc_it->second.entry_copy; + } + return read_entry(oid, NULL); +} + +bool blockstore_heap_t::unlock_entry(object_id oid, uint64_t lsn) +{ + auto mvcc_it = object_mvcc.find((heap_object_lsn_t){ .oid = oid, .lsn = lsn }); + if (mvcc_it == object_mvcc.end()) + { + return false; + } + mvcc_it->second.readers--; + if (!mvcc_it->second.readers) + { + if (mvcc_it->second.entry_copy) + { + // Free refcounted data & buffer blocks + heap_object_t *obj = (heap_object_t*)mvcc_it->second.entry_copy; + free_object_space(obj->inode, obj->get_writes(), (heap_write_t*)obj->next(), BS_HEAP_FREE_MVCC); + bool is_last_mvcc = true; + if (mvcc_it != object_mvcc.end()) + { + // object_mvcc may contain multiple copied entries, but always in a whole sequence + auto next_it = std::next(mvcc_it); + if (next_it->first.oid == oid && next_it->second.entry_copy) + is_last_mvcc = false; + } + if (is_last_mvcc && mvcc_it != object_mvcc.begin()) + { + auto prev_it = std::prev(mvcc_it); + if (prev_it->first.oid == oid && prev_it->second.entry_copy) + is_last_mvcc = false; + } + if (is_last_mvcc) + { + // Free data references from the newest object version when the last MVCC is freed + heap_object_t *new_obj = read_entry(oid, NULL); + if (new_obj) + { + free_object_space(new_obj->inode, new_obj->get_writes(), (heap_write_t*)new_obj->next(), BS_HEAP_FREE_MAIN); + } + } + free(mvcc_it->second.entry_copy); + } + object_mvcc.erase(mvcc_it); + } + return true; +} + +heap_object_t *blockstore_heap_t::read_entry(object_id oid, uint32_t *block_num_ptr, bool for_update) +{ + auto pool_pg_id = get_pg_id(oid.inode, oid.stripe); + auto & pg_index = block_index[pool_pg_id]; + auto inode_it = pg_index.find(oid.inode); + if (inode_it == pg_index.end()) + { + return NULL; + } + auto stripe_it = inode_it->second.find(oid.stripe); + if (stripe_it == inode_it->second.end()) + { + return NULL; + } + uint64_t block_pos = stripe_it->second; + uint32_t block_num = block_pos / dsk->meta_block_size; + assert(block_info[block_num].data != NULL); + heap_object_t *obj = (heap_object_t*)(block_info[block_num].data + (block_pos % dsk->meta_block_size)); + assert(obj->inode == oid.inode && obj->stripe == oid.stripe); + if (block_num_ptr) + { + *block_num_ptr = block_num; + } + if (for_update) + { + mvcc_save_copy(obj); + } + return obj; +} + +void blockstore_heap_t::get_compact_range(heap_object_t *obj, uint64_t max_lsn, heap_write_t **begin_wr, heap_write_t **end_wr) +{ + *begin_wr = NULL; + *end_wr = NULL; + heap_write_t *wr = obj->get_writes(); + for (uint16_t wr_i = 0; wr_i < obj->write_count; wr_i++, wr = wr->next(this)) + { + if (wr->is_compacted(max_lsn)) + { + *begin_wr = wr; + } + if (*begin_wr) + { + bool is_last = (wr_i == obj->write_count-1); + if (is_last) + { + *end_wr = wr; + } + // all subsequent small write entries must also be compacted + assert(wr->is_allowed_before_compacted(UINT64_MAX, is_last)); + } + } +} + +bool blockstore_heap_t::compact_object_to(heap_object_t *obj, uint64_t compact_lsn, heap_object_t *to_obj, uint8_t *new_csums) +{ + heap_write_t *wr = obj->get_writes(); + assert(obj->write_count <= 1024); + heap_write_t *compacted_wrs[obj->write_count]; + int compacted_wr_count = 0; + bool has_more = false; + bool skip = false; + heap_write_t *big_wr = NULL; + for (uint16_t wr_i = 0; wr_i < obj->write_count; wr_i++) + { + if (wr->is_compacted(compact_lsn)) + { + compacted_wrs[compacted_wr_count++] = wr; + } + else if (wr->needs_compact(compact_lsn)) + { + has_more = true; + } + if (compacted_wr_count) + { + bool is_last = (wr_i == obj->write_count-1); + if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) + { + big_wr = wr; + } + // all subsequent small write entries must also be compacted + assert(wr->is_allowed_before_compacted(compact_lsn, is_last)); + if (!new_csums && !wr->can_be_collapsed(this)) + { + skip = true; + } + } + wr = wr->next(this); + } + if (compacted_wr_count == 0 || skip) + { + return false; + } + if (!has_more) + { + compact_queue_lsn.erase((object_id){ .inode = obj->inode, .stripe = obj->stripe }); + } + free_object_space(obj->inode, compacted_wrs[0], big_wr); + // Generate a collapsed BIG_WRITE entry + uint8_t collapsed_buf[max_write_entry_size]; + heap_write_t *collapsed_wr = (heap_write_t*)collapsed_buf; + collapsed_wr->lsn = compacted_wrs[0]->lsn; + collapsed_wr->version = compacted_wrs[0]->version; + collapsed_wr->offset = big_wr->offset; + collapsed_wr->len = big_wr->offset + big_wr->len; + collapsed_wr->location = big_wr->location; + collapsed_wr->flags = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE; + memcpy(collapsed_wr->get_ext_bitmap(this), compacted_wrs[0]->get_ext_bitmap(this), dsk->clean_entry_bitmap_size); + memcpy(collapsed_wr->get_int_bitmap(this), big_wr->get_int_bitmap(this), dsk->clean_entry_bitmap_size); + for (int i = 0; i < compacted_wr_count; i++) + { + auto cur_wr = compacted_wrs[i]; + if (collapsed_wr->offset > cur_wr->offset) + { + collapsed_wr->offset = cur_wr->offset; + } + if (collapsed_wr->len < cur_wr->offset+cur_wr->len) + { + collapsed_wr->len = cur_wr->offset+cur_wr->len; + } + } + collapsed_wr->len -= collapsed_wr->offset; + assert(collapsed_wr->get_size(this) <= max_write_entry_size); + uint8_t *int_bmp = collapsed_wr->get_int_bitmap(this); + uint8_t *csums = collapsed_wr->get_checksums(this); + const uint32_t csum_size = (dsk->data_csum_type & 0xFF); + for (int i = compacted_wr_count-1; i >= 0; i--) + { + auto cur_wr = compacted_wrs[i]; + bitmap_set(int_bmp, cur_wr->offset, cur_wr->len, dsk->bitmap_granularity); + // copy checksums + if (csums && !new_csums) + { + assert(i == compacted_wr_count-1 || + (cur_wr->offset % dsk->csum_block_size) == 0 && + (cur_wr->len % dsk->csum_block_size) == 0); + memcpy(csums + (cur_wr->offset/dsk->csum_block_size - collapsed_wr->offset/dsk->csum_block_size)*csum_size, + cur_wr->get_checksums(this), cur_wr->len/dsk->csum_block_size*csum_size); + } + } + if (csums && new_csums) + { + memcpy(csums, new_csums, collapsed_wr->get_csum_size(this)); + } + // Copy it over the old entries + uint32_t copy_wr_bytes = (uint8_t*)compacted_wrs[0] - (uint8_t*)obj; + if (to_obj != obj) + { + memmove(to_obj, obj, copy_wr_bytes); + } + memcpy((uint8_t*)to_obj + copy_wr_bytes, collapsed_wr, collapsed_wr->get_size(this)); + to_obj->write_count -= compacted_wr_count; + to_obj->size = (uint8_t*)compacted_wrs[0] - (uint8_t*)obj + collapsed_wr->get_size(this); + to_obj->crc32c = to_obj->calc_crc32c(); + return true; +} + +void blockstore_heap_t::compact_block(uint32_t block_num, object_id skip_oid) +{ + const uint8_t *data = block_info[block_num].data; + assert(data); + const heap_object_t *block_end = (heap_object_t *)((uint8_t*)data + block_info[block_num].used_space); + heap_object_t *obj = (heap_object_t *)data; + heap_object_t *to_obj = (heap_object_t *)data; + while (obj < block_end && obj->size) + { + heap_object_t *next_obj = obj->next(); + if (obj->inode == skip_oid.inode && obj->stripe == skip_oid.stripe) + { + obj = next_obj; + continue; + } + bool compacted = compact_object_to(obj, compacted_lsn, to_obj, NULL); + if (to_obj != obj) + { + block_index[get_pg_id(obj->inode, obj->stripe)][obj->inode][obj->stripe] = (uint64_t)block_num*dsk->meta_block_size + ((uint8_t*)to_obj - data); + if (!compacted) + { + memmove(to_obj, obj, obj->size); + } + } + to_obj = to_obj->next(); + obj = next_obj; + } + uint32_t new_used_space = (uint8_t*)to_obj - (uint8_t*)data; + if (new_used_space < block_info[block_num].used_space) + { + memset((void*)to_obj, 0, block_info[block_num].used_space - new_used_space); + } + unmark_allocated_block(block_num); + block_info[block_num].used_space = new_used_space; + block_info[block_num].virtual_free_space = 0; + mark_allocated_block(block_num); +} + +int blockstore_heap_t::get_block_for_new_object(uint32_t & out_block_num) +{ + // Blocks with at least target_block_free_space are tried first in number order + uint64_t block_num = meta_alloc->find_free(); + if (block_num >= block_info.size()) + { + // Blocks with less than target_block_free_space are tried second, in free space order + auto u_it = used_alloc_queue.begin(); + if (u_it == used_alloc_queue.end() || u_it->free_space < sizeof(heap_object_t) + 2*max_write_entry_size) + { + return compact_queue.size() ? EAGAIN : ENOSPC; + } + block_num = u_it->block_num; + } + out_block_num = block_num; + return 0; +} + +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 + if ((wr->flags & BS_HEAP_TYPE) != BS_HEAP_BIG_WRITE && + (wr->flags & BS_HEAP_TYPE) != BS_HEAP_TOMBSTONE) + { + return EINVAL; + } + if (!wr->version) + { + wr->version = 1; + } + uint32_t wr_size = wr->get_size(this); + // Allocate block + uint32_t block_num = 0; + int res = get_block_for_new_object(block_num); + if (res != 0) + { + 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); + } + if (modified_block) + { + *modified_block = block_num; + } + // Compact block + if (block_info[block_num].virtual_free_space) + { + compact_block(block_num, {}); + } + block_index[get_pg_id(oid.inode, oid.stripe)][oid.inode][oid.stripe] = (uint64_t)block_num*dsk->meta_block_size + inf.used_space; + // and just append the object entry + heap_object_t *new_entry = (heap_object_t *)(inf.data + inf.used_space); + new_entry->inode = oid.inode; + new_entry->stripe = oid.stripe; + new_entry->write_count = 1; + heap_write_t *new_wr = new_entry->get_writes(); + memcpy(new_wr, wr, wr_size); + new_wr->lsn = ++next_lsn; + if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) + { + uint8_t *int_bitmap = new_wr->get_int_bitmap(this); + memset(int_bitmap, 0, dsk->clean_entry_bitmap_size); + bitmap_set(int_bitmap, wr->offset, wr->len, dsk->bitmap_granularity); + } + if (wr->needs_compact(0) && + compact_queue_lsn.find(oid) == compact_queue_lsn.end()) + { + compact_queue.push_back(oid); + compact_queue_lsn[oid] = new_wr->lsn; + } + new_entry->size = sizeof(heap_object_t) + wr_size; + new_entry->crc32c = new_entry->calc_crc32c(); + unmark_allocated_block(block_num); + inf.used_space += new_entry->size; + mark_allocated_block(block_num); + return 0; +} + +heap_object_t *blockstore_heap_t::mvcc_save_copy(heap_object_t *obj) +{ + auto oid = (object_id){ .inode = obj->inode, .stripe = obj->stripe }; + auto lsn = obj->get_writes()->lsn; + auto mvcc_it = object_mvcc.find((heap_object_lsn_t){ .oid = oid, .lsn = lsn }); + if (mvcc_it == object_mvcc.end()) + { + return NULL; + } + assert(!mvcc_it->second.entry_copy); + heap_object_t *obj_copy = (heap_object_t*)malloc_or_die(obj->size); + memcpy(obj_copy, obj, obj->size); + mvcc_it->second.entry_copy = obj_copy; + uint32_t add_ref = 1; + bool for_obj = false; + // save_copy is performed when the object is modified, so object_mvcc may only + // contain 1 version with entry_copy == NULL + if (mvcc_it == object_mvcc.begin() || std::prev(mvcc_it)->first.oid != oid) + { + // Init refcounts for the copy and for the object itself, when it's the first MVCC entry + add_ref = 2; + for_obj = true; + } + for (auto wr = obj->get_writes(); wr < (heap_write_t*)obj->next(); wr = wr->next(this)) + { + if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) + { + mvcc_data_refs[wr->location] += add_ref; + if (wr->flags & BS_HEAP_STABLE) + { + if (!for_obj) + { + break; + } + add_ref = 1; + } + } + else if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE) + { + mvcc_buffer_refs[wr->location] += add_ref; + } + } + return mvcc_it->second.entry_copy; +} + +int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, heap_write_t *wr, uint32_t *modified_block) +{ + auto oid = (object_id){ .inode = obj->inode, .stripe = obj->stripe }; + uint32_t wr_size = wr->get_size(this); + auto & inf = block_info.at(block_num); + assert(inf.data); + bool is_overwrite = (wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) || wr->flags == (BS_HEAP_TOMBSTONE|BS_HEAP_STABLE)); + uint32_t new_object_size = (is_overwrite ? sizeof(heap_object_t)+wr_size : obj->size+wr_size); + if (dsk->meta_block_size-inf.used_space+inf.virtual_free_space+obj->size < new_object_size) + { + // Something in the block has to be compacted + return compact_queue.size() ? EAGAIN : ENOSPC; + } + if ((obj->get_writes()->flags & BS_HEAP_TYPE) == BS_HEAP_TOMBSTONE && !is_overwrite) + { + // Small overwrites are only allowed over live objects + return EINVAL; + } + if (!(obj->get_writes()->flags & BS_HEAP_STABLE) && (wr->flags & BS_HEAP_STABLE)) + { + // Stable overwrites are not allowed over unstable + return EINVAL; + } + if (wr->version <= obj->get_writes()->version) + { + if (!wr->version) + { + wr->version = obj->get_writes()->version + 1; + } + else + { + // Overwrites with a smaller version are forbidden + return EINVAL; + } + } + if (modified_block) + { + *modified_block = block_num; + } + // Save a copy of the object + bool free_copy = false; + heap_object_t *obj_copy = mvcc_save_copy(obj); + bool tracking_active = !!obj_copy; + if (!tracking_active) + { + auto mvcc_it = object_mvcc.lower_bound((heap_object_lsn_t){ .oid = oid, .lsn = 0 }); + tracking_active = (mvcc_it != object_mvcc.end() && mvcc_it->first.oid == oid && mvcc_it->second.entry_copy); + } + if (!obj_copy) + { + obj_copy = (heap_object_t*)malloc_or_die(obj->size); + memcpy(obj_copy, obj, obj->size); + free_copy = true; + } + if (tracking_active) + { + // MVCC reference tracking is in action for the object, increase the refcount + if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) + { + mvcc_data_refs[wr->location]++; + } + else if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE) + { + mvcc_buffer_refs[wr->location]++; + } + } + // Compact block, skipping the object at the same time + if (is_overwrite) + { + free_object_space(obj->inode, obj->get_writes(), (heap_write_t*)obj->next()); + } + compact_block(block_num, oid); + // Remove block from allocation maps + unmark_allocated_block(block_num); + // Add the object to block again + obj = (heap_object_t*)(inf.data + inf.used_space); + memcpy(obj, obj_copy, sizeof(heap_object_t)); + block_index[get_pg_id(obj->inode, obj->stripe)][obj->inode][obj->stripe] = (uint32_t)block_num*dsk->meta_block_size + inf.used_space; + heap_write_t *new_wr = obj->get_writes(); + memcpy(new_wr, wr, wr_size); + new_wr->lsn = ++next_lsn; + if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) + { + uint8_t *int_bitmap = new_wr->get_int_bitmap(this); + memset(int_bitmap, 0, dsk->clean_entry_bitmap_size); + bitmap_set(int_bitmap, wr->offset, wr->len, dsk->bitmap_granularity); + } + if (wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) || + wr->flags == (BS_HEAP_TOMBSTONE|BS_HEAP_STABLE)) + { + obj->write_count = 1; + obj->size = sizeof(heap_object_t) + wr_size; + } + else + { + memcpy((uint8_t*)new_wr + wr_size, obj_copy->get_writes(), obj_copy->size - sizeof(heap_object_t)); + obj->write_count++; + obj->size += wr_size; + } + if (wr->needs_compact(0) && + compact_queue_lsn.find(oid) == compact_queue_lsn.end()) + { + compact_queue.push_back(oid); + compact_queue_lsn[oid] = new_wr->lsn; + } + obj->crc32c = obj->calc_crc32c(); + inf.used_space += obj->size; + mark_allocated_block(block_num); + if (free_copy) + { + free(obj_copy); + } + return 0; +} + +int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block) +{ + uint32_t block_num = 0; + heap_object_t *obj = read_entry(oid, &block_num); + if (!obj) + { + return add_object(oid, wr, modified_block); + } + return update_object(block_num, obj, wr, modified_block); +} + +int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block) +{ + uint32_t block_num = 0; + heap_object_t *obj = read_entry(oid, &block_num); + if (!obj) + { + // No such object + return ENOENT; + } + auto & inf = block_info.at(block_num); + assert(inf.data); + if (inf.virtual_free_space) + { + compact_block(block_num, {}); + obj = read_entry(oid, &block_num); + assert(obj); + } + assert(obj->write_count > 0); + uint32_t unstable_idx = UINT32_MAX; + uint32_t unstable_big_idx = UINT32_MAX; + heap_write_t *unstable_big_wr = NULL; + heap_write_t *wr = obj->get_writes(); + uint32_t wr_i; + if (wr->version < version) + { + // No such version + return ENOENT; + } + for (wr = obj->get_writes(), wr_i = 0; wr_i < obj->write_count; wr_i++, wr = wr->next(this)) + { + if (!(wr->flags & BS_HEAP_STABLE) && wr->version <= version) + { + unstable_idx = wr_i; + if (unstable_big_idx == UINT32_MAX && + ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE || + (wr->flags & BS_HEAP_TYPE) == BS_HEAP_TOMBSTONE)) + { + unstable_big_wr = wr; + unstable_big_idx = wr_i; + } + } + } + if (unstable_idx == UINT32_MAX) + { + // Version is already stable + return 0; + } + if (modified_block) + { + *modified_block = block_num; + } + // Save a copy of the object + mvcc_save_copy(obj); + if (unstable_big_idx != UINT32_MAX && unstable_big_idx+1 < obj->write_count) + { + // Remove previous stable entry series + unmark_allocated_block(block_num); + assert(unstable_big_wr); + free_object_space(obj->inode, unstable_big_wr->next(this), (heap_write_t*)obj->next()); + auto after_wr = unstable_big_wr->next(this); + uint32_t to_copy = inf.used_space - ((uint8_t*)obj + obj->size - (uint8_t*)inf.data); + uint32_t to_erase = (uint8_t*)obj + obj->size - (uint8_t*)after_wr; + memmove((void*)after_wr, obj->next(), to_copy); + memset(inf.data+inf.used_space-to_erase, 0, to_erase); + obj->size -= to_erase; + obj->write_count = unstable_big_idx+1; + inf.used_space -= to_erase; + reindex_block(block_num, obj->next()); + mark_allocated_block(block_num); + } + // Set the stability flag + uint64_t to_compact = 0; + for (wr = obj->get_writes(), wr_i = 0; wr_i < obj->write_count; wr_i++, wr = wr->next(this)) + { + if (!(wr->flags & BS_HEAP_STABLE) && wr->version <= version) + { + wr->flags |= BS_HEAP_STABLE; + } + if (wr->needs_compact(0)) + { + to_compact = wr->lsn; + } + } + if (to_compact && compact_queue_lsn.find(oid) == compact_queue_lsn.end()) + { + compact_queue.push_back(oid); + compact_queue_lsn[oid] = to_compact; + } + obj->crc32c = obj->calc_crc32c(); + return 0; +} + +int blockstore_heap_t::post_rollback(object_id oid, uint64_t version, uint32_t *modified_block) +{ + uint32_t block_num = 0; + heap_object_t *obj = read_entry(oid, &block_num); + if (!obj) + { + // No such object + return ENOENT; + } + auto & inf = block_info.at(block_num); + assert(inf.data); + if (inf.virtual_free_space) + { + compact_block(block_num, {}); + obj = read_entry(oid, &block_num); + assert(obj); + } + assert(obj->write_count > 0); + uint32_t wr_i; + heap_write_t *wr = obj->get_writes(); + if (wr->version < version) + { + // No such version + return ENOENT; + } + if (wr->version == version && (wr->flags & BS_HEAP_STABLE)) + { + // Already rolled back + return 0; + } + for (wr_i = 0; wr_i < obj->write_count && wr->version > version; wr_i++, wr = wr->next(this)) + { + if (wr->flags & BS_HEAP_STABLE) + { + // Already committed, can't rollback + return EBUSY; + } + } + if (modified_block) + { + *modified_block = block_num; + } + mvcc_save_copy(obj); + if (wr_i >= obj->write_count) + { + erase_object(block_num, obj); + } + else + { + unmark_allocated_block(block_num); + // Erase head versions + heap_write_t *first_wr = obj->get_writes(); + free_object_space(obj->inode, first_wr, wr); + uint32_t to_copy = inf.used_space - ((uint8_t*)wr - (uint8_t*)inf.data); + uint32_t to_erase = (uint8_t*)wr - (uint8_t*)first_wr; + memmove((void*)first_wr, wr, to_copy); + memset(inf.data+inf.used_space-to_erase, 0, to_erase); + obj->size -= to_erase; + obj->write_count -= wr_i; + obj->crc32c = obj->calc_crc32c(); + inf.used_space -= to_erase; + mark_allocated_block(block_num); + reindex_block(block_num, obj->next()); + } + return 0; +} + +int blockstore_heap_t::post_delete(object_id oid, uint32_t *modified_block) +{ + uint32_t block_num = 0; + heap_object_t *obj = read_entry(oid, &block_num); + if (!obj) + { + // No such object + return ENOENT; + } + if (modified_block) + { + *modified_block = block_num; + } + mvcc_save_copy(obj); + auto & inf = block_info.at(block_num); + assert(inf.data); + if (inf.virtual_free_space) + { + free_object_space(obj->inode, obj->get_writes(), (heap_write_t*)obj->next()); + compact_block(block_num, oid); + erase_block_index(oid.inode, oid.stripe); + } + else + { + erase_object(block_num, obj); + } + return 0; +} + +int blockstore_heap_t::get_next_compact(object_id & oid) +{ + auto begin_it = compact_queue.begin(), compact_it = begin_it; + for (; compact_it != compact_queue.end(); compact_it++) + { + auto lsn_it = compact_queue_lsn.find(*compact_it); + if (lsn_it != compact_queue_lsn.end()) + { + oid = *compact_it; + compact_queue.erase(begin_it, compact_it+1); + compact_queue_lsn.erase(lsn_it); + return 0; + } + } + compact_queue.clear(); + return ENOENT; +} + +// FIXME try to use virtual_free_space when possible +int blockstore_heap_t::compact_object(object_id oid, uint64_t compact_lsn, uint8_t *new_csums) +{ + uint32_t block_num = 0; + heap_object_t *obj = read_entry(oid, &block_num); + if (!obj) + { + // No such object + return ENOENT; + } + mvcc_save_copy(obj); + auto & inf = block_info.at(block_num); + uint32_t old_size = obj->size; + int res = EAGAIN; + if (compact_object_to(obj, compact_lsn, obj, new_csums)) + { + unmark_allocated_block(block_num); + uint32_t new_size = obj->size; + uint32_t to_copy = inf.used_space - ((uint8_t*)obj + old_size - (uint8_t*)inf.data); + memmove((uint8_t*)obj + new_size, (uint8_t*)obj + old_size, to_copy); + memset(inf.data+inf.used_space-(old_size-new_size), 0, old_size-new_size); + res = 0; + inf.used_space -= old_size-new_size; + reindex_block(block_num, obj); + mark_allocated_block(block_num); + } + return res; +} + +void blockstore_heap_t::free_object_space(inode_t inode, heap_write_t *from, heap_write_t *to, int mode) +{ + for (heap_write_t *wr = from; wr < to; wr = wr->next(this)) + { + if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) + { + auto ref_it = mvcc_data_refs.find(wr->location); + if (ref_it != mvcc_data_refs.end()) + { + assert(ref_it->second > 0); + ref_it->second--; + if (!ref_it->second) + { + mvcc_data_refs.erase(ref_it); + ref_it = mvcc_data_refs.end(); + } + } + if (ref_it == mvcc_data_refs.end() && mode != BS_HEAP_FREE_MAIN) + { + assert(data_alloc->get(wr->location >> dsk->block_order)); + data_alloc->set(wr->location >> dsk->block_order, false); + auto & space = inode_space_stats[inode]; + assert(space >= dsk->data_block_size); + space -= dsk->data_block_size; + data_used_space -= dsk->data_block_size; + if (!space) + inode_space_stats.erase(inode); + } + if (mode == BS_HEAP_FREE_MVCC && (wr->flags & BS_HEAP_STABLE)) + { + // Stop at the last visible version + break; + } + } + else if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE) + { + auto ref_it = mvcc_buffer_refs.find(wr->location); + if (ref_it != mvcc_buffer_refs.end()) + { + assert(ref_it->second > 0); + ref_it->second--; + if (!ref_it->second) + { + mvcc_buffer_refs.erase(ref_it); + ref_it = mvcc_buffer_refs.end(); + } + } + if (ref_it == mvcc_buffer_refs.end() && mode != BS_HEAP_FREE_MAIN) + { + free_buffer_area(inode, wr->location, wr->len); + } + } + } +} + +void blockstore_heap_t::reindex_block(uint32_t block_num, heap_object_t *from_obj) +{ + auto & inf = block_info.at(block_num); + for (heap_object_t *obj = from_obj; obj < (heap_object_t*)(inf.data + inf.used_space); obj = obj->next()) + { + assert(obj->size > 0); + block_index[get_pg_id(obj->inode, obj->stripe)][obj->inode][obj->stripe] = (uint32_t)block_num*dsk->meta_block_size + ((uint8_t*)obj - inf.data); + } +} + +void blockstore_heap_t::erase_block_index(inode_t inode, uint64_t stripe) +{ + auto & pg_index = block_index[get_pg_id(inode, stripe)]; + auto & inode_index = pg_index[inode]; + inode_index.erase(stripe); + if (!inode_index.size()) + { + pg_index.erase(inode); + } +} + +void blockstore_heap_t::erase_object(uint32_t block_num, heap_object_t *obj) +{ + auto & inf = block_info.at(block_num); + unmark_allocated_block(block_num); + // Erase object + erase_block_index(obj->inode, obj->stripe); + free_object_space(obj->inode, obj->get_writes(), (heap_write_t*)((uint8_t*)obj + obj->size)); + uint32_t to_copy = inf.used_space - ((uint8_t*)obj + obj->size - (uint8_t*)inf.data); + uint32_t to_erase = obj->size; + memmove(obj, (uint8_t*)obj + to_erase, to_copy); + memset(inf.data+inf.used_space-to_erase, 0, to_erase); + inf.used_space -= to_erase; + mark_allocated_block(block_num); + reindex_block(block_num, obj); +} + +void blockstore_heap_t::unmark_allocated_block(uint32_t block_num) +{ + auto & inf = block_info.at(block_num); + meta_used_space -= inf.used_space-inf.virtual_free_space; + if (inf.used_space-inf.virtual_free_space > dsk->meta_block_size-target_block_free_space) + { + meta_alloc_count--; + meta_alloc->set(block_num, false); + used_alloc_queue.erase((heap_block_free_t){ + .block_num = block_num, + .free_space = (uint32_t)(dsk->meta_block_size-inf.used_space+inf.virtual_free_space), + }); + } +} + +void blockstore_heap_t::mark_allocated_block(uint32_t block_num) +{ + auto & inf = block_info.at(block_num); + meta_used_space += inf.used_space-inf.virtual_free_space; + if (inf.used_space-inf.virtual_free_space > dsk->meta_block_size-target_block_free_space) + { + meta_alloc_count++; + meta_alloc->set(block_num, true); + used_alloc_queue.insert((heap_block_free_t){ + .block_num = block_num, + .free_space = (uint32_t)(dsk->meta_block_size-inf.used_space+inf.virtual_free_space), + }); + } +} + +int blockstore_heap_t::list_objects(uint32_t pg_num, uint64_t min_inode, uint64_t max_inode, + obj_ver_id **result_list, size_t *stable_count, size_t *unstable_count) +{ + obj_ver_id *res = NULL; + size_t res_size = 0, res_alloc = 0; + obj_ver_id *unstable = NULL; + size_t unstable_size = 0, unstable_alloc = 0; + uint64_t pool_id = (min_inode >> (64-POOL_ID_BITS)); + if (pool_id == 0 || pool_id != (max_inode >> (64-POOL_ID_BITS))) + { + return EINVAL; + } + auto sh_it = pool_shard_settings.find(pool_id); + uint32_t pg_count = (sh_it != pool_shard_settings.end() ? sh_it->second.pg_count : 0); + if (pg_num == 0 || pg_num > (pg_count == 0 ? 1 : pg_count)) + { + return EINVAL; + } + uint64_t pool_pg_id = (pool_id << (64-POOL_ID_BITS)) | (pg_count == 0 ? 0 : pg_num); + auto first_it = block_index[pool_pg_id].lower_bound(min_inode); + auto last_it = block_index[pool_pg_id].upper_bound(max_inode); + for (auto inode_it = first_it; inode_it != last_it; inode_it++) + { + for (auto & stripe_pair: inode_it->second) + { + auto oid = (object_id){ .inode = inode_it->first, .stripe = stripe_pair.first }; + const uint64_t block_pos = stripe_pair.second; + const uint32_t block_num = block_pos / dsk->meta_block_size; + heap_object_t *obj = (heap_object_t*)(block_info[block_num].data + (block_pos % dsk->meta_block_size)); + assert(obj->inode == oid.inode && obj->stripe == oid.stripe); + heap_write_t *first_wr = obj->get_writes(); + heap_write_t *last_wr = (heap_write_t*)((uint8_t*)obj + obj->size); + uint64_t stable_version = 0; + for (heap_write_t *wr = first_wr; wr < last_wr; wr = wr->next(this)) + { + if (wr->flags & BS_HEAP_STABLE) + { + stable_version = wr->version; + break; + } + } + if (!(first_wr->flags & BS_HEAP_STABLE)) + { + if (unstable_size >= unstable_alloc) + { + unstable_alloc = (!unstable_alloc ? 128 : unstable_alloc*2); + unstable = (obj_ver_id*)realloc_or_die(unstable, sizeof(obj_ver_id) * unstable_alloc); + } + unstable[unstable_size++] = (obj_ver_id){ .oid = oid, .version = stable_version }; + } + if (stable_version) + { + if (res_size >= res_alloc) + { + res_alloc = (!res_alloc ? 128 : res_alloc*2); + res = (obj_ver_id*)realloc_or_die(res, sizeof(obj_ver_id) * res_alloc); + } + res[res_size++] = (obj_ver_id){ .oid = oid, .version = stable_version }; + } + } + } + if (unstable_size) + { + if (res_size+unstable_size > res_alloc) + { + res_alloc = res_size+unstable_size; + res = (obj_ver_id*)realloc_or_die(res, sizeof(obj_ver_id) * res_alloc); + } + memcpy(res + res_size, unstable, sizeof(obj_ver_id) * unstable_size); + free(unstable); + unstable = NULL; + } + *result_list = res; + *stable_count = res_size; + *unstable_count = unstable_size; + return 0; +} + +uint64_t blockstore_heap_t::find_free_data() +{ + uint64_t loc = data_alloc->find_free(); + if (loc != UINT64_MAX) + { + loc = loc << dsk->block_order; + } + return loc; +} + +bool blockstore_heap_t::is_data_used(uint64_t location) +{ + return data_alloc->get(location >> dsk->block_order); +} + +void blockstore_heap_t::use_data(inode_t inode, uint64_t location) +{ + assert(!data_alloc->get(location >> dsk->block_order)); + data_alloc->set(location >> dsk->block_order, true); + inode_space_stats[inode] += dsk->data_block_size; + data_used_space += dsk->data_block_size; +} + +uint64_t blockstore_heap_t::find_free_buffer_area(uint64_t size) +{ + auto free_it = buffer_by_size.lower_bound((heap_extent_t){ .start = 0, .end = size }); + if (free_it == buffer_by_size.end()) + { + return UINT64_MAX; + } + return free_it->start; +} + +bool blockstore_heap_t::is_buffer_area_free(uint64_t location, uint64_t size) +{ + auto free_it = buffer_by_end.lower_bound((heap_extent_t){ .end = location+size }); + return (free_it != buffer_by_end.end() && free_it->start <= location); +} + +void blockstore_heap_t::use_buffer_area(inode_t inode, uint64_t location, uint64_t size) +{ + auto free_it = buffer_by_end.lower_bound((heap_extent_t){ .end = location+size }); + assert(free_it != buffer_by_end.end() && free_it->start <= location && free_it->end >= location+size); + heap_extent_t extent = *free_it; + buffer_by_end.erase(free_it); + buffer_by_size.erase(extent); + if (extent.start == location) + { + extent.start += size; + buffer_by_end.insert(extent); + buffer_by_size.insert(extent); + } + else if (extent.end == location+size) + { + extent.end -= size; + buffer_by_end.insert(extent); + buffer_by_size.insert(extent); + } + else + { + buffer_by_end.insert((heap_extent_t){ .start = extent.start, .end = location }); + buffer_by_size.insert((heap_extent_t){ .start = extent.start, .end = location }); + buffer_by_end.insert((heap_extent_t){ .start = location+size, .end = extent.end }); + buffer_by_size.insert((heap_extent_t){ .start = location+size, .end = extent.end }); + } + buffer_area_used_space += size; +} + +void blockstore_heap_t::free_buffer_area(inode_t inode, uint64_t location, uint64_t size) +{ + auto next_it = buffer_by_end.lower_bound((heap_extent_t){ .end = location+size }); + auto prev_it = next_it == buffer_by_end.begin() ? buffer_by_end.end() : std::prev(next_it); + assert(next_it == buffer_by_end.end() || next_it->start >= location+size); + assert(prev_it == buffer_by_end.end() || prev_it->end <= location); + bool merge_prev = (prev_it != buffer_by_end.end() && prev_it->end == location); + bool merge_next = (next_it != buffer_by_end.end() && next_it->start == location+size); + uint64_t prev_start = merge_prev ? prev_it->start : 0; + uint64_t next_end = merge_next ? next_it->end : 0; + if (merge_prev && merge_next) + { + buffer_by_size.erase(*prev_it); + buffer_by_size.erase(*next_it); + buffer_by_end.erase(prev_it); + buffer_by_end.erase(next_it); + buffer_by_end.insert((heap_extent_t){ .start = prev_start, .end = next_end }); + buffer_by_size.insert((heap_extent_t){ .start = prev_start, .end = next_end }); + } + else if (merge_prev) + { + buffer_by_size.erase(*prev_it); + buffer_by_end.erase(prev_it); + buffer_by_end.insert((heap_extent_t){ .start = prev_start, .end = location+size }); + buffer_by_size.insert((heap_extent_t){ .start = prev_start, .end = location+size }); + } + else if (merge_next) + { + buffer_by_size.erase(*next_it); + buffer_by_end.erase(next_it); + buffer_by_end.insert((heap_extent_t){ .start = location, .end = next_end }); + buffer_by_size.insert((heap_extent_t){ .start = location, .end = next_end }); + } + else + { + buffer_by_end.insert((heap_extent_t){ .start = location, .end = location+size }); + buffer_by_size.insert((heap_extent_t){ .start = location, .end = location+size }); + } + buffer_area_used_space -= size; +} + +uint64_t blockstore_heap_t::get_buffer_area_used_space() +{ + return buffer_area_used_space; +} + +uint8_t *blockstore_heap_t::get_meta_block(uint32_t block_num) +{ + auto & inf = block_info.at(block_num); + return inf.data; +} + +uint32_t blockstore_heap_t::get_meta_block_used_space(uint32_t block_num) +{ + auto & inf = block_info.at(block_num); + return inf.used_space-inf.virtual_free_space; +} + +uint64_t blockstore_heap_t::get_data_used_space() +{ + return data_used_space; +} + +const std::map & blockstore_heap_t::get_inode_space_stats() +{ + return inode_space_stats; +} + +uint64_t blockstore_heap_t::get_meta_total_space() +{ + return (uint64_t)meta_block_count*dsk->meta_block_size; +} + +uint64_t blockstore_heap_t::get_meta_used_space() +{ + return meta_used_space; +} + +uint32_t blockstore_heap_t::get_meta_nearfull_blocks() +{ + return meta_alloc_count; +} + +uint32_t blockstore_heap_t::get_compact_queue_size() +{ + return compact_queue.size(); +} + +uint32_t blockstore_heap_t::get_max_write_entry_size() +{ + return max_write_entry_size; +} + +void blockstore_heap_t::set_fail_on_warn(bool fail) +{ + fail_on_warn = fail; +} diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h new file mode 100644 index 00000000..141d6f5a --- /dev/null +++ b/src/blockstore/blockstore_heap.h @@ -0,0 +1,275 @@ +// Metadata storage version 3 ("heap") +// Copyright (c) Vitaliy Filippov, 2025+ +// License: VNPL-1.1 (see README.md for details) + +#pragma once + +#include +#include +#include +#include + +#include "../client/object_id.h" +#include "../../cpp-btree/btree_map.h" +#include "blockstore_disk.h" + +struct pool_shard_settings_t +{ + uint32_t pg_count; + uint32_t pg_stripe_size; +}; + +#define BS_HEAP_TYPE 3 +#define BS_HEAP_SMALL_WRITE 1 +#define BS_HEAP_BIG_WRITE 2 +#define BS_HEAP_TOMBSTONE 3 +#define BS_HEAP_STABLE 4 + +class blockstore_heap_t; + +struct __attribute__((__packed__)) heap_write_t +{ + uint64_t lsn = 0; + uint64_t version = 0; + uint32_t offset = 0; + uint32_t len = 0; + uint64_t location = 0; + uint8_t flags = 0; // 1|2|3 = small|big|tombstone, 4|0 = stable|unstable + + // uint8_t[] external_bitmap + // uint8_t[] internal_bitmap + // uint32_t[] checksums + + heap_write_t *next(blockstore_heap_t *heap); + uint32_t get_size(blockstore_heap_t *heap); + uint32_t get_csum_size(blockstore_heap_t *heap); + bool needs_recheck(blockstore_heap_t *heap); + bool needs_compact(uint64_t compacted_lsn); + bool is_compacted(uint64_t compacted_lsn); + bool can_be_collapsed(blockstore_heap_t *heap); + bool is_allowed_before_compacted(uint64_t compacted_lsn, bool is_last_entry); + uint8_t *get_ext_bitmap(blockstore_heap_t *heap); + uint8_t *get_int_bitmap(blockstore_heap_t *heap); + uint8_t *get_checksums(blockstore_heap_t *heap); + uint32_t *get_checksum(blockstore_heap_t *heap); +}; + +struct __attribute__((__packed__)) heap_object_t +{ + uint16_t size = 0; + uint32_t crc32c = 0; + uint64_t inode = 0; + uint64_t stripe = 0; + uint16_t write_count = 0; + + // Newest entries are stored first to simplify scanning + // heap_write_t[] writes + + heap_object_t *next(); + heap_write_t *get_writes(); + uint32_t calc_crc32c(); +}; + +struct heap_object_lsn_t +{ + object_id oid; + uint64_t lsn; +}; + +inline bool operator < (const heap_object_lsn_t & a, const heap_object_lsn_t & b) +{ + return a.oid < b.oid || a.oid == b.oid && a.lsn < b.lsn; +} + +struct heap_object_mvcc_t +{ + uint32_t readers = 0; + heap_object_t *entry_copy = NULL; +}; + +struct __attribute__((__packed__)) heap_block_info_t +{ + uint32_t used_space = 0; + uint32_t virtual_free_space = 0; + uint8_t *data = NULL; +}; + +struct __attribute__((__packed__)) heap_block_free_t +{ + uint32_t block_num = 0; + uint32_t free_space = 0; +}; + +inline bool operator < (const heap_block_free_t & a, const heap_block_free_t & b) +{ + return a.free_space > b.free_space || a.free_space == b.free_space && a.block_num < b.block_num; +} + +struct heap_extent_t +{ + uint64_t start = 0; + uint64_t end = 0; +}; + +struct heap_less_extent_by_end +{ + const bool operator()(const heap_extent_t & a, const heap_extent_t & b) const + { + return a.end < b.end; + } +}; + +struct heap_less_extent_by_size +{ + const bool operator()(const heap_extent_t & a, const heap_extent_t & b) const + { + return a.end-a.start < b.end-b.start || a.end-a.start == b.end-b.start && a.start < b.start; + } +}; + +class blockstore_heap_t +{ + friend class heap_write_t; + friend class heap_object_t; + + blockstore_disk_t *dsk = NULL; + uint8_t* buffer_area = NULL; + bool fail_on_warn = false; + int log_level = 0; + + const uint32_t meta_block_count = 0; + uint32_t target_block_free_space = 800; + + uint64_t next_lsn = 0; + uint64_t compacted_lsn = 0; + std::map pool_shard_settings; + // PG => inode => stripe => block number + std::map>> block_index; + std::deque compact_queue; + std::map compact_queue_lsn; + std::vector block_info; + allocator_t *data_alloc = NULL; + allocator_t *meta_alloc = NULL; + uint32_t meta_alloc_count = 0; + uint64_t meta_used_space = 0; + std::set buffer_by_end; + std::set buffer_by_size; + std::set used_alloc_queue; + std::map object_mvcc; + std::map mvcc_data_refs; + std::map mvcc_buffer_refs; + std::map inode_space_stats; + uint64_t buffer_area_used_space = 0; + uint64_t data_used_space = 0; + + std::deque recheck_queue; + int recheck_in_progress = 0; + bool in_recheck = false; + std::function)> recheck_cb; + int recheck_queue_depth = 0; + + 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, object_id skip_oid); + bool compact_object_to(heap_object_t *obj, uint64_t lsn, heap_object_t *to_obj, uint8_t *new_csums); + heap_object_t *mvcc_save_copy(heap_object_t *obj); + int add_object(object_id oid, 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); + void erase_object(uint32_t block_num, heap_object_t *obj); + void reindex_block(uint32_t block_num, heap_object_t *from_obj); + void erase_block_index(inode_t inode, uint64_t stripe); + void free_object_space(inode_t inode, heap_write_t *from, heap_write_t *to, int mode = 0); + void unmark_allocated_block(uint32_t block_num); + void mark_allocated_block(uint32_t block_num); + +public: + blockstore_heap_t(blockstore_disk_t *dsk, uint8_t *buffer_area, int log_level = 0); + ~blockstore_heap_t(); + // set initially compacted lsn - should be done before loading + void set_compacted_lsn(uint64_t compacted_lsn); + uint64_t get_compacted_lsn(); + // load data from the disk, returns count of loaded write entries + uint64_t load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf); + // finish loading + void finish_load(); + // recheck small write data after reading the database from disk + bool recheck_small_writes(std::function)> read_buffer, int queue_depth); + // initialize metadata area (fill it with empty data) + // returns 0 when done, EAGAIN when the caller has to wait more + int initialize(); + // read from the metadata area + // returns 0 when done, EAGAIN when the caller has to wait more + int read(); + // reshard database according to the pool's PG count + void reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size); + // read an object entry and lock it against removal + // in the future, may become asynchronous + heap_object_t *lock_and_read_entry(object_id oid, uint64_t & lsn); + // re-read a locked object entry with the given lsn (pointer may be invalidated) + heap_object_t *read_locked_entry(object_id oid, uint64_t lsn); + // read an object entry without locking it + heap_object_t *read_entry(object_id oid, uint32_t *block_num_ptr, bool for_update = false); + // unlock an entry + bool unlock_entry(object_id oid, uint64_t lsn); + // set or verify checksums in a write request + bool calc_checksums(heap_write_t *wr, uint8_t *data, bool set); + // set or verify raw block checksums + bool calc_block_checksums(uint32_t *block_csums, uint8_t *data, uint8_t *bitmap, uint32_t start, uint32_t end, + 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); + // 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); + // rollback an unstable object version + // return 0 if OK, ENOENT if not exists, EBUSY if already stable + int post_rollback(object_id oid, uint64_t version, uint32_t *modified_block); + // forget an object + // return error code + int post_delete(object_id oid, uint32_t *modified_block); + // get the next object to compact + // guaranteed to return objects in min lsn order + // returns 0 if OK, ENOENT if nothing to compact + int get_next_compact(object_id & oid); + // get the range of an object eligible for compaction + void get_compact_range(heap_object_t *obj, uint64_t max_lsn, heap_write_t **begin_wr, heap_write_t **end_wr); + // mark an object as compacted up to the given lsn + int compact_object(object_id oid, uint64_t lsn, uint8_t *new_csums); + // retrieve object listing from a PG + int list_objects(uint32_t pg_num, uint64_t min_inode, uint64_t max_inode, + obj_ver_id **result_list, size_t *stable_count, size_t *unstable_count); + // set a block number for a new object and returns error status: 0, EAGAIN or ENOSPC + int get_block_for_new_object(uint32_t & out_block_num); + + // data device block allocator functions + uint64_t find_free_data(); + bool is_data_used(uint64_t location); + void use_data(inode_t inode, uint64_t location); + + // buffer device allocator functions + uint64_t find_free_buffer_area(uint64_t size); + bool is_buffer_area_free(uint64_t location, uint64_t size); + void use_buffer_area(inode_t inode, uint64_t location, uint64_t size); + void free_buffer_area(inode_t inode, uint64_t location, uint64_t size); + uint64_t get_buffer_area_used_space(); + + // get metadata block data buffer and used space + uint8_t *get_meta_block(uint32_t block_num); + uint32_t get_meta_block_used_space(uint32_t block_num); + + // get space usage statistics + uint64_t get_data_used_space(); + const std::map & get_inode_space_stats(); + uint64_t get_meta_total_space(); + uint64_t get_meta_used_space(); + uint32_t get_meta_nearfull_blocks(); + uint32_t get_compact_queue_size(); + + // get maximum size for a temporary heap_write_t buffer + uint32_t get_max_write_entry_size(); + + // only for tests + void set_fail_on_warn(bool fail); +}; diff --git a/src/test/CMakeLists.txt b/src/test/CMakeLists.txt index f32152df..59a74fbd 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -34,6 +34,19 @@ add_executable(test_allocator EXCLUDE_FROM_ALL test_allocator.cpp ../util/alloca add_dependencies(build_tests test_allocator) add_test(NAME test_allocator COMMAND test_allocator) +# test_heap +add_executable(test_heap + test_heap.cpp + ../blockstore/blockstore_heap.cpp + ../util/crc32c.c + ../util/allocator.cpp + ../blockstore/blockstore_disk.cpp + ../util/str_util.cpp +) +add_test(NAME test_heap COMMAND test_heap) +target_compile_options(test_heap PRIVATE -coverage) +target_link_options(test_heap PRIVATE -coverage) + # test_cas add_executable(test_cas test_cas.cpp diff --git a/src/test/osd_test.cpp b/src/test/osd_test.cpp index e7f5f453..a4b9c0a4 100644 --- a/src/test/osd_test.cpp +++ b/src/test/osd_test.cpp @@ -195,7 +195,7 @@ uint64_t test_read(int connect_fd, uint64_t inode, uint64_t stripe, uint64_t ver op.sec_rw.version = version; op.sec_rw.offset = offset; op.sec_rw.len = len; - void *data = memalign(MEM_ALIGNMENT, op.sec_rw.len); + void *data = memalign_or_die(MEM_ALIGNMENT, op.sec_rw.len); write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); int r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); if (!check_reply(r, op, reply, op.sec_rw.len)) @@ -222,7 +222,7 @@ uint64_t test_read(int connect_fd, uint64_t inode, uint64_t stripe, uint64_t ver { return 0; } - data = memalign(MEM_ALIGNMENT, sizeof(obj_ver_id)*reply.hdr.retval); + data = memalign_or_die(MEM_ALIGNMENT, sizeof(obj_ver_id)*reply.hdr.retval); r = read_blocking(connect_fd, data, sizeof(obj_ver_id)*reply.hdr.retval); if (r != sizeof(obj_ver_id)*reply.hdr.retval) { @@ -255,7 +255,7 @@ uint64_t test_write(int connect_fd, uint64_t inode, uint64_t stripe, uint64_t ve op.sec_rw.version = version; op.sec_rw.offset = 0; op.sec_rw.len = 128*1024; - void *data = memalign(MEM_ALIGNMENT, op.sec_rw.len); + void *data = memalign_or_die(MEM_ALIGNMENT, op.sec_rw.len); for (int i = 0; i < (op.sec_rw.len)/sizeof(uint64_t); i++) ((uint64_t*)data)[i] = pattern; write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); @@ -290,7 +290,7 @@ void* test_primary_read(int connect_fd, uint64_t inode, uint64_t offset, uint64_ op.rw.inode = inode; op.rw.offset = offset; op.rw.len = len; - void *data = memalign(MEM_ALIGNMENT, len); + void *data = memalign_or_die(MEM_ALIGNMENT, len); write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); int r = read_blocking(connect_fd, reply.buf, OSD_PACKET_SIZE); if (!check_reply(r, op, reply, len)) @@ -318,7 +318,7 @@ void test_primary_write(int connect_fd, uint64_t inode, uint64_t offset, uint64_ op.rw.inode = inode; op.rw.offset = offset; op.rw.len = len; - void *data = memalign(MEM_ALIGNMENT, len); + void *data = memalign_or_die(MEM_ALIGNMENT, len); set_pattern(data, len, pattern); write_blocking(connect_fd, op.buf, OSD_PACKET_SIZE); write_blocking(connect_fd, data, len); @@ -364,7 +364,7 @@ void test_list_stab(int connect_fd) assert(check_reply(r, op, reply, -1)); int total_count = reply.hdr.retval; int stable_count = reply.sec_list.stable_count; - obj_ver_id *data = (obj_ver_id*)malloc(total_count * sizeof(obj_ver_id)); + obj_ver_id *data = (obj_ver_id*)malloc_or_die(total_count * sizeof(obj_ver_id)); assert(data); assert(read_blocking(connect_fd, data, total_count * sizeof(obj_ver_id)) == (total_count * sizeof(obj_ver_id))); int last_start = stable_count; @@ -382,7 +382,7 @@ void test_list_stab(int connect_fd) last_start = i; } } - obj_ver_id *data2 = (obj_ver_id*)malloc(sizeof(obj_ver_id) * 32); + obj_ver_id *data2 = (obj_ver_id*)malloc_or_die(sizeof(obj_ver_id) * 32); assert(data2); free(data2); free(data); diff --git a/src/test/test_heap.cpp b/src/test/test_heap.cpp new file mode 100644 index 00000000..4940f6a0 --- /dev/null +++ b/src/test/test_heap.cpp @@ -0,0 +1,1355 @@ +// Copyright (c) Vitaliy Filippov, 2019+ +// License: VNPL-1.1 (see README.md for details) + +#include +#include +#include +#include "../util/malloc_or_die.h" +#include "../util/allocator.h" +#include "blockstore_heap.h" +#include "../util/crc32c.h" + +int _test_do_big_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64_t inode, uint64_t stripe, uint64_t version, uint64_t location, + bool stable = true, uint32_t offset = 0, uint32_t len = 0) +{ + if (!offset && !len) + len = dsk.data_block_size; + object_id oid = { .inode = INODE_WITH_POOL(1, inode), .stripe = stripe }; + uint8_t wr_buf[heap.get_max_write_entry_size()]; + heap_write_t *wr = (heap_write_t*)wr_buf; + wr->version = version; + wr->offset = offset; + wr->len = len; + wr->location = location; + wr->flags = BS_HEAP_BIG_WRITE | (stable ? BS_HEAP_STABLE : 0); + assert(heap.get_max_write_entry_size() >= wr->get_size(&heap)); + assert(wr->get_size(&heap) == sizeof(heap_write_t) + 2*dsk.clean_entry_bitmap_size + (dsk.csum_block_size + ? ((offset+len+dsk.csum_block_size-1)/dsk.csum_block_size - offset/dsk.csum_block_size)*4 : 0)); + memset(wr->get_ext_bitmap(&heap), 0xff, dsk.clean_entry_bitmap_size); + if (dsk.csum_block_size) + memset(wr->get_checksums(&heap), 0xab, dsk.data_block_size/dsk.csum_block_size*4); + uint32_t mblock; + return heap.post_write(oid, wr, &mblock); +} + +void _test_big_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64_t inode, uint64_t stripe, uint64_t version, uint64_t location, + bool stable = true, uint32_t offset = 0, uint32_t len = 0) +{ + heap.use_data(INODE_WITH_POOL(1, inode), location); // blocks are allocated before write and outside the heap_t + int res = _test_do_big_write(heap, dsk, inode, stripe, version, location, stable, offset, len); + assert(res == 0); + assert(heap.is_data_used(location)); +} + +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) +{ + object_id oid = { .inode = INODE_WITH_POOL(1, inode), .stripe = stripe }; + uint8_t wr_buf[heap.get_max_write_entry_size()]; + heap_write_t *wr = (heap_write_t*)wr_buf; + wr->version = version; + wr->offset = offset; + wr->len = len; + wr->location = location; + wr->flags = BS_HEAP_SMALL_WRITE | (stable ? BS_HEAP_STABLE : 0); + assert(wr->get_size(&heap) == sizeof(heap_write_t) + dsk.clean_entry_bitmap_size + (dsk.csum_block_size + ? ((offset+len+dsk.csum_block_size-1)/dsk.csum_block_size - offset/dsk.csum_block_size)*4 : 4)); + memset(wr->get_ext_bitmap(&heap), 0xff, dsk.clean_entry_bitmap_size); + assert(!wr->get_int_bitmap(&heap)); + if (checksums) + { + if (dsk.csum_block_size) + memcpy(wr->get_checksums(&heap), checksums, wr->get_csum_size(&heap)); + else + *wr->get_checksum(&heap) = *checksums; + } + else if (dsk.csum_block_size) + 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); +} + +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) +{ + 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); + assert(res == 0); + assert(!heap.is_buffer_area_free(location, len)); +} + +void _test_init(blockstore_disk_t & dsk, bool csum) +{ + std::map config; + if (csum) + config["data_csum_type"] = "crc32c"; + dsk.parse_config(config); + dsk.data_device_size = 1*1024*1024*1024; + dsk.meta_device_size = 4*1024*1024; + dsk.journal_device_size = 4*1024*1024; + dsk.data_fd = 0; + dsk.meta_fd = 1; + dsk.journal_fd = 2; + dsk.calc_lengths(); +} + +void test_mvcc(bool csum) +{ + blockstore_disk_t dsk; + _test_init(dsk, csum); + std::vector buffer_area(dsk.journal_device_size); + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + // write, read, modify, check basic mvcc + { + assert(_test_do_small_write(heap, dsk, 1, 0, 1, 0, 4096, 0) == EINVAL); + + assert(heap.find_free_data() == 0); + + _test_big_write(heap, dsk, 1, 0, 1, 0); + assert(heap.get_meta_block_used_space(0) == sizeof(heap_object_t) + sizeof(heap_write_t) + + 2*dsk.clean_entry_bitmap_size + (dsk.csum_block_size ? dsk.data_block_size/dsk.csum_block_size*4 : 0)); + assert(heap.get_meta_used_space() == heap.get_meta_block_used_space(0)); + + assert(heap.find_free_data() == 0x20000); + + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + uint64_t lsn = 0; + heap_object_t *obj = heap.lock_and_read_entry(oid, lsn); + assert(obj); + assert(lsn >= 1); + assert(obj->write_count == 1); + heap_write_t *wr = obj->get_writes(); + assert(wr->lsn == lsn); + assert(wr->version == 1); + assert(wr->offset == 0); + assert(wr->len == dsk.data_block_size); + assert(wr->location == 0); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + uint64_t old_size = obj->size; + + assert(heap.read_locked_entry(oid, lsn) == obj); + + assert(_test_do_small_write(heap, dsk, 1, 0, 1, 0, 4096, 0) == EINVAL); + + _test_small_write(heap, dsk, 1, 0, 2, 8192, 4096, 16384, true); + obj = heap.read_entry(oid, NULL); + assert(heap.get_meta_block_used_space(0) == old_size + obj->get_writes()->get_size(&heap)); + + assert(!heap.read_locked_entry(oid, UINT64_MAX)); + obj = heap.read_locked_entry(oid, lsn); + assert(obj); + assert(obj->write_count == 1); + wr = obj->get_writes(); + assert(wr->lsn == lsn); + assert(wr->version == 1); + assert(wr->offset == 0); + assert(wr->len == dsk.data_block_size); + assert(wr->location == 0); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 2); + wr = obj->get_writes(); + assert(wr->lsn > lsn); + assert(wr->version == 2); + assert(wr->offset == 8192); + assert(wr->len == 4096); + assert(wr->location == 16384); + assert(wr->flags == BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); + assert(!wr->get_int_bitmap(&heap)); + wr = wr->next(&heap); + assert(wr->lsn == lsn); + assert(wr->version == 1); + assert(wr->offset == 0); + assert(wr->len == dsk.data_block_size); + assert(wr->location == 0); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + assert(!heap.unlock_entry(oid, UINT64_MAX)); + assert(heap.unlock_entry(oid, lsn)); + } + + printf("OK test_mvcc %s\n", csum ? "csum" : "no_csum"); +} + +void test_update(bool csum) +{ + blockstore_disk_t dsk; + _test_init(dsk, csum); + std::vector buffer_area(dsk.journal_device_size); + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + { + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + + _test_big_write(heap, dsk, 1, 0x20000, 1, 0x40000); + + _test_small_write(heap, dsk, 1, 0, 2, 8192, 4096, 16384, true); + } + + printf("OK test_update %s\n", csum ? "csum" : "no_csum"); +} + +void test_delete(bool csum) +{ + blockstore_disk_t dsk; + _test_init(dsk, csum); + std::vector buffer_area(dsk.journal_device_size); + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + { + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + + _test_big_write(heap, dsk, 1, 0x20000, 1, 0x40000); + + auto & space = heap.get_inode_space_stats(); + assert(space.at(INODE_WITH_POOL(1, 1)) == 0x40000); + assert(heap.get_data_used_space() == 0x40000); + + object_id oid = { .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }; + uint32_t mblock; + int res = heap.post_delete(oid, &mblock); + assert(res == ENOENT); + assert(mblock == 0); + + oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + res = heap.post_delete(oid, &mblock); + assert(res == 0); + + uint64_t lsn = 0; + heap_object_t *obj = heap.lock_and_read_entry(oid, lsn); + assert(!obj); + } + + printf("OK test_delete %s\n", csum ? "csum" : "no_csum"); +} + +void test_compact(bool csum, bool stable) +{ + int res; + blockstore_disk_t dsk; + _test_init(dsk, csum); + std::vector buffer_area(dsk.journal_device_size); + + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + _test_big_write(heap, dsk, 1, 0, 1, 0x20000, true, 0, 4096); + + // write unstable - stabilize - compact + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + uint64_t lsn = 0; + heap_object_t *obj = heap.lock_and_read_entry(oid, lsn); + assert(obj); + assert(obj->write_count == 1); + assert(obj->get_writes()->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + uint8_t ref_int_bitmap[dsk.clean_entry_bitmap_size]; + memset(ref_int_bitmap, 0, dsk.clean_entry_bitmap_size); + bitmap_set(ref_int_bitmap, 0, 4096, 4096); + assert(!memcmp(obj->get_writes()->get_int_bitmap(&heap), ref_int_bitmap, dsk.clean_entry_bitmap_size)); + uint64_t old_size = obj->size; + + _test_small_write(heap, dsk, 1, 0, 3, 8192, 4096, 16384, stable); + obj = heap.read_entry(oid, NULL); + uint64_t wr_size = obj->get_writes()->get_size(&heap); + assert(heap.get_meta_block_used_space(0) == old_size + wr_size); + + _test_big_write(heap, dsk, 2, 0, 1, 0x40000, true, 0, 4096); + + obj = heap.read_locked_entry(oid, lsn); + assert(obj); + assert(obj->write_count == 1); + assert(obj->get_writes()->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + uint32_t mblock; + object_id compact_oid = {}; + if (!stable) + { + res = heap.get_next_compact(compact_oid); + assert(res == ENOENT); + + res = heap.post_stabilize({ .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }, 3, NULL); + assert(res == ENOENT); + res = heap.post_stabilize(oid, 5, NULL); + assert(res == ENOENT); + res = heap.post_stabilize(oid, 1, &mblock); + assert(res == 0); + res = heap.post_stabilize(oid, 3, &mblock); + assert(res == 0); + assert(mblock == 0); + assert(heap.get_meta_block_used_space(0) == 2*old_size + wr_size); + } + + assert(heap.get_compact_queue_size() == 1); + res = heap.get_next_compact(compact_oid); + assert(res == 0); + assert(oid == compact_oid); + + heap_write_t *compact_begin = NULL, *compact_end = NULL; + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 2); + heap.get_compact_range(obj, UINT64_MAX, &compact_begin, &compact_end); + assert(compact_begin == obj->get_writes()); + assert(compact_end == obj->get_writes()->next(&heap)); + + res = heap.compact_object((object_id){ .inode = INODE_WITH_POOL(1, 3), .stripe = 0 }, compact_begin->lsn, NULL); + assert(res == ENOENT); + + res = heap.compact_object(compact_oid, compact_begin->lsn, NULL); + assert(res == 0); + assert(heap.get_meta_block_used_space(0) == 2*old_size + (dsk.csum_block_size ? 8 /* two extra csum blocks */ : 0)); + + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 1); + assert(obj->get_writes()->version == 3); + bitmap_set(ref_int_bitmap, 8192, 4096, 4096); + assert(!memcmp(obj->get_writes()->get_int_bitmap(&heap), ref_int_bitmap, dsk.clean_entry_bitmap_size)); + + obj = heap.read_entry({ .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }, NULL); + assert(obj); + assert(obj->write_count == 1); + assert(obj->get_writes()->version == 1); + + int unlock_res = heap.unlock_entry(oid, lsn); + assert(unlock_res); + + printf("OK test_compact %s %s\n", stable ? "stable" : "unstable", csum ? "csum" : "no_csum"); +} + +void test_modify_bitmap() +{ + 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(); + + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + + uint64_t lsn = 0; + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + heap_object_t *obj = heap.lock_and_read_entry(oid, lsn); + assert(obj); + + uint32_t modified_block = 1; + obj = heap.read_entry(oid, &modified_block, true); + assert(obj); + assert(modified_block == 0); + uint8_t *bmp = obj->get_writes()->get_int_bitmap(&heap); + bitmap_clear(bmp, 4096, 16384, dsk.bitmap_granularity); + obj->crc32c = obj->calc_crc32c(); + + uint8_t ref_int_bitmap[dsk.clean_entry_bitmap_size]; + memset(ref_int_bitmap, 0xFF, dsk.clean_entry_bitmap_size); + + obj = heap.read_locked_entry(oid, lsn); + assert(obj); + assert(!memcmp(obj->get_writes()->get_int_bitmap(&heap), ref_int_bitmap, dsk.clean_entry_bitmap_size)); + + obj = heap.read_entry(oid, NULL); + assert(obj); + bitmap_clear(ref_int_bitmap, 4096, 16384, dsk.bitmap_granularity); + assert(!memcmp(obj->get_writes()->get_int_bitmap(&heap), ref_int_bitmap, dsk.clean_entry_bitmap_size)); + + int unlock_res = heap.unlock_entry(oid, lsn); + assert(unlock_res); + + printf("OK test_modify_bitmap\n"); +} + +void test_recheck(bool async, bool csum) +{ + blockstore_disk_t dsk; + _test_init(dsk, csum); + std::vector buffer_area(dsk.journal_device_size); + std::vector tmp; + + memset(buffer_area.data(), 0xab, 4096); + uint32_t buf_csum = crc32c(0, buffer_area.data(), 4096); + + // write + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + // object 1 + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + _test_small_write(heap, dsk, 1, 0, 2, 8192, 4096, 16384, true, &buf_csum); + + // object 2 + _test_big_write(heap, dsk, 2, 0, 1, 0x40000); + _test_small_write(heap, dsk, 2, 0, 2, 8192, 4096, 20480, true, &buf_csum); + + // persist + assert(heap.get_meta_block_used_space(0) > 0); + tmp.resize(dsk.meta_block_size); + memcpy(tmp.data(), heap.get_meta_block(0), dsk.meta_block_size); + } + + // reload heap + { + memset(buffer_area.data()+16384, 0, 4096); // invalid data + memset(buffer_area.data()+20480, 0xab, 4096); // valid data + + blockstore_heap_t heap(&dsk, async ? NULL : buffer_area.data()); + heap.load_blocks(0, dsk.meta_block_size, tmp.data()); + heap.finish_load(); + + if (async) + { + int calls = 0; + bool done = heap.recheck_small_writes([&](uint64_t offset, uint64_t len, uint8_t *buf, std::function cb) + { + calls++; + if (len) + { + assert(len == 4096); + assert(offset == 16384 || offset == 20480); + assert(cb); + memcpy(buf, buffer_area.data()+offset, len); + cb(); + } + }, 1); + assert(done); + assert(calls == 3); + } + + // read object 1 - big_write should be there but small_write should be rechecked and removed + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + uint64_t lsn = 0; + heap_object_t *obj = heap.lock_and_read_entry(oid, lsn); + assert(obj); + assert(obj->write_count == 1); + heap_write_t *wr = obj->get_writes(); + assert(wr->lsn == lsn); + assert(wr->version == 1); + assert(wr->offset == 0); + assert(wr->len == dsk.data_block_size); + assert(wr->location == 0x20000); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + // read object 2 - both writes should be present + oid = { .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }; + obj = heap.lock_and_read_entry(oid, lsn); + assert(obj); + assert(obj->write_count == 2); + wr = obj->get_writes(); + assert(wr->lsn == lsn); + assert(wr->version == 2); + assert(wr->offset == 8192); + assert(wr->len == 4096); + assert(wr->location == 20480); + assert(wr->flags == BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); + } + + printf("OK test_recheck %s %s\n", async ? "async" : "sync", csum ? "csum" : "no_csum"); +} + +void test_corruption() +{ + int res; + blockstore_disk_t dsk; + _test_init(dsk, false); + std::vector buffer_area(dsk.journal_device_size); + std::vector tmp; + + // write + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + // big_write + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + + // check size + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + heap_object_t *obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->size == sizeof(heap_object_t) + sizeof(heap_write_t) + dsk.clean_entry_bitmap_size*2); + + // big_write object 2 + _test_big_write(heap, dsk, 1, 0x20000, 1, 0x40000); + + // big_write object 3 + _test_big_write(heap, dsk, 1, 0x40000, 1, 0x60000); + + // tombstone object 4 + oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0x60000 }; + uint8_t wr_buf[heap.get_max_write_entry_size()]; + heap_write_t *wr = (heap_write_t*)wr_buf; + wr->version = 2; + wr->offset = 0; + wr->len = 0; + wr->location = 0; + wr->flags = BS_HEAP_TOMBSTONE|BS_HEAP_STABLE; + assert(!wr->get_checksums(&heap)); + res = heap.post_write(oid, wr, 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); + assert(res == EINVAL); + + // persist + assert(heap.get_meta_block_used_space(0) > 0); + assert(heap.get_meta_block_used_space(1) == 0); + tmp.resize(dsk.meta_block_size); + memcpy(tmp.data(), heap.get_meta_block(0), dsk.meta_block_size); + } + + // reload heap with corruption + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.set_fail_on_warn(true); + tmp.data()[10]++; // corrupt the first object + heap.load_blocks(0, dsk.meta_block_size, tmp.data()); + heap.finish_load(); + + // read object - object should be not present (checksum is invalid) + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + uint64_t lsn = 0; + heap_object_t *obj = heap.lock_and_read_entry(oid, lsn); + assert(!obj); + + // object 2 should be present + oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0x20000 }; + obj = heap.lock_and_read_entry(oid, lsn); + assert(obj); + assert(obj->write_count == 1); + heap_write_t *wr = obj->get_writes(); + assert(wr->location == 0x40000); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + // object 3 should be present + oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0x40000 }; + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 1); + wr = obj->get_writes(); + assert(wr->location == 0x60000); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + // object 4 should be a tombstone + oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0x60000 }; + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 1); + wr = obj->get_writes(); + assert(wr->flags == BS_HEAP_TOMBSTONE|BS_HEAP_STABLE); + } + + printf("OK test_corruption\n"); +} + +void test_full_overwrite(bool stable) +{ + int res; + blockstore_disk_t dsk; + _test_init(dsk, false); + std::vector buffer_area(dsk.journal_device_size); + + // write + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + // big_write + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + + // read it to test mvcc + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + uint64_t read_lsn = 0; + heap_object_t *obj = heap.lock_and_read_entry(oid, read_lsn); + assert(obj); + + // small_write + _test_small_write(heap, dsk, 1, 0, 2, 8192, 4096, 16384, true); + + // big_write again + _test_big_write(heap, dsk, 1, 0, 3, 0x40000, stable, 16384, 4096); + if (stable) + { + assert(heap.is_buffer_area_free(16384, 4096)); // should be freed because it's not in MVCC + } + assert(heap.is_data_used(0x20000)); // should NOT be freed - still referenced by MVCC + + // free mvcc + heap.unlock_entry(oid, read_lsn); + if (stable) + { + assert(!heap.is_data_used(0x20000)); // should now be freed + } + + // small_write again + if (!stable) + { + res = _test_do_small_write(heap, dsk, 1, 0, 4, 20480, 4096, 20480, true); + assert(res == EINVAL); + } + _test_small_write(heap, dsk, 1, 0, 4, 20480, 4096, 20480, stable); + + if (!stable) + { + res = heap.post_stabilize(oid, 4, NULL); + assert(res == 0); + } + + // read object + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 2); + heap_write_t *wr = obj->get_writes(); + assert(wr->version == 4); + assert(wr->location == 20480); + assert(wr->flags == BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); + wr = wr->next(&heap); + assert(wr->version == 3); + assert(wr->location == 0x40000); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + // check that the data block 0x20000 is freed and 0x40000 is used + assert(!heap.is_data_used(0x20000)); + assert(heap.is_data_used(0x40000)); + assert(heap.is_buffer_area_free(16384, 4096)); + assert(!heap.is_buffer_area_free(20480, 4096)); + } + + printf("OK test_full_overwrite %s\n", stable ? "stable" : "unstable"); +} + +void test_reshard_list() +{ + int res; + blockstore_disk_t dsk; + _test_init(dsk, false); + std::vector buffer_area(dsk.journal_device_size); + + // write + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + _test_big_write(heap, dsk, 1, 0x20000, 1, 0x40000); + _test_big_write(heap, dsk, 1, 0x40000, 1, 0); + _test_big_write(heap, dsk, 2, 0x60000, 1, 0x60000); + _test_big_write(heap, dsk, 2, 0x60000, 2, 0x80000, false); + + obj_ver_id *listing = NULL; + size_t stable_count = 0, unstable_count = 0; + res = heap.list_objects(1, INODE_WITH_POOL(0, 1), INODE_WITH_POOL(1, 1), &listing, &stable_count, &unstable_count); + assert(res == EINVAL); + res = heap.list_objects(1, INODE_WITH_POOL(1, 1), INODE_WITH_POOL(2, 1), &listing, &stable_count, &unstable_count); + assert(res == EINVAL); + res = heap.list_objects(2, INODE_WITH_POOL(1, 1), INODE_WITH_POOL(1, 1), &listing, &stable_count, &unstable_count); + assert(res == EINVAL); + + res = heap.list_objects(1, INODE_WITH_POOL(1, 1), INODE_WITH_POOL(1, UINT64_MAX), &listing, &stable_count, &unstable_count); + assert(res == 0); + assert(stable_count == 4); + assert(unstable_count == 1); + free(listing); + listing = NULL; + + res = heap.list_objects(1, INODE_WITH_POOL(1, 1), INODE_WITH_POOL(1, 1), &listing, &stable_count, &unstable_count); + assert(res == 0); + assert(stable_count == 3); + assert(unstable_count == 0); + free(listing); + listing = NULL; + + heap.reshard(1, 2, 0x20000); + + assert(heap.read_entry((object_id){ .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }, NULL)); + assert(heap.read_entry((object_id){ .inode = INODE_WITH_POOL(1, 1), .stripe = 0x20000 }, NULL)); + assert(heap.read_entry((object_id){ .inode = INODE_WITH_POOL(1, 1), .stripe = 0x40000 }, NULL)); + assert(heap.read_entry((object_id){ .inode = INODE_WITH_POOL(1, 2), .stripe = 0x60000 }, NULL)); + assert(!heap.read_entry((object_id){ .inode = INODE_WITH_POOL(1, 2), .stripe = 0x80000 }, NULL)); + + res = heap.list_objects(3, INODE_WITH_POOL(1, 1), INODE_WITH_POOL(1, 1), &listing, &stable_count, &unstable_count); + assert(res == EINVAL); + res = heap.list_objects(1, INODE_WITH_POOL(1, 1), INODE_WITH_POOL(1, 1), &listing, &stable_count, &unstable_count); + assert(res == 0); + assert(stable_count == 2); + assert(unstable_count == 0); + free(listing); + listing = NULL; + + res = heap.list_objects(2, INODE_WITH_POOL(1, 1), INODE_WITH_POOL(1, UINT64_MAX), &listing, &stable_count, &unstable_count); + assert(res == 0); + assert(stable_count == 2); + assert(unstable_count == 1); + free(listing); + listing = NULL; + } + + printf("OK test_reshard_list\n"); +} + +void _test_invalid_data_setup(blockstore_disk_t & dsk, std::vector & buffer_area, std::vector & tmp) +{ + tmp.resize(dsk.meta_block_size*2); + + heap_object_t *obj = (heap_object_t*)tmp.data(); + obj->size = sizeof(heap_object_t) + sizeof(heap_write_t) + dsk.clean_entry_bitmap_size; + obj->inode = INODE_WITH_POOL(1, 1); + obj->write_count = 1; + heap_write_t *wr = obj->get_writes(); + wr->lsn = 1; + wr->version = 1; + wr->flags = BS_HEAP_TOMBSTONE; + obj->crc32c = obj->calc_crc32c(); + + obj = obj->next(); + obj->size = sizeof(heap_object_t) + sizeof(heap_write_t) + dsk.clean_entry_bitmap_size; + obj->inode = INODE_WITH_POOL(1, 3); + obj->stripe = 0; + obj->write_count = 1; + wr = obj->get_writes(); + wr->lsn = 1; + wr->version = 1; + wr->flags = BS_HEAP_TOMBSTONE; + obj->crc32c = obj->calc_crc32c(); + + obj = (heap_object_t*)(tmp.data() + dsk.meta_block_size); + obj->inode = INODE_WITH_POOL(1, 2); + obj->write_count = 1; + wr = obj->get_writes(); + wr->lsn = 2; + wr->version = 1; + wr->flags = BS_HEAP_TOMBSTONE; + obj->size = sizeof(heap_object_t) + sizeof(heap_write_t) + dsk.clean_entry_bitmap_size; + obj->crc32c = obj->calc_crc32c(); +} + +void test_invalid_data() +{ + blockstore_disk_t dsk; + _test_init(dsk, false); + std::vector buffer_area(dsk.journal_device_size); + std::vector tmp; + + // Too small object + { + _test_invalid_data_setup(dsk, buffer_area, tmp); + heap_object_t *obj = (heap_object_t*)tmp.data(); + obj->size = sizeof(heap_object_t)-1; + obj->crc32c = obj->calc_crc32c(); + + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.load_blocks(0, dsk.meta_block_size*2, tmp.data()); + heap.finish_load(); + + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + assert(!heap.read_entry(oid, NULL)); + + oid = { .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }; + assert(heap.read_entry(oid, NULL)); + } + + // Too large object + { + _test_invalid_data_setup(dsk, buffer_area, tmp); + heap_object_t *obj = (heap_object_t*)tmp.data(); + obj->size = dsk.meta_block_size+1; + obj->crc32c = obj->calc_crc32c(); + + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.load_blocks(0, dsk.meta_block_size*2, tmp.data()); + heap.finish_load(); + + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + assert(!heap.read_entry(oid, NULL)); + + oid = { .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }; + assert(heap.read_entry(oid, NULL)); + } + + // No writes + { + _test_invalid_data_setup(dsk, buffer_area, tmp); + heap_object_t *obj = (heap_object_t*)tmp.data(); + obj->write_count = 0; + obj->crc32c = obj->calc_crc32c(); + + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.load_blocks(0, dsk.meta_block_size*2, tmp.data()); + heap.finish_load(); + + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + assert(!heap.read_entry(oid, NULL)); + + oid = { .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }; + assert(heap.read_entry(oid, NULL)); + + oid = { .inode = INODE_WITH_POOL(1, 3), .stripe = 0 }; + assert(heap.read_entry(oid, NULL)); + } + + // Bad crc32c + { + _test_invalid_data_setup(dsk, buffer_area, tmp); + heap_object_t *obj = (heap_object_t*)tmp.data(); + obj->write_count = 1; + obj->crc32c = obj->calc_crc32c()+1; + + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.load_blocks(0, dsk.meta_block_size*2, tmp.data()); + heap.finish_load(); + + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + assert(!heap.read_entry(oid, NULL)); + + oid = { .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }; + assert(heap.read_entry(oid, NULL)); + + oid = { .inode = INODE_WITH_POOL(1, 3), .stripe = 0 }; + assert(heap.read_entry(oid, NULL)); + } + + // Object write size exceeds object size + { + _test_invalid_data_setup(dsk, buffer_area, tmp); + heap_object_t *obj = (heap_object_t*)tmp.data(); + obj->get_writes()->flags = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE; + obj->crc32c = obj->calc_crc32c(); + + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.load_blocks(0, dsk.meta_block_size*2, tmp.data()); + heap.finish_load(); + + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + assert(!heap.read_entry(oid, NULL)); + + oid = { .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }; + assert(heap.read_entry(oid, NULL)); + + oid = { .inode = INODE_WITH_POOL(1, 3), .stripe = 0 }; + assert(heap.read_entry(oid, NULL)); + } + + printf("OK test_invalid_data\n"); +} + +void test_destructor_mvcc() +{ + 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(); + + // some writes + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + + // read it to test mvcc + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + uint64_t read_lsn = 0; + heap_object_t *obj = heap.lock_and_read_entry(oid, read_lsn); + assert(obj); + + _test_small_write(heap, dsk, 1, 0, 2, 8192, 4096, 16384, true); + } + + printf("OK test_destructor_mvcc\n"); +} + +void test_rollback() +{ + int res; + blockstore_disk_t dsk; + _test_init(dsk, false); + std::vector buffer_area(dsk.journal_device_size); + std::vector tmp; + + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + // some writes + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + _test_small_write(heap, dsk, 1, 0, 2, 8192, 4096, 16384, true); + + // read it to test mvcc + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + uint64_t read_lsn = 0; + heap_object_t *obj = heap.lock_and_read_entry(oid, read_lsn); + assert(obj); + + // already stable + uint32_t mblock; + res = heap.post_rollback(oid, 2, &mblock); + assert(res == 0); + res = heap.post_rollback(oid, 1, NULL); + assert(res == EBUSY); + + // unstable writes + _test_big_write(heap, dsk, 1, 0, 3, 0x40000, false, 16384, 4096); + _test_small_write(heap, dsk, 1, 0, 4, 20480, 4096, 20480, false); + + // second read + uint64_t read2_lsn = 0; + obj = heap.lock_and_read_entry(oid, read2_lsn); + assert(obj); + + // rollback + assert(heap.is_data_used(0x20000)); + assert(heap.is_data_used(0x40000)); + assert(!heap.is_buffer_area_free(16384, 4096)); + assert(!heap.is_buffer_area_free(20480, 4096)); + res = heap.post_rollback({ .inode = INODE_WITH_POOL(1, 1), .stripe = 0x20000 }, 2, NULL); + assert(res == ENOENT); + res = heap.post_rollback(oid, 5, NULL); + assert(res == ENOENT); + res = heap.post_rollback(oid, 2, NULL); + assert(res == 0); + assert(heap.is_data_used(0x20000)); + assert(heap.is_data_used(0x40000)); + assert(!heap.is_buffer_area_free(16384, 4096)); + assert(!heap.is_buffer_area_free(20480, 4096)); + + // free second mvcc + heap.unlock_entry(oid, read2_lsn); + assert(heap.is_data_used(0x20000)); + assert(!heap.is_data_used(0x40000)); + assert(!heap.is_buffer_area_free(16384, 4096)); + assert(heap.is_buffer_area_free(20480, 4096)); + + // free first mvcc + heap.unlock_entry(oid, read_lsn); + + // check object data + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 2); + heap_write_t *wr = obj->get_writes(); + assert(wr->version == 2); + assert(wr->location == 16384); + assert(wr->len == 4096); + assert(wr->flags == BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); + wr = wr->next(&heap); + assert(wr->version == 1); + assert(wr->location == 0x20000); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + assert(heap.is_data_used(0x20000)); + assert(!heap.is_data_used(0x40000)); + assert(!heap.is_buffer_area_free(16384, 4096)); + assert(heap.is_buffer_area_free(20480, 4096)); + } + + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0x20000 }; + _test_big_write(heap, dsk, 1, 0x20000, 1, 0x20000, false); + + res = heap.post_rollback(oid, 0, NULL); + assert(res == 0); + + assert(!heap.read_entry(oid, NULL)); + assert(!heap.is_data_used(0x20000)); + } + + printf("OK test_rollback\n"); +} + +void test_alloc_buffer() +{ + 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(); + + uint64_t pos; + + pos = heap.find_free_buffer_area(4*1024*1024+1); + assert(pos == UINT64_MAX); + + pos = heap.find_free_buffer_area(4*1024*1024); + assert(pos == 0); + + pos = heap.find_free_buffer_area(1024*1024); + assert(pos == 0); + heap.use_buffer_area(1, 0, 1024*1024); + assert(heap.get_buffer_area_used_space() == 1024*1024); + + assert(!heap.is_buffer_area_free(1024, 1024)); + assert(heap.is_buffer_area_free(1024*1024+1024, 1024)); + + pos = heap.find_free_buffer_area(1024*1024); + assert(pos == 1024*1024); + heap.use_buffer_area(1, 1024*1024, 1024*1024); + assert(heap.get_buffer_area_used_space() == 2*1024*1024); + + heap.free_buffer_area(1, 0, 1024*1024); + assert(heap.get_buffer_area_used_space() == 1024*1024); + + pos = heap.find_free_buffer_area(2*1024*1024+1); + assert(pos == UINT64_MAX); + + heap.free_buffer_area(1, 1024*1024, 1024*1024); + assert(heap.get_buffer_area_used_space() == 0); + + pos = heap.find_free_buffer_area(4*1024*1024); + assert(pos == 0); + + heap.use_buffer_area(1, 3*1024*1024, 1024*1024); + heap.free_buffer_area(1, 3*1024*1024, 1024*1024); + + printf("OK test_alloc_buffer\n"); +} + +void test_full_alloc() +{ + 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 == 193); + assert(small_write_size == 41); + uint32_t b_4s = (big_write_size + 4*small_write_size); // 357 + for (int j = 0; j < 4; j++) + { + assert(heap.get_meta_nearfull_blocks() == j); + for (int i = j*10; i < j*10+10; i++) + { + _test_big_write(heap, dsk, 1, i*0x20000, 1, i*0x20000); + _test_small_write(heap, dsk, 1, i*0x20000, 2, 8192, 4096, i*16384, true); + _test_small_write(heap, dsk, 1, i*0x20000, 3, 8192, 4096, i*16384+4096, true); + _test_small_write(heap, dsk, 1, i*0x20000, 4, 8192, 4096, i*16384+2*4096, true); + _test_small_write(heap, dsk, 1, i*0x20000, 5, 8192, 4096, i*16384+3*4096, true); + assert(heap.get_meta_block_used_space(0) == (i < 10 ? i+1 : 10)*b_4s); + assert(heap.get_meta_block_used_space(1) == (i < 10 ? 0 : (i < 20 ? i-9 : 10)*b_4s)); + assert(heap.get_meta_block_used_space(2) == (i < 20 ? 0 : (i < 30 ? i-19 : 10)*b_4s)); + assert(heap.get_meta_block_used_space(3) == (i < 30 ? 0 : (i < 40 ? i-29 : 10)*b_4s)); + } + } + + // After filling all blocks to (4096-800), most free blocks should start to be allocated first + for (int i = 0; i < 4; i++) + { + assert(heap.get_meta_nearfull_blocks() == 4); + _test_big_write(heap, dsk, 1, (40+i)*0x20000, 1, (40+i)*0x20000); + assert(heap.get_meta_block_used_space(i) == (10*b_4s + big_write_size)); + } + + // New writes are prevented if it may lead to inability to overwrite any object + // - i.e. if the block doesn't have at least free space as the result + assert(_test_do_big_write(heap, dsk, 1, 44*0x20000, 1, 44*0x20000) == EAGAIN); + + // Overwrites are, however, allowed until the block is almost empty + for (int i = 0; i < 8; i++) + { + assert(_test_do_small_write(heap, dsk, 1, 0, 6+i, 0, 4096, 44*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, 14, 0, 4096, 44*16384+8*4096) == EAGAIN); + + // Check that used_alloc_queue doesn't return used blocks + { + // object from block 2 + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 11*0x20000 }; + int res = heap.post_delete(oid, NULL); + assert(res == 0); + + uint32_t block_num = 0; + assert(!heap.read_entry(oid, &block_num)); + + _test_big_write(heap, dsk, 1, 11*0x20000, 6, 11*0x20000); + assert(heap.read_entry(oid, &block_num)); + assert(block_num == 1); + } + + printf("OK test_full_alloc\n"); +} + +void test_duplicate() +{ + blockstore_disk_t dsk; + _test_init(dsk, false); + std::vector buffer_area(dsk.journal_device_size); + std::vector tmp; + + tmp.resize(dsk.meta_block_size*2); + + // write + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + // big_write + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + + // persist + assert(heap.get_meta_block_used_space(0) > 0); + memcpy(tmp.data(), heap.get_meta_block(0), dsk.meta_block_size); + + // update object + _test_big_write(heap, dsk, 1, 0, 2, 0x40000); + + // persist again to block 2 + assert(heap.get_meta_block_used_space(0) > 0); + memcpy(tmp.data()+dsk.meta_block_size, heap.get_meta_block(0), dsk.meta_block_size); + } + + // reload heap with duplicate + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.set_fail_on_warn(true); + heap.load_blocks(0, 2*dsk.meta_block_size, tmp.data()); + heap.finish_load(); + + // read object - version 2 should be present + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + heap_object_t *obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 1); + heap_write_t *wr = obj->get_writes(); + assert(wr->version == 2); + assert(wr->location == 0x40000); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + assert(heap.get_meta_block_used_space(0) == 0); + assert(heap.get_meta_block_used_space(1) == obj->size); + assert(heap.is_data_used(0x40000)); + assert(!heap.is_data_used(0x20000)); + } + + // reload heap with duplicate in different order + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.set_fail_on_warn(true); + heap.load_blocks(dsk.meta_block_size, dsk.meta_block_size, tmp.data()+dsk.meta_block_size); + heap.load_blocks(0, dsk.meta_block_size, tmp.data()); + heap.finish_load(); + + // read object - version 2 should be present + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + heap_object_t *obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 1); + heap_write_t *wr = obj->get_writes(); + assert(wr->version == 2); + assert(wr->location == 0x40000); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + assert(heap.get_meta_block_used_space(0) == 0); + assert(heap.get_meta_block_used_space(1) == obj->size); + assert(heap.is_data_used(0x40000)); + assert(!heap.is_data_used(0x20000)); + } + + printf("OK test_duplicate\n"); +} + +void test_autocompact(bool csum) +{ + int res; + blockstore_disk_t dsk; + _test_init(dsk, csum); + std::vector buffer_area(dsk.journal_device_size); + std::vector tmp; + + tmp.resize(dsk.meta_block_size); + + uint32_t big_write_size = 0, small_write_size = 0; + + // write + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.finish_load(); + + // some writes + uint32_t buffer_csum = crc32c(0, buffer_area.data()+4*4096, 4096); + _test_big_write(heap, dsk, 1, 0, 1, 0x20000); + _test_small_write(heap, dsk, 1, 0, 2, 4096, 4096, 4*4096, true, &buffer_csum); + _test_small_write(heap, dsk, 1, 0, 3, 3*4096, 4096, 5*4096, true, &buffer_csum); + _test_small_write(heap, dsk, 1, 0, 4, 5*4096, 4096, 6*4096, true, &buffer_csum); + _test_small_write(heap, dsk, 1, 0, 5, 7*4096, 4096, 7*4096, true, &buffer_csum); + + _test_big_write(heap, dsk, 1, 0x40000, 1, 0x60000); + + // check lsn + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + heap_object_t *obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 5); + assert(obj->get_writes()->lsn == 5); + + big_write_size = obj->get_writes()->next(&heap)->next(&heap)->next(&heap)->next(&heap)->get_size(&heap); + small_write_size = obj->get_writes()->get_size(&heap); + + // persist + assert(heap.get_meta_block_used_space(0) == 2*sizeof(heap_object_t) + 2*big_write_size + 4*small_write_size); + memcpy(tmp.data(), heap.get_meta_block(0), dsk.meta_block_size); + } + + // reload heap with autocompaction + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.set_fail_on_warn(true); + heap.set_compacted_lsn(3); + assert(heap.get_compacted_lsn() == 3); + heap.load_blocks(0, dsk.meta_block_size, tmp.data()); + heap.finish_load(); + + // read object - all entries should be present first... + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + heap_object_t *obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 5); + assert(obj->get_writes()->lsn == 5); + + assert(heap.get_meta_block_used_space(0) == 2*sizeof(heap_object_t) + 2*big_write_size + 2*small_write_size); + + // object should be in the compaction queue + object_id compact_oid = {}; + res = heap.get_next_compact(compact_oid); + assert(res == 0); + assert(compact_oid == oid); + + assert(heap.is_data_used(0x20000)); + assert(!heap.is_buffer_area_free(4*4096, 4096)); + assert(!heap.is_buffer_area_free(5*4096, 4096)); + assert(!heap.is_buffer_area_free(6*4096, 4096)); + assert(!heap.is_buffer_area_free(7*4096, 4096)); + + _test_big_write(heap, dsk, 1, 0x20000, 1, 0x40000); + oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0x20000 }; + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 1); + assert(obj->get_writes()->lsn == 7); + + // two entries should be auto-compacted on the first write + oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 3); + heap_write_t *wr = obj->get_writes(); + assert(wr->lsn == 5); + assert(wr->version == 5); + assert(wr->offset == 7*4096); + assert(wr->len == 4096); + assert(wr->location == 7*4096); + assert(wr->flags == BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); + wr = wr->next(&heap); + assert(wr->lsn == 4); + assert(wr->version == 4); + assert(wr->offset == 5*4096); + assert(wr->len == 4096); + assert(wr->location == 6*4096); + assert(wr->flags == BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); + wr = wr->next(&heap); + assert(wr->lsn == 3); + assert(wr->version == 3); + assert(wr->offset == 0); + assert(wr->len == dsk.data_block_size); + assert(wr->location == 0x20000); + assert(wr->flags == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + + // check that blocks are auto-freed + assert(heap.is_data_used(0x20000)); + assert(heap.is_buffer_area_free(4*4096, 4096)); + assert(heap.is_buffer_area_free(5*4096, 4096)); + assert(!heap.is_buffer_area_free(6*4096, 4096)); + assert(!heap.is_buffer_area_free(7*4096, 4096)); + + assert(heap.get_meta_block_used_space(0) == 3*sizeof(heap_object_t) + 3*big_write_size + 2*small_write_size); + } + + // same on delete + { + blockstore_heap_t heap(&dsk, buffer_area.data()); + heap.set_fail_on_warn(true); + heap.set_compacted_lsn(3); + heap.load_blocks(0, dsk.meta_block_size, tmp.data()); + heap.finish_load(); + + object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + heap_object_t *obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 5); + + oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0x40000 }; + int res = heap.post_delete(oid, NULL); + assert(res == 0); + + oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 }; + obj = heap.read_entry(oid, NULL); + assert(obj); + assert(obj->write_count == 3); + heap_write_t *wr = obj->get_writes(); + assert(wr->lsn == 5); + assert(wr->next(&heap)->lsn == 4); + + assert(heap.get_meta_block_used_space(0) == sizeof(heap_object_t) + big_write_size + 2*small_write_size); + } + + printf("OK test_autocompact %s\n", csum ? "csum" : "no_csum"); +} + +int main(int narg, char *args[]) +{ + test_mvcc(true); + test_mvcc(false); + test_update(true); + test_update(false); + test_delete(true); + test_delete(false); + test_compact(true, true); + test_compact(true, false); + test_compact(false, true); + test_compact(false, false); + test_modify_bitmap(); + test_recheck(false, true); + test_recheck(false, false); + test_recheck(true, true); + test_recheck(true, false); + test_corruption(); + test_full_overwrite(true); + test_full_overwrite(false); + test_reshard_list(); + test_invalid_data(); + test_destructor_mvcc(); + test_rollback(); + test_alloc_buffer(); + test_full_alloc(); + test_duplicate(); + test_autocompact(true); + test_autocompact(false); + return 0; +} diff --git a/src/util/malloc_or_die.h b/src/util/malloc_or_die.h index fb28274c..3aa7c37d 100644 --- a/src/util/malloc_or_die.h +++ b/src/util/malloc_or_die.h @@ -13,6 +13,11 @@ #pragma GCC visibility push(default) +// Memory allocation alignment (page size is usually optimal) +#ifndef MEM_ALIGNMENT +#define MEM_ALIGNMENT 4096 +#endif + inline void* memalign_or_die(size_t alignment, size_t size) { void *buf = memalign(alignment, size);