Add a test with fsync

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(23))
-        if (!write_meta_block(24))
         {
             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->dsk.disable_journal_fsync && bs->dsk.disable_meta_fsync && bs->dsk.disable_data_fsync || !bs->unsynced_big_write_count && !bs->unsynced_small_write_count)
-    if (!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;
+        wait_state = wait_base+0;
-        await_sqe(0);
+        return 0;
-        data->iov = { 0 };
-        data->callback = simple_callback_w;
-        io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC);
-        wait_count++;
     }
-    if (bs->unsynced_small_write_count > 0 && !bs->dsk.disable_journal_fsync && bs->dsk.journal_fd != bs->dsk.meta_fd)
+resume_1:
-    {
-        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:
     if (wait_count > 0)
     {
-        wait_state = wait_base+2;
+        wait_state = wait_base+1;
         return 0;
     }
     bs->heap->mark_lsn_fsynced(compact_lsn);

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)
+    return len > 0 && lsn > heap->compacted_lsn &&
-        || flags == BS_HEAP_SMALL_WRITE || flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE));
+        ((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::function<void(bool is_data, ui
         {
             if (wr->needs_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::function<void(bool is_data, ui
                         }
                         if (wr && !calc_checksums(wr, buf, false))
                         {
+                            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++;
+                            }
                             // 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) &&
+    if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE &&
-        first_wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) &&
+        (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

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;
+    ring_loop_i *ringloop = NULL;
-    timerfd_manager_t *tfd;
+    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

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;

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_TYPE) == BS_HEAP_BIG_WRITE ||
-            wr->flags == (BS_HEAP_TOMBSTONE|BS_HEAP_STABLE))
+            (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<copy_buffer_t> & 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<copy_buffer_t> & read_
 uint32_t blockstore_impl_t::prepare_read_simple(std::vector<copy_buffer_t> & 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<copy_buffer_t> & 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)
             {

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 (!submit_fsyncs(PRIV(op)->pending_ops))
-    if (!dsk.disable_meta_fsync)
     {
-        // fsync meta
+        PRIV(op)->wait_detail = 1;
-        BS_SUBMIT_GET_SQE(sqe, data);
+        PRIV(op)->wait_for = WAIT_SQE;
-        io_uring_prep_fsync(sqe, dsk.meta_fd, IORING_FSYNC_DATASYNC);
+        return 0;
-        data->iov = { 0 };
-        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
-        PRIV(op)->pending_ops++;
     }
-    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)
     {

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 &&
+    else if (op->len > 0 && op->len <= dsk.atomic_write_size &&
-        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) ||
+        (op->opcode == BS_OP_WRITE_STABLE &&
-        obj->get_writes()->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) && dsk.disable_data_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 && 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);

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)

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);
 };

src/test/test_blockstore.cpp

@@ -2,42 +2,145 @@
 // License: VNPL-1.1 (see README.md for details)
 
 #include <malloc.h>
+#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)
+    disk_mock_t *meta_disk = NULL;
-    {
+    ring_loop_mock_t *ringloop = NULL;
-        assert(sqe->fd == MOCK_DATA_FD);
+    timerfd_manager_t *tfd = NULL;
-        bool ok = data_disk->submit(sqe);
+    blockstore_impl_t *bs = NULL;
-        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);
 
-    // Wait for blockstore init
+    ~bs_test_t()
-    while (!bs->is_started())
+    {
-        ringloop->loop();
+        destroy();
-    printf("init completed\n");
+    }
+
+    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)
+    test.exec_op(&op);
-    {
-        printf("op completed code=%lu retval=%d\n", op->opcode, op->retval);
-        done = true;
-    };
-    bs->enqueue_op(&op);
-    while (!done)
-        ringloop->loop();
     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);
+    test.exec_op(&op);
-    while (!done)
-        ringloop->loop();
     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);
+    test.exec_op(&op);
-    while (!done)
-        ringloop->loop();
     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);
+    test.exec_op(&op);
-    while (!done)
-        ringloop->loop();
     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
+static void test_fsync(bool separate_meta)
-    while (!bs->is_safe_to_stop())
+{
-        ringloop->loop();
+    printf("\n-- test_fsync%s\n", separate_meta ? " separate_meta" : "");
-    delete bs;
+
-    delete tfd;
+    bs_test_t test;
-    delete data_disk;
+    test.default_cfg();
-    delete ringloop;
+    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;
 }