Integrate moving objects

src/blockstore/blockstore_heap.cpp

@@ -1999,6 +1999,14 @@ void blockstore_heap_t::use_data(inode_t inode, uint64_t location)
     data_used_space += dsk->data_block_size;
 }
 
+void blockstore_heap_t::free_data(inode_t inode, uint64_t location)
+{
+    assert(data_alloc->get(location / dsk->data_block_size));
+    data_alloc->set(location / dsk->data_block_size, false);
+    inode_space_stats[inode] -= dsk->data_block_size;
+    data_used_space -= dsk->data_block_size;
+}
+
 uint64_t blockstore_heap_t::find_free_buffer_area(uint64_t size)
 {
     assert(!(size % dsk->bitmap_granularity));

src/blockstore/blockstore_heap.h

@@ -275,6 +275,7 @@ public:
     uint64_t find_free_data();
     bool is_data_used(uint64_t location);
     void use_data(inode_t inode, uint64_t location);
+    void free_data(inode_t inode, uint64_t location);
 
     // buffer device allocator functions
     uint64_t find_free_buffer_area(uint64_t size);

src/blockstore/blockstore_impl.h

@@ -168,7 +168,7 @@ public:
     bool enqueue_write(blockstore_op_t *op);
     void prepare_meta_block_write(blockstore_op_t *op, uint64_t modified_block, io_uring_sqe *sqe = NULL);
     int dequeue_write(blockstore_op_t *op);
-    bool make_big_write(blockstore_op_t *op, uint32_t offset, uint32_t len, uint32_t *modified_block);
+    int make_big_write(blockstore_op_t *op, uint32_t offset, uint32_t len, uint32_t *modified_block, uint32_t *moved_from_block);
     int continue_write(blockstore_op_t *op);
     void handle_write_event(ring_data_t *data, blockstore_op_t *op);
 

src/blockstore/blockstore_write.cpp

@@ -39,7 +39,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         return 0;
     }
     PRIV(op)->is_big = false;
-    uint32_t modified_block = 0;
+    uint32_t modified_block = UINT32_MAX, moved_from_block = UINT32_MAX;
     heap_object_t *obj = heap->read_entry(op->oid, &modified_block);
     if (op->opcode == BS_OP_DELETE)
     {
@@ -94,8 +94,10 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         if (!dsk.disable_data_fsync && dsk.disable_meta_fsync)
         {
             // Do big_write as an INTENT to avoid data fsync
-            bool ok = make_big_write(op, 0, 0, &modified_block);
+            int res = make_big_write(op, 0, 0, &modified_block, &moved_from_block);
-            assert(ok);
+            assert(res == 0);
+            if (moved_from_block != UINT32_MAX)
+                prepare_meta_block_write(op, moved_from_block);
             obj = heap->read_entry(op->oid, &modified_block);
             heap->mark_lsn_completed(PRIV(op)->lsn);
             goto process_intent;
@@ -165,23 +167,26 @@ process_intent:
         if (op->bitmap)
             memcpy(wr->get_ext_bitmap(heap), op->bitmap, dsk.clean_entry_bitmap_size);
         heap->calc_checksums(wr, (uint8_t*)op->buf, true);
-        int res = heap->post_write(modified_block, op->oid, obj, wr);
+        int res = heap->post_write(modified_block, op->oid, obj, wr, &moved_from_block);
-        if (res == ENOSPC)
+        if (res == EAGAIN)
         {
-            if (!heap->get_inflight_queue_size())
+            assert(heap->get_inflight_queue_size());
-            {
-                // no space
-                op->retval = -ENOSPC;
-                FINISH_OP(op);
-                return 2;
-            }
             PRIV(op)->wait_for = WAIT_COMPACTION;
             PRIV(op)->wait_detail = flusher->get_compact_counter();
             flusher->request_trim();
             return 0;
         }
+        else if (res == ENOSPC)
+        {
+            // no space
+            op->retval = -ENOSPC;
+            FINISH_OP(op);
+            return 2;
+        }
         assert(res == 0);
         PRIV(op)->lsn = wr->lsn;
+        if (moved_from_block != UINT32_MAX)
+            prepare_meta_block_write(op, moved_from_block);
         prepare_meta_block_write(op, modified_block);
         PRIV(op)->op_state = 9;
         write_iodepth++;
@@ -211,25 +216,28 @@ process_intent:
         if (op->bitmap)
             memcpy(wr->get_ext_bitmap(heap), op->bitmap, dsk.clean_entry_bitmap_size);
         heap->calc_checksums(wr, (uint8_t*)op->buf, true);
-        int res = heap->post_write(modified_block, op->oid, obj, wr);
+        int res = heap->post_write(modified_block, op->oid, obj, wr, &moved_from_block);
-        if (res == ENOSPC)
+        if (res == EAGAIN)
         {
-            if (!heap->get_inflight_queue_size())
+            assert(heap->get_inflight_queue_size());
-            {
-                // no space
-                op->retval = -ENOSPC;
-                FINISH_OP(op);
-                return 2;
-            }
             PRIV(op)->wait_for = WAIT_COMPACTION;
             PRIV(op)->wait_detail = flusher->get_compact_counter();
             flusher->request_trim();
             return 0;
         }
+        else if (res == ENOSPC)
+        {
+            // no space
+            op->retval = -ENOSPC;
+            FINISH_OP(op);
+            return 2;
+        }
         assert(res == 0);
         PRIV(op)->lsn = wr->lsn;
         if (op->len)
             heap->use_buffer_area(op->oid.inode, loc, op->len);
+        if (moved_from_block != UINT32_MAX)
+            prepare_meta_block_write(op, moved_from_block);
         prepare_meta_block_write(op, modified_block);
         if (op->len > 0)
         {
@@ -253,7 +261,7 @@ process_intent:
     return 1;
 }
 
-bool blockstore_impl_t::make_big_write(blockstore_op_t *op, uint32_t offset, uint32_t len, uint32_t *modified_block)
+int blockstore_impl_t::make_big_write(blockstore_op_t *op, uint32_t offset, uint32_t len, uint32_t *modified_block, uint32_t *moved_from_block)
 {
     uint8_t wr_buf[heap->get_max_write_entry_size()];
     heap_write_t *wr = (heap_write_t*)wr_buf;
@@ -267,12 +275,12 @@ bool blockstore_impl_t::make_big_write(blockstore_op_t *op, uint32_t offset, uin
     memset(wr->get_int_bitmap(heap), 0, dsk.clean_entry_bitmap_size);
     bitmap_set(wr->get_int_bitmap(heap), offset, len, dsk.bitmap_granularity);
     heap->calc_checksums(wr, (uint8_t*)op->buf, true);
-    int res = heap->post_write(op->oid, wr, modified_block);
+    int res = heap->post_write(op->oid, wr, modified_block, moved_from_block);
-    if (res == ENOSPC)
+    if (res != 0)
-        return false;
+        return res;
     assert(res == 0);
     PRIV(op)->lsn = wr->lsn;
-    return true;
+    return 0;
 }
 
 int blockstore_impl_t::continue_write(blockstore_op_t *op)
@@ -341,13 +349,24 @@ resume_12:
     }
 resume_4:
     {
-        uint32_t modified_block = 0;
+        uint32_t modified_block = UINT32_MAX, moved_from_block = UINT32_MAX;
-        if (!make_big_write(op, op->offset, op->len, &modified_block))
+        int res = make_big_write(op, op->offset, op->len, &modified_block, &moved_from_block);
+        if (res == EAGAIN)
         {
             PRIV(op)->wait_for = WAIT_COMPACTION;
             PRIV(op)->wait_detail = flusher->get_compact_counter();
             return 1;
         }
+        else if (res == ENOSPC)
+        {
+            heap->free_data(op->oid.inode, PRIV(op)->location);
+            write_iodepth--;
+            op->retval = -ENOSPC;
+            FINISH_OP(op);
+            return 2;
+        }
+        if (moved_from_block != UINT32_MAX)
+            prepare_meta_block_write(op, moved_from_block);
         prepare_meta_block_write(op, modified_block);
         PRIV(op)->op_state = 5;
         return 1;