From 622cc559249d0ee1b7dfd60b608a60880404d0d4 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Thu, 26 Jun 2025 01:29:50 +0300 Subject: [PATCH] Use linked list heap in blockstore code WIP, still slower than the old version :-E --- src/blockstore/blockstore_flush.cpp | 27 ++++++++++++++++------- src/blockstore/blockstore_flush.h | 3 +++ src/blockstore/blockstore_impl.cpp | 4 ++-- src/blockstore/blockstore_read.cpp | 4 ++-- src/blockstore/blockstore_write.cpp | 33 +++++++++++++++++------------ 5 files changed, 46 insertions(+), 25 deletions(-) diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index e8023d10..245cdd4c 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -62,6 +62,16 @@ journal_flusher_co::~journal_flusher_co() } } +int journal_flusher_t::get_active() +{ + return active_flushers; +} + +uint64_t journal_flusher_t::get_counter() +{ + return compact_counter; +} + bool journal_flusher_t::is_active() { return active_flushers > 0 || bs->heap->get_compact_queue_size() > (force_start > 0 ? 0 : bs->flusher_start_threshold); @@ -107,8 +117,11 @@ void journal_flusher_t::loop() cur_flusher_count--; } } + int prev_active = active_flushers; for (int i = 0; is_active() && i < cur_flusher_count; i++) co[i].loop(); + if (prev_active && !active_flushers && force_start > 0) + bs->ringloop->wakeup(); } #define await_sqe(label) \ @@ -168,7 +181,7 @@ resume_1: goto resume_0; } compact_lsn = begin_wr->lsn; - assert(end_wr < (heap_write_t*)cur_obj->next() && end_wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)); + assert(!end_wr->next() && end_wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)); clean_loc = end_wr->location; // "Lock" object for flushing repeat_it = flusher->sync_to_repeat.find(cur_oid); @@ -194,7 +207,7 @@ resume_1: flusher->active_flushers++; // Scan versions to flush read_vec.clear(); - for (auto wr = begin_wr; wr != end_wr; wr = wr->next(bs->heap)) + for (auto wr = begin_wr; wr != end_wr; wr = wr->next()) { min_compact_lsn = wr->lsn; bs->prepare_read(read_vec, cur_obj, wr, 0, bs->dsk.data_block_size); @@ -206,8 +219,7 @@ resume_1: overwrite_end = read_vec[read_vec.size()-1].offset + read_vec[read_vec.size()-1].len; } read_to_fill_incomplete = false; - if (bs->dsk.csum_block_size > bs->dsk.bitmap_granularity && - end_wr < (heap_write_t*)cur_obj->next()) + if (bs->dsk.csum_block_size > bs->dsk.bitmap_granularity && end_wr->next()) { // Read original checksum blocks to calculate padded checksums if required fill_partial_checksum_blocks(); @@ -271,13 +283,12 @@ resume_12: resume_13: if (copy_count && !fsync_batch(false, 11)) return false; + bs->heap->unlock_entry(cur_oid, cur_lsn); // Modify the metadata entry; don't write anything. Metadata block will be written on the next write calc_block_checksums(); bs->heap->compact_object(cur_oid, cur_lsn, new_data_csums); // Done, free all buffers free_buffers(); - // Unlock entry and free referenced block only after fsync - bs->heap->unlock_entry(cur_oid, cur_lsn); #ifdef BLOCKSTORE_DEBUG printf("Compacted %jx:%jx v%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, cur_version, copy_count); #endif @@ -296,16 +307,16 @@ release_oid: repeat_it = flusher->sync_to_repeat.find(cur_oid); do_repeat = (repeat_it != flusher->sync_to_repeat.end() && repeat_it->second > cur_version); flusher->sync_to_repeat.erase(repeat_it); + flusher->active_flushers--; if (do_repeat) { // Flush the same object again goto resume_1; } // All done - flusher->active_flushers--; + flusher->compact_counter++; wait_state = 0; goto resume_0; - return true; } void journal_flusher_co::iterate_partial_overwrites(std::function cb) diff --git a/src/blockstore/blockstore_flush.h b/src/blockstore/blockstore_flush.h index e96589b5..7a605e19 100644 --- a/src/blockstore/blockstore_flush.h +++ b/src/blockstore/blockstore_flush.h @@ -95,6 +95,7 @@ class journal_flusher_t friend class journal_flusher_co; int advance_lsn_counter = 0; + uint64_t compact_counter = 0; int active_flushers = 0; int syncing_flushers = 0; @@ -105,6 +106,8 @@ public: journal_flusher_t(blockstore_impl_t *bs); ~journal_flusher_t(); void loop(); + int get_active(); + uint64_t get_counter(); bool is_active(); void request_trim(); void release_trim(); diff --git a/src/blockstore/blockstore_impl.cpp b/src/blockstore/blockstore_impl.cpp index 0ecb78b9..197000eb 100644 --- a/src/blockstore/blockstore_impl.cpp +++ b/src/blockstore/blockstore_impl.cpp @@ -244,11 +244,11 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op) } else if (PRIV(op)->wait_for == WAIT_COMPACTION) { - if (heap->get_compact_queue_size() >= PRIV(op)->wait_detail) + if (flusher->get_counter() <= PRIV(op)->wait_detail) { // do not submit #ifdef BLOCKSTORE_DEBUG - printf("Still waiting to reduce compaction queue size below %ju\n", PRIV(op)->wait_detail); + printf("Still waiting for more flushes\n"); #endif return; } diff --git a/src/blockstore/blockstore_read.cpp b/src/blockstore/blockstore_read.cpp index f6870867..6c8eedf7 100644 --- a/src/blockstore/blockstore_read.cpp +++ b/src/blockstore/blockstore_read.cpp @@ -19,7 +19,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op) PRIV(op)->pending_ops = 0; auto & rv = PRIV(op)->read_vec; uint64_t result_version = 0; - for (heap_write_t *wr = obj->get_writes(); wr < (heap_write_t*)obj->next(); wr = wr->next(heap)) + for (auto wr = obj->get_writes(); wr; wr = wr->next()) { if (op->version < wr->version) { @@ -362,7 +362,7 @@ int blockstore_impl_t::read_bitmap(object_id oid, uint64_t target_version, void heap_object_t *obj = heap->read_entry(oid, NULL); if (obj) { - for (heap_write_t *wr = obj->get_writes(); wr < (heap_write_t*)obj->next(); wr = wr->next(heap)) + for (auto wr = obj->get_writes(); wr; wr = wr->next()) { if (target_version < wr->version) { diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index cba251cc..2ec340b0 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -28,6 +28,8 @@ void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, int retval) (other_op->opcode == BS_OP_WRITE || other_op->opcode == BS_OP_WRITE_STABLE)) { // Mark operations to cancel them + if (PRIV(other_op)->op_state != 0 && PRIV(other_op)->op_state != 100) + write_iodepth--; PRIV(other_op)->op_state = 100; other_op->retval = retval; } @@ -85,6 +87,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) prepare_meta_block_write(op, modified_block); PRIV(op)->pending_ops++; PRIV(op)->op_state = 5; + write_iodepth++; } // FIXME: Allow to do initial writes as buffered, not redirected // FIXME: Allow to do direct writes over holes @@ -97,20 +100,18 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) if (loc == UINT64_MAX || !obj && heap->get_block_for_new_object(tmp_block) != 0) { - auto queue_size = heap->get_compact_queue_size(); - if (!queue_size) + if (!heap->get_compact_queue_size() && !flusher->get_active()) { // no space cancel_all_writes(op, -ENOSPC); return 2; } PRIV(op)->wait_for = WAIT_COMPACTION; - PRIV(op)->wait_detail = queue_size; + PRIV(op)->wait_detail = flusher->get_counter(); flusher->request_trim(); return 0; } BS_SUBMIT_GET_SQE(sqe, data); - write_iodepth++; PRIV(op)->location = loc; #ifdef BLOCKSTORE_DEBUG printf( @@ -141,6 +142,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) PRIV(op)->pending_ops = 1; unsynced_big_write_count++; PRIV(op)->op_state = 1; + write_iodepth++; } else { @@ -150,13 +152,12 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) if (loc == UINT64_MAX) { PRIV(op)->wait_for = WAIT_COMPACTION; - PRIV(op)->wait_detail = heap->get_compact_queue_size(); + PRIV(op)->wait_detail = flusher->get_counter(); flusher->request_trim(); return 0; } // There is sufficient space. Check SQE(s) BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0)); - write_iodepth++; uint8_t wr_buf[heap->get_max_write_entry_size()]; heap_write_t *wr = (heap_write_t*)wr_buf; wr->version = op->version; @@ -173,12 +174,15 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) int res = heap->post_write(op->oid, wr, &modified_block); if (res == ENOSPC) { - cancel_all_writes(op, -ENOSPC); - return 2; - } - else if (res == EAGAIN) - { - // Pause submission, wait for compaction + if (!heap->get_compact_queue_size() && !flusher->get_active()) + { + // no space + cancel_all_writes(op, -ENOSPC); + return 2; + } + PRIV(op)->wait_for = WAIT_COMPACTION; + PRIV(op)->wait_detail = flusher->get_counter(); + flusher->request_trim(); return 0; } assert(res == 0); @@ -202,11 +206,13 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) if (!PRIV(op)->pending_ops) { PRIV(op)->op_state = 6; + write_iodepth++; return continue_write(op); } else { PRIV(op)->op_state = 5; + write_iodepth++; } } return 1; @@ -260,7 +266,8 @@ resume_4: int res = heap->post_write(op->oid, wr, &modified_block); if (res == ENOSPC) { - // wait for compaction + PRIV(op)->wait_for = WAIT_COMPACTION; + PRIV(op)->wait_detail = flusher->get_counter(); return 1; } assert(res == 0);