Fix intent writes with padded checksums

src/blockstore/blockstore_flush.cpp

@@ -238,9 +238,14 @@ resume_1:
     flusher->active_flushers++;
     // Scan versions to flush
     read_vec.clear();
+    unaligned_intent = NULL;
     for (auto wr = begin_wr; wr != end_wr; wr = wr->next())
     {
         bs->prepare_read(read_vec, cur_obj, wr, 0, bs->dsk.data_block_size);
+        if (wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) && !wr->can_be_collapsed(bs->heap))
+        {
+            unaligned_intent = wr;
+        }
     }
     overwrite_start = overwrite_end = 0;
     if (read_vec.size() > 0)
@@ -249,7 +254,7 @@ resume_1:
         overwrite_end = read_vec[read_vec.size()-1].offset + read_vec[read_vec.size()-1].len;
     }
     read_to_fill_incomplete = false;
-    if (bs->dsk.csum_block_size > bs->dsk.bitmap_granularity && end_wr->next())
+    if (bs->dsk.csum_block_size > bs->dsk.bitmap_granularity)
     {
         // Read original checksum blocks to calculate padded checksums if required
         fill_partial_checksum_blocks();
@@ -257,6 +262,8 @@ resume_1:
         {
             flusher->wanting_meta_fsync++;
         }
+        // Unaligned intent checksums have to be recalculated by reading data blocks from the disk
+        fill_unaligned_intent_checksums();
     }
     // Read buffered data
     cur_obj = NULL;
@@ -423,6 +430,7 @@ void journal_flusher_co::fill_partial_checksum_blocks()
     {
         read_to_fill_incomplete = true;
         int out_pos = read_vec.size();
+        // FIXME: Take end_wr bitmap into account here, now it's incorrect
         bs->prepare_disk_read(read_vec, out_pos, 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,
@@ -437,6 +445,23 @@ void journal_flusher_co::fill_partial_checksum_blocks()
     });
 }
 
+void journal_flusher_co::fill_unaligned_intent_checksums()
+{
+    if (!unaligned_intent)
+    {
+        return;
+    }
+    assert(unaligned_intent->next() == end_wr);
+    uint32_t blk_start = unaligned_intent->offset, blk_end = unaligned_intent->offset + unaligned_intent->len;
+    blk_start = (blk_start / bs->dsk.csum_block_size) * bs->dsk.csum_block_size;
+    blk_end = ((blk_end-1) / bs->dsk.csum_block_size + 1) * bs->dsk.csum_block_size;
+    bs->find_holes(read_vec, blk_start, blk_end, [&](int & pos, uint32_t start, uint32_t end)
+    {
+        // FIXME: Take end_wr bitmap + unaligned_intent range into account here too
+        bs->prepare_disk_read(read_vec, pos, cur_obj, end_wr, start, end, start, end, COPY_BUF_SKIP_CSUM);
+    });
+}
+
 void journal_flusher_co::free_buffers()
 {
     for (auto it = read_vec.begin(); it != read_vec.end(); it++)

src/blockstore/blockstore_flush.h

@@ -44,7 +44,7 @@ class journal_flusher_co
     uint64_t compact_lsn;
     uint64_t cur_version;
     heap_object_t *cur_obj;
-    heap_write_t *begin_wr, *end_wr;
+    heap_write_t *begin_wr, *end_wr, *unaligned_intent;
     uint32_t modified_block;
     bool should_repeat;
 
@@ -64,6 +64,7 @@ class journal_flusher_co
     void iterate_partial_overwrites(std::function<int(int, uint32_t, uint32_t)> cb);
     void iterate_checksum_holes(std::function<void(int, uint32_t, uint32_t)> cb);
     void fill_partial_checksum_blocks();
+    void fill_unaligned_intent_checksums();
     void free_buffers();
     int check_and_punch_checksums();
     void calc_block_checksums();

src/blockstore/blockstore_heap.cpp

@@ -59,9 +59,11 @@ bool heap_write_t::needs_recheck(blockstore_heap_t *heap)
         || flags == BS_HEAP_SMALL_WRITE || flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE));
 }
 
-bool heap_write_t::needs_compact(uint64_t compacted_lsn)
+bool heap_write_t::needs_compact(blockstore_heap_t *heap)
 {
-    return lsn > compacted_lsn && flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE);
+    return (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) ||
+        flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) && heap->dsk->csum_block_size > heap->dsk->bitmap_granularity &&
+        ((offset % heap->dsk->csum_block_size) || (len % heap->dsk->csum_block_size)));
 }
 
 bool heap_write_t::is_compacted(uint64_t compacted_lsn)
@@ -460,7 +462,7 @@ skip_object:
                 }
                 if (wr->lsn > this->compacted_lsn)
                 {
-                    tmp_compact_queue.push_back((tmp_compact_item_t){ .oid = oid, .lsn = wr->lsn, .compact = wr->needs_compact(0) });
+                    tmp_compact_queue.push_back((tmp_compact_item_t){ .oid = oid, .lsn = wr->lsn, .compact = wr->needs_compact(this) });
                 }
             }
             if (lsn > next_lsn)
@@ -1140,7 +1142,7 @@ int blockstore_heap_t::add_object(object_id oid, heap_write_t *wr, uint32_t *mod
     new_wr->size = wr_size;
     new_wr->lsn = ++next_lsn;
     wr->lsn = new_wr->lsn;
-    push_inflight_lsn(oid, new_wr->lsn, new_wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
+    push_inflight_lsn(oid, new_wr->lsn, new_wr->needs_compact(this) ? HEAP_INFLIGHT_COMPACTABLE : 0);
     if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
     {
         uint8_t *int_bitmap = new_wr->get_int_bitmap(this);
@@ -1242,7 +1244,7 @@ void blockstore_heap_t::mark_overwritten(uint64_t over_lsn, uint64_t inode, heap
 {
     while (wr && wr != end_wr)
     {
-        if (wr->needs_compact(0))
+        if (wr->needs_compact(this))
         {
             mark_lsn_compacted(wr->lsn, true);
         }
@@ -1283,6 +1285,18 @@ 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)
+    {
+        // Unstable intent writes 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) &&
+        !first_wr->can_be_collapsed(this))
+    {
+        // Intent writes are not allowed over noncollapsible intent writes
+        return EINVAL;
+    }
     if (wr->version < first_wr->version)
     {
         // Overwrites with a smaller version are forbidden
@@ -1329,10 +1343,9 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea
     else if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE &&
         (first_wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE)
     {
-        assert(wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE));
         auto second_wr = first_wr->next();
         bitmap_set(second_wr->get_int_bitmap(this), first_wr->offset, first_wr->len, dsk->bitmap_granularity);
-        if (dsk->csum_block_size && wr->can_be_collapsed(this))
+        if (dsk->csum_block_size)
         {
             const uint32_t csum_size = (dsk->data_csum_type & 0xFF);
             memcpy(second_wr->get_checksums(this) + first_wr->offset/dsk->csum_block_size*csum_size,
@@ -1346,7 +1359,7 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea
         new_wr->next_pos = ((uint8_t*)obj + obj->write_pos) - (uint8_t*)new_wr;
     }
     wr->lsn = new_wr->lsn;
-    push_inflight_lsn(oid, new_wr->lsn, new_wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
+    push_inflight_lsn(oid, new_wr->lsn, new_wr->needs_compact(this) ? HEAP_INFLIGHT_COMPACTABLE : 0);
     if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
     {
         uint8_t *int_bitmap = new_wr->get_int_bitmap(this);
@@ -1453,7 +1466,7 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t
         {
             wr->flags |= BS_HEAP_STABLE;
             wr->lsn = last_lsn--;
-            push_inflight_lsn(oid, wr->lsn, wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
+            push_inflight_lsn(oid, wr->lsn, wr->needs_compact(this) ? HEAP_INFLIGHT_COMPACTABLE : 0);
         }
     }
     obj->crc32c = obj->calc_crc32c();
@@ -1696,7 +1709,7 @@ void blockstore_heap_t::erase_object(uint32_t block_num, heap_object_t *obj, uin
     {
         for (auto wr = obj->get_writes(); wr; wr = wr->next())
         {
-            if (wr->needs_compact(0))
+            if (wr->needs_compact(this))
             {
                 mark_lsn_compacted(wr->lsn, true);
             }

src/blockstore/blockstore_heap.h

@@ -49,7 +49,7 @@ struct __attribute__((__packed__)) heap_write_t
     uint32_t get_size(blockstore_heap_t *heap);
     uint32_t get_csum_size(blockstore_heap_t *heap);
     bool needs_recheck(blockstore_heap_t *heap);
-    bool needs_compact(uint64_t compacted_lsn);
+    bool needs_compact(blockstore_heap_t *heap);
     bool is_compacted(uint64_t compacted_lsn);
     bool can_be_collapsed(blockstore_heap_t *heap);
     bool is_allowed_before_compacted(uint64_t compacted_lsn, bool is_last_entry);

src/blockstore/blockstore_impl.h

@@ -162,7 +162,7 @@ class blockstore_impl_t: public blockstore_i
     void prepare_disk_read(std::vector<copy_buffer_t> & read_vec, int & pos, heap_object_t *obj, heap_write_t *wr,
         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&, bool, uint32_t, uint32_t)> callback);
+        std::function<void(int&, uint32_t, uint32_t)> callback);
     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

@@ -161,17 +161,14 @@ uint32_t blockstore_impl_t::prepare_read_with_bitmaps(std::vector<copy_buffer_t>
 uint32_t blockstore_impl_t::prepare_read_zero(std::vector<copy_buffer_t> & read_vec, uint32_t start, uint32_t end)
 {
     uint32_t res = 0;
-    find_holes(read_vec, start, end, [&](int & pos, bool alloc, uint32_t start, uint32_t end)
+    find_holes(read_vec, start, end, [&](int & pos, uint32_t start, uint32_t end)
     {
-        if (!alloc)
-        {
-            res += end-start;
-            read_vec.insert(read_vec.begin() + (pos++), (copy_buffer_t){
-                .copy_flags = COPY_BUF_ZERO,
-                .offset = start,
-                .len = end-start,
-            });
-        }
+        res += end-start;
+        read_vec.insert(read_vec.begin() + (pos++), (copy_buffer_t){
+            .copy_flags = COPY_BUF_ZERO,
+            .offset = start,
+            .len = end-start,
+        });
     });
     return res;
 }
@@ -179,12 +176,8 @@ 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;
-    find_holes(read_vec, start, end, [&](int & pos, bool alloc, uint32_t start, uint32_t end)
+    find_holes(read_vec, start, end, [&](int & pos, uint32_t start, uint32_t end)
     {
-        if (alloc)
-        {
-            return;
-        }
         res += end-start;
         if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE && dsk.inmemory_journal)
         {
@@ -280,7 +273,7 @@ void blockstore_impl_t::prepare_disk_read(std::vector<copy_buffer_t> & read_vec,
 
 void blockstore_impl_t::find_holes(std::vector<copy_buffer_t> & read_vec,
     uint32_t item_start, uint32_t item_end,
-    std::function<void(int&, bool, uint32_t, uint32_t)> callback)
+    std::function<void(int&, uint32_t, uint32_t)> callback)
 {
     auto cur_start = item_start;
     int i = 0;
@@ -290,14 +283,14 @@ void blockstore_impl_t::find_holes(std::vector<copy_buffer_t> & read_vec,
         if (i >= read_vec.size() || (read_vec[i].copy_flags & COPY_BUF_CSUM_FILL) || read_vec[i].offset >= item_end)
         {
             // Hole (at end): cur_start .. item_end
-            callback(i, false, cur_start, item_end);
+            callback(i, cur_start, item_end);
             break;
         }
         else if (read_vec[i].offset > cur_start)
         {
             // Hole: cur_start .. min(read_vec[i].offset, item_end)
             auto cur_end = read_vec[i].offset > item_end ? item_end : read_vec[i].offset;
-            callback(i, false, cur_start, cur_end);
+            callback(i, cur_start, cur_end);
             cur_start = cur_end;
         }
         else if (read_vec[i].offset + read_vec[i].len > cur_start)
@@ -305,7 +298,7 @@ void blockstore_impl_t::find_holes(std::vector<copy_buffer_t> & read_vec,
             // Allocated: cur_start .. min(read_vec[i].offset + read_vec[i].len, item_end)
             auto cur_end = read_vec[i].offset + read_vec[i].len;
             cur_end = cur_end > item_end ? item_end : cur_end;
-            callback(i, true, cur_start, cur_end);
+            //callback(i, true, cur_start, cur_end);
             cur_start = cur_end;
             i++;
         }

src/blockstore/blockstore_write.cpp

@@ -161,6 +161,13 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     // 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 &&
+        // 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 ||
+            !(op->offset % dsk.csum_block_size) &&
+            !(op->len % dsk.csum_block_size) &&
+            (obj->get_writes()->flags != (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) ||
+            obj->get_writes()->can_be_collapsed(heap))) &&
         // One intent-write is allowed even with fsyncs because BIG_WRITE is always counted as fsynced
         (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))