Do not read incomplete intent writes

src/blockstore/blockstore_heap.cpp

@@ -2190,6 +2190,15 @@ void blockstore_heap_t::apply_inflight(heap_inflight_lsn_t & inflight)
     }
 }
 
+bool blockstore_heap_t::is_lsn_completed(uint64_t lsn)
+{
+    if (lsn <= completed_lsn)
+        return true;
+    assert(lsn-first_inflight_lsn < inflight_lsn.size());
+    auto it = inflight_lsn.begin() + (lsn-first_inflight_lsn);
+    return (it->flags & HEAP_INFLIGHT_DONE);
+}
+
 uint64_t blockstore_heap_t::get_completed_lsn()
 {
     return completed_lsn;

src/blockstore/blockstore_heap.h

@@ -314,6 +314,7 @@ public:
     void start_block_write(uint32_t block_num);
     void complete_block_write(uint32_t block_num);
     void complete_lsn_write(uint64_t lsn);
+    bool is_lsn_completed(uint64_t lsn);
     uint64_t get_completed_lsn();
     uint64_t get_fsynced_lsn();
     void mark_lsn_fsynced(uint64_t lsn);

src/blockstore/blockstore_read.cpp

@@ -22,7 +22,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
     uint64_t result_version = 0;
     bool found = false;
     uint32_t skip_csum = 0;
-    uint32_t blk_start = 0, blk_end = 0;
+    uint32_t blk_start = op->offset, blk_end = op->offset+op->len;
     bool need_skip = dsk.csum_block_size > dsk.bitmap_granularity && !perfect_csum_update;
     if (need_skip)
     {
@@ -32,12 +32,28 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
         if (blk_end % dsk.csum_block_size)
             blk_end += dsk.csum_block_size - (blk_end % dsk.csum_block_size);
     }
+    bool need_wait = false;
     heap->iterate_with_stable(obj, obj->lsn, [&](heap_entry_t *wr, bool stable)
     {
         if (wr->type() == BS_HEAP_DELETE)
         {
             return false;
         }
+        if (!heap->is_lsn_completed(wr->lsn))
+        {
+            if (wr->type() == BS_HEAP_BIG_INTENT && wr->big_intent().offset < blk_end && wr->big_intent().offset+wr->big_intent().len > blk_start ||
+                wr->type() == BS_HEAP_INTENT_WRITE && wr->small().offset < blk_end && wr->small().offset+wr->small().len > blk_start)
+            {
+                // Wait until intent write is completed
+                need_wait = true;
+                return false;
+            }
+            else if (wr->type() == BS_HEAP_SMALL_WRITE && wr->small().offset < blk_end && wr->small().offset+wr->small().len > blk_start)
+            {
+                // Skip entry and read the previous one
+                return true;
+            }
+        }
         if (op->version >= wr->version && !found)
         {
             found = true;
@@ -47,12 +63,6 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
                 memcpy(op->bitmap, wr->get_ext_bitmap(heap), dsk.clean_entry_bitmap_size);
             }
         }
-        if (need_skip && wr->lsn < heap->get_completed_lsn() &&
-            (wr->type() == BS_HEAP_BIG_INTENT && wr->big_intent().offset < blk_end && wr->big_intent().offset+wr->big_intent().len > blk_start ||
-            wr->type() == BS_HEAP_INTENT_WRITE && wr->small().offset < blk_end && wr->small().offset+wr->small().len > blk_start))
-        {
-            skip_csum = COPY_BUF_SKIP_CSUM;
-        }
         if (op->version >= wr->version)
         {
             fulfilled += prepare_read(PRIV(op)->read_vec, obj, wr, op->offset, op->offset+op->len,
@@ -65,13 +75,23 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
                 return false;
             }
         }
-        if (need_skip && (wr->type() == BS_HEAP_SMALL_WRITE || wr->type() == BS_HEAP_INTENT_WRITE) &&
+        if (need_skip && wr->type() == BS_HEAP_SMALL_WRITE &&
             wr->small().offset < blk_end && wr->small().offset+wr->small().len > blk_start)
         {
+            // Small write may mutate big write checksums during flush
             skip_csum = COPY_BUF_SKIP_CSUM;
         }
         return true;
     });
+    if (need_wait)
+    {
+undo_wait:
+        // Need to wait. undo added requests, unlock lsn
+        heap->unlock_entry(op->oid);
+        free_read_buffers(rv);
+        rv.clear();
+        return 0;
+    }
     if (!found)
     {
         // May happen if there are entries but all of them are > requested version
@@ -84,11 +104,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
     assert(fulfilled == op->len);
     if (!fulfill_read(op))
     {
-        // Need to wait. undo added requests, unlock lsn
-        heap->unlock_entry(op->oid);
-        free_read_buffers(rv);
-        rv.clear();
-        return 0;
+        goto undo_wait;
     }
     op->version = result_version;
     if (!PRIV(op)->pending_ops)