From 852734270e47e8e5d60745b816788fc574dbaf02 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Tue, 21 Oct 2025 00:55:15 +0300 Subject: [PATCH] wip --- src/blockstore/blockstore_heap.cpp | 156 +++++++++++----------------- src/blockstore/blockstore_heap.h | 1 + src/blockstore/blockstore_impl.cpp | 12 ++- src/blockstore/blockstore_impl.h | 8 +- src/blockstore/blockstore_write.cpp | 5 +- src/disk_tool/disk_tool_meta.cpp | 4 +- src/disk_tool/disk_tool_resize.cpp | 23 ++-- 7 files changed, 90 insertions(+), 119 deletions(-) diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index d869bd72..bc9b0d1d 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -310,56 +310,20 @@ int blockstore_heap_t::load_blocks(uint64_t disk_offset, uint64_t size, uint8_t entries_loaded++; auto & inode_idx = block_index[get_pg_id(wr->inode, wr->stripe)][wr->inode]; auto & idx = inode_idx[wr->stripe]; - if (!idx.ptr) + insert_list_item(idx, li); + if (li->next && (li->next->entry.is_overwrite() || li->next->entry.is_garbage())) { - idx.ptr = li; + // Mark as garbage + wr->set_garbage(); + block_info[li->block_num].has_garbage = true; } - else + else if (wr->is_overwrite()) { - auto prev_li = idx.ptr; - if (!prev_li || prev_li->entry.is_before(wr)) + // Mark all previous entries as garbage + for (auto prev_li = li->prev; prev_li; prev_li = prev_li->prev) { - li->prev = prev_li; - if (prev_li) - { - prev_li->next = li; - } - if (wr->is_overwrite()) - { - // Mark all previous entries as garbage - while (prev_li) - { - prev_li->entry.set_garbage(); - block_info[prev_li->block_num].has_garbage = true; // FIXME modify_alloc - prev_li = prev_li->prev; - } - } - // Insert on top - idx.ptr = li; - } - else - { - while (true) - { - auto prev_prev = prev_li->prev; - if (!prev_prev || prev_prev->entry.is_before(wr)) - { - break; - } - prev_li = prev_prev; - } - if (prev_li->entry.is_overwrite()) - { - // Mark as garbage - wr->set_garbage(); - block_info[li->block_num].has_garbage = true; - } - // Insert before - li->prev = prev_li->prev; - if (prev_li->prev) - prev_li->prev->next = li; - prev_li->prev = li; - li->next = prev_li; + prev_li->entry.set_garbage(); + block_info[prev_li->block_num].has_garbage = true; // FIXME modify_alloc } } if (!wr->is_garbage()) @@ -390,16 +354,22 @@ void blockstore_heap_t::fill_recheck_queue() { for (auto & op: ip.second) { - auto wr = &op.second.ptr->entry; - bool prev_intent = false; - while (wr) + auto obj = &op.second.ptr->entry; + if (obj->type() == BS_HEAP_INTENT_WRITE) { - if ((wr->type() == BS_HEAP_SMALL_WRITE || wr->type() == BS_HEAP_INTENT_WRITE && !prev_intent) && wr->small().len > 0) + // Recheck only the latest intent_write + recheck_queue.push_back(obj); + } + else + { + // Or a series of small_writes + for (auto wr = obj; wr && wr->type() == BS_HEAP_SMALL_WRITE; wr = prev(wr)) { - recheck_queue.push_back(wr); + if (wr->small().len > 0) + { + recheck_queue.push_back(wr); + } } - prev_intent = wr->type() == BS_HEAP_INTENT_WRITE; - wr = prev(wr); } } } @@ -440,7 +410,7 @@ void blockstore_heap_t::mark_used_blocks() void blockstore_heap_t::recheck_buffer(heap_entry_t *cwr, uint8_t *buf) { - if (cwr->size & FREE_SPACE_BIT) + if (cwr->size & FREE_SPACE_BIT) // FIXME { // Already freed return; @@ -516,22 +486,22 @@ bool blockstore_heap_t::recheck_small_writes(std::functionsize & FREE_SPACE_BIT) - { - // Already freed - continue; - } bool is_intent = wr->type() == BS_HEAP_INTENT_WRITE; uint64_t loc = wr->small().location; if (is_intent) { auto prev_wr = prev(wr); - assert(prev_wr && (prev_wr->entry_type == (BS_HEAP_BIG_WRITE | (wr->entry_type & BS_HEAP_STABLE)) || prev_wr->entry_type == wr->entry_type)); + if (!prev_wr || prev_wr->entry_type != (BS_HEAP_BIG_WRITE | (wr->entry_type & BS_HEAP_STABLE)) && prev_wr->entry_type != wr->entry_type) + { + fprintf(stderr, "Error: intent_write entry %jx:%jx v%ju l%ju is not written over a big_write\n", + wr->inode, wr->stripe, wr->version, wr->lsn); + exit(1); + } loc = wr->small().offset + prev_wr->big_location(this); } if (log_level > 5) { - fprintf(stderr, "Notice: rechecking %u bytes at %ju in %s area (lsn %lu)\n", + fprintf(stderr, "Notice: rechecking %u bytes at %ju in %s area (lsn %ju)\n", wr->small().len, loc, is_intent ? "data" : "buffer", wr->lsn); } if (!is_intent && buffer_area) @@ -906,6 +876,36 @@ int blockstore_heap_t::allocate_entry(uint32_t entry_size, uint32_t *block_num, return 0; } +void blockstore_heap_t::insert_list_item(heap_idx_t & idx, heap_list_item_t *li) +{ + auto old_head = idx.ptr; + if (old_head && !old_head->entry.is_before(&li->entry)) + { + // BIG_WRITE may be inserted into the middle of the sequence during compaction + // and it overrides SMALL_WRITEs and COMMITs with the same LSN + // However, all entries of other types (say DELETE) override previous ones + auto prev_li = old_head->prev; + while (prev_li && !prev_li->entry.is_before(&li->entry)) + { + prev_li = prev_li->prev; + } + // Insert
  • between and + li->prev = prev_li; + if (prev_li) + prev_li->next = li; + old_head->prev = li; + li->next = old_head; + } + else + { + li->prev = idx.ptr; + li->next = NULL; + if (idx.ptr) + idx.ptr->next = li; + idx.ptr = li; + } +} + int blockstore_heap_t::add_entry(uint32_t wr_size, uint32_t *modified_block, bool allow_last_free, std::function fill_entry) { @@ -936,38 +936,7 @@ int blockstore_heap_t::add_entry(uint32_t wr_size, uint32_t *modified_block, (new_wr->is_overwrite() ? HEAP_INFLIGHT_COMPACTED : 0) | (new_wr->is_compactable() ? HEAP_INFLIGHT_COMPACTABLE : 0)); auto & idx = block_index[get_pg_id(oid.inode, oid.stripe)][oid.inode][oid.stripe]; - auto old_head = idx.ptr; - if (old_head && !old_head->entry.is_before(new_wr)) - { - // BIG_WRITE may be inserted into the middle of the sequence during compaction - // and it overrides SMALL_WRITEs and COMMITs with the same LSN - // However, all entries of other types (say DELETE) override previous ones - auto next_li = old_head; - while (true) - { - if (!next_li->prev || next_li->prev->entry.is_before(new_wr)) - break; - next_li = next_li->prev; - } - auto prev_wr = &next_li->prev->entry; - // may be an identical big_write entry when we "punch holes" in the bitmap - assert(prev_wr && prev_wr->type() != BS_HEAP_DELETE && - (prev_wr->type() != BS_HEAP_BIG_WRITE || prev_wr->version == new_wr->version)); - // Insert between and - li->prev = next_li->prev; - if (next_li->prev) - next_li->prev->next = li; - next_li->prev = li; - li->next = next_li; - } - else - { - li->prev = idx.ptr; - li->next = NULL; - if (idx.ptr) - idx.ptr->next = li; - idx.ptr = li; - } + insert_list_item(idx, li); li->block_num = block_num; new_wr->size = wr_size; new_wr->crc32c = new_wr->calc_crc32c(); @@ -1712,6 +1681,7 @@ void blockstore_heap_t::get_meta_block(uint32_t block_num, uint8_t *buffer) memcpy(buffer+pos, &li->entry, li->entry.size); pos += li->entry.size; } + assert(pos <= dsk->meta_block_size); memset(buffer+pos, 0, dsk->meta_block_size-pos); } diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index c3e50410..2680cbe4 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -180,6 +180,7 @@ class blockstore_heap_t void defragment_block(uint32_t block_num); int allocate_entry(uint32_t entry_size, uint32_t *block_num, bool allow_last_free); + void insert_list_item(heap_idx_t & idx, heap_list_item_t *li); int add_entry(uint32_t wr_size, uint32_t *modified_block, bool allow_last_free, std::function fill_entry); int add_simple(heap_entry_t *obj, uint64_t version, uint32_t *modified_block, uint32_t entry_type); diff --git a/src/blockstore/blockstore_impl.cpp b/src/blockstore/blockstore_impl.cpp index 80c95b29..0d5cea00 100644 --- a/src/blockstore/blockstore_impl.cpp +++ b/src/blockstore/blockstore_impl.cpp @@ -192,16 +192,18 @@ void blockstore_impl_t::loop() { flusher->loop(); } + for (auto & block_num: pending_modified_blocks) + { + auto & mb = modified_blocks[block_num]; + heap->get_meta_block(block_num, mb.buf); + heap->start_block_write(block_num); + mb.sent = true; + } int ret = ringloop->submit(); if (ret < 0) { throw std::runtime_error(std::string("io_uring_submit: ") + strerror(-ret)); } - for (auto & block_num: pending_modified_blocks) - { - heap->start_block_write(block_num); - modified_blocks[block_num] = true; - } pending_modified_blocks.clear(); if ((initial_ring_space - ringloop->space_left()) > 0) { diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index a05d0c90..9d31b99f 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -71,6 +71,12 @@ struct blockstore_op_private_t timespec tv_begin; }; +struct bs_modified_block_t +{ + bool sent; + uint8_t *buf; +}; + class blockstore_impl_t: public blockstore_i { public: @@ -117,7 +123,7 @@ public: uint8_t *zero_object = NULL; std::vector pending_modified_blocks; - robin_hood::unordered_flat_map modified_blocks; + robin_hood::unordered_flat_map modified_blocks; journal_flusher_t *flusher; int write_iodepth = 0; diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index 0a543d53..1476f8bd 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -17,11 +17,8 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block) return; io_uring_sqe *sqe = get_sqe(); assert(sqe != NULL); - pending_modified_blocks.push_back(modified_block); - modified_blocks[modified_block] = false; ring_data_t *data = ((ring_data_t*)sqe->user_data); uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size); - heap->get_meta_block(modified_block, buf); data->iov = (struct iovec){ buf, (size_t)dsk.meta_block_size }; data->callback = [this, modified_block, buf](ring_data_t *data) { @@ -39,6 +36,8 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block) io_uring_prep_writev( sqe, dsk.meta_fd, &data->iov, 1, dsk.meta_offset + (modified_block+1)*dsk.meta_block_size ); + pending_modified_blocks.push_back(modified_block); + modified_blocks[modified_block] = { .sent = false, .buf = buf }; } bool blockstore_impl_t::meta_block_is_pending(uint32_t modified_block) diff --git a/src/disk_tool/disk_tool_meta.cpp b/src/disk_tool/disk_tool_meta.cpp index 7bee467c..54e6700e 100644 --- a/src/disk_tool/disk_tool_meta.cpp +++ b/src/disk_tool/disk_tool_meta.cpp @@ -108,9 +108,9 @@ close_error: { uint64_t read_len = buf_size < dsk.meta_area_size-meta_pos ? buf_size : dsk.meta_area_size-meta_pos; read_blocking(dsk.meta_fd, data, read_len); - heap->read_blocks(meta_pos-dsk.meta_block_size, read_len, data, [&](heap_entry_t *obj) + heap->read_blocks(meta_pos-dsk.meta_block_size, read_len, data, [&](uint32_t block_num, heap_entry_t *obj) { - obj_fn(heap, obj, ((uint8_t*)obj-data+meta_pos)/dsk.meta_block_size); + obj_fn(heap, obj, block_num); }, [](uint32_t, uint32_t, uint8_t*){}); meta_pos += read_len; } diff --git a/src/disk_tool/disk_tool_resize.cpp b/src/disk_tool/disk_tool_resize.cpp index 526e8157..e5ed0c70 100644 --- a/src/disk_tool/disk_tool_resize.cpp +++ b/src/disk_tool/disk_tool_resize.cpp @@ -733,25 +733,18 @@ int disk_tool_t::resize_write_new_meta() { assert(new_heap); uint32_t new_meta_blocks = new_meta_len / dsk.meta_block_size - 1; - uint8_t *zero_block = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size); - memset(zero_block, 0, dsk.meta_block_size); - std::vector iov; - iov.reserve(IOV_MAX); - iov.push_back((iovec){ .iov_base = new_meta_hdr, .iov_len = dsk.meta_block_size }); - for (uint32_t i = 0; i < new_meta_blocks; i++) + const uint32_t nb = 1024; + uint8_t *data = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size*nb); + write_blocking(new_meta_fd, new_meta_hdr, dsk.meta_block_size); + for (uint32_t i = 0; i < new_meta_blocks; ) { - uint8_t *data = new_heap->get_meta_block(i); - iov.push_back((iovec){ .iov_base = data ? data : zero_block, .iov_len = dsk.meta_block_size }); - if (iov.size() >= IOV_MAX) + uint32_t j = 0; + for (j = 0; j < nb && i < new_meta_blocks; j++, i++) { - writev_blocking(new_meta_fd, iov.data(), iov.size()); - iov.clear(); + new_heap->get_meta_block(i, data + j*dsk.meta_block_size); } + write_blocking(new_meta_fd, data, j*dsk.meta_block_size); } - if (iov.size() > 0) - writev_blocking(new_meta_fd, iov.data(), iov.size()); - free(zero_block); - zero_block = NULL; } fsync(new_meta_fd); close(new_meta_fd);