Prevent compaction of incomplete object writes

src/blockstore/blockstore_flush.cpp

@@ -177,7 +177,7 @@ resume_0:
     }
 resume_1:
     should_repeat = false;
-    cur_obj = bs->heap->lock_and_read_entry(cur_oid, cur_lsn);
+    cur_obj = bs->heap->lock_and_read_entry(cur_oid, copy_id);
     if (!cur_obj)
     {
         // Object does not exist
@@ -185,14 +185,18 @@ resume_1:
     }
     cur_version = cur_obj->get_writes()->version;
     // Find the range to compact
-    bs->heap->get_compact_range(cur_obj, cur_lsn, &begin_wr, &end_wr);
+    compact_lsn = bs->heap->get_completed_lsn();
+    if (auto cm_it = bs->committing_lsn.find(cur_oid); cm_it != bs->committing_lsn.end())
+    {
+        compact_lsn = cm_it->second;
+    }
+    bs->heap->get_compact_range(cur_obj, compact_lsn, &begin_wr, &end_wr);
     if (!begin_wr)
     {
         // Nothing to flush
-        bs->heap->unlock_entry(cur_oid, cur_lsn);
+        bs->heap->unlock_entry(cur_oid, copy_id);
         goto resume_0;
     }
-    compact_lsn = begin_wr->lsn;
     assert(!end_wr->next() && end_wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE));
     clean_loc = end_wr->location;
 #ifdef BLOCKSTORE_DEBUG
@@ -277,10 +281,10 @@ resume_12:
 resume_13:
     if (copy_count && !fsync_batch(false, 11))
         return false;
-    bs->heap->unlock_entry(cur_oid, cur_lsn);
+    bs->heap->unlock_entry(cur_oid, copy_id);
     // Modify the metadata entry; don't write anything. Metadata block will be written on the next write
     calc_block_checksums();
-    bs->heap->compact_object(cur_oid, cur_lsn, new_data_csums);
+    bs->heap->compact_object(cur_oid, compact_lsn, new_data_csums);
     // Done, free all buffers
     free_buffers();
 #ifdef BLOCKSTORE_DEBUG
@@ -398,7 +402,7 @@ int journal_flusher_co::check_and_punch_checksums()
         return 0;
     }
     // Verify data checksums
-    cur_obj = bs->heap->read_locked_entry(cur_oid, cur_lsn);
+    cur_obj = bs->heap->read_locked_entry(cur_oid, copy_id);
     bool csum_ok = true;
     for (int i = 0; i < read_vec.size(); i++)
     {
@@ -436,7 +440,7 @@ int journal_flusher_co::check_and_punch_checksums()
         // Object is deleted, abort compaction
         return ENOENT;
     }
-    bs->heap->get_compact_range(cur_obj, cur_lsn, &begin_wr, &end_wr);
+    bs->heap->get_compact_range(cur_obj, compact_lsn, &begin_wr, &end_wr);
     if (!begin_wr || begin_wr->lsn != compact_lsn)
     {
         // Object is overwritten, abort compaction
@@ -476,7 +480,7 @@ void journal_flusher_co::calc_block_checksums()
     if (bs->dsk.csum_block_size <= bs->dsk.bitmap_granularity)
         return;
     new_data_csums = csum_buf + overwrite_start/bs->dsk.csum_block_size * (bs->dsk.data_csum_type & 0xFF);
-    cur_obj = bs->heap->read_locked_entry(cur_oid, cur_lsn);
+    cur_obj = bs->heap->read_locked_entry(cur_oid, copy_id);
     uint64_t block_offset = 0;
     uint32_t block_done = 0;
     uint32_t block_csum = 0;

src/blockstore/blockstore_flush.h

@@ -49,7 +49,7 @@ class journal_flusher_co
     std::function<void(ring_data_t*)> simple_callback_r, simple_callback_w;
 
     object_id cur_oid;
-    uint64_t cur_lsn;
+    uint64_t copy_id;
     uint64_t compact_lsn;
     uint64_t min_compact_lsn;
     uint64_t cur_version;

src/blockstore/blockstore_heap.cpp

@@ -12,6 +12,9 @@
 #define BS_HEAP_FREE_MAIN 2
 #define FREE_SPACE_BIT 0x8000
 
+#define HEAP_INFLIGHT_DONE 1
+#define HEAP_INFLIGHT_COMPACTABLE 2
+
 heap_write_t *heap_write_t::next()
 {
     return (next_pos ? (heap_write_t*)((uint8_t*)this + next_pos) : NULL);
@@ -584,7 +587,7 @@ skip_object:
                 }
                 if (wr->needs_compact(this->compacted_lsn))
                 {
-                    compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn });
+                    tmp_compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn });
                 }
                 else if (wr->is_compacted(this->compacted_lsn))
                 {
@@ -596,7 +599,7 @@ skip_object:
                     {
                         // We can't just collapse the object entry when csum_block_size is larger
                         // than bitmap_granularity, so we add the object into the compact queue
-                        compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn });
+                        tmp_compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn });
                     }
                 }
             }
@@ -640,10 +643,15 @@ skip_object:
 
 void blockstore_heap_t::finish_load()
 {
-    std::sort(compact_queue.begin(), compact_queue.end(), [this](const heap_object_lsn_t & a, const heap_object_lsn_t & b)
+    std::sort(tmp_compact_queue.begin(), tmp_compact_queue.end(), [this](const heap_object_lsn_t & a, const heap_object_lsn_t & b)
     {
         return a.lsn < b.lsn;
     });
+    for (auto & e: tmp_compact_queue)
+    {
+        compact_queue.push_back(e.oid);
+    }
+    tmp_compact_queue.clear();
 }
 
 bool blockstore_heap_t::calc_checksums(heap_write_t *wr, uint8_t *data, bool set)
@@ -1241,16 +1249,14 @@ int blockstore_heap_t::add_object(object_id oid, heap_write_t *wr, uint32_t *mod
     new_wr->next_pos = 0;
     new_wr->size = wr_size;
     new_wr->lsn = ++next_lsn;
+    wr->lsn = new_wr->lsn;
+    push_inflight_lsn(new_wr->lsn, oid, wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
     if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
     {
         uint8_t *int_bitmap = new_wr->get_int_bitmap(this);
         memset(int_bitmap, 0, dsk->clean_entry_bitmap_size);
         bitmap_set(int_bitmap, wr->offset, wr->len, dsk->bitmap_granularity);
     }
-    if (wr->needs_compact(0))
-    {
-        compact_queue.push_back({ .oid = oid, .lsn = new_wr->lsn });
-    }
     new_entry->size = sizeof(heap_object_t);
     new_entry->crc32c = new_entry->calc_crc32c();
     add_used_space(block_num, sizeof(heap_object_t) + wr_size);
@@ -1402,6 +1408,8 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea
     }
     new_wr->size = wr_size;
     new_wr->lsn = ++next_lsn;
+    wr->lsn = new_wr->lsn;
+    push_inflight_lsn(new_wr->lsn, oid, wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
     if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
     {
         uint8_t *int_bitmap = new_wr->get_int_bitmap(this);
@@ -1410,11 +1418,6 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea
     }
     obj->write_pos = offset - ((uint8_t*)obj - inf.data);
     obj->crc32c = obj->calc_crc32c();
-    // Add to compaction queue
-    if (new_wr->needs_compact(0))
-    {
-        compact_queue.push_back({ .oid = oid, .lsn = new_wr->lsn });
-    }
     // Change block free space
     add_used_space(block_num, used_delta);
     return 0;
@@ -1431,7 +1434,7 @@ int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *mod
     return update_object(block_num, obj, wr, modified_block);
 }
 
-int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block)
+int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *before_compact_lsn, uint64_t *to_compact_lsn)
 {
     uint32_t block_num = 0;
     heap_object_t *obj = read_entry(oid, &block_num);
@@ -1445,6 +1448,7 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t
     heap_write_t *unstable_wr = NULL;
     heap_write_t *unstable_big_wr = NULL;
     heap_write_t *wr = obj->get_writes();
+    heap_write_t *pre_wr = NULL;
     if (wr->version < version)
     {
         // No such version
@@ -1452,7 +1456,12 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t
     }
     for (; wr; wr = wr->next())
     {
-        if (!(wr->flags & BS_HEAP_STABLE) && wr->version <= version)
+        if ((wr->flags & BS_HEAP_STABLE))
+        {
+            pre_wr = wr;
+            break;
+        }
+        else if (wr->version <= version)
         {
             unstable_wr = wr;
             if (!unstable_big_wr &&
@@ -1472,6 +1481,10 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t
     {
         *modified_block = block_num;
     }
+    if (before_compact_lsn && pre_wr && pre_wr->needs_compact(0))
+    {
+        *before_compact_lsn = pre_wr->lsn;
+    }
     // Save a copy of the object
     mvcc_save_copy(obj);
     if (unstable_big_wr && unstable_big_wr->next())
@@ -1494,9 +1507,9 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t
             to_compact = wr->lsn;
         }
     }
-    if (to_compact)
+    if (to_compact_lsn)
     {
-        compact_queue.push_back({ .oid = oid, .lsn = to_compact });
+        *to_compact_lsn = to_compact;
     }
     obj->crc32c = obj->calc_crc32c();
     return 0;
@@ -1577,7 +1590,7 @@ int blockstore_heap_t::get_next_compact(object_id & oid)
 {
     while (compact_queue.size())
     {
-        oid = compact_queue.front().oid;
+        oid = compact_queue.front();
         compact_queue.pop_front();
         return 0;
     }
@@ -1897,3 +1910,46 @@ void blockstore_heap_t::set_fail_on_warn(bool fail)
 {
     fail_on_warn = fail;
 }
+
+void blockstore_heap_t::push_inflight_lsn(uint64_t lsn, object_id oid, uint64_t flags)
+{
+    if (!inflight_lsn.size())
+    {
+        first_inflight_lsn = lsn;
+    }
+    else
+    {
+        assert(lsn == first_inflight_lsn+inflight_lsn.size());
+    }
+    inflight_lsn.push_back((heap_inflight_lsn_t){ .oid = oid, .flags = flags });
+}
+
+void blockstore_heap_t::complete_lsn(uint64_t lsn)
+{
+    assert(lsn >= first_inflight_lsn && lsn < first_inflight_lsn+inflight_lsn.size());
+    assert(!(inflight_lsn[lsn - first_inflight_lsn].flags & HEAP_INFLIGHT_DONE));
+    inflight_lsn[lsn - first_inflight_lsn].flags |= HEAP_INFLIGHT_DONE;
+    if (lsn == first_inflight_lsn)
+    {
+        while (inflight_lsn.size() && (inflight_lsn[0].flags & HEAP_INFLIGHT_DONE))
+        {
+            if ((inflight_lsn[0].flags & HEAP_INFLIGHT_COMPACTABLE))
+            {
+                compact_queue.push_back(inflight_lsn[0].oid);
+            }
+            inflight_lsn.pop_front();
+            first_inflight_lsn++;
+        }
+        completed_lsn = first_inflight_lsn-1;
+    }
+}
+
+uint64_t blockstore_heap_t::get_completed_lsn()
+{
+    return completed_lsn;
+}
+
+void blockstore_heap_t::add_to_compact_queue(object_id oid)
+{
+    compact_queue.push_back(oid);
+}

src/blockstore/blockstore_heap.h

@@ -97,6 +97,12 @@ struct __attribute__((__packed__)) heap_block_info_t
     uint8_t *data = NULL;
 };
 
+struct heap_inflight_lsn_t
+{
+    object_id oid;
+    uint64_t flags;
+};
+
 struct multilist_alloc_t
 {
     const uint32_t count, maxn;
@@ -148,6 +154,11 @@ class blockstore_heap_t
     uint64_t buffer_area_used_space = 0;
     uint64_t data_used_space = 0;
 
+    std::deque<heap_inflight_lsn_t> inflight_lsn;
+    uint64_t first_inflight_lsn = 0;
+    uint64_t completed_lsn = 0;
+
+    std::vector<heap_object_lsn_t> tmp_compact_queue;
     std::deque<object_id> recheck_queue;
     int recheck_in_progress = 0;
     bool in_recheck = false;
@@ -209,7 +220,7 @@ public:
     int post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block);
     // stabilize an unstable object version
     // return 0 if OK, ENOENT if not exists
-    int post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block);
+    int post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *before_compact_lsn, uint64_t *to_compact_lsn);
     // rollback an unstable object version
     // return 0 if OK, ENOENT if not exists, EBUSY if already stable
     int post_rollback(object_id oid, uint64_t version, uint32_t *modified_block);
@@ -230,6 +241,12 @@ public:
     // set a block number for a new object and returns error status: 0, EAGAIN or ENOSPC
     int get_block_for_new_object(uint32_t & out_block_num);
 
+    // inflight write tracking
+    void push_inflight_lsn(uint64_t lsn, object_id oid, uint64_t flags);
+    void complete_lsn(uint64_t lsn);
+    uint64_t get_completed_lsn();
+    void add_to_compact_queue(object_id oid);
+
     // data device block allocator functions
     uint64_t find_free_data();
     bool is_data_used(uint64_t location);

src/blockstore/blockstore_impl.h

@@ -53,8 +53,10 @@ struct blockstore_op_private_t
     int pending_ops;
     int op_state;
 
-    // Read
+    // Read, write
     uint64_t lsn;
+
+    // Read
     std::vector<copy_buffer_t> read_vec;
 
     // Write
@@ -63,6 +65,7 @@ struct blockstore_op_private_t
 
     // Stabilize/rollback
     int stab_pos;
+    std::vector<uint64_t> to_compact;
 
     // Write
     struct iovec iov_zerofill[3];
@@ -114,6 +117,7 @@ class blockstore_impl_t: public blockstore_i
 
     journal_flusher_t *flusher;
     int write_iodepth = 0;
+    std::unordered_map<object_id, uint64_t> committing_lsn;
 
     bool live = false, queue_stall = false;
     ring_loop_t *ringloop;

src/blockstore/blockstore_stable.cpp

@@ -16,6 +16,7 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
     assert(!priv->op_state);
     priv->stab_pos = 0;
     op->retval = 0;
+    priv->to_compact.resize(op->len);
     while (priv->stab_pos < op->len)
     {
         io_uring_sqe *sqe = get_sqe();
@@ -27,9 +28,11 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
             priv->wait_for = WAIT_SQE;
             return 0;
         }
-        uint32_t modified_block;
+        uint32_t modified_block = 0;
+        uint64_t before_compact = 0;
+        uint64_t to_compact = 0;
         int res = op->opcode == BS_OP_STABLE
-            ? heap->post_stabilize(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block)
+            ? heap->post_stabilize(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block, &before_compact, &to_compact)
             : heap->post_rollback(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block);
         if (res != 0)
         {
@@ -37,8 +40,16 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
             FINISH_OP(op);
             return 2;
         }
-        prepare_meta_block_write(op, modified_block);
+        if (modified_block)
-        priv->pending_ops++;
+        {
+            if (to_compact)
+            {
+                priv->to_compact[priv->stab_pos] = true;
+                committing_lsn[v[priv->stab_pos].oid] = before_compact;
+            }
+            prepare_meta_block_write(op, modified_block);
+            priv->pending_ops++;
+        }
         priv->stab_pos++;
     }
 resume_1:
@@ -63,6 +74,18 @@ resume_3:
         return 0;
     }
 resume_4:
+    for (int i = 0; i < op->len; i++)
+    {
+        if (priv->to_compact[i])
+        {
+            // Add to compact queue only when metadata writes are finished
+            heap->add_to_compact_queue(v[i].oid);
+            auto cm_it = committing_lsn.find(v[i].oid);
+            assert(cm_it != committing_lsn.end());
+            if (cm_it->second <= priv->to_compact[i])
+                committing_lsn.erase(cm_it);
+        }
+    }
     // Done. Don't touch op->retval - if anything resulted in ENOENT, return it as is
     FINISH_OP(op);
     return 2;

src/blockstore/blockstore_write.cpp

@@ -25,13 +25,10 @@ void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, int retval)
             found = true;
         }
         else if (found && other_op->oid == op->oid &&
-            (other_op->opcode == BS_OP_WRITE || other_op->opcode == BS_OP_WRITE_STABLE))
+            (other_op->opcode == BS_OP_WRITE || other_op->opcode == BS_OP_WRITE_STABLE) &&
+            !PRIV(other_op)->op_state)
         {
             // Mark operations to cancel them
-            if (PRIV(other_op)->op_state != 0 && PRIV(other_op)->op_state != 100)
-            {
-                write_iodepth--;
-            }
             PRIV(other_op)->op_state = 100;
             other_op->retval = retval;
         }
@@ -103,7 +100,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         if (loc == UINT64_MAX ||
             !obj && heap->get_block_for_new_object(tmp_block) != 0)
         {
-            if (!heap->get_compact_queue_size() && !flusher->get_active())
+            if (!heap->get_compact_queue_size() && !write_iodepth && !flusher->get_active())
             {
                 // no space
                 cancel_all_writes(op, -ENOSPC);
@@ -179,7 +176,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         int res = heap->post_write(op->oid, wr, &modified_block);
         if (res == ENOSPC)
         {
-            if (!heap->get_compact_queue_size() && !flusher->get_active())
+            if (!heap->get_compact_queue_size() && !write_iodepth && !flusher->get_active())
             {
                 // no space
                 cancel_all_writes(op, -ENOSPC);
@@ -191,6 +188,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             return 0;
         }
         assert(res == 0);
+        PRIV(op)->lsn = wr->lsn;
         prepare_meta_block_write(op, modified_block);
         unsynced_small_write_count++;
         PRIV(op)->op_state = 9;
@@ -225,7 +223,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         int res = heap->post_write(op->oid, wr, &modified_block);
         if (res == ENOSPC)
         {
-            if (!heap->get_compact_queue_size() && !flusher->get_active())
+            if (!heap->get_compact_queue_size() && !write_iodepth && !flusher->get_active())
             {
                 // no space
                 cancel_all_writes(op, -ENOSPC);
@@ -237,6 +235,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             return 0;
         }
         assert(res == 0);
+        PRIV(op)->lsn = wr->lsn;
         heap->use_buffer_area(op->oid.inode, loc, op->len);
         prepare_meta_block_write(op, modified_block);
         if (op->len > 0)
@@ -317,6 +316,7 @@ resume_4:
             return 1;
         }
         assert(res == 0);
+        PRIV(op)->lsn = wr->lsn;
         prepare_meta_block_write(op, modified_block);
         PRIV(op)->op_state = 5;
         return 1;
@@ -361,6 +361,7 @@ resume_6:
 resume_8:
     // Acknowledge write
     op->retval = op->len;
+    heap->complete_lsn(PRIV(op)->lsn);
     write_iodepth--;
     FINISH_OP(op);
     return 2;

src/test/test_heap.cpp

@@ -434,19 +434,25 @@ void test_compact(bool csum, bool stable)
         res = heap.get_next_compact(compact_oid);
         assert(res == ENOENT);
 
-        res = heap.post_stabilize({ .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }, 3, NULL);
+        uint64_t to_lsn = 0, before_lsn = 0;
+        res = heap.post_stabilize({ .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }, 3, NULL, &before_lsn, &to_lsn);
         assert(res == ENOENT);
-        res = heap.post_stabilize(oid, 5, NULL);
+        res = heap.post_stabilize(oid, 5, NULL, &before_lsn, &to_lsn);
         assert(res == ENOENT);
-        res = heap.post_stabilize(oid, 1, &mblock);
+        res = heap.post_stabilize(oid, 1, &mblock, &before_lsn, &to_lsn);
         assert(res == 0);
-        res = heap.post_stabilize(oid, 3, &mblock);
+        assert(before_lsn == 0);
+        assert(to_lsn == 0);
+        res = heap.post_stabilize(oid, 3, &mblock, &before_lsn, &to_lsn);
         assert(res == 0);
         assert(mblock == 0);
+        assert(before_lsn == 0);
+        assert(to_lsn == 2);
         assert(check_used_space(heap, dsk, 0));
         assert(heap.get_meta_block_used_space(0) == 2*old_size + wr_size);
     }
 
+    heap.add_to_compact_queue(oid);
     assert(heap.get_compact_queue_size() == 1);
     res = heap.get_next_compact(compact_oid);
     assert(res == 0);
@@ -762,7 +768,7 @@ void test_full_overwrite(bool stable)
 
         if (!stable)
         {
-            res = heap.post_stabilize(oid, 4, NULL);
+            res = heap.post_stabilize(oid, 4, NULL, NULL, NULL);
             assert(res == 0);
         }