From 98bdcfbff977fb662ffc4bfa591946d866344ec0 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Fri, 29 Aug 2025 01:03:42 +0300 Subject: [PATCH] Add read_blocks() API --- src/blockstore/blockstore_disk.cpp | 12 ++++ src/blockstore/blockstore_heap.cpp | 89 ++++++++++++++++++------------ src/blockstore/blockstore_heap.h | 2 + 3 files changed, 68 insertions(+), 35 deletions(-) diff --git a/src/blockstore/blockstore_disk.cpp b/src/blockstore/blockstore_disk.cpp index 70d0fad4..02a65f98 100644 --- a/src/blockstore/blockstore_disk.cpp +++ b/src/blockstore/blockstore_disk.cpp @@ -364,6 +364,10 @@ static int bs_openmode(const std::string & mode) void blockstore_disk_t::open_data() { + if (data_fd >= 0) + { + throw std::runtime_error("data device is already opened"); + } data_fd = mock_mode ? MOCK_DATA_FD : open(data_device.c_str(), bs_openmode(data_io) | O_RDWR); if (data_fd == -1) { @@ -392,6 +396,10 @@ void blockstore_disk_t::open_data() void blockstore_disk_t::open_meta() { + if (meta_fd >= 0) + { + throw std::runtime_error("metadata device is already opened"); + } if (meta_device != data_device || meta_io != data_io) { meta_fd = mock_mode ? MOCK_META_FD : open(meta_device.c_str(), bs_openmode(meta_io) | O_RDWR); @@ -433,6 +441,10 @@ void blockstore_disk_t::open_meta() void blockstore_disk_t::open_journal() { + if (journal_fd >= 0) + { + throw std::runtime_error("journal device is already opened"); + } if (journal_device != meta_device || journal_io != meta_io) { journal_fd = mock_mode ? MOCK_JOURNAL_FD : open(journal_device.c_str(), bs_openmode(journal_io) | O_RDWR); diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index 0b1e4cbe..cba79518 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -243,14 +243,14 @@ static uint32_t free_writes(heap_write_t *wr, heap_write_t *to) } // EASY PEASY LEMON SQUEEZIE -uint64_t blockstore_heap_t::load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf) +void blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf, + std::function handle_object, std::function handle_block) { - 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 block_offset = 0, used_space = 0; + uint32_t block_offset = 0; std::set offsets_seen; while (block_offset < dsk->meta_block_size - 2) { @@ -463,56 +463,75 @@ skip_object: { compact_object_to(obj, compacted_lsn, NULL, false); } - // Allocate space - used_space += obj->size; - for (auto wr = obj->get_writes(); wr; wr = wr->next()) - { - used_space += wr->size; - if (wr->type() == BS_HEAP_SMALL_WRITE) - { - use_buffer_area(obj->inode, wr->location, wr->len); - } - else if (wr->type() == BS_HEAP_BIG_WRITE) - { - // Mark data block as used - use_data(obj->inode, wr->location); - } - if (wr->lsn > this->compacted_lsn) - { - tmp_compact_queue.push_back((tmp_compact_item_t){ .oid = oid, .lsn = wr->lsn, .compact = wr->needs_compact(this) }); - } - if (wr->lsn > next_lsn) - { - next_lsn = wr->lsn; - } - } 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 += wr_i; + handle_object(obj); block_offset += obj->size; } + handle_block(block_num, block_offset, buf+buf_offset); + } +} + +uint64_t blockstore_heap_t::load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf) +{ + uint64_t entries_loaded = 0; + uint32_t used_space = 0; + read_blocks(disk_offset, size, buf, [&](heap_object_t *obj) + { + // Allocate space + used_space += obj->size; + uint32_t wr_i = 0; + for (auto wr = obj->get_writes(); wr; wr = wr->next(), wr_i++) + { + used_space += wr->size; + if (wr->type() == BS_HEAP_SMALL_WRITE) + { + use_buffer_area(obj->inode, wr->location, wr->len); + } + else if (wr->type() == BS_HEAP_BIG_WRITE) + { + // Mark data block as used + use_data(obj->inode, wr->location); + } + if (wr->lsn > this->compacted_lsn) + { + tmp_compact_queue.push_back((tmp_compact_item_t){ + .oid = (object_id){ .inode = obj->inode, .stripe = obj->stripe }, + .lsn = wr->lsn, + .compact = wr->needs_compact(this), + }); + } + if (wr->lsn > next_lsn) + { + next_lsn = wr->lsn; + } + } + // 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] = (uint8_t*)obj - buf + disk_offset; + entries_loaded += wr_i; + }, [&](uint32_t block_num, uint32_t last_offset, uint8_t *buf) + { uint8_t *copy = NULL; - if (block_offset > 0) + if (used_space > 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); + memcpy(copy, buf, last_offset); + memset(copy+last_offset, 0, dsk->meta_block_size-last_offset); } block_info[block_num] = { .used_space = 0, .data = copy, }; - if (block_offset > 0) + if (used_space > 0) { add_used_space(block_num, used_space); } - } + used_space = 0; + }); return entries_loaded; } diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index eec666a6..16afc0d6 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -198,6 +198,8 @@ public: void set_compacted_lsn(uint64_t compacted_lsn); uint64_t get_compacted_lsn(); // load data from the disk, returns count of loaded write entries + void read_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf, + std::function handle_object, std::function handle_block); uint64_t load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf); // finish loading void finish_load();