Complete small/intent writes explicitly, not implicitly on metadata block write

src/blockstore/blockstore_heap.cpp

@@ -23,6 +23,7 @@
 #define HEAP_INFLIGHT_COMPACTABLE 2
 #define HEAP_INFLIGHT_COMPACTED 4
 #define HEAP_INFLIGHT_GC 8
+#define HEAP_INFLIGHT_EXPLICIT 16
 
 static inline heap_list_item_t *list_item(heap_entry_t *wr)
 {
@@ -697,7 +698,7 @@ bool blockstore_heap_t::recheck_small_writes(std::function<void(bool is_data, ui
         else if (wr->type() == BS_HEAP_BIG_INTENT)
         {
             auto & bi = wr->big_intent();
-            loc = bi.block_num * dsk->data_block_size + bi.offset;
+            loc = (uint64_t)bi.block_num * dsk->data_block_size + bi.offset;
             len = bi.len;
             from_data = true;
         }
@@ -1147,7 +1148,7 @@ void blockstore_heap_t::insert_list_item(heap_list_item_t *li)
 }
 
 int blockstore_heap_t::add_entry(uint32_t wr_size, uint32_t *modified_block,
-    bool allow_last_free, std::function<void(heap_entry_t *wr)> fill_entry)
+    bool allow_last_free, bool explicit_complete, std::function<void(heap_entry_t *wr)> fill_entry)
 {
     uint32_t block_num;
     int res = allocate_entry(wr_size, &block_num, allow_last_free);
@@ -1172,6 +1173,7 @@ int blockstore_heap_t::add_entry(uint32_t wr_size, uint32_t *modified_block,
     inf.mod_lsn_to = next_lsn;
     // Remember the object as dirty and remove older entries when this block is written and fsynced
     push_inflight_lsn(next_lsn, new_wr,
+        (explicit_complete ? HEAP_INFLIGHT_EXPLICIT : 0) |
         (new_wr->is_overwrite() ? HEAP_INFLIGHT_COMPACTED : 0) |
         (new_wr->is_compactable() ? HEAP_INFLIGHT_COMPACTABLE : 0));
     insert_list_item(li);
@@ -1183,17 +1185,19 @@ int blockstore_heap_t::add_entry(uint32_t wr_size, uint32_t *modified_block,
 
 // 1st step: post a write
 
-int blockstore_heap_t::add_small_write(object_id oid, heap_entry_t *old_head, uint16_t type, uint64_t version,
+int blockstore_heap_t::add_small_write(object_id oid, heap_entry_t **obj_ptr, uint16_t type, uint64_t version,
     uint32_t offset, uint32_t len, uint64_t location, uint8_t *bitmap, uint8_t *data, uint32_t *modified_block)
 {
-    if (!old_head || old_head->type() == BS_HEAP_DELETE || old_head->version > version ||
+    auto obj = *obj_ptr;
+    if (!obj || obj->type() == BS_HEAP_DELETE || obj->version > version ||
         type != (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) && type != BS_HEAP_SMALL_WRITE && type != (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) ||
-        (type & BS_HEAP_STABLE) && !(old_head->entry_type & BS_HEAP_STABLE))
+        (type & BS_HEAP_STABLE) && !(obj->entry_type & BS_HEAP_STABLE))
     {
         return EINVAL;
     }
     uint32_t wr_size = get_small_entry_size(offset, len);
-    return add_entry(wr_size, modified_block, false, [&](heap_entry_t *wr)
+    // Small writes are written in parallel with buffered data so they require explicit_complete
+    return add_entry(wr_size, modified_block, false, true, [&](heap_entry_t *wr)
     {
         wr->entry_type = type;
         wr->inode = oid.inode;
@@ -1204,11 +1208,12 @@ int blockstore_heap_t::add_small_write(object_id oid, heap_entry_t *old_head, ui
         wr->small().location = location;
         if (bitmap)
             memcpy(wr->get_ext_bitmap(this), bitmap, dsk->clean_entry_bitmap_size);
-        else if (old_head)
+        else if (obj)
-            memcpy(wr->get_ext_bitmap(this), old_head->get_ext_bitmap(this), dsk->clean_entry_bitmap_size);
+            memcpy(wr->get_ext_bitmap(this), obj->get_ext_bitmap(this), dsk->clean_entry_bitmap_size);
         else
             memset(wr->get_ext_bitmap(this), 0, dsk->clean_entry_bitmap_size);
         calc_checksums(wr, (uint8_t*)data, true);
+        *obj_ptr = wr;
     });
 }
 
@@ -1220,7 +1225,8 @@ int blockstore_heap_t::add_big_write(object_id oid, heap_entry_t *old_head, bool
         return EINVAL;
     }
     uint32_t wr_size = get_big_entry_size();
-    return add_entry(wr_size, modified_block, false, [&](heap_entry_t *wr)
+    // Big writes are written after writing data so they don't require explicit_complete
+    return add_entry(wr_size, modified_block, false, false, [&](heap_entry_t *wr)
     {
         wr->entry_type = BS_HEAP_BIG_WRITE | (stable ? BS_HEAP_STABLE : 0);
         wr->inode = oid.inode;
@@ -1241,18 +1247,20 @@ int blockstore_heap_t::add_big_write(object_id oid, heap_entry_t *old_head, bool
     });
 }
 
-int blockstore_heap_t::add_big_intent(object_id oid, heap_entry_t *old_head, uint64_t version,
+int blockstore_heap_t::add_big_intent(object_id oid, heap_entry_t **obj_ptr, uint64_t version,
     uint32_t offset, uint32_t len, uint8_t *bitmap, uint8_t *data, uint8_t *checksums, uint32_t *modified_block)
 {
-    if (!old_head ||
+    auto obj = *obj_ptr;
-        old_head->entry_type != (BS_HEAP_BIG_INTENT|BS_HEAP_STABLE) &&
+    if (!obj ||
-        old_head->entry_type != (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) ||
+        obj->entry_type != (BS_HEAP_BIG_INTENT|BS_HEAP_STABLE) &&
+        obj->entry_type != (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) ||
         dsk->csum_block_size > dsk->bitmap_granularity && !checksums)
     {
         return EINVAL;
     }
     uint32_t wr_size = get_big_intent_entry_size();
-    return add_entry(wr_size, modified_block, false, [&](heap_entry_t *wr)
+    // Big intents are written before writing data so they require explicit_complete
+    return add_entry(wr_size, modified_block, false, true, [&](heap_entry_t *wr)
     {
         wr->entry_type = BS_HEAP_BIG_INTENT | BS_HEAP_STABLE;
         wr->inode = oid.inode;
@@ -1261,14 +1269,14 @@ int blockstore_heap_t::add_big_intent(object_id oid, heap_entry_t *old_head, uin
         auto & bi = wr->big_intent();
         bi.offset = offset;
         bi.len = len;
-        bi.block_num = (old_head->type() == BS_HEAP_BIG_INTENT
+        bi.block_num = (obj->type() == BS_HEAP_BIG_INTENT
-            ? old_head->big_intent().block_num
+            ? obj->big_intent().block_num
-            : old_head->big().block_num);
+            : obj->big().block_num);
         if (bitmap)
             memcpy(wr->get_ext_bitmap(this), bitmap, dsk->clean_entry_bitmap_size);
         else
-            memcpy(wr->get_ext_bitmap(this), old_head->get_ext_bitmap(this), dsk->clean_entry_bitmap_size);
+            memcpy(wr->get_ext_bitmap(this), obj->get_ext_bitmap(this), dsk->clean_entry_bitmap_size);
-        memcpy(wr->get_int_bitmap(this), old_head->get_int_bitmap(this), dsk->clean_entry_bitmap_size);
+        memcpy(wr->get_int_bitmap(this), obj->get_int_bitmap(this), dsk->clean_entry_bitmap_size);
         bitmap_set(wr->get_int_bitmap(this), offset, len, dsk->bitmap_granularity);
         if (dsk->data_csum_type)
         {
@@ -1276,12 +1284,13 @@ int blockstore_heap_t::add_big_intent(object_id oid, heap_entry_t *old_head, uin
                 memcpy(wr->get_checksums(this), checksums, dsk->clean_entry_bitmap_size);
             else
             {
-                memcpy(wr->get_checksums(this), old_head->get_checksums(this), dsk->clean_entry_bitmap_size);
+                memcpy(wr->get_checksums(this), obj->get_checksums(this), dsk->clean_entry_bitmap_size);
                 calc_checksums(wr, (uint8_t*)data, true, offset, len);
             }
         }
         else
             calc_checksums(wr, (uint8_t*)data, true);
+        *obj_ptr = wr;
     });
 }
 
@@ -1290,7 +1299,7 @@ int blockstore_heap_t::add_compact(heap_entry_t *obj, uint64_t compact_version,
 {
     if (do_delete)
     {
-        return add_entry(get_simple_entry_size(), modified_block, false, [&](heap_entry_t *wr)
+        return add_entry(get_simple_entry_size(), modified_block, false, false, [&](heap_entry_t *wr)
         {
             wr->entry_type = BS_HEAP_DELETE|BS_HEAP_STABLE;
             wr->inode = obj->inode;
@@ -1300,7 +1309,8 @@ int blockstore_heap_t::add_compact(heap_entry_t *obj, uint64_t compact_version,
         });
     }
     uint32_t wr_size = get_big_entry_size();
-    return add_entry(wr_size, modified_block, true, [&](heap_entry_t *new_wr)
+    // Compaction entry is added after copying data so it doesn't require explicit_complete
+    return add_entry(wr_size, modified_block, true, false, [&](heap_entry_t *new_wr)
     {
         new_wr->entry_type = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE;
         new_wr->inode = obj->inode;
@@ -1335,7 +1345,8 @@ int blockstore_heap_t::punch_holes(heap_entry_t *wr, uint8_t *new_bitmap, uint8_
 int blockstore_heap_t::add_simple(heap_entry_t *obj, uint64_t version, uint32_t *modified_block, uint32_t entry_type)
 {
     uint32_t wr_size = get_simple_entry_size();
-    return add_entry(wr_size, modified_block, false, [&](heap_entry_t *wr)
+    // Simple entries don't have data so they don't require explicit_complete
+    return add_entry(wr_size, modified_block, false, false, [&](heap_entry_t *wr)
     {
         wr->entry_type = entry_type;
         wr->inode = obj->inode;
@@ -1518,12 +1529,22 @@ void blockstore_heap_t::complete_block_write(uint32_t block_num)
         for (uint64_t lsn = mod_lsn; lsn <= mod_lsn_to; lsn++, it++)
         {
             assert(!(it->flags & HEAP_INFLIGHT_DONE));
-            it->flags |= HEAP_INFLIGHT_DONE;
+            if (!(it->flags & HEAP_INFLIGHT_EXPLICIT))
+                it->flags |= HEAP_INFLIGHT_DONE;
         }
         mark_completed_lsns(mod_lsn);
     }
 }
 
+void blockstore_heap_t::complete_lsn_write(uint64_t lsn)
+{
+    auto it = inflight_lsn.begin() + (lsn-first_inflight_lsn);
+    assert(!(it->flags & HEAP_INFLIGHT_DONE));
+    assert(it->flags & HEAP_INFLIGHT_EXPLICIT);
+    it->flags |= HEAP_INFLIGHT_DONE;
+    mark_completed_lsns(lsn);
+}
+
 void blockstore_heap_t::mark_garbage_up_to(heap_entry_t *wr)
 {
     auto mvcc_it = object_mvcc.find((object_id){ .inode = wr->inode, .stripe = wr->stripe });

src/blockstore/blockstore_heap.h

@@ -213,7 +213,7 @@ class blockstore_heap_t
     int allocate_entry(uint32_t entry_size, uint32_t *block_num, bool allow_last_free);
     void insert_list_item(heap_list_item_t *li);
     int add_entry(uint32_t wr_size, uint32_t *modified_block, bool allow_last_free,
-        std::function<void(heap_entry_t *wr)> fill_entry);
+        bool explicit_complete, std::function<void(heap_entry_t *wr)> fill_entry);
     int add_simple(heap_entry_t *obj, uint64_t version, uint32_t *modified_block, uint32_t entry_type);
     uint32_t meta_alloc_pos(const heap_block_info_t & inf);
     void modify_alloc(uint32_t block_num, std::function<void(heap_block_info_t &)> change_cb);
@@ -260,13 +260,13 @@ public:
         bool set, std::function<void(uint32_t, uint32_t, uint32_t)> bad_block_cb);
     // adds a small_write or intent_write entry to an object
     // return 0 if OK, or maybe ENOSPC
-    int add_small_write(object_id oid, heap_entry_t *old_head, uint16_t type, uint64_t version,
+    int add_small_write(object_id oid, heap_entry_t **obj_ptr, uint16_t type, uint64_t version,
         uint32_t offset, uint32_t len, uint64_t location, uint8_t *bitmap, uint8_t *data, uint32_t *modified_block);
     // adds a big_write (overwrite) entry to an object
     int add_big_write(object_id oid, heap_entry_t *old_head, bool stable, uint64_t version,
         uint32_t offset, uint32_t len, uint64_t location, uint8_t *bitmap, uint8_t *data, uint32_t *modified_block);
     // adds a big_intent (atomic partial modification) entry to an object
-    int add_big_intent(object_id oid, heap_entry_t *old_head, uint64_t version,
+    int add_big_intent(object_id oid, heap_entry_t **obj_ptr, uint64_t version,
         uint32_t offset, uint32_t len, uint8_t *bitmap, uint8_t *data, uint8_t *checksums, uint32_t *modified_block);
     // adds a compacted up to <version> entry to an object
     int add_compact(heap_entry_t *obj, uint64_t compact_version, uint64_t compact_lsn, uint64_t compact_location,
@@ -299,6 +299,7 @@ public:
     // inflight write tracking
     void start_block_write(uint32_t block_num);
     void complete_block_write(uint32_t block_num);
+    void complete_lsn_write(uint64_t lsn);
     uint64_t get_completed_lsn();
     uint64_t get_fsynced_lsn();
     void mark_lsn_fsynced(uint64_t lsn);

src/blockstore/blockstore_impl.h

@@ -47,8 +47,11 @@ struct blockstore_op_private_t
     // Read
     std::vector<copy_buffer_t> read_vec;
 
+    // Read, write
+    uint64_t lsn;
+
     // Write
-    uint64_t location; // or lsn for read
+    uint64_t location;
     uint32_t write_type;
 
     // Stabilize, rollback

src/blockstore/blockstore_read.cpp

@@ -17,7 +17,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
     }
     uint32_t fulfilled = 0;
     PRIV(op)->pending_ops = 0;
-    PRIV(op)->location = 0;
+    PRIV(op)->lsn = 0;
     auto & rv = PRIV(op)->read_vec;
     uint64_t result_version = 0;
     bool found = false;
@@ -26,7 +26,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
     bool need_skip = dsk.csum_block_size > dsk.bitmap_granularity && !perfect_csum_update;
     if (need_skip)
     {
-        PRIV(op)->location = obj->lsn;
+        PRIV(op)->lsn = obj->lsn;
         blk_start = op->offset - op->offset%dsk.csum_block_size;
         blk_end = op->offset + op->len;
         if (blk_end % dsk.csum_block_size)
@@ -395,10 +395,10 @@ void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op
 bool blockstore_impl_t::verify_read_checksums(blockstore_op_t *op)
 {
     bool skip_all = false;
-    if (PRIV(op)->location)
+    if (PRIV(op)->lsn)
     {
         heap_entry_t *obj = heap->read_entry(op->oid);
-        if (obj->lsn != PRIV(op)->location) // check top lsn
+        if (obj->lsn != PRIV(op)->lsn) // check top lsn
         {
             skip_all = true;
         }

src/blockstore/blockstore_write.cpp

@@ -211,8 +211,12 @@ enospc:
             // FIXME: Support RMW mode for csum_block_size > bitmap_granularity
             PRIV(op)->write_type = BS_HEAP_BIG_INTENT;
             PRIV(op)->location = obj->big_location(heap);
-            res = heap->add_big_intent(op->oid, obj, op->version, op->offset, op->len, op->bitmap,
+            res = heap->add_big_intent(op->oid, &obj, op->version, op->offset, op->len, op->bitmap,
                 (uint8_t*)op->buf, NULL, &PRIV(op)->modified_block);
+            if (res == ENOSPC)
+                goto enospc;
+            assert(res == 0);
+            PRIV(op)->lsn = obj->lsn;
         }
         else
         {
@@ -224,12 +228,13 @@ enospc:
             }
             assert(wr && (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_BIG_INTENT));
             PRIV(op)->location = wr->big_location(heap);
-            res = heap->add_small_write(op->oid, obj, (BS_HEAP_INTENT_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)),
+            res = heap->add_small_write(op->oid, &obj, (BS_HEAP_INTENT_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)),
                 op->version, op->offset, op->len, 0, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
+            if (res == ENOSPC)
+                goto enospc;
+            assert(res == 0);
+            PRIV(op)->lsn = obj->lsn;
         }
-        if (res == ENOSPC)
-            goto enospc;
-        assert(res == 0);
         prepare_meta_block_write(PRIV(op)->modified_block);
         intent_write_counter++;
         PRIV(op)->pending_ops++;
@@ -251,11 +256,12 @@ enospc:
         }
         // There is sufficient space. Check SQE(s)
         BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0));
-        int res = heap->add_small_write(op->oid, obj, (BS_HEAP_SMALL_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)),
+        int res = heap->add_small_write(op->oid, &obj, (BS_HEAP_SMALL_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)),
             op->version, op->offset, op->len, loc, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
         if (res == ENOSPC)
             goto enospc;
         assert(res == 0);
+        PRIV(op)->lsn = obj->lsn;
         if (op->len)
             heap->use_buffer_area(op->oid.inode, loc, op->len);
         prepare_meta_block_write(PRIV(op)->modified_block);
@@ -418,6 +424,10 @@ resume_8:
     printf("Ack write %jx:%jx v%ju\n", op->oid.inode, op->oid.stripe, op->version);
 #endif
     op->retval = op->len;
+    if (PRIV(op)->write_type == BS_HEAP_BIG_INTENT ||
+        PRIV(op)->write_type == BS_HEAP_INTENT_WRITE ||
+        PRIV(op)->write_type == BS_HEAP_SMALL_WRITE)
+        heap->complete_lsn_write(PRIV(op)->lsn);
     if (PRIV(op)->write_type == BS_HEAP_BIG_WRITE ||
         PRIV(op)->write_type == BS_HEAP_BIG_INTENT ||
         PRIV(op)->write_type == BS_HEAP_INTENT_WRITE)

src/test/test_heap.cpp

@@ -73,10 +73,13 @@ void _test_big_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64_t
 }
 
 int _test_do_small_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64_t inode, uint64_t stripe, uint64_t version,
-    uint32_t offset, uint32_t len, uint64_t location, bool stable, uint8_t *data, bool is_intent = false, uint32_t *mblock = NULL)
+    uint32_t offset, uint32_t len, uint64_t location, bool stable, uint8_t *data, bool is_intent = false, uint32_t *mblock = NULL, heap_entry_t **obj = NULL)
 {
     object_id oid = { .inode = INODE_WITH_POOL(1, inode), .stripe = stripe };
-    heap_entry_t *obj = heap.read_entry(oid);
+    heap_entry_t *local_obj;
+    if (!obj)
+        obj = &local_obj;
+    *obj = heap.read_entry(oid);
     uint16_t type = (is_intent ? BS_HEAP_INTENT_WRITE : BS_HEAP_SMALL_WRITE) | (stable ? BS_HEAP_STABLE : 0);
     uint8_t ext_bitmap[dsk.clean_entry_bitmap_size];
     memset(ext_bitmap, 0xff, dsk.clean_entry_bitmap_size);
@@ -90,13 +93,15 @@ void _test_small_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64
     if (!is_intent)
         heap.use_buffer_area(INODE_WITH_POOL(1, inode), location, len); // blocks are allocated before write and outside the heap_t
     uint32_t mblock = 999999;
-    int res = _test_do_small_write(heap, dsk, inode, stripe, version, offset, len, location, stable, data, is_intent, &mblock);
+    heap_entry_t *obj = NULL;
+    int res = _test_do_small_write(heap, dsk, inode, stripe, version, offset, len, location, stable, data, is_intent, &mblock, &obj);
     assert(res == 0);
     if (!is_intent)
         assert(!heap.is_buffer_area_free(location, len));
     assert(mblock == expected_mblock || expected_mblock == UINT32_MAX);
     heap.start_block_write(mblock);
     heap.complete_block_write(mblock);
+    heap.complete_lsn_write(obj->lsn);
 }
 
 void _test_init(blockstore_disk_t & dsk, bool csum, std::function<void(std::map<std::string, std::string> &)> cfg_cb = NULL)
@@ -1234,6 +1239,8 @@ void test_intent_write(bool csum)
     printf("OK test_intent_write %s\n", csum ? "csum" : "no_csum");
 }
 
+// FIXME: Add a test for big_intent, incl. explicit_complete with big_intent over big_write over deletion over big_write :)
+
 int main(int narg, char *args[])
 {
     test_mvcc(false);