From eb3df895e05f3003b3a09be169a0330ebdf07df5 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Tue, 3 Feb 2026 03:03:20 +0300 Subject: [PATCH] WIP refactor dequeue_write/continue_write into a single "coroutine" --- src/blockstore/blockstore_heap.cpp | 5 + src/blockstore/blockstore_write.cpp | 252 +++++++++++++++++++--------- 2 files changed, 178 insertions(+), 79 deletions(-) diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index 2a22d7dd..09c05681 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -1347,6 +1347,7 @@ int blockstore_heap_t::add_small_write(object_id oid, heap_entry_t **obj_ptr, ui // Small writes are written in parallel with buffered data so they require explicit_complete return add_entry(wr_size, modified_block, false, true, [&](heap_entry_t *wr) { + printf("add_small_write t%u %lx:%lx l%lu v%lu %u +%u loc:%lx\n", type, oid.inode, oid.stripe, wr->lsn, version, offset, len, location); wr->entry_type = type; wr->inode = oid.inode; wr->stripe = oid.stripe; @@ -1376,6 +1377,7 @@ int blockstore_heap_t::add_big_write(object_id oid, heap_entry_t *old_head, bool // Big writes are written after writing data so they don't require explicit_complete return add_entry(wr_size, modified_block, false, false, [&](heap_entry_t *wr) { + printf("add_big_write %lx:%lx l%lu v%lu loc:%lx\n", oid.inode, oid.stripe, wr->lsn, version, location); wr->entry_type = BS_HEAP_BIG_WRITE | (stable ? BS_HEAP_STABLE : 0); wr->inode = oid.inode; wr->stripe = oid.stripe; @@ -1402,6 +1404,7 @@ int blockstore_heap_t::add_redirect_intent(object_id oid, heap_entry_t **obj_ptr // Big-redirect intents, just like regular big writes, are written after writing data so they don't require explicit_complete return add_entry(wr_size, modified_block, false, false, [&](heap_entry_t *wr) { + printf("add_redir_intent %lx:%lx l%lu v%lu %u +%u loc:%lx\n", oid.inode, oid.stripe, wr->lsn, version, offset, len, location); wr->entry_type = BS_HEAP_BIG_INTENT|BS_HEAP_STABLE; wr->inode = oid.inode; wr->stripe = oid.stripe; @@ -1438,6 +1441,7 @@ int blockstore_heap_t::add_big_intent(object_id oid, heap_entry_t **obj_ptr, uin // Big intents are written before writing data so they require explicit_complete return add_entry(wr_size, modified_block, false, true, [&](heap_entry_t *wr) { + printf("add_big_intent %lx:%lx l%lu v%lu %u +%u loc:%lx\n", oid.inode, oid.stripe, wr->lsn, version, offset, len, obj->big_location(this)); wr->entry_type = BS_HEAP_BIG_INTENT | BS_HEAP_STABLE; wr->inode = oid.inode; wr->stripe = oid.stripe; @@ -1493,6 +1497,7 @@ int blockstore_heap_t::add_compact(heap_entry_t *obj, uint64_t compact_version, // Compaction entry is added after copying data so it doesn't require explicit_complete return add_entry(wr_size, modified_block, true, false, [&](heap_entry_t *new_wr) { + printf("add_compact %lx:%lx l%lu v%lu loc:%lx\n", obj->inode, obj->stripe, compact_lsn, compact_version, compact_location); new_wr->entry_type = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE; new_wr->inode = obj->inode; new_wr->stripe = obj->stripe; diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index 5dd48ac1..7381dbb7 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -13,10 +13,13 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op) return true; } -void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block) +bool blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block) { - if (modified_blocks.find(modified_block) != modified_blocks.end()) - return; + auto mod_it = modified_blocks.find(modified_block); + if (mod_it != modified_blocks.end()) + { + return !mod_it->second.sent; + } io_uring_sqe *sqe = get_sqe(); assert(sqe != NULL); ring_data_t *data = ((ring_data_t*)sqe->user_data); @@ -41,6 +44,7 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block) unsynced_meta_write_count++; pending_modified_blocks.push_back(modified_block); modified_blocks[modified_block] = { .sent = false, .buf = buf }; + return true; } bool blockstore_impl_t::meta_block_is_pending(uint32_t modified_block) @@ -121,6 +125,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) heap_entry_t *obj = heap->read_entry(op->oid); if (op->opcode == BS_OP_DELETE) { + return continue_delete(op, 0); // Delete if (!obj || obj->type() == BS_HEAP_DELETE) { @@ -133,17 +138,38 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) BS_SUBMIT_CHECK_SQES(1); int res = heap->add_delete(obj, &PRIV(op)->modified_block); if (res == ENOSPC) + { goto enospc; + } assert(res == 0); - prepare_meta_block_write(PRIV(op)->modified_block); - PRIV(op)->pending_ops++; - PRIV(op)->op_state = 5; write_iodepth++; +resume_1: + while (!prepare_meta_block_write(PRIV(op)->modified_block)) + { + PRIV(op)->op_state = 1; + return 1; + } +rseume_2: + while (meta_block_is_pending(PRIV(op)->modified_block)) + { + PRIV(op)->op_state = 2; + return 1; + } +resume_3: +resume_4: + if (!throttle_write(op, 3)) + { + return 1; + } + write_iodepth--; + ack_write(op); + return 2; } // FIXME: Allow to do initial writes as buffered, not redirected // FIXME: Allow to do direct writes over holes else if (!obj || obj->type() == BS_HEAP_DELETE || op->offset == 0 && op->len == dsk.data_block_size) { + return continue_big_write(op, 10); // Big (redirect) write PRIV(op)->write_type = dsk.disable_data_fsync || op->opcode != BS_OP_WRITE_STABLE ? BS_HEAP_BIG_WRITE : _REDIRECT_INTENT; BS_SUBMIT_CHECK_SQES(1); @@ -163,6 +189,7 @@ enospc: flusher->request_trim(); return 0; } + write_iodepth++; uint64_t loc = PRIV(op)->location; #ifdef BLOCKSTORE_DEBUG printf( @@ -176,18 +203,72 @@ enospc: data->iov = (struct iovec){ op->buf, op->len }; data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); }; io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + loc + op->offset); + if (PRIV(op)->write_type == BS_HEAP_BIG_WRITE) + inflight_big++; PRIV(op)->pending_ops++; - write_iodepth++; +resume_10: + if (PRIV(op)->pending_ops > 0) + { + PRIV(op)->op_state = 10; + return 1; + } if (PRIV(op)->write_type == BS_HEAP_BIG_WRITE) { - PRIV(op)->op_state = 1; - inflight_big++; + inflight_big--; +resume_11: +resume_12: +resume_13: + if (!fsync_big_write(op, 11)) + return 1; + } + heap_entry_t *obj = heap->read_entry(op->oid); + int res = 0; + if (PRIV(op)->write_type == _REDIRECT_INTENT) + { + res = heap->add_redirect_intent(op->oid, &obj, op->version, op->offset, op->len, + PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block); } else - PRIV(op)->op_state = 3; + { + res = heap->add_big_write(op->oid, obj, op->opcode == BS_OP_WRITE_STABLE, + op->version, op->offset, op->len, PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block); + } + if (res == ENOSPC) + { + if (!heap->get_to_compact_count()) + { + // no space + heap->free_data(op->oid.inode, PRIV(op)->location); + write_iodepth--; + op->retval = -ENOSPC; + FINISH_OP(op); + return 2; + } + PRIV(op)->wait_for = WAIT_COMPACTION; + PRIV(op)->wait_detail = heap->get_compacted_count(); + flusher->request_trim(); + return 0; + } + assert(res == 0); +resume_14: + while (!prepare_meta_block_write(PRIV(op)->modified_block)) + { + PRIV(op)->op_state = 14; + return 1; + } +resume_15: + while (meta_block_is_pending(PRIV(op)->modified_block)) + { + PRIV(op)->op_state = 15; + return 1; + } + write_iodepth--; + ack_write(op); + return 2; } else if (intent_write_allowed(op, obj)) { + return continue_intent_write(op, 20); // Direct intent-write BS_SUBMIT_CHECK_SQES(1); int res = 0; @@ -223,13 +304,41 @@ enospc: assert(res == 0); PRIV(op)->lsn = obj->lsn; } - prepare_meta_block_write(PRIV(op)->modified_block); - PRIV(op)->pending_ops++; - PRIV(op)->op_state = 9; write_iodepth++; +resume_20: + while (!prepare_meta_block_write(PRIV(op)->modified_block)) + { + PRIV(op)->op_state = 20; + return 1; + } +resume_21: + while (meta_block_is_pending(PRIV(op)->modified_block)) + { + PRIV(op)->op_state = 21; + return 1; + } + // Direct intent-write + // LSN is not marked as completed so big_write won't be freed + BS_SUBMIT_GET_SQE(sqe, data); + data->iov = (struct iovec){ op->buf, op->len }; + data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); }; + io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + PRIV(op)->location + op->offset); + if (dsk.use_atomic_flag) + sqe->rw_flags = RWF_ATOMIC; + PRIV(op)->pending_ops++; +resume_22: + if (PRIV(op)->pending_ops > 0) + { + PRIV(op)->op_state = 22; + return 1; + } + write_iodepth--; + ack_write(op); + return 2; } else { + return continue_small_write(op, 30); // Small (buffered) overwrite // First check if there is free buffer space PRIV(op)->write_type = BS_HEAP_SMALL_WRITE; @@ -242,7 +351,7 @@ enospc: return 0; } // There is sufficient space. Check SQE(s) - BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0)); + BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0)); ---> refactor too int res = heap->add_small_write(op->oid, &obj, (BS_HEAP_SMALL_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)), op->version, op->offset, op->len, loc, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block); if (res == ENOSPC) @@ -311,27 +420,46 @@ again: goto again; } resume_2: + + +resume_4: +resume_6: + +resume_8: + ack + return 2; +resume_10: + return 1; +} + +bool blockstore_impl_t::fsync_big_write(blockstore_op_t *op, int base_state) +{ + if (PRIV(op)->state == base_state) + goto resume_0; + else if (PRIV(op)->state == base_state+1) + goto resume_1; + else if (PRIV(op)->state == base_state+2) + goto resume_2; // We must fsync all big writes to avoid complex write workflows // It's OK for all HDDs and for server SSDs, but slightly worse for desktop SSDs - inflight_big--; if (!dsk.disable_data_fsync) { // fsync data in a batch -resume_11: +resume_0: if (inflight_big > 0) { - PRIV(op)->op_state = 11; - return 1; + PRIV(op)->op_state = base_state; + return false; } if (fsyncing_data) { -resume_12: +resume_1: if (fsyncing_data) { - PRIV(op)->op_state = 12; - return 1; + PRIV(op)->op_state = base_state+1; + return false; } - goto resume_4; + return true; } fsyncing_data = true; BS_SUBMIT_GET_SQE(sqe, data); @@ -343,47 +471,23 @@ resume_12: handle_write_event(data, op); }; PRIV(op)->pending_ops++; - PRIV(op)->op_state = 3; - return 1; +resume_2: + if (PRIV(op)->pending_ops > 0) + { + PRIV(op)->op_state = base_state+2; + return false; + } } -resume_4: - { - auto obj = heap->read_entry(op->oid); - int res = 0; - if (PRIV(op)->write_type == _REDIRECT_INTENT) - { - res = heap->add_redirect_intent(op->oid, &obj, op->version, op->offset, op->len, - PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block); - } - else - { - res = heap->add_big_write(op->oid, obj, op->opcode == BS_OP_WRITE_STABLE, - op->version, op->offset, op->len, PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block); - } - if (res == ENOSPC) - { - if (!heap->get_to_compact_count()) - { - // no space - heap->free_data(op->oid.inode, PRIV(op)->location); - write_iodepth--; - op->retval = -ENOSPC; - FINISH_OP(op); - return 2; - } - PRIV(op)->wait_for = WAIT_COMPACTION; - PRIV(op)->wait_detail = heap->get_compacted_count(); - flusher->request_trim(); - return 0; - } - assert(res == 0); - prepare_meta_block_write(PRIV(op)->modified_block); - PRIV(op)->pending_ops++; - PRIV(op)->op_state = 5; - return 1; - } -resume_6: + return true; +} + +bool blockstore_impl_t::throttle_write(blockstore_op_t *op, int base_state) +{ // Apply throttling to not fill the journal too quickly for the SSD+HDD case + if (PRIV(op)->op_state >= base_state+1) + { + return true; + } if (PRIV(op)->write_type == BS_HEAP_SMALL_WRITE && throttle_small_writes) { // Apply throttling @@ -404,17 +508,21 @@ resume_6: if (ref_us > exec_us + throttle_threshold_us) { // Pause reply - PRIV(op)->op_state = 7; + PRIV(op)->op_state = base_state; // Remember that the timer can in theory be called right here tfd->set_timer_us(ref_us-exec_us, false, [this, op](int timer_id) { - PRIV(op)->op_state = 8; + PRIV(op)->op_state++; ringloop->wakeup(); }); - return 1; + return false; } } -resume_8: + return true; +} + +void blockstore_impl_t::ack_write(blockstore_op_t *op) +{ // Acknowledge write #ifdef BLOCKSTORE_DEBUG printf("Ack write %jx:%jx v%ju\n", op->oid.inode, op->oid.stripe, op->version); @@ -441,21 +549,7 @@ resume_8: unsynced_data_write_count++; intent_write_counter++; } - write_iodepth--; FINISH_OP(op); - return 2; -resume_10: - // Direct intent-write - // LSN is not marked as completed so big_write won't be freed - BS_SUBMIT_GET_SQE(sqe, data); - data->iov = (struct iovec){ op->buf, op->len }; - data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); }; - io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + PRIV(op)->location + op->offset); - if (dsk.use_atomic_flag) - sqe->rw_flags = RWF_ATOMIC; - PRIV(op)->pending_ops++; - PRIV(op)->op_state = 7; - return 1; } void blockstore_impl_t::handle_write_event(ring_data_t *data, blockstore_op_t *op)