diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index b6c3e9a7..6d33ec8b 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -169,7 +169,6 @@ bool journal_flusher_co::loop() else if (wait_state == 22) goto resume_22; else if (wait_state == 23) goto resume_23; else if (wait_state == 24) goto resume_24; - else if (wait_state == 25) goto resume_25; resume_0: wait_state = 0; cur_oid = {}; @@ -179,7 +178,6 @@ resume_0: { resume_21: resume_22: -resume_23: res = fsync_buffer(21); if (!res) { @@ -335,9 +333,9 @@ resume_10: calc_block_checksums(); if (read_to_fill_incomplete) { +resume_23: resume_24: -resume_25: - if (!write_meta_block(24)) + if (!write_meta_block(23)) { return false; } @@ -698,8 +696,7 @@ int journal_flusher_co::fsync_buffer(int wait_base) { if (wait_state == wait_base) goto resume_0; else if (wait_state == wait_base+1) goto resume_1; - else if (wait_state == wait_base+2) goto resume_2; - if (!bs->unsynced_big_write_count && !bs->unsynced_small_write_count) + if (bs->dsk.disable_journal_fsync && bs->dsk.disable_meta_fsync && bs->dsk.disable_data_fsync || !bs->unsynced_big_write_count && !bs->unsynced_small_write_count) { return 1; } @@ -707,32 +704,20 @@ int journal_flusher_co::fsync_buffer(int wait_base) { return 0; } - compact_lsn = bs->heap->get_completed_lsn(); flusher->active_flushers++; flusher->syncing_buffer++; +resume_0: assert(!wait_count); - if (!bs->dsk.disable_meta_fsync) + compact_lsn = bs->heap->get_completed_lsn(); + if (!bs->submit_fsyncs(wait_count)) { - bs->unsynced_big_write_count = 0; - await_sqe(0); - data->iov = { 0 }; - data->callback = simple_callback_w; - io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC); - wait_count++; + wait_state = wait_base+0; + return 0; } - if (bs->unsynced_small_write_count > 0 && !bs->dsk.disable_journal_fsync && bs->dsk.journal_fd != bs->dsk.meta_fd) - { - bs->unsynced_small_write_count = 0; - await_sqe(1); - data->iov = { 0 }; - data->callback = simple_callback_w; - io_uring_prep_fsync(sqe, bs->dsk.journal_fd, IORING_FSYNC_DATASYNC); - wait_count++; - } -resume_2: +resume_1: if (wait_count > 0) { - wait_state = wait_base+2; + wait_state = wait_base+1; return 0; } bs->heap->mark_lsn_fsynced(compact_lsn); diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index f4d78f38..c63b656a 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -55,8 +55,9 @@ uint32_t heap_write_t::get_csum_size(blockstore_heap_t *heap) bool heap_write_t::needs_recheck(blockstore_heap_t *heap) { - return len > 0 && lsn > heap->compacted_lsn && (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) - || flags == BS_HEAP_SMALL_WRITE || flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)); + return len > 0 && lsn > heap->compacted_lsn && + ((flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE || + (flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE); } bool heap_write_t::needs_compact(blockstore_heap_t *heap) @@ -418,6 +419,14 @@ skip_object: else if (!calc_checksums(wr, buffer_area + wr->location, false)) { // entry is invalid (not fully written before OSD crash) - remove it and all newer (previous) entries too + if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE && + wr->next() && (wr->next()->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE && + wr->next()->version == wr->version) + { + // BIG_WRITE+INTENT_WRITE pair + wr = wr->next(); + wr_i++; + } remove_wr = wr; remove_i = wr_i; } @@ -636,12 +645,12 @@ bool blockstore_heap_t::recheck_small_writes(std::functionneeds_recheck(this)) { - bool is_intent = (wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)); + bool is_intent = (wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE; uint64_t loc = wr->location; if (is_intent) { auto next_wr = wr->next(); - assert(next_wr && next_wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)); + assert(next_wr && next_wr->flags == (BS_HEAP_BIG_WRITE | (wr->flags & BS_HEAP_STABLE))); loc = wr->offset + next_wr->location; } recheck_in_progress++; @@ -664,6 +673,14 @@ bool blockstore_heap_t::recheck_small_writes(std::functionflags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE && + wr->next() && (wr->next()->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE && + wr->next()->version == wr->version) + { + // BIG_WRITE+INTENT_WRITE pair + wr = wr->next(); + wr_i++; + } // Erase all writes to the object from this one to the newest if (!wr->next_pos) { @@ -1291,13 +1308,13 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea // Stable overwrites are not allowed over unstable return EINVAL; } - if (wr->flags == BS_HEAP_INTENT_WRITE) + if (wr->flags == BS_HEAP_INTENT_WRITE && (first_wr->flags & BS_HEAP_STABLE)) { - // Unstable intent writes are not allowed + // Unstable intent writes over stable are not allowed return EINVAL; } - if (wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) && - first_wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) && + if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE && + (first_wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE && !first_wr->can_be_collapsed(this)) { // Intent writes are not allowed over noncollapsible intent writes diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index 4cf71a83..10c0ca62 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -124,10 +124,10 @@ class blockstore_impl_t: public blockstore_i bool fsyncing_data = false; bool live = false, queue_stall = false; - ring_loop_i *ringloop; - timerfd_manager_t *tfd; + ring_loop_i *ringloop = NULL; + timerfd_manager_t *tfd = NULL; - bool stop_sync_submitted; + bool stop_sync_submitted = false; inline struct io_uring_sqe* get_sqe() { @@ -177,6 +177,7 @@ class blockstore_impl_t: public blockstore_i // Sync int continue_sync(blockstore_op_t *op); + bool submit_fsyncs(int & wait_count); int do_sync(blockstore_op_t *op, int base_state); // Stabilize diff --git a/src/blockstore/blockstore_init.cpp b/src/blockstore/blockstore_init.cpp index 5f5f6b66..a8e5b6ff 100644 --- a/src/blockstore/blockstore_init.cpp +++ b/src/blockstore/blockstore_init.cpp @@ -3,6 +3,7 @@ #include "blockstore_impl.h" #include "blockstore_internal.h" +#include "str_util.h" #include "crc32c.h" #define INIT_META_EMPTY 0 @@ -16,14 +17,6 @@ throw std::runtime_error("io_uring is full during initialization");\ data = ((ring_data_t*)sqe->user_data) -static bool iszero(uint64_t *buf, int len) -{ - for (int i = 0; i < len; i++) - if (buf[i] != 0) - return false; - return true; -} - blockstore_init_meta::blockstore_init_meta(blockstore_impl_t *bs) { this->bs = bs; @@ -74,7 +67,7 @@ resume_1: wait_state = 1; return 1; } - if (iszero((uint64_t*)bs->meta_superblock, bs->dsk.meta_block_size / sizeof(uint64_t))) + if (is_zero((uint64_t*)bs->meta_superblock, bs->dsk.meta_block_size)) { { blockstore_meta_header_v3_t *hdr = (blockstore_meta_header_v3_t *)bs->meta_superblock; diff --git a/src/blockstore/blockstore_read.cpp b/src/blockstore/blockstore_read.cpp index 090c2e70..60919398 100644 --- a/src/blockstore/blockstore_read.cpp +++ b/src/blockstore/blockstore_read.cpp @@ -35,8 +35,8 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op) } fulfilled += prepare_read(PRIV(op)->read_vec, obj, wr, op->offset, op->offset+op->len); if (fulfilled == op->len || - wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) || - wr->flags == (BS_HEAP_TOMBSTONE|BS_HEAP_STABLE)) + (wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE || + (wr->flags & BS_HEAP_TYPE) == BS_HEAP_TOMBSTONE) { break; } @@ -111,10 +111,6 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *op) uint32_t blockstore_impl_t::prepare_read(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end) { - if (wr->offset >= end || wr->offset+wr->len <= start) - { - return 0; - } start = start < wr->offset ? wr->offset : start; end = end > wr->offset+wr->len ? wr->offset+wr->len : end; if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) @@ -176,6 +172,10 @@ uint32_t blockstore_impl_t::prepare_read_zero(std::vector & read_ uint32_t blockstore_impl_t::prepare_read_simple(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end) { uint32_t res = 0; + if (wr->offset >= end || wr->offset+wr->len <= start) + { + return 0; + } find_holes(read_vec, start, end, [&](int & pos, uint32_t start, uint32_t end) { res += end-start; @@ -244,7 +244,7 @@ void blockstore_impl_t::prepare_disk_read(std::vector & read_vec, 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 - assert(wr->flags != (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) || wr->next()->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)); + assert((wr->flags & BS_HEAP_TYPE) != BS_HEAP_INTENT_WRITE || (wr->next()->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE); copy_buffer_t vec = { .copy_flags = ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE ? COPY_BUF_JOURNAL : COPY_BUF_DATA) | copy_flags, .offset = start, @@ -352,7 +352,9 @@ bool blockstore_impl_t::verify_read_checksums(blockstore_op_t *op) memcpy(op->buf + vec.offset - op->offset, vec.buf + vec.offset - blk_start, vec.len); } uint8_t *buf = vec.buf ? vec.buf : (op->buf + vec.offset - op->offset); - uint32_t *csums = (uint32_t*)(wr->get_checksums(heap) + (blk_start/dsk.csum_block_size)*(dsk.data_csum_type & 0xFF)); + uint32_t *csums = (uint32_t*)(wr->get_checksums(heap) + + (blk_start/dsk.csum_block_size)*(dsk.data_csum_type & 0xFF) + - (((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) ? 0 : (wr->offset/dsk.csum_block_size)*(dsk.data_csum_type & 0xFF))); if (!heap->calc_block_checksums(csums, buf, wr->get_int_bitmap(heap), blk_start, blk_end, false, [&](uint32_t mismatch_pos, uint32_t expected_csum, uint32_t real_csum) { diff --git a/src/blockstore/blockstore_sync.cpp b/src/blockstore/blockstore_sync.cpp index 8a84f1d6..91a809b1 100644 --- a/src/blockstore/blockstore_sync.cpp +++ b/src/blockstore/blockstore_sync.cpp @@ -18,6 +18,66 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op) return res; } +bool blockstore_impl_t::submit_fsyncs(int & wait_count) +{ + int n = ((unsynced_small_write_count > 0 || unsynced_big_write_count > 0) && !dsk.disable_meta_fsync) + + (unsynced_small_write_count > 0 && !dsk.disable_journal_fsync && dsk.journal_fd != dsk.meta_fd) + + (unsynced_big_write_count > 0 && !dsk.disable_data_fsync && dsk.data_fd != dsk.meta_fd && dsk.data_fd != dsk.journal_fd); + if (ringloop->space_left() < n) + { + return false; + } + if (!n) + { + return true; + } + auto cb = [this, & wait_count](ring_data_t *data) + { + if (data->res != 0) + disk_error_abort("sync meta", data->res, 0); + wait_count--; + assert(wait_count >= 0); + if (!wait_count) + ringloop->wakeup(); + }; + if (!dsk.disable_meta_fsync) + { + // fsync meta + io_uring_sqe *sqe = get_sqe(); + assert(sqe); + ring_data_t *data = ((ring_data_t*)sqe->user_data); + io_uring_prep_fsync(sqe, dsk.meta_fd, IORING_FSYNC_DATASYNC); + data->iov = { 0 }; + data->callback = cb; + wait_count++; + } + if (unsynced_small_write_count > 0 && !dsk.disable_journal_fsync && dsk.meta_fd != dsk.journal_fd) + { + // fsync buffer + io_uring_sqe *sqe = get_sqe(); + assert(sqe); + ring_data_t *data = ((ring_data_t*)sqe->user_data); + io_uring_prep_fsync(sqe, dsk.journal_fd, IORING_FSYNC_DATASYNC); + data->iov = { 0 }; + data->callback = cb; + wait_count++; + } + if (unsynced_big_write_count > 0 && !dsk.disable_data_fsync && dsk.data_fd != dsk.meta_fd && dsk.data_fd != dsk.journal_fd) + { + // fsync data + io_uring_sqe *sqe = get_sqe(); + assert(sqe); + ring_data_t *data = ((ring_data_t*)sqe->user_data); + io_uring_prep_fsync(sqe, dsk.data_fd, IORING_FSYNC_DATASYNC); + data->iov = { 0 }; + data->callback = cb; + wait_count++; + } + unsynced_big_write_count = 0; + unsynced_small_write_count = 0; + return true; +} + int blockstore_impl_t::do_sync(blockstore_op_t *op, int base_state) { int op_state = PRIV(op)->op_state - base_state; @@ -29,34 +89,19 @@ int blockstore_impl_t::do_sync(blockstore_op_t *op, int base_state) // Wait for flusher-initiated sync return 0; } - if (dsk.disable_journal_fsync && dsk.disable_meta_fsync || !unsynced_big_write_count && !unsynced_small_write_count) + if (dsk.disable_journal_fsync && dsk.disable_meta_fsync && dsk.disable_data_fsync || !unsynced_big_write_count && !unsynced_small_write_count) { // We can return immediately because sync only syncs previous writes unsynced_big_write_count = unsynced_small_write_count = 0; return 2; } PRIV(op)->lsn = heap->get_completed_lsn(); - stop_sync_submitted = false; - if (!dsk.disable_meta_fsync) + if (!submit_fsyncs(PRIV(op)->pending_ops)) { - // fsync meta - BS_SUBMIT_GET_SQE(sqe, data); - io_uring_prep_fsync(sqe, dsk.meta_fd, IORING_FSYNC_DATASYNC); - data->iov = { 0 }; - data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); }; - PRIV(op)->pending_ops++; + PRIV(op)->wait_detail = 1; + PRIV(op)->wait_for = WAIT_SQE; + return 0; } - if (unsynced_small_write_count > 0 && !dsk.disable_journal_fsync && dsk.meta_fd != dsk.journal_fd) - { - // fsync buffer - BS_SUBMIT_GET_SQE(sqe, data); - io_uring_prep_fsync(sqe, dsk.journal_fd, IORING_FSYNC_DATASYNC); - data->iov = { 0 }; - data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); }; - PRIV(op)->pending_ops++; - } - unsynced_big_write_count = 0; - unsynced_small_write_count = 0; resume_1: if (PRIV(op)->pending_ops > 0) { diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index 8204000b..08055b0e 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -123,9 +123,9 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) ); #endif heap->use_data(op->oid.inode, PRIV(op)->location); - if (!dsk.disable_data_fsync) + if (!dsk.disable_data_fsync && dsk.disable_meta_fsync) { - // Do big_write as an INTENT to avoid fsync + // Do big_write as an INTENT to avoid data fsync bool ok = make_big_write(op, 0, 0, &modified_block); assert(ok); obj = heap->read_entry(op->oid, &modified_block); @@ -160,8 +160,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) } // Only one INTENT_WRITE is allowed at a time, but in fact, // parallel writes to the same object are forbidden anyway - else if (op->opcode == BS_OP_WRITE_STABLE && - op->len > 0 && op->len <= dsk.atomic_write_size && + else if (op->len > 0 && op->len <= dsk.atomic_write_size && // Intent-writes are disabled if "absolutely correct during compaction" checksum validation algorithm is enabled // We could also do RMW here when padded_csum_update is enabled, but it's unclear if we need it (!padded_csum_update || dsk.csum_block_size <= dsk.bitmap_granularity || @@ -171,8 +170,10 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) obj->get_writes()->can_be_collapsed(heap))) && // One intent-write is allowed even with fsyncs because BIG_WRITE is always counted as fsynced dsk.disable_meta_fsync && - (obj->get_writes()->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) || - obj->get_writes()->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) && dsk.disable_data_fsync)) + (op->opcode == BS_OP_WRITE_STABLE && + (obj->get_writes()->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) || + obj->get_writes()->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) && dsk.disable_data_fsync) || + op->opcode == BS_OP_WRITE && obj->get_writes()->flags == BS_HEAP_BIG_WRITE)) { // Direct intent-write BS_SUBMIT_CHECK_SQES(1); @@ -191,7 +192,7 @@ process_intent: wr->offset = op->offset; wr->len = op->len; wr->location = 0; - wr->flags = BS_HEAP_INTENT_WRITE | BS_HEAP_STABLE; + wr->flags = BS_HEAP_INTENT_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0); if (op->bitmap) memcpy(wr->get_ext_bitmap(heap), op->bitmap, dsk.clean_entry_bitmap_size); heap->calc_checksums(wr, (uint8_t*)op->buf, true); diff --git a/src/test/ringloop_mock.cpp b/src/test/ringloop_mock.cpp index 6294fd3d..82a8bb1e 100644 --- a/src/test/ringloop_mock.cpp +++ b/src/test/ringloop_mock.cpp @@ -229,8 +229,18 @@ void disk_mock_t::erase_buffers(uint64_t begin, uint64_t end) } } +void disk_mock_t::clear(size_t offset, size_t len) +{ + if (offset < size) + { + memset(data+offset, 0, len < size-offset ? len : size-offset); + } +} + void disk_mock_t::discard_buffers(bool all, uint32_t seed) { + if (trace) + printf("disk: discard buffers all=%d seed=%u\n", all, seed); if (all) { for (auto & b: buffers) @@ -267,7 +277,7 @@ ssize_t disk_mock_t::copy_from_sqe(io_uring_sqe *sqe, uint8_t *to, uint64_t base } size_t cur = (off + v[i].iov_len > size ? size-off : v[i].iov_len); if (trace) - printf("write %zu+%zu from %jx\n", off, cur, (uint64_t)v[i].iov_base); + printf("disk: write %zu+%zu from %jx\n", off, cur, (uint64_t)v[i].iov_base); memcpy(to + off - base_offset, v[i].iov_base, cur); off += v[i].iov_len; } @@ -320,7 +330,7 @@ bool disk_mock_t::submit(io_uring_sqe *sqe) { size_t cur = (off + v[i].iov_len > size ? size-off : v[i].iov_len); if (trace) - printf("read %zu+%zu to %jx\n", off, cur, (uint64_t)v[i].iov_base); + printf("disk: read %zu+%zu to %jx\n", off, cur, (uint64_t)v[i].iov_base); if (buffers.size()) read_item((uint8_t*)v[i].iov_base, off, cur); else @@ -361,6 +371,8 @@ bool disk_mock_t::submit(io_uring_sqe *sqe) } else if (sqe->opcode == IORING_OP_FSYNC) { + if (trace) + printf("disk: fsync\n"); if (buffers.size()) { for (auto & b: buffers) diff --git a/src/test/ringloop_mock.h b/src/test/ringloop_mock.h index d2ac8f1a..b655dac1 100644 --- a/src/test/ringloop_mock.h +++ b/src/test/ringloop_mock.h @@ -56,6 +56,7 @@ public: bool trace = false; disk_mock_t(ring_loop_mock_t *loop, size_t size, bool buffered); ~disk_mock_t(); + void clear(size_t offset, size_t len); void discard_buffers(bool all, uint32_t seed); bool submit(io_uring_sqe *sqe); }; diff --git a/src/test/test_blockstore.cpp b/src/test/test_blockstore.cpp index ff1819ee..6b064acf 100644 --- a/src/test/test_blockstore.cpp +++ b/src/test/test_blockstore.cpp @@ -2,42 +2,145 @@ // License: VNPL-1.1 (see README.md for details) #include +#include "str_util.h" #include "ringloop_mock.h" #include "blockstore_impl.h" -int main(int narg, char *args[]) +struct bs_test_t { blockstore_config_t config; - config["data_device"] = "./test_data.bin"; - config["data_device_size"] = "1073741824"; - config["data_device_sect"] = "4096"; - config["meta_offset"] = "0"; - config["journal_offset"] = "16777216"; - config["data_offset"] = "33554432"; - config["disable_data_fsync"] = "1"; - config["immediate_commit"] = "all"; - config["log_level"] = "10"; - config["data_csum_type"] = "crc32c"; - config["csum_block_size"] = "4096"; - disk_mock_t *data_disk = NULL; - ring_loop_mock_t *ringloop = new ring_loop_mock_t(RINGLOOP_DEFAULT_SIZE, [&](io_uring_sqe *sqe) - { - assert(sqe->fd == MOCK_DATA_FD); - bool ok = data_disk->submit(sqe); - assert(ok); - }); - timerfd_manager_t *tfd = new timerfd_manager_t(nullptr); - data_disk = new disk_mock_t(ringloop, 1073741824, false); - blockstore_impl_t *bs = new blockstore_impl_t(config, ringloop, tfd, true); + disk_mock_t *meta_disk = NULL; + ring_loop_mock_t *ringloop = NULL; + timerfd_manager_t *tfd = NULL; + blockstore_impl_t *bs = NULL; - // Wait for blockstore init - while (!bs->is_started()) - ringloop->loop(); - printf("init completed\n"); + ~bs_test_t() + { + destroy(); + } + + void destroy_bs() + { + if (bs) + { + delete bs; + bs = NULL; + } + } + + void destroy() + { + while (bs && !bs->is_safe_to_stop()) + ringloop->loop(); + destroy_bs(); + if (tfd) + { + delete tfd; + tfd = NULL; + } + if (meta_disk) + { + delete meta_disk; + meta_disk = NULL; + } + if (data_disk) + { + delete data_disk; + data_disk = NULL; + } + if (ringloop) + { + delete ringloop; + ringloop = NULL; + } + } + + void default_cfg() + { + config["data_device"] = "./test_data.bin"; + config["data_device_size"] = "1073741824"; + config["data_device_sect"] = "4096"; + config["meta_offset"] = "0"; + config["journal_offset"] = "16777216"; + config["data_offset"] = "33554432"; + config["disable_data_fsync"] = "1"; + config["immediate_commit"] = "all"; + config["log_level"] = "10"; + config["data_csum_type"] = "crc32c"; + config["csum_block_size"] = "4096"; + } + + void init() + { + if (!ringloop) + { + ringloop = new ring_loop_mock_t(RINGLOOP_DEFAULT_SIZE, [&](io_uring_sqe *sqe) + { + if (sqe->fd == MOCK_DATA_FD) + { + bool ok = data_disk->submit(sqe); + assert(ok); + } + else if (sqe->fd == MOCK_META_FD) + { + bool ok = meta_disk->submit(sqe); + assert(ok); + } + else + { + assert(0); + } + }); + } + if (!tfd) + { + tfd = new timerfd_manager_t(nullptr); + } + if (!data_disk) + { + data_disk = new disk_mock_t(ringloop, 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->clear(0, meta_size); + } + if (!bs) + { + bs = new blockstore_impl_t(config, ringloop, tfd, true); + while (!bs->is_started()) + ringloop->loop(); + printf("blockstore initialized\n"); + } + } + + void exec_op(blockstore_op_t *op) + { + bool done = false; + op->callback = [&](blockstore_op_t *op) + { + printf("op opcode=%lu completed retval=%d\n", op->opcode, op->retval); + done = true; + }; + bs->enqueue_op(op); + while (!done) + ringloop->loop(); + op->callback = nullptr; + } +}; + +static void test_simple() +{ + printf("\n-- test_simple\n"); + + bs_test_t test; + test.default_cfg(); + test.init(); // Write - bool done = false; blockstore_op_t op; uint64_t version = 0; op.opcode = BS_OP_WRITE; @@ -47,51 +150,35 @@ int main(int narg, char *args[]) op.len = 4096; op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024); memset(op.buf, 0xaa, 4096); - op.callback = [&](blockstore_op_t *op) - { - printf("op completed code=%lu retval=%d\n", op->opcode, op->retval); - done = true; - }; - bs->enqueue_op(&op); - while (!done) - ringloop->loop(); + test.exec_op(&op); assert(op.retval == op.len); // Sync printf("version %ju written, syncing\n", op.version); - done = false; version = op.version; op.opcode = BS_OP_SYNC; - bs->enqueue_op(&op); - while (!done) - ringloop->loop(); + test.exec_op(&op); assert(op.retval == 0); // Commit printf("commit version %ju\n", version); - done = false; op.opcode = BS_OP_STABLE; op.len = 1; *((obj_ver_id*)op.buf) = { .oid = { .inode = 1, .stripe = 0 }, .version = version, }; - bs->enqueue_op(&op); - while (!done) - ringloop->loop(); + test.exec_op(&op); assert(op.retval == 0); // Read printf("reading 0-128K\n"); - done = false; op.opcode = BS_OP_READ; op.oid = { .inode = 1, .stripe = 0 }; op.version = UINT64_MAX; op.offset = 0; op.len = 128*1024; - bs->enqueue_op(&op); - while (!done) - ringloop->loop(); + test.exec_op(&op); assert(op.retval == op.len); uint8_t *cmp = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024); memset(cmp, 0, 128*1024); @@ -106,13 +193,91 @@ int main(int narg, char *args[]) free(cmp); free(op.buf); +} - // Destroy - while (!bs->is_safe_to_stop()) - ringloop->loop(); - delete bs; - delete tfd; - delete data_disk; - delete ringloop; +static void test_fsync(bool separate_meta) +{ + printf("\n-- test_fsync%s\n", separate_meta ? " separate_meta" : ""); + + bs_test_t test; + test.default_cfg(); + test.config["disable_data_fsync"] = "0"; + test.config["immediate_commit"] = "none"; + if (separate_meta) + { + test.config["meta_device"] = "./test_meta.bin"; + test.config["disable_meta_fsync"] = "1"; + test.config["meta_device_size"] = "33554432"; + test.config["meta_device_sect"] = "4096"; + test.config["data_offset"] = "0"; + } + 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 = 16384; + 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); + + // Destroy and restart without sync + printf("destroying\n"); + test.destroy_bs(); + test.data_disk->discard_buffers(true, 0); + test.init(); + + // Check ENOENT + printf("checking for ENOENT\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 == -ENOENT); + + // Write again + printf("writing again\n"); + test.exec_op(&op); + assert(op.retval == op.len); + + // Sync + printf("version %ju written, syncing\n", op.version); + op.opcode = BS_OP_SYNC; + test.exec_op(&op); + assert(op.retval == 0); + + // Discard and restart again + printf("destroying again\n"); + test.destroy_bs(); + test.data_disk->discard_buffers(true, 0); + test.init(); + + // Check that it's present now + printf("checking for OK\n"); + op2.version = UINT64_MAX; + test.exec_op(&op2); + assert(op2.retval == op2.len); + assert(is_zero(op2.buf, 16*1024)); + assert(memcmp(op2.buf+16*1024, op.buf, 4*1024) == 0); + assert(is_zero(op2.buf+20*1024, 108*1024)); + + free(op.buf); + free(op2.buf); +} + +int main(int narg, char *args[]) +{ + test_simple(); + test_fsync(false); + test_fsync(true); return 0; }