From e0d2705294e1ac4e577d70055234a55313de9713 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Fri, 18 Jul 2025 22:06:21 +0300 Subject: [PATCH] Add 2 tests for intent writes --- src/blockstore/blockstore_flush.cpp | 57 ++++++----- src/blockstore/blockstore_heap.cpp | 2 + src/blockstore/blockstore_impl.h | 9 +- src/blockstore/blockstore_internal.h | 14 +-- src/blockstore/blockstore_read.cpp | 20 ++-- src/blockstore/blockstore_stable.cpp | 1 + src/blockstore/blockstore_write.cpp | 13 +-- src/test/test_blockstore.cpp | 142 +++++++++++++++++++++++++++ 8 files changed, 204 insertions(+), 54 deletions(-) diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index 6d33ec8b..3120a7a4 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -56,6 +56,7 @@ journal_flusher_t::~journal_flusher_t() journal_flusher_co::~journal_flusher_co() { + free_buffers(); if (csum_buf) { free(csum_buf); @@ -235,7 +236,7 @@ resume_1: #endif flusher->active_flushers++; // Scan versions to flush - read_vec.clear(); + free_buffers(); for (auto wr = begin_wr; wr != end_wr; wr = wr->next()) { bs->prepare_read(read_vec, cur_obj, wr, 0, bs->dsk.data_block_size); @@ -295,18 +296,20 @@ resume_8: else if (res == ENOENT || res == EDOM) { // Abort compaction - goto release_oid; + flusher->active_flushers--; + goto resume_0; } assert(res == 0); // Submit data writes copy_count = 0; for (i = 0; i < read_vec.size(); i++) { - if (read_vec[i].copy_flags & COPY_BUF_JOURNAL) + if ((read_vec[i].copy_flags & COPY_BUF_JOURNAL) && + !(read_vec[i].copy_flags & COPY_BUF_COALESCED)) { assert(read_vec[i].buf); await_sqe(9); - data->iov = (struct iovec){ read_vec[i].buf, (size_t)read_vec[i].len }; + data->iov = (struct iovec){ (bs->dsk.inmemory_journal ? bs->buffer_area + read_vec[i].disk_offset : read_vec[i].buf), (size_t)read_vec[i].len }; data->callback = simple_callback_w; io_uring_prep_writev(sqe, bs->dsk.data_fd, &data->iov, 1, bs->dsk.data_offset + clean_loc + read_vec[i].offset); wait_count++; @@ -328,7 +331,7 @@ resume_10: if (!cur_obj) { // Abort compaction - goto release_oid; + goto resume_0; } calc_block_checksums(); if (read_to_fill_incomplete) @@ -341,8 +344,7 @@ resume_24: } } bs->heap->mark_object_compacted(cur_obj, compact_lsn); - // Done, free all buffers - free_buffers(); + // Done #ifdef BLOCKSTORE_DEBUG printf("Compacted %jx:%jx l%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, compact_lsn, copy_count); #endif @@ -360,7 +362,6 @@ resume_15: if (!trim_lsn(11)) return false; } -release_oid: if (should_repeat) { // Flush the same object again @@ -374,20 +375,23 @@ void journal_flusher_co::iterate_partial_overwrites(std::function prev_end) { - i += cb(prev, prev_begin, prev_end); + if (prev_end > prev_begin && ((prev_begin % bs->dsk.csum_block_size) || (prev_end % bs->dsk.csum_block_size))) + { + i += cb(prev, prev_begin, prev_end); + } prev = i; prev_begin = read_vec[i].offset; } prev_end = read_vec[i].offset + read_vec[i].len; } } - if (prev_end > prev_begin) + if (prev_end > prev_begin && ((prev_begin % bs->dsk.csum_block_size) || (prev_end % bs->dsk.csum_block_size))) { cb(prev, prev_begin, prev_end); } @@ -401,14 +405,12 @@ void journal_flusher_co::iterate_checksum_holes(std::functiondsk.csum_block_size) && (prev_begin / bs->dsk.csum_block_size) != (prev_end / bs->dsk.csum_block_size)) { - cb(pos, prev_begin, prev_begin + bs->dsk.csum_block_size - prev_begin%bs->dsk.csum_block_size); + cb(pos, prev_begin - prev_begin%bs->dsk.csum_block_size, prev_begin); r++; } - if ((prev_end % bs->dsk.csum_block_size) || - (prev_begin % bs->dsk.csum_block_size) && - (prev_end / bs->dsk.csum_block_size) == (prev_begin / bs->dsk.csum_block_size)) + if (prev_end % bs->dsk.csum_block_size) { - cb(i, prev_end - (prev_end % bs->dsk.csum_block_size ? (prev_end % bs->dsk.csum_block_size) : bs->dsk.csum_block_size), prev_end); + cb(i, prev_end, prev_end - (prev_end % bs->dsk.csum_block_size) + bs->dsk.csum_block_size); r++; } return r; @@ -420,17 +422,18 @@ void journal_flusher_co::fill_partial_checksum_blocks() iterate_checksum_holes([&](int vec_pos, uint32_t hole_start, uint32_t hole_end) { read_to_fill_incomplete = true; - int out_pos = read_vec.size(); - bs->prepare_disk_read(read_vec, out_pos, cur_obj, end_wr, + bs->prepare_disk_read(read_vec, read_vec.size(), cur_obj, end_wr, hole_start - hole_start % bs->dsk.csum_block_size, hole_start - hole_start % bs->dsk.csum_block_size + bs->dsk.csum_block_size, hole_start - hole_start % bs->dsk.csum_block_size, hole_start - hole_start % bs->dsk.csum_block_size + bs->dsk.csum_block_size, COPY_BUF_CSUM_FILL | (bs->padded_csum_update ? 0 : COPY_BUF_SKIP_CSUM)); - out_pos--; + auto & vec = read_vec[read_vec.size()-1]; + if (!vec.buf) + vec.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, vec.disk_len); read_vec.insert(read_vec.begin()+vec_pos, (copy_buffer_t){ .copy_flags = COPY_BUF_JOURNAL|COPY_BUF_COALESCED, .offset = hole_start, .len = hole_end-hole_start, - .buf = read_vec[out_pos].buf + hole_start - read_vec[out_pos].offset, + .buf = vec.buf + hole_start - vec.offset, }); }); } @@ -458,7 +461,7 @@ int journal_flusher_co::check_and_punch_checksums() return 0; } // Verify data checksums - cur_obj = bs->heap->read_locked_entry(cur_oid, copy_id); + cur_obj = bs->heap->read_locked_entry(cur_oid, copy_id); // FIXME locks can be removed from flusher bool csum_ok = true; for (int i = 0; i < read_vec.size(); i++) { @@ -469,9 +472,11 @@ int journal_flusher_co::check_and_punch_checksums() while (wr && wr->lsn != vec.wr_lsn) wr = wr->next(); assert(wr); + uint32_t *csums = (uint32_t*)(wr->get_checksums(bs->heap) + + (vec.offset/bs->dsk.csum_block_size)*(bs->dsk.data_csum_type & 0xFF) + - (((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) ? 0 : (wr->offset/bs->dsk.csum_block_size)*(bs->dsk.data_csum_type & 0xFF))); bs->heap->calc_block_checksums( - (uint32_t*)((uint8_t*)wr->get_checksums(bs->heap) + vec.offset/bs->dsk.csum_block_size*(bs->dsk.data_csum_type & 0xFF)), - vec.buf, wr->get_int_bitmap(bs->heap), vec.offset, vec.offset+vec.len, false, + csums, vec.buf, wr->get_int_bitmap(bs->heap), vec.offset, vec.offset+vec.len, false, [&](uint32_t mismatch_pos, uint32_t expected_csum, uint32_t real_csum) { printf("Checksum mismatch in object %jx:%jx v%ju in %s area at offset 0x%jx: got %08x, expected %08x\n", @@ -546,7 +551,7 @@ void journal_flusher_co::calc_block_checksums() for (auto it = read_vec.begin(); it != read_vec.end(); it++) { if (it->copy_flags & COPY_BUF_CSUM_FILL) - break; + continue; if (block_done == 0) { // `read_vec` should contain aligned items, possibly split into pieces @@ -637,6 +642,8 @@ bool journal_flusher_co::read_buffered(int wait_base) { await_sqe(0); auto & vec = read_vec[i]; + if (!vec.buf) + vec.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, vec.disk_len); data->iov = (struct iovec){ vec.buf, (size_t)vec.disk_len }; wait_count++; io_uring_prep_readv( diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index c63b656a..677b253c 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -1366,7 +1366,9 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea else if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE && (first_wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE) { + // FIXME: All other types of writes should also purge&merge the intent write auto second_wr = first_wr->next(); + second_wr->version = first_wr->version; bitmap_set(second_wr->get_int_bitmap(this), first_wr->offset, first_wr->len, dsk->bitmap_granularity); if (dsk->csum_block_size) { diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index 10c0ca62..7913f890 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -76,6 +76,7 @@ struct blockstore_op_private_t class blockstore_impl_t: public blockstore_i { +public: blockstore_disk_t dsk; /******* OPTIONS *******/ @@ -134,10 +135,6 @@ class blockstore_impl_t: public blockstore_i return ringloop->get_sqe(); } - friend class blockstore_init_meta; - friend class journal_flusher_t; - friend class journal_flusher_co; - void open_data(); void open_meta(); void open_journal(); @@ -159,7 +156,7 @@ class blockstore_impl_t: public blockstore_i uint32_t prepare_read_with_bitmaps(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end); uint32_t prepare_read_zero(std::vector & read_vec, uint32_t start, uint32_t end); uint32_t prepare_read_simple(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end); - void prepare_disk_read(std::vector & read_vec, int & pos, heap_object_t *obj, heap_write_t *wr, + void prepare_disk_read(std::vector & read_vec, int pos, heap_object_t *obj, heap_write_t *wr, uint32_t blk_start, uint32_t blk_end, uint32_t start, uint32_t end, uint32_t copy_flags); void find_holes(std::vector & read_vec, uint32_t item_start, uint32_t item_end, std::function callback); @@ -186,7 +183,7 @@ class blockstore_impl_t: public blockstore_i // List void process_list(blockstore_op_t *op); -public: +/*public:*/ blockstore_impl_t(blockstore_config_t & config, ring_loop_i *ringloop, timerfd_manager_t *tfd, bool mock_mode = false); ~blockstore_impl_t(); diff --git a/src/blockstore/blockstore_internal.h b/src/blockstore/blockstore_internal.h index 3f226c1d..89569d43 100644 --- a/src/blockstore/blockstore_internal.h +++ b/src/blockstore/blockstore_internal.h @@ -41,10 +41,10 @@ // Suspend operation until there are bytes of free space in the journal on disk #define WAIT_COMPACTION 2 -#define COPY_BUF_JOURNAL 1 -#define COPY_BUF_DATA 2 -#define COPY_BUF_ZERO 4 -#define COPY_BUF_CSUM_FILL 8 -#define COPY_BUF_COALESCED 16 -#define COPY_BUF_PADDED 32 -#define COPY_BUF_SKIP_CSUM 64 +#define COPY_BUF_JOURNAL 0x01 +#define COPY_BUF_DATA 0x02 +#define COPY_BUF_ZERO 0x04 +#define COPY_BUF_CSUM_FILL 0x08 +#define COPY_BUF_COALESCED 0x10 +#define COPY_BUF_PADDED 0x20 +#define COPY_BUF_SKIP_CSUM 0x40 diff --git a/src/blockstore/blockstore_read.cpp b/src/blockstore/blockstore_read.cpp index 60919398..2f9032a9 100644 --- a/src/blockstore/blockstore_read.cpp +++ b/src/blockstore/blockstore_read.cpp @@ -195,7 +195,7 @@ uint32_t blockstore_impl_t::prepare_read_simple(std::vector & rea else if (dsk.csum_block_size <= dsk.bitmap_granularity) { // simple disk read - prepare_disk_read(read_vec, pos, obj, wr, start, end, start, end, 0); + prepare_disk_read(read_vec, pos++, obj, wr, start, end, start, end, 0); } else { @@ -221,7 +221,7 @@ uint32_t blockstore_impl_t::prepare_read_simple(std::vector & rea blk_end == end && blk_start == start) { // single block, two partial blocks, or any number of full blocks - prepare_disk_read(read_vec, pos, obj, wr, blk_start, blk_end, start, end, skip_csum); + prepare_disk_read(read_vec, pos++, obj, wr, blk_start, blk_end, start, end, skip_csum); } else { @@ -229,18 +229,18 @@ uint32_t blockstore_impl_t::prepare_read_simple(std::vector & rea uint32_t full_start = (blk_start != start ? blk_start+dsk.csum_block_size : blk_start); uint32_t full_end = (blk_end != end ? blk_end-dsk.csum_block_size : blk_end); if (blk_start != start) - prepare_disk_read(read_vec, pos, obj, wr, blk_start, full_start, start, full_start, skip_csum); + prepare_disk_read(read_vec, pos++, obj, wr, blk_start, full_start, start, full_start, skip_csum); if (full_start > full_end) - prepare_disk_read(read_vec, pos, obj, wr, full_start, full_end, full_start, full_end, skip_csum); + prepare_disk_read(read_vec, pos++, obj, wr, full_start, full_end, full_start, full_end, skip_csum); if (blk_end != end) - prepare_disk_read(read_vec, pos, obj, wr, full_end, blk_end, full_end, end, skip_csum); + prepare_disk_read(read_vec, pos++, obj, wr, full_end, blk_end, full_end, end, skip_csum); } } }); return res; } -void blockstore_impl_t::prepare_disk_read(std::vector & read_vec, int & pos, heap_object_t *obj, heap_write_t *wr, +void blockstore_impl_t::prepare_disk_read(std::vector & read_vec, int pos, heap_object_t *obj, heap_write_t *wr, uint32_t blk_start, uint32_t blk_end, uint32_t start, uint32_t end, uint32_t copy_flags) { // Only one INTENT_WRITE is allowed at a time @@ -256,19 +256,19 @@ void blockstore_impl_t::prepare_disk_read(std::vector & read_vec, if (blk_start != start || blk_end != end) { vec.copy_flags |= COPY_BUF_PADDED; - if (read_vec.size() > pos && (read_vec[pos].copy_flags & ~COPY_BUF_CSUM_FILL) == vec.copy_flags && - read_vec[pos].offset >= blk_start && read_vec[pos].offset+read_vec[pos].len <= blk_end) + if (pos > 0 && read_vec.size() >= pos && (read_vec[pos-1].copy_flags & ~COPY_BUF_CSUM_FILL) == vec.copy_flags && + read_vec[pos-1].offset >= blk_start && read_vec[pos-1].offset+read_vec[pos-1].len <= blk_end) { // This is the same block as the previous one, we can read it only once vec.copy_flags |= COPY_BUF_COALESCED; - vec.buf = read_vec[pos].buf; + vec.buf = read_vec[pos-1].buf; } else { vec.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, blk_end-blk_start); } } - read_vec.insert(read_vec.begin() + (pos++), vec); + read_vec.insert(read_vec.begin() + pos, vec); } void blockstore_impl_t::find_holes(std::vector & read_vec, diff --git a/src/blockstore/blockstore_stable.cpp b/src/blockstore/blockstore_stable.cpp index 66bebcf7..6e5e25a7 100644 --- a/src/blockstore/blockstore_stable.cpp +++ b/src/blockstore/blockstore_stable.cpp @@ -17,6 +17,7 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op) // Modify in-memory state and assign contiguous LSNs priv->stab_pos = 0; priv->lsn = priv->to_lsn = 0; + op->retval = 0; while (priv->stab_pos < op->len) { uint32_t modified_block = 0; diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index 08055b0e..c80b39fa 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -177,13 +177,14 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) { // Direct intent-write BS_SUBMIT_CHECK_SQES(1); - for (auto wr = obj->get_writes(); wr; wr = wr->next()) + if ((obj->get_writes()->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) { - assert(wr->flags != BS_HEAP_BIG_WRITE); - if (wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)) - { - PRIV(op)->location = wr->location; - } + PRIV(op)->location = obj->get_writes()->location; + } + else + { + assert((obj->get_writes()->next()->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE); + PRIV(op)->location = obj->get_writes()->next()->location; } process_intent: uint8_t wr_buf[heap->get_max_write_entry_size()]; diff --git a/src/test/test_blockstore.cpp b/src/test/test_blockstore.cpp index 6b064acf..03d378d1 100644 --- a/src/test/test_blockstore.cpp +++ b/src/test/test_blockstore.cpp @@ -132,6 +132,14 @@ struct bs_test_t } }; +static bool memcheck(uint8_t *buf, uint8_t byte, size_t len) +{ + for (size_t i = 0; i < len; i++) + if (buf[i] != byte) + return false; + return true; +} + static void test_simple() { printf("\n-- test_simple\n"); @@ -274,10 +282,144 @@ static void test_fsync(bool separate_meta) free(op2.buf); } +static void test_intent_over_unstable() +{ + printf("\n-- test_intent_over_unstable\n"); + + bs_test_t test; + test.default_cfg(); + test.init(); + + // Write + printf("writing\n"); + blockstore_op_t op; + op.opcode = BS_OP_WRITE; + op.oid = { .inode = 1, .stripe = 0 }; + op.version = 1; + op.offset = 20480; + op.len = 4096; + op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096); + memset(op.buf, 0xaa, 4096); + test.exec_op(&op); + assert(op.retval == op.len); + + // Write again + printf("writing again\n"); + op.version = 2; + op.offset = 28*1024; + test.exec_op(&op); + assert(op.retval == op.len); + + free(op.buf); +} + +static void test_padded_csum_intent() +{ + printf("\n-- test_padded_csum_intent\n"); + + bs_test_t test; + test.default_cfg(); + test.config["csum_block_size"] = "16384"; + test.init(); + + // Write + printf("writing\n"); + blockstore_op_t op; + op.opcode = BS_OP_WRITE; + op.oid = { .inode = 1, .stripe = 0 }; + op.version = 1; + op.offset = 8192; + op.len = 4096; + op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096); + memset(op.buf, 0xaa, 4096); + test.exec_op(&op); + assert(op.retval == op.len); + + // Read + printf("reading\n"); + blockstore_op_t op2; + op2.opcode = BS_OP_READ; + op2.oid = { .inode = 1, .stripe = 0 }; + op2.version = UINT64_MAX; + op2.offset = 0; + op2.len = 128*1024; + op2.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024); + test.exec_op(&op2); + assert(op2.retval == op2.len); + assert(is_zero(op2.buf, 8*1024)); + assert(memcmp(op2.buf+8*1024, op.buf, 4*1024) == 0); + assert(is_zero(op2.buf+12*1024, 116*1024)); + + // Write again (intent) + printf("writing (intent)\n"); + op.version = 2; + op.offset = 28*1024; + memset(op.buf, 0xbb, 4096); + test.exec_op(&op); + assert(op.retval == op.len); + + // Write again (small because uncompactable) + printf("writing (small)\n"); + op.version = 3; + op.offset = 60*1024; + memset(op.buf, 0xcc, 4096); + test.exec_op(&op); + assert(op.retval == op.len); + + // Check that these are really big+intent+small writes + // (intent is not collapsible because of csum_block_size > bitmap_granularity) + heap_object_t *obj = test.bs->heap->read_entry((object_id){ .inode = 1, .stripe = 0 }, NULL); + assert(obj); + assert(!obj->get_writes()->next()->next()->next()); + assert(obj->get_writes()->flags == BS_HEAP_SMALL_WRITE); + assert(obj->get_writes()->next()->flags == BS_HEAP_INTENT_WRITE); + assert(obj->get_writes()->next()->next()->flags == BS_HEAP_BIG_WRITE); + + // Commit + printf("commit version 3\n"); + op.opcode = BS_OP_STABLE; + op.len = 1; + *((obj_ver_id*)op.buf) = { + .oid = { .inode = 1, .stripe = 0 }, + .version = 3, + }; + test.exec_op(&op); + assert(op.retval == 0); + assert(test.bs->heap->get_compact_queue_size()); + + // Trigger & wait compaction + test.bs->flusher->request_trim(); + // FIXME: Не зацикливаться при обломе + while (test.bs->heap->get_compact_queue_size()) + test.ringloop->loop(); + test.bs->flusher->release_trim(); + + // Read again and check + printf("reading compacted\n"); + op2.version = UINT64_MAX; + test.exec_op(&op2); + assert(op2.retval == op2.len); + assert(memcheck(op2.buf, 0, 8*1024)); + assert(memcheck(op2.buf+8*1024, 0xaa, 4*1024)); + assert(memcheck(op2.buf+12*1024, 0, 16*1024)); + assert(memcheck(op2.buf+28*1024, 0xbb, 4*1024)); + assert(memcheck(op2.buf+32*1024, 0, 28*1024)); + assert(memcheck(op2.buf+60*1024, 0xcc, 4*1024)); + assert(memcheck(op2.buf+64*1024, 0, 64*1024)); + + obj = test.bs->heap->read_entry((object_id){ .inode = 1, .stripe = 0 }, NULL); + assert(!obj->get_writes()->next()); + + free(op.buf); + free(op2.buf); +} + int main(int narg, char *args[]) { test_simple(); test_fsync(false); test_fsync(true); + test_intent_over_unstable(); + test_padded_csum_intent(); return 0; }