Support moving objects between blocks

src/blockstore/blockstore_heap.cpp

@@ -381,7 +381,7 @@ skip_object:
                 {
                     fprintf(stderr, "Warning: Object %jx:%jx in metadata block %u at %u is a newer duplicate (lsn %lu < %lu), overriding\n",
                         obj->inode, obj->stripe, block_num, block_offset, dup_lsn, lsn);
-                    erase_object(dup_block, dup_obj, 0, false);
+                    init_erase(dup_block, dup_obj);
                 }
             }
             // Verify checksums
@@ -739,7 +739,7 @@ bool blockstore_heap_t::recheck_small_writes(std::function<void(bool is_data, ui
                             {
                                 fprintf(stderr, "Notice: the whole object %jx:%jx only has unfinished writes, rolling back\n",
                                     obj->inode, obj->stripe);
-                                erase_object(block_num, obj, 0, false);
+                                init_erase(block_num, obj);
                             }
                             else
                             {
@@ -1047,7 +1047,7 @@ uint32_t blockstore_heap_t::compact_object_to(heap_object_t *obj, uint64_t compa
     return freed;
 }
 
-void blockstore_heap_t::compact_block(uint32_t block_num)
+void blockstore_heap_t::defragment_block(uint32_t block_num)
 {
     auto & inf = block_info[block_num];
     assert(inf.data);
@@ -1159,6 +1159,30 @@ uint32_t blockstore_heap_t::find_block_run(heap_block_info_t & inf, uint32_t spa
     return UINT32_MAX;
 }
 
+uint32_t blockstore_heap_t::block_has_compactable(uint8_t *data)
+{
+    uint32_t sum = 0;
+    uint8_t *end = data + dsk->meta_block_size;
+    while (data < end)
+    {
+        uint16_t region_marker = *((uint16_t*)data);
+        assert(region_marker);
+        if (!(region_marker & FREE_SPACE_BIT) &&
+            region_marker > sizeof(heap_object_t))
+        {
+            heap_write_t *wr = (heap_write_t*)data;
+            if (wr->flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) ||
+                wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE))
+            {
+                // May be freed in the future
+                sum += wr->size;
+            }
+        }
+        data += (region_marker & ~FREE_SPACE_BIT);
+    }
+    return sum;
+}
+
 uint32_t blockstore_heap_t::find_block_space(uint32_t block_num, uint32_t space)
 {
     auto & inf = block_info.at(block_num);
@@ -1177,10 +1201,20 @@ uint32_t blockstore_heap_t::find_block_space(uint32_t block_num, uint32_t space)
             return res;
         }
     }
-    compact_block(block_num);
+    defragment_block(block_num);
     return find_block_run(inf, space);
 }
 
+void blockstore_heap_t::allocate_block(heap_block_info_t & inf)
+{
+    if (!inf.data)
+    {
+        inf.data = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk->meta_block_size);
+        memset(inf.data, 0, dsk->meta_block_size);
+        *((uint16_t*)inf.data) = FREE_SPACE_BIT | dsk->meta_block_size;
+    }
+}
+
 int blockstore_heap_t::add_object(object_id oid, heap_write_t *wr, uint32_t *modified_block)
 {
     // By now, initial small_writes are not allowed
@@ -1198,12 +1232,7 @@ int blockstore_heap_t::add_object(object_id oid, heap_write_t *wr, uint32_t *mod
         return res;
     }
     auto & inf = block_info.at(block_num);
-    if (!inf.data)
+    allocate_block(inf);
-    {
-        inf.data = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk->meta_block_size);
-        memset(inf.data, 0, dsk->meta_block_size);
-        *((uint16_t*)inf.data) = FREE_SPACE_BIT | dsk->meta_block_size;
-    }
     if (modified_block)
     {
         *modified_block = block_num;
@@ -1243,6 +1272,20 @@ bool blockstore_heap_t::mvcc_check_tracking(object_id oid)
     return (mvcc_it->first.oid == oid && mvcc_it->second.entry_copy);
 }
 
+void blockstore_heap_t::copy_full_object(uint8_t *dst, heap_object_t *obj)
+{
+    memcpy(dst, obj, sizeof(heap_object_t));
+    ((heap_object_t*)dst)->write_pos = obj->size;
+    dst += obj->size;
+    for (auto wr = obj->get_writes(); wr; wr = wr->next())
+    {
+        memcpy(dst, wr, wr->size);
+        if (wr->next_pos)
+            ((heap_write_t*)dst)->next_pos = wr->size;
+        dst += wr->size;
+    }
+}
+
 // returns tracking_active, i.e. true if there exists at least one copied MVCC version of the object
 // it's used for reference tracking because tracking_active=false means that there is 1 implicit reference
 // for heap_writes of the current version of the object and true means that there isn't
@@ -1273,17 +1316,8 @@ bool blockstore_heap_t::mvcc_save_copy(heap_object_t *obj)
         total_size += wr->size;
     }
     heap_object_t *obj_copy = (heap_object_t*)malloc_or_die(total_size);
-    memcpy(obj_copy, obj, sizeof(heap_object_t));
+    copy_full_object((uint8_t*)obj_copy, obj);
     mvcc_it->second.entry_copy = obj_copy;
-    total_size = obj->size;
-    obj_copy->write_pos = obj->size;
-    for (auto wr = obj->get_writes(); wr; wr = wr->next())
-    {
-        auto new_wr = (heap_write_t*)((uint8_t*)obj_copy + total_size);
-        memcpy((uint8_t*)new_wr, wr, wr->size);
-        new_wr->next_pos = wr->next_pos ? wr->size : 0;
-        total_size += wr->size;
-    }
     uint32_t add_ref = 1;
     bool for_obj = false;
     // save_copy is performed when the object is modified, so object_mvcc may only
@@ -1339,18 +1373,11 @@ void blockstore_heap_t::mark_overwritten(uint64_t over_lsn, uint64_t inode, heap
     }
 }
 
-int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, heap_write_t *wr, uint32_t *modified_block)
+int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, heap_write_t *wr, uint32_t *modified_block, uint32_t *moved_from_block)
 {
     const auto oid = (object_id){ .inode = obj->inode, .stripe = obj->stripe };
-    const uint32_t wr_size = wr->get_size(this);
+    // First some validation
-    auto & inf = block_info.at(block_num);
-    assert(inf.data);
     bool is_overwrite = (wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) || wr->flags == (BS_HEAP_TOMBSTONE|BS_HEAP_STABLE));
-    if (dsk->meta_block_size-inf.used_space < wr_size+2)
-    {
-        // Something in the block has to be compacted
-        return ENOSPC;
-    }
     auto first_wr = obj->get_writes();
     if (first_wr->type() == BS_HEAP_TOMBSTONE && !is_overwrite)
     {
@@ -1379,6 +1406,48 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea
         // Overwrites with a smaller version are forbidden
         return EINVAL;
     }
+    // Then a free space check
+    const uint32_t wr_size = wr->get_size(this);
+    auto *inf = &block_info.at(block_num);
+    assert(inf->data);
+    if (inf->used_space+wr_size > dsk->meta_block_size-2)
+    {
+        // Something in the block has to be compacted
+        if (block_has_compactable(inf->data) >= inf->used_space+wr_size-(dsk->meta_block_size-2))
+        {
+            return EAGAIN;
+        }
+        // Otherwise, move the object
+        uint32_t new_block = 0;
+        int res = get_block_for_new_object(new_block);
+        if (res == ENOSPC)
+        {
+            return ENOSPC;
+        }
+        uint32_t full_size = obj->size;
+        for (auto wr = obj->get_writes(); wr; wr = wr->next())
+        {
+            full_size += wr->size;
+        }
+        inf = &block_info.at(new_block);
+        if (inf->used_space+full_size+wr_size > dsk->meta_block_size-2)
+        {
+            return ENOSPC;
+        }
+        allocate_block(*inf);
+        if (moved_from_block)
+        {
+            *moved_from_block = block_num;
+        }
+        uint32_t new_offset = find_block_space(new_block, full_size);
+        assert(new_offset != UINT32_MAX);
+        copy_full_object(inf->data + new_offset, obj);
+        erase_object(block_num, obj, 0, false);
+        block_num = new_block;
+        obj = (heap_object_t*)(inf->data + new_offset);
+        block_index[get_pg_id(oid.inode, oid.stripe)][oid.inode][oid.stripe] = (uint64_t)new_block*dsk->meta_block_size + new_offset;
+        add_used_space(new_block, full_size);
+    }
     if (modified_block)
     {
         *modified_block = block_num;
@@ -1397,16 +1466,16 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea
             mvcc_buffer_refs[wr->location]++;
         }
     }
-    const uint8_t *old_data = inf.data;
+    const uint8_t *old_data = inf->data;
     const uint32_t offset = find_block_space(block_num, wr_size);
-    if (old_data != inf.data)
+    if (old_data != inf->data)
     {
         obj = read_entry(oid, NULL);
         first_wr = obj->get_writes();
     }
     assert(offset != UINT32_MAX);
-    memcpy(inf.data + offset, wr, wr_size);
+    memcpy(inf->data + offset, wr, wr_size);
-    heap_write_t *new_wr = (heap_write_t*)(inf.data + offset);
+    heap_write_t *new_wr = (heap_write_t*)(inf->data + offset);
     new_wr->size = wr_size;
     new_wr->lsn = ++next_lsn;
     int32_t used_delta = wr_size;
@@ -1441,14 +1510,14 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea
     }
     wr->lsn = new_wr->lsn;
     push_inflight_lsn(oid, new_wr->lsn, new_wr->needs_compact(this) ? HEAP_INFLIGHT_COMPACTABLE : 0);
-    obj->write_pos = offset - ((uint8_t*)obj - inf.data);
+    obj->write_pos = offset - ((uint8_t*)obj - inf->data);
     obj->crc32c = obj->calc_crc32c();
     // Change block free space
     add_used_space(block_num, used_delta);
     return 0;
 }
 
-int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block)
+int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block, uint32_t *moved_from_block)
 {
     uint32_t block_num = 0;
     heap_object_t *obj = read_entry(oid, &block_num);
@@ -1456,16 +1525,16 @@ int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *mod
     {
         return add_object(oid, wr, modified_block);
     }
-    return update_object(block_num, obj, wr, modified_block);
+    return update_object(block_num, obj, wr, modified_block, moved_from_block);
 }
 
-int blockstore_heap_t::post_write(uint32_t & block_num, object_id oid, heap_object_t *obj, heap_write_t *wr)
+int blockstore_heap_t::post_write(uint32_t & block_num, object_id oid, heap_object_t *obj, heap_write_t *wr, uint32_t *moved_from_block)
 {
     if (!obj)
     {
         return add_object(oid, wr, &block_num);
     }
-    return update_object(block_num, obj, wr, &block_num);
+    return update_object(block_num, obj, wr, &block_num, moved_from_block);
 }
 
 int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *new_lsn, uint64_t *new_to_lsn)
@@ -1713,6 +1782,7 @@ void blockstore_heap_t::deref_data(uint64_t inode, uint64_t location, bool free_
 
 void blockstore_heap_t::deref_buffer(uint64_t inode, uint64_t location, uint32_t len, bool free_at_0)
 {
+    assert(len > 0);
     auto ref_it = mvcc_buffer_refs.find(location);
     if (ref_it != mvcc_buffer_refs.end())
     {
@@ -1777,24 +1847,22 @@ void blockstore_heap_t::erase_block_index(inode_t inode, uint64_t stripe)
     }
 }
 
+void blockstore_heap_t::init_erase(uint32_t block_num, heap_object_t *obj)
+{
+    for (auto wr = obj->get_writes(); wr; wr = wr->next())
+    {
+        if (wr->needs_compact(this))
+            mark_lsn_compacted(wr->lsn, true);
+    }
+    free_object_space(obj->inode, obj->get_writes(), NULL);
+    erase_object(block_num, obj, 0, false);
+}
+
 void blockstore_heap_t::erase_object(uint32_t block_num, heap_object_t *obj, uint64_t lsn, bool tracking_active)
 {
     // Erase object
-    if (!lsn)
+    if (lsn > 0)
-    {
-        for (auto wr = obj->get_writes(); wr; wr = wr->next())
-        {
-            if (wr->needs_compact(this))
-            {
-                mark_lsn_compacted(wr->lsn, true);
-            }
-        }
-        free_object_space(obj->inode, obj->get_writes(), NULL);
-    }
-    else
-    {
         mark_overwritten(lsn, obj->inode, obj->get_writes(), NULL, tracking_active);
-    }
     erase_block_index(obj->inode, obj->stripe);
     auto freed = free_writes(obj->get_writes(), NULL);
     auto obj_size = obj->size;

src/blockstore/blockstore_heap.h

@@ -172,15 +172,19 @@ class blockstore_heap_t
     const uint32_t max_write_entry_size;
 
     uint64_t get_pg_id(inode_t inode, uint64_t stripe);
-    void compact_block(uint32_t block_num);
+    void defragment_block(uint32_t block_num);
     uint32_t find_block_run(heap_block_info_t & block, uint32_t space);
     uint32_t find_block_space(uint32_t block_num, uint32_t space);
+    uint32_t block_has_compactable(uint8_t *data);
     uint32_t compact_object_to(heap_object_t *obj, uint64_t lsn, uint8_t *new_csums, bool do_free);
+    void copy_full_object(uint8_t *dst, heap_object_t *obj);
     bool mvcc_save_copy(heap_object_t *obj);
     bool mvcc_check_tracking(object_id oid);
+    void allocate_block(heap_block_info_t & inf);
     int add_object(object_id oid, heap_write_t *wr, uint32_t *modified_block);
     void mark_overwritten(uint64_t over_lsn, uint64_t inode, heap_write_t *wr, heap_write_t *end_wr, bool tracking_active);
-    int update_object(uint32_t block_num, heap_object_t *obj, heap_write_t *wr, uint32_t *modified_block);
+    int update_object(uint32_t block_num, heap_object_t *obj, heap_write_t *wr, uint32_t *modified_block, uint32_t *moved_from_block);
+    void init_erase(uint32_t block_num, heap_object_t *obj);
     void erase_object(uint32_t block_num, heap_object_t *obj, uint64_t lsn, bool tracking_active);
     void reindex_block(uint32_t block_num, heap_object_t *from_obj);
     void erase_block_index(inode_t inode, uint64_t stripe);
@@ -232,8 +236,8 @@ public:
         bool set, std::function<void(uint32_t, uint32_t, uint32_t)> bad_block_cb);
     // auto-compacts the object, then adds a write entry to it and to the compaction queue
     // return 0 if OK, or maybe ENOSPC
-    int post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block);
+    int post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block, uint32_t *moved_from_block);
-    int post_write(uint32_t & block_num, object_id oid, heap_object_t *obj, heap_write_t *wr);
+    int post_write(uint32_t & block_num, object_id oid, heap_object_t *obj, heap_write_t *wr, uint32_t *moved_from_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, uint64_t *new_lsn, uint64_t *new_to_lsn);

src/test/test_heap.cpp

@@ -83,8 +83,8 @@ int _test_do_big_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64
         else
             memset(wr->get_checksums(&heap), 0xde, dsk.data_block_size/dsk.csum_block_size*4);
     }
-    uint32_t mblock;
+    uint32_t mblock, mfblock;
-    return heap.post_write(oid, wr, &mblock);
+    return heap.post_write(oid, wr, &mblock, &mfblock);
 }
 
 void _test_big_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint64_t inode, uint64_t stripe, uint64_t version, uint64_t location,
@@ -97,7 +97,8 @@ 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 = true, uint32_t *checksums = NULL, bool is_intent = false)
+    uint32_t offset, uint32_t len, uint64_t location, bool stable = true, uint32_t *checksums = NULL, bool is_intent = false,
+    uint32_t *mblock = NULL, uint32_t *mfblock = NULL)
 {
     object_id oid = { .inode = INODE_WITH_POOL(1, inode), .stripe = stripe };
     uint8_t wr_buf[heap.get_max_write_entry_size()];
@@ -122,16 +123,16 @@ int _test_do_small_write(blockstore_heap_t & heap, blockstore_disk_t & dsk, uint
         memset(wr->get_checksums(&heap), 0xab, ((offset+len+dsk.csum_block_size-1)/dsk.csum_block_size - offset/dsk.csum_block_size)*4);
     else
         *wr->get_checksum(&heap) = 0xabababab;
-    uint32_t mblock;
+    return heap.post_write(oid, wr, mblock, mfblock);
-    return heap.post_write(oid, wr, &mblock);
 }
 
 void _test_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 = true, uint32_t *checksums = NULL, bool is_intent = false)
+    uint32_t offset, uint32_t len, uint64_t location, bool stable = true, uint32_t *checksums = NULL, bool is_intent = false,
+    uint32_t *mblock = NULL, uint32_t *mfblock = NULL)
 {
     if (!is_intent)
         heap.use_buffer_area(INODE_WITH_POOL(1, inode), location, len); // blocks are allocated before write and outside the heap_t
-    int res = _test_do_small_write(heap, dsk, inode, stripe, version, offset, len, location, stable, checksums, is_intent);
+    int res = _test_do_small_write(heap, dsk, inode, stripe, version, offset, len, location, stable, checksums, is_intent, mblock, mfblock);
     assert(res == 0);
     if (!is_intent)
         assert(!heap.is_buffer_area_free(location, len));
@@ -615,13 +616,13 @@ void test_corruption()
         wr->location = 0;
         wr->flags = BS_HEAP_TOMBSTONE|BS_HEAP_STABLE;
         assert(!wr->get_checksums(&heap));
-        res = heap.post_write(oid, wr, NULL);
+        res = heap.post_write(oid, wr, NULL, NULL);
         assert(res == 0);
 
         // try to do a small_write over a tombstone to fail
         wr->version = 3;
         wr->flags = BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE;
-        res = heap.post_write(oid, wr, NULL);
+        res = heap.post_write(oid, wr, NULL, NULL);
         assert(res == EINVAL);
 
         // persist
@@ -1320,7 +1321,7 @@ void test_full_alloc()
         assert(_test_do_small_write(heap, dsk, 1, 0, 6+i, 0, 4096, epb*4*16384+i*4096) == 0);
     }
     assert(dsk.meta_block_size-heap.get_meta_block_used_space(0) < big_write_size);
-    assert(_test_do_small_write(heap, dsk, 1, 0, 12, 0, 4096, 48*16384+8*4096) == ENOSPC);
+    assert(_test_do_small_write(heap, dsk, 1, 0, 12, 0, 4096, 48*16384+8*4096) == EAGAIN);
 
     // Check that used_alloc_queue doesn't return used blocks
     {
@@ -1558,6 +1559,62 @@ void test_intent_write(bool csum)
     printf("OK test_intent_write %s\n", csum ? "csum" : "no_csum");
 }
 
+void test_move()
+{
+    blockstore_disk_t dsk;
+    std::map<std::string, std::string> config;
+    config["data_csum_type"] = "crc32c";
+    dsk.parse_config(config);
+    dsk.data_device_size = 8*1024*1024;
+    dsk.meta_device_size = 5*4096;
+    dsk.journal_device_size = 4*1024*1024;
+    dsk.data_fd = 0;
+    dsk.meta_fd = 1;
+    dsk.journal_fd = 2;
+    dsk.calc_lengths();
+    std::vector<uint8_t> buffer_area(dsk.journal_device_size);
+
+    blockstore_heap_t heap(&dsk, buffer_area.data());
+    heap.finish_load();
+    assert(heap.get_meta_total_space() == 4*4096);
+
+    uint32_t big_write_size = (sizeof(heap_object_t) + sizeof(heap_write_t) + 2*dsk.clean_entry_bitmap_size + dsk.data_block_size/dsk.csum_block_size*4);
+    uint32_t small_write_size = (sizeof(heap_write_t) + dsk.clean_entry_bitmap_size + 4);
+    assert(big_write_size == 197);
+    assert(small_write_size == 45);
+
+    // Fill block 1 almost completely with unstable small writes
+    _test_big_write(heap, dsk, 1, 0*0x20000, 1, 0*0x20000);
+    _test_big_write(heap, dsk, 1, 1*0x20000, 1, 1*0x20000);
+    int i = 0;
+    while (i < (dsk.meta_block_size-2*big_write_size-2)/small_write_size/2)
+    {
+        _test_small_write(heap, dsk, 1, 0*0x20000, 2+i, (i*4096) % dsk.data_block_size, 4096, 2*i*4096, false);
+        _test_small_write(heap, dsk, 1, 1*0x20000, 2+i, (i*4096) % dsk.data_block_size, 4096, (2*i+1)*4096, false);
+        i++;
+    }
+    assert(heap.get_meta_block_used_space(0) > 0);
+    assert(!heap.get_meta_block_used_space(1));
+    assert(!heap.get_meta_block_used_space(2));
+    assert(!heap.get_meta_block_used_space(3));
+
+    // Next small_write should auto-move an object
+    uint32_t mblock = UINT32_MAX, mfblock = UINT32_MAX;
+    _test_small_write(heap, dsk, 1, 0*0x20000, 2+i, (i*4096) % dsk.data_block_size, 4096, 2*i*4096, false, NULL, false, &mblock, &mfblock);
+    assert(mblock == 1 && mfblock == 0);
+    assert(heap.get_meta_block_used_space(0) == big_write_size+i*small_write_size);
+    assert(heap.get_meta_block_used_space(1) == big_write_size+(i+1)*small_write_size);
+
+    // Check that the object is still readable
+    object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
+    heap_object_t *obj = heap.read_entry(oid, NULL);
+    assert(obj);
+    assert(count_writes(obj) == 2+i);
+    assert(obj->get_writes()->version == 2+i);
+
+    printf("OK test_move\n");
+}
+
 int main(int narg, char *args[])
 {
     test_mvcc(true);
@@ -1594,5 +1651,6 @@ int main(int narg, char *args[])
     test_autocompact(false);
     test_intent_write(true);
     test_intent_write(false);
+    test_move();
     return 0;
 }