From e709657de4750214e2708397cd5c06a883d0e93f Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Sun, 2 Nov 2025 00:22:23 +0300 Subject: [PATCH] Save completed_lsn in the superblock --- src/blockstore/blockstore_flush.cpp | 48 +++++++++++++++++++++++++++-- src/blockstore/blockstore_flush.h | 2 +- src/blockstore/blockstore_heap.cpp | 13 ++++++-- src/blockstore/blockstore_heap.h | 1 + src/blockstore/blockstore_impl.h | 13 +------- src/blockstore/blockstore_init.cpp | 1 + src/blockstore/blockstore_open.cpp | 2 +- src/blockstore/blockstore_write.cpp | 1 + src/blockstore/ondisk_formats.h | 2 +- src/disk_tool/disk_tool_meta.cpp | 2 +- src/disk_tool/disk_tool_resize.cpp | 2 +- 11 files changed, 65 insertions(+), 22 deletions(-) diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index 5ade09fe..866abe19 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -18,7 +18,6 @@ journal_flusher_t::journal_flusher_t(blockstore_impl_t *bs) this->cur_flusher_count = bs->min_flusher_count; this->target_flusher_count = bs->min_flusher_count; active_flushers = 0; - advance_lsn_counter = 0; co = new journal_flusher_co[max_flusher_count]; for (int i = 0; i < max_flusher_count; i++) { @@ -132,8 +131,12 @@ void journal_flusher_t::loop() } } int prev_active = active_flushers; - for (int i = 0; (active_flushers > 0 || force_start > 0 || bs->heap->get_to_compact_count() > bs->flusher_start_threshold) && i < cur_flusher_count; i++) + for (int i = 0; (active_flushers > 0 || force_start > 0 || + bs->heap->get_to_compact_count() > bs->flusher_start_threshold || + i == 0 && bs->intent_write_counter >= bs->journal_trim_interval) && i < cur_flusher_count; i++) + { co[i].loop(); + } if (prev_active && !active_flushers && force_start > 0) bs->ringloop->wakeup(); } @@ -170,11 +173,22 @@ bool journal_flusher_co::loop() else if (wait_state == 14) goto resume_14; else if (wait_state == 15) goto resume_15; else if (wait_state == 16) goto resume_16; + else if (wait_state == 17) goto resume_17; + else if (wait_state == 18) goto resume_18; resume_0: wait_state = 0; wait_count = 0; cur_oid = {}; res = bs->heap->get_next_compact(cur_oid); + // Advance fsynced_lsn every intent writes + if ((bs->intent_write_counter >= bs->journal_trim_interval) && co_id == 0) + { + bs->intent_write_counter = 0; +resume_17: +resume_18: + if (!trim_lsn(17)) + return false; + } if (res == ENOENT && flusher->force_start > 0 && co_id == 0 && (!bs->dsk.disable_journal_fsync || !bs->dsk.disable_meta_fsync)) { @@ -740,3 +754,33 @@ resume_2: flusher->syncing_buffer--; return true; } + +bool journal_flusher_co::trim_lsn(int wait_base) +{ + if (wait_state == wait_base) goto resume_0; + else if (wait_state == wait_base+1) goto resume_1; + fsynced_lsn = bs->heap->get_fsynced_lsn(); + if (((blockstore_meta_header_v3_t*)bs->meta_superblock)->completed_lsn == fsynced_lsn) + { + return true; + } + flusher->active_flushers++; + ((blockstore_meta_header_v3_t*)bs->meta_superblock)->completed_lsn = fsynced_lsn; + ((blockstore_meta_header_v3_t*)bs->meta_superblock)->set_crc32c(); + await_sqe(0); + data->iov = (struct iovec){ bs->meta_superblock, (size_t)bs->dsk.meta_block_size }; + data->callback = simple_callback_w; + io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset); + // Update superblock with datasync + sqe->rw_flags = RWF_DSYNC; + wait_count++; +resume_1: + if (wait_count > 0) + { + wait_state = wait_base+1; + return false; + } + flusher->compact_counter++; + flusher->active_flushers--; + return true; +} diff --git a/src/blockstore/blockstore_flush.h b/src/blockstore/blockstore_flush.h index a5d5b0e9..f8fdcbc6 100644 --- a/src/blockstore/blockstore_flush.h +++ b/src/blockstore/blockstore_flush.h @@ -71,6 +71,7 @@ class journal_flusher_co bool read_buffered(int wait_base); bool fsync_meta(int wait_base); bool fsync_buffer(int wait_base); + bool trim_lsn(int wait_base); public: journal_flusher_co(); ~journal_flusher_co(); @@ -86,7 +87,6 @@ class journal_flusher_t blockstore_impl_t *bs; friend class journal_flusher_co; - int advance_lsn_counter = 0; uint64_t compact_counter = 0; robin_hood::unordered_flat_set flushing; diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index cbebf42a..d2735af5 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -255,6 +255,11 @@ blockstore_heap_t::~blockstore_heap_t() delete buffer_alloc; } +void blockstore_heap_t::start_load(uint64_t completed_lsn) +{ + this->completed_lsn = completed_lsn; +} + int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uint8_t *buf, std::function handle_write, std::function handle_block) { @@ -385,8 +390,11 @@ void blockstore_heap_t::fill_recheck_queue() if (obj->type() == BS_HEAP_INTENT_WRITE || obj->type() == BS_HEAP_BIG_INTENT) { // Recheck only the latest intent_write - // FIXME Save checked_lsn in the superblock - recheck_queue.push_back(obj); + if (obj->lsn > completed_lsn) + { + // Do not recheck if it's already marked as completed in the superblock + recheck_queue.push_back(obj); + } } else { @@ -1344,7 +1352,6 @@ void blockstore_heap_t::start_block_write(uint32_t block_num) assert(!inf.is_writing); inf.is_writing = true; }); - // FIXME also get_meta_block } void blockstore_heap_t::complete_block_write(uint32_t block_num) diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index e4faec88..e631a82f 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -211,6 +211,7 @@ class blockstore_heap_t public: blockstore_heap_t(blockstore_disk_t *dsk, uint8_t *buffer_area, int log_level = 0); ~blockstore_heap_t(); + void start_load(uint64_t completed_lsn); // load data from the disk, returns EDOM on corruption int read_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf, std::function handle_write, diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index 9d31b99f..46d1d1a1 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -29,18 +29,6 @@ class blockstore_impl_t; //#define BLOCKSTORE_DEBUG -// - Sync must be submitted after previous writes/deletes (not before!) -// - Reads may be submitted in parallel with writes/deletes because we use MVCC -// - Writes may be submitted in any order, because they don't overlap. Each write -// goes into a new location - either on the journal device or on the data device -// - Stable (stabilize) must be submitted after sync of that object is completed -// It's even OK to return an error to the caller if that object is not synced yet -// - compacted_lsn should be moved forward only after all versions are moved to the main storage -// - If an operation can not be submitted because the ring is full -// we should stop submission of other operations. Otherwise some "scatter" reads -// may end up blocked for a long time. -// Otherwise, the submission order is free. - #include "blockstore_init.h" #include "blockstore_flush.h" @@ -128,6 +116,7 @@ public: journal_flusher_t *flusher; int write_iodepth = 0; int inflight_big = 0; + int intent_write_counter = 0; bool fsyncing_data = false; bool live = false, queue_stall = false; diff --git a/src/blockstore/blockstore_init.cpp b/src/blockstore/blockstore_init.cpp index 8a39829d..131ff408 100644 --- a/src/blockstore/blockstore_init.cpp +++ b/src/blockstore/blockstore_init.cpp @@ -155,6 +155,7 @@ resume_1: bs->dsk.check_lengths(); } bs->init(); + bs->heap->start_load(((blockstore_meta_header_v3_t *)bs->meta_superblock)->completed_lsn); if (bs->dsk.inmemory_journal) { // Read buffer area diff --git a/src/blockstore/blockstore_open.cpp b/src/blockstore/blockstore_open.cpp index d4c6f5e0..3b9a520f 100644 --- a/src/blockstore/blockstore_open.cpp +++ b/src/blockstore/blockstore_open.cpp @@ -42,7 +42,7 @@ void blockstore_impl_t::parse_config(blockstore_config_t & config, bool init) } if (!journal_trim_interval) { - journal_trim_interval = 1024; + journal_trim_interval = 4096; } if (!flusher_start_threshold) { diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index 5098afe8..02d88bde 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -237,6 +237,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) } assert(res == 0); prepare_meta_block_write(PRIV(op)->modified_block); + intent_write_counter++; PRIV(op)->pending_ops++; PRIV(op)->op_state = 9; write_iodepth++; diff --git a/src/blockstore/ondisk_formats.h b/src/blockstore/ondisk_formats.h index 22bdc05e..86aeb3af 100644 --- a/src/blockstore/ondisk_formats.h +++ b/src/blockstore/ondisk_formats.h @@ -185,7 +185,7 @@ struct __attribute__((__packed__)) blockstore_meta_header_v3_t uint32_t data_csum_type; uint32_t csum_block_size; uint32_t header_csum; - uint64_t compacted_lsn; + uint64_t completed_lsn; void set_crc32c(); }; diff --git a/src/disk_tool/disk_tool_meta.cpp b/src/disk_tool/disk_tool_meta.cpp index 54e6700e..cd66fe3d 100644 --- a/src/disk_tool/disk_tool_meta.cpp +++ b/src/disk_tool/disk_tool_meta.cpp @@ -275,7 +275,7 @@ int disk_tool_t::dump_meta() if (dump_as_old) { hdr->version = BLOCKSTORE_META_FORMAT_V2; - hdr->compacted_lsn = 0; + hdr->completed_lsn = 0; hdr->header_csum = 0; hdr->header_csum = crc32c(0, hdr, sizeof(blockstore_meta_header_v2_t)); } diff --git a/src/disk_tool/disk_tool_resize.cpp b/src/disk_tool/disk_tool_resize.cpp index e5ed0c70..fe015154 100644 --- a/src/disk_tool/disk_tool_resize.cpp +++ b/src/disk_tool/disk_tool_resize.cpp @@ -554,7 +554,7 @@ int disk_tool_t::resize_rebuild_meta() new_meta_hdr->bitmap_granularity = dsk.bitmap_granularity ? dsk.bitmap_granularity : 4096; new_meta_hdr->data_csum_type = dsk.data_csum_type; new_meta_hdr->csum_block_size = dsk.csum_block_size; - new_meta_hdr->compacted_lsn = hdr->compacted_lsn; + new_meta_hdr->completed_lsn = hdr->completed_lsn; new_meta_hdr->header_csum = 0; new_meta_hdr->header_csum = crc32c(0, new_meta_hdr, new_meta_hdr->version == BLOCKSTORE_META_FORMAT_HEAP ? sizeof(blockstore_meta_header_v3_t) : sizeof(blockstore_meta_header_v2_t));