From 5dc0b421467fe8b19aaf0338221739802df478b7 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Sat, 5 Jul 2025 14:53:37 +0300 Subject: [PATCH] Use new LSNs on stabilize --- src/blockstore/blockstore_flush.cpp | 4 -- src/blockstore/blockstore_impl.h | 11 +++-- src/blockstore/blockstore_stable.cpp | 70 ++++++++++++++-------------- src/blockstore/blockstore_write.cpp | 9 ++-- 4 files changed, 48 insertions(+), 46 deletions(-) diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index cc91a43a..88c6dca4 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -186,10 +186,6 @@ resume_1: cur_version = cur_obj->get_writes()->version; // Find the range to compact compact_lsn = bs->heap->get_completed_lsn(); - if (auto cm_it = bs->committing_lsn.find(cur_oid); cm_it != bs->committing_lsn.end()) - { - compact_lsn = cm_it->second; - } bs->heap->get_compact_range(cur_obj, compact_lsn, &begin_wr, &end_wr); if (!begin_wr) { diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index 1956168c..a198d6e5 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -53,7 +53,7 @@ struct blockstore_op_private_t int pending_ops; int op_state; - // Read, write + // Read, write, stabilize uint64_t lsn; // Read @@ -63,9 +63,11 @@ struct blockstore_op_private_t uint64_t location; bool is_big; - // Stabilize/rollback + // Stabilize, rollback int stab_pos; - std::vector to_compact; + + // Stabilize + uint64_t to_lsn; // Write struct iovec iov_zerofill[3]; @@ -117,7 +119,6 @@ class blockstore_impl_t: public blockstore_i journal_flusher_t *flusher; int write_iodepth = 0; - std::unordered_map committing_lsn; bool live = false, queue_stall = false; ring_loop_t *ringloop; @@ -165,7 +166,7 @@ class blockstore_impl_t: public blockstore_i // Write bool enqueue_write(blockstore_op_t *op); void cancel_all_writes(blockstore_op_t *op, int retval); - void prepare_meta_block_write(blockstore_op_t *op, uint64_t modified_block); + void prepare_meta_block_write(blockstore_op_t *op, uint64_t modified_block, io_uring_sqe *sqe = NULL); int dequeue_write(blockstore_op_t *op); int continue_write(blockstore_op_t *op); void handle_write_event(ring_data_t *data, blockstore_op_t *op); diff --git a/src/blockstore/blockstore_stable.cpp b/src/blockstore/blockstore_stable.cpp index 3e08a325..a0038232 100644 --- a/src/blockstore/blockstore_stable.cpp +++ b/src/blockstore/blockstore_stable.cpp @@ -14,45 +14,53 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op) else if (priv->op_state == 3) goto resume_3; else if (priv->op_state == 4) goto resume_4; assert(!priv->op_state); + // Modify in-memory state and assign contiguous LSNs priv->stab_pos = 0; - op->retval = 0; - priv->to_compact.resize(op->len); + priv->lsn = priv->to_lsn = 0; while (priv->stab_pos < op->len) { - io_uring_sqe *sqe = get_sqe(); - if (!sqe) - { - if (priv->pending_ops > 0) - return 1; - priv->wait_detail = 1; - priv->wait_for = WAIT_SQE; - return 0; - } uint32_t modified_block = 0; - uint64_t before_compact = 0; - uint64_t to_compact = 0; + uint64_t new_lsn = 0; + uint64_t new_to_lsn = 0; int res = op->opcode == BS_OP_STABLE - ? heap->post_stabilize(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block, &before_compact, &to_compact) + ? heap->post_stabilize(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block, &new_lsn, &new_to_lsn) : heap->post_rollback(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block); if (res != 0) { + assert(res == ENOENT || res == EBUSY); op->retval = -res; - FINISH_OP(op); - return 2; } - if (modified_block) + if (new_lsn) { - if (to_compact) - { - priv->to_compact[priv->stab_pos] = true; - committing_lsn[v[priv->stab_pos].oid] = before_compact; - } - prepare_meta_block_write(op, modified_block); - priv->pending_ops++; + if (!priv->lsn) + priv->lsn = new_lsn; + priv->to_lsn = new_to_lsn; } priv->stab_pos++; } + // Submit metadata writes + priv->stab_pos = 0; resume_1: + priv->op_state = 1; + while (priv->stab_pos < op->len) + { + uint32_t block_num = 0; + heap_object_t *obj = heap->read_entry(v[priv->stab_pos].oid, &block_num); + if (obj) + { + io_uring_sqe *sqe = get_sqe(); + if (!sqe) + { + if (priv->pending_ops > 0) + return 1; + priv->wait_detail = 1; + priv->wait_for = WAIT_SQE; + return 0; + } + prepare_meta_block_write(op, block_num, sqe); + } + priv->stab_pos++; + } if (priv->pending_ops > 0) { priv->op_state = 1; @@ -74,17 +82,11 @@ resume_3: return 0; } resume_4: - for (int i = 0; i < op->len; i++) + // Mark writes as completed to allow compaction + // FIXME: Also mark as fsynced + for (uint64_t lsn = priv->lsn; lsn <= priv->to_lsn; lsn++) { - if (priv->to_compact[i]) - { - // Add to compact queue only when metadata writes are finished - heap->add_to_compact_queue(v[i].oid); - auto cm_it = committing_lsn.find(v[i].oid); - assert(cm_it != committing_lsn.end()); - if (cm_it->second <= priv->to_compact[i]) - committing_lsn.erase(cm_it); - } + heap->complete_lsn(lsn); } // Done. Don't touch op->retval - if anything resulted in ENOENT, return it as is FINISH_OP(op); diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index a222497f..381f93cd 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -37,10 +37,13 @@ void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, int retval) FINISH_OP(op); } -void blockstore_impl_t::prepare_meta_block_write(blockstore_op_t *op, uint64_t modified_block) +void blockstore_impl_t::prepare_meta_block_write(blockstore_op_t *op, uint64_t modified_block, io_uring_sqe *sqe) { - io_uring_sqe *sqe = get_sqe(); - assert(sqe != NULL); + if (!sqe) + { + sqe = get_sqe(); + assert(sqe != NULL); + } ring_data_t *data = ((ring_data_t*)sqe->user_data); data->iov = (struct iovec){ heap->get_meta_block(modified_block), (size_t)dsk.meta_block_size }; data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };