diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index 49145855..66f31247 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -410,7 +410,7 @@ void journal_flusher_co::iterate_checksum_holes(std::functiondsk.csum_block_size); if (blk_begin < big_start) blk_begin = big_start; - cb(pos, blk_begin, prev_begin); + cb(pos++, blk_begin, prev_begin); r++; } if ((prev_end % bs->dsk.csum_block_size) && prev_end < big_end) @@ -418,7 +418,7 @@ void journal_flusher_co::iterate_checksum_holes(std::functiondsk.csum_block_size) + bs->dsk.csum_block_size; if (blk_end > big_end) blk_end = big_end; - cb(i, prev_end, blk_end); + cb(++pos, prev_end, blk_end); r++; } return r; @@ -430,10 +430,12 @@ 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; -// bs->prepare_read(read_vec, cur_obj, end_wr, hole_start, hole_end); + uint32_t blk_begin = (hole_start - hole_start % bs->dsk.csum_block_size); 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, + blk_begin < big_start ? big_start : blk_begin, + (blk_begin + bs->dsk.csum_block_size) > big_end ? big_end : (blk_begin + bs->dsk.csum_block_size), + blk_begin < big_start ? big_start : blk_begin, + (blk_begin + bs->dsk.csum_block_size) > big_end ? big_end : (blk_begin + bs->dsk.csum_block_size), COPY_BUF_CSUM_FILL | (bs->perfect_csum_update ? 0 : COPY_BUF_SKIP_CSUM)); auto & vec = read_vec[read_vec.size()-1]; if (!vec.buf) @@ -469,7 +471,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); // FIXME locks can be removed from flusher + cur_obj = bs->heap->read_locked_entry(cur_oid, copy_id); bool csum_ok = true; for (int i = 0; i < read_vec.size(); i++) { @@ -507,8 +509,7 @@ int journal_flusher_co::check_and_punch_checksums() // Nothing to do return 0; } - // FIXME: Do it before read_buffered? - cur_obj = bs->heap->read_entry(cur_oid, &modified_block, true); + cur_obj = bs->heap->read_entry(cur_oid, &modified_block); if (!cur_obj) { // Object is deleted, abort compaction diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index 5d86831f..357153d4 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -160,6 +160,7 @@ public: 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); + void free_read_buffers(std::vector & rv); void handle_read_event(ring_data_t *data, blockstore_op_t *op); bool verify_read_checksums(blockstore_op_t *op); diff --git a/src/blockstore/blockstore_read.cpp b/src/blockstore/blockstore_read.cpp index 8bc812d7..29493a11 100644 --- a/src/blockstore/blockstore_read.cpp +++ b/src/blockstore/blockstore_read.cpp @@ -54,17 +54,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op) { // Need to wait. undo added requests, unlock lsn heap->unlock_entry(op->oid, PRIV(op)->lsn); - if (dsk.csum_block_size > dsk.bitmap_granularity) - { - for (auto & vec: rv) - { - if (!(vec.copy_flags & COPY_BUF_COALESCED) && vec.buf) - { - free(vec.buf); - vec.buf = NULL; - } - } - } + free_read_buffers(rv); rv.clear(); return 0; } @@ -73,6 +63,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op) { // everything is fulfilled from memory op->retval = op->len; + free_read_buffers(rv); FINISH_OP(op); return 2; } @@ -305,6 +296,21 @@ void blockstore_impl_t::find_holes(std::vector & read_vec, } } +void blockstore_impl_t::free_read_buffers(std::vector & rv) +{ + if (dsk.csum_block_size > dsk.bitmap_granularity) + { + for (auto & vec: rv) + { + if (!(vec.copy_flags & COPY_BUF_COALESCED) && vec.buf) + { + free(vec.buf); + vec.buf = NULL; + } + } + } +} + void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op) { live = true; @@ -322,6 +328,7 @@ void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op else if (op->retval == 0) op->retval = op->len; heap->unlock_entry(op->oid, PRIV(op)->lsn); + free_read_buffers(PRIV(op)->read_vec); FINISH_OP(op); } } diff --git a/src/test/ringloop_mock.cpp b/src/test/ringloop_mock.cpp index 82a8bb1e..336d1a38 100644 --- a/src/test/ringloop_mock.cpp +++ b/src/test/ringloop_mock.cpp @@ -166,9 +166,8 @@ void ring_loop_mock_t::mark_completed(ring_data_t *data) wakeup(); } -disk_mock_t::disk_mock_t(ring_loop_mock_t *loop, size_t size, bool buffered) +disk_mock_t::disk_mock_t(size_t size, bool buffered) { - this->loop = loop; this->size = size; this->data = (uint8_t*)malloc_or_die(size); this->buffered = buffered; @@ -392,6 +391,5 @@ bool disk_mock_t::submit(io_uring_sqe *sqe) // 1) reads submitted in parallel to writes (not after completing the write) should return old or new data randomly // 2) parallel operation completions should be delivered in random order // 3) when fsync is enabled, write cache should be sometimes lost during a simulated power outage - loop->mark_completed(userdata); return true; } diff --git a/src/test/ringloop_mock.h b/src/test/ringloop_mock.h index b655dac1..639df98a 100644 --- a/src/test/ringloop_mock.h +++ b/src/test/ringloop_mock.h @@ -47,14 +47,13 @@ class disk_mock_t std::map buffers; size_t size = 0; bool buffered = false; - ring_loop_mock_t *loop = NULL; void erase_buffers(uint64_t begin, uint64_t end); ssize_t copy_from_sqe(io_uring_sqe *sqe, uint8_t *to, uint64_t base_offset); void read_item(uint8_t *to, uint64_t offset, uint64_t len); public: bool trace = false; - disk_mock_t(ring_loop_mock_t *loop, size_t size, bool buffered); + disk_mock_t(size_t size, bool buffered); ~disk_mock_t(); void clear(size_t offset, size_t len); void discard_buffers(bool all, uint32_t seed); diff --git a/src/test/test_blockstore.cpp b/src/test/test_blockstore.cpp index 6eed7173..3dd5cf8e 100644 --- a/src/test/test_blockstore.cpp +++ b/src/test/test_blockstore.cpp @@ -11,6 +11,7 @@ struct bs_test_t blockstore_config_t config; disk_mock_t *data_disk = NULL; disk_mock_t *meta_disk = NULL; + std::function sqe_handler; ring_loop_mock_t *ringloop = NULL; timerfd_manager_t *tfd = NULL; blockstore_impl_t *bs = NULL; @@ -77,15 +78,20 @@ struct bs_test_t { ringloop = new ring_loop_mock_t(RINGLOOP_DEFAULT_SIZE, [&](io_uring_sqe *sqe) { - if (sqe->fd == MOCK_DATA_FD) + if (sqe_handler && sqe_handler(sqe)) + { + } + else if (sqe->fd == MOCK_DATA_FD) { bool ok = data_disk->submit(sqe); assert(ok); + ringloop->mark_completed((ring_data_t*)sqe->user_data); } else if (sqe->fd == MOCK_META_FD) { bool ok = meta_disk->submit(sqe); assert(ok); + ringloop->mark_completed((ring_data_t*)sqe->user_data); } else { @@ -99,13 +105,13 @@ struct bs_test_t } if (!data_disk) { - data_disk = new disk_mock_t(ringloop, parse_size(config["data_device_size"]), config["disable_data_fsync"] != "1"); + data_disk = new disk_mock_t(parse_size(config["data_device_size"]), config["disable_data_fsync"] != "1"); data_disk->clear(0, parse_size(config["data_offset"])); } uint64_t meta_size = parse_size(config["meta_device_size"]); if (meta_size && !meta_disk) { - meta_disk = new disk_mock_t(ringloop, meta_size, config["disable_meta_fsync"] != "1"); + meta_disk = new disk_mock_t(meta_size, config["disable_meta_fsync"] != "1"); meta_disk->clear(0, meta_size); } if (!bs) @@ -419,6 +425,88 @@ static void test_padded_csum_intent(bool perfect) free(op2.buf); } +static void test_padded_csum_parallel_read(bool perfect, uint32_t offset) +{ + printf("\n-- test_padded_csum_parallel_read%s offset=%u\n", perfect ? " perfect_csum_update" : "", offset); + + bs_test_t test; + test.default_cfg(); + test.config["csum_block_size"] = "16384"; + test.config["atomic_write_size"] = "0"; + if (perfect) + test.config["perfect_csum_update"] = "1"; + test.init(); + + // Write + printf("writing (initial)\n"); + blockstore_op_t op; + op.opcode = BS_OP_WRITE_STABLE; + op.oid = { .inode = 1, .stripe = 0 }; + op.version = 1; + op.offset = 8192; + op.len = 16384; + op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 16384); + memset(op.buf, 0xaa, 16384); + test.exec_op(&op); + assert(op.retval == op.len); + + // Write 2 + printf("writing (%u+%u)\n", offset, 4096); + op.version = 2; + op.offset = offset; + op.len = 4096; + memset(op.buf, 0xbb, 4096); + test.exec_op(&op); + assert(op.retval == op.len); + + // Trigger & wait compaction + test.bs->flusher->request_trim(); + std::vector flush_writes; + test.sqe_handler = [&](io_uring_sqe *sqe) + { + if (sqe->fd == MOCK_DATA_FD && sqe->opcode == IORING_OP_WRITEV && + sqe->off >= test.bs->dsk.data_offset) + { + bool ok = test.data_disk->submit(sqe); + assert(ok); + flush_writes.push_back((ring_data_t*)sqe->user_data); + return true; + } + return false; + }; + // Wait for 2 flusher writes, execute and pause them + while (test.bs->heap->get_compact_queue_size() && flush_writes.size() < 1) + test.ringloop->loop(); + while (test.bs->flusher->is_active() && flush_writes.size() < 1) + test.ringloop->loop(); + // Run a read operation in parallel - it shouldn't complain about checksum errors + printf("reading in parallel\n"); + blockstore_op_t op2; + op2.opcode = BS_OP_READ; + op2.oid = { .inode = 1, .stripe = 0 }; + op2.version = 1; + 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); + // Continue flushing + test.sqe_handler = NULL; + for (auto & w: flush_writes) + test.ringloop->mark_completed(w); + flush_writes.clear(); + while (test.bs->heap->get_compact_queue_size() && flush_writes.size() < 2) + test.ringloop->loop(); + while (test.bs->flusher->is_active() && flush_writes.size() < 2) + test.ringloop->loop(); + test.bs->flusher->release_trim(); + // Check that compaction succeeded + assert(!test.bs->heap->get_to_compact_count()); + + free(op.buf); + free(op2.buf); +} + int main(int narg, char *args[]) { test_simple(); @@ -427,5 +515,9 @@ int main(int narg, char *args[]) test_intent_over_unstable(); test_padded_csum_intent(false); test_padded_csum_intent(true); + test_padded_csum_parallel_read(false, 8192); + test_padded_csum_parallel_read(true, 8192); + test_padded_csum_parallel_read(false, 16384); + test_padded_csum_parallel_read(true, 16384); return 0; }