Add a test for parallel reads with block checksums

src/blockstore/blockstore_flush.cpp

@@ -410,7 +410,7 @@ void journal_flusher_co::iterate_checksum_holes(std::function<void(int, uint32_t
             uint32_t blk_begin = (prev_begin - prev_begin%bs->dsk.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::function<void(int, uint32_t
             uint32_t blk_end = prev_end - (prev_end % bs->dsk.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

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<copy_buffer_t> & read_vec, uint32_t item_start, uint32_t item_end,
         std::function<void(int&, uint32_t, uint32_t)> callback);
+    void free_read_buffers(std::vector<copy_buffer_t> & rv);
     void handle_read_event(ring_data_t *data, blockstore_op_t *op);
     bool verify_read_checksums(blockstore_op_t *op);
 

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<copy_buffer_t> & read_vec,
     }
 }
 
+void blockstore_impl_t::free_read_buffers(std::vector<copy_buffer_t> & 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);
     }
 }

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

src/test/ringloop_mock.h

@@ -47,14 +47,13 @@ class disk_mock_t
     std::map<uint64_t, iovec> 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);

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<bool(io_uring_sqe*)> 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<ring_data_t*> 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;
 }