Speedup blockstore_heap loading by postponing only objects with a lot of versions

src/blockstore/blockstore_heap.cpp

@@ -28,6 +28,7 @@
 
 #define IMAP_MALLOC_LOW_BITS ((size_t)0x0F)
 #define IMAP_MAX_LOW 16
+#define POSTPONE_INSERT_COUNT 10
 
 #define list_item_overhead(a) (((a) + sizeof(heap_list_item_t) - sizeof(heap_entry_t) + sizeof(void*) + 15) & ~15)
 
@@ -126,26 +127,26 @@ uint32_t heap_entry_t::get_size(blockstore_heap_t *heap)
     return heap->get_simple_entry_size();
 }
 
-bool heap_entry_t::is_overwrite()
+bool heap_entry_t::is_overwrite() const
 {
     return ((entry_type & ~BS_HEAP_GARBAGE) == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) ||
         (entry_type & ~BS_HEAP_GARBAGE) == (BS_HEAP_BIG_INTENT|BS_HEAP_STABLE) ||
         (entry_type & ~BS_HEAP_GARBAGE) == (BS_HEAP_DELETE|BS_HEAP_STABLE));
 }
 
-bool heap_entry_t::is_compactable()
+bool heap_entry_t::is_compactable() const
 {
     return !is_overwrite() && (entry_type & BS_HEAP_STABLE) ||
         (entry_type & ~BS_HEAP_GARBAGE) == BS_HEAP_COMMIT ||
         (entry_type & ~BS_HEAP_GARBAGE) == BS_HEAP_ROLLBACK;
 }
 
-bool heap_entry_t::is_before(heap_entry_t *other)
+bool heap_entry_t::is_before(const heap_entry_t *other) const
 {
     return lsn < other->lsn || lsn == other->lsn && !is_overwrite() && other->is_overwrite();
 }
 
-bool heap_entry_t::is_garbage()
+bool heap_entry_t::is_garbage() const
 {
     return (entry_type & BS_HEAP_GARBAGE);
 }
@@ -441,7 +442,7 @@ int blockstore_heap_t::load_blocks(uint64_t disk_offset, uint64_t size, uint8_t
             next_lsn = wr->lsn;
         }
         entries_loaded++;
-        loaded_list_items.push_back(li);
+        insert_list_items(&li, 1, true);
         modify_alloc(block_num, [&](heap_block_info_t & inf)
         {
             if (!inf.entries.size())
@@ -547,18 +548,26 @@ bool blockstore_heap_t::validate_object(heap_entry_t *obj)
 
 void blockstore_heap_t::finish_load()
 {
-    if (loaded_list_items.size())
+    if (postponed_items.size())
     {
-        // Sort everything and load in correct order
+        // Sort "postponed" items and load in batches
-        std::sort(loaded_list_items.begin(), loaded_list_items.end(), [this](const heap_list_item_t* a, const heap_list_item_t* b)
+        std::sort(postponed_items.begin(), postponed_items.end(), [this](const heap_list_item_t* a, const heap_list_item_t* b)
         {
-            return a->entry.lsn < b->entry.lsn;
+            return a->entry.inode < b->entry.inode || a->entry.inode == b->entry.inode &&
+                (a->entry.stripe < b->entry.stripe || a->entry.stripe == b->entry.stripe &&
+                    !a->entry.is_before(&b->entry)); // object ASC, lsn DESC
         });
-        for (auto & li: loaded_list_items)
+        size_t s = 0, e, n = postponed_items.size();
+        for (e = 1; e <= n; e++)
         {
-            insert_list_item(li);
+            if (e >= n || postponed_items[e]->entry.inode != postponed_items[s]->entry.inode &&
+                postponed_items[e]->entry.stripe != postponed_items[s]->entry.stripe)
+            {
+                insert_list_items(postponed_items.data()+s, e-s, false);
+                s = e;
+            }
         }
-        loaded_list_items.clear();
+        postponed_items.clear();
     }
 }
 
@@ -1474,43 +1483,51 @@ int blockstore_heap_t::allocate_entry(uint32_t entry_size, uint32_t *block_num,
     return 0;
 }
 
-void blockstore_heap_t::insert_list_item(heap_list_item_t *li)
+void blockstore_heap_t::insert_list_items(heap_list_item_t** v, size_t count, bool postpone)
 {
-    auto & inode_idx = block_index[get_pg_id(li->entry.inode, li->entry.stripe)][li->entry.inode];
+    auto wr = &v[0]->entry;
+    auto & inode_idx = block_index[get_pg_id(wr->inode, wr->stripe)][wr->inode];
     heap_inode_map_t::iterator li_it;
     heap_list_item_t *old_head = NULL;
     if (inode_idx)
-        inode_map_get(inode_idx, li_it, old_head, li->entry.stripe);
+        inode_map_get(inode_idx, li_it, old_head, wr->stripe);
-    if (old_head && !old_head->entry.is_before(&li->entry))
+    heap_list_item_t *next_li = NULL;
+    heap_list_item_t *prev_li = old_head;
+    int skips = 0;
+    // Merge entry array and inode_idx linked list (both sorted in newest first order)
+    for (size_t i = 0; i < count; i++)
     {
         // BIG_WRITE may be inserted into the middle of the sequence during compaction
         // and it overrides SMALL_WRITEs and COMMITs with the same LSN
         // However, all entries of other types (say DELETE) override previous ones
-        auto next_li = old_head;
+        auto li = v[i];
-        auto prev_li = old_head->prev;
         while (prev_li && !prev_li->entry.is_before(&li->entry))
         {
             next_li = prev_li;
             prev_li = prev_li->prev;
+            skips++;
+        }
+        if (postpone && skips > POSTPONE_INSERT_COUNT)
+        {
+            postponed_items.push_back(li);
+            return;
+        }
+        if (next_li == NULL)
+        {
+            // Replace the latest entry pointer
+            if (old_head)
+                inode_map_replace(inode_idx, li_it, li);
+            else
+                inode_map_put(inode_idx, li);
         }
         // Insert <li> between <next_li> and <prev_li>
+        li->next = next_li;
+        if (next_li)
+            next_li->prev = li;
         li->prev = prev_li;
         if (prev_li)
             prev_li->next = li;
-        next_li->prev = li;
+        next_li = li;
-        li->next = next_li;
-    }
-    else
-    {
-        li->prev = old_head;
-        li->next = NULL;
-        if (old_head)
-        {
-            old_head->next = li;
-            inode_map_replace(inode_idx, li_it, li);
-        }
-        else
-            inode_map_put(inode_idx, li);
     }
 }
 
@@ -1542,7 +1559,7 @@ int blockstore_heap_t::add_entry(uint32_t wr_size, uint32_t *modified_block,
         (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);
+    insert_list_items(&li, 1, false);
     li->block_num = block_num;
     new_wr->size = wr_size;
     new_wr->crc32c = new_wr->calc_crc32c();

src/blockstore/blockstore_heap.h

@@ -57,11 +57,11 @@ struct __attribute__((__packed__)) heap_entry_t
     inline heap_small_write_t& small() { return *(heap_small_write_t*)this; }
     inline heap_big_write_t& big() { return *(heap_big_write_t*)this; }
     inline heap_big_intent_t& big_intent() { return *(heap_big_intent_t*)this; }
-    bool is_garbage();
+    bool is_garbage() const;
     void set_garbage();
-    bool is_overwrite();
+    bool is_overwrite() const;
-    bool is_compactable();
+    bool is_compactable() const;
-    bool is_before(heap_entry_t *other);
+    bool is_before(const heap_entry_t *other) const;
     uint32_t get_size(blockstore_heap_t *heap);
     uint8_t *get_ext_bitmap(blockstore_heap_t *heap);
     uint8_t *get_int_bitmap(blockstore_heap_t *heap);
@@ -208,7 +208,7 @@ class blockstore_heap_t
 
     bool marked_used_blocks = false;
     bool recheck_queue_filled = false;
-    std::vector<heap_list_item_t*> loaded_list_items;
+    std::vector<heap_list_item_t*> postponed_items;
     std::set<uint32_t> recheck_modified_blocks;
     std::deque<heap_entry_t*> recheck_queue;
     int recheck_in_progress = 0;
@@ -230,7 +230,7 @@ class blockstore_heap_t
 
     void gc_block(heap_block_info_t & inf);
     int allocate_entry(uint32_t entry_size, uint32_t *block_num, bool allow_last_free);
-    void insert_list_item(heap_list_item_t *li);
+    void insert_list_items(heap_list_item_t** v, size_t count, bool postpone);
     void remove_list_item(heap_list_item_t *li);
     void unlink_list_item(heap_list_item_t *li);
     int add_entry(uint32_t wr_size, uint32_t *modified_block, bool allow_last_free,

src/test/test_heap.cpp

@@ -2338,6 +2338,139 @@ void test_skip_double_claim()
     }
 }
 
+void test_postpone_load()
+{
+    blockstore_disk_t dsk;
+    // FIXME dsk.readonly = true;
+    _test_init(dsk, false);
+    std::vector<uint8_t> tmp(dsk.meta_block_size*10);
+    std::vector<uint8_t> buffer_area(dsk.journal_device_size);
+
+    {
+        blockstore_heap_t heap(&dsk, buffer_area.data(), 10);
+
+        size_t total_size = 0;
+        auto wr1 = (heap_entry_t*)(tmp.data() + total_size);
+        wr1->size = heap.get_big_entry_size();
+        wr1->entry_type = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE;
+        wr1->lsn = 1;
+        wr1->inode = INODE_WITH_POOL(1, 1);
+        wr1->stripe = 0;
+        wr1->version = 1;
+        wr1->set_big_location(&heap, 0x20000);
+        memset(wr1->get_ext_bitmap(&heap), 0xff, dsk.clean_entry_bitmap_size);
+        wr1->crc32c = wr1->calc_crc32c();
+        total_size += wr1->size;
+
+        assert(total_size+heap.get_big_entry_size() <= dsk.meta_block_size);
+        wr1 = (heap_entry_t*)(tmp.data() + total_size);
+        wr1->size = heap.get_big_entry_size();
+        wr1->entry_type = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE;
+        wr1->lsn = 20; // 20 but compacted - newest entry
+        wr1->inode = INODE_WITH_POOL(1, 1);
+        wr1->stripe = 0;
+        wr1->version = 1;
+        wr1->set_big_location(&heap, 0x20000);
+        memset(wr1->get_ext_bitmap(&heap), 0xff, dsk.clean_entry_bitmap_size);
+        wr1->crc32c = wr1->calc_crc32c();
+        total_size += wr1->size;
+
+        uint32_t small_size = heap.get_small_entry_size(0, 4096);
+        auto add_small = [&](uint64_t lsn)
+        {
+            assert(total_size+small_size <= dsk.meta_block_size);
+            auto wr2 = (heap_entry_t*)(tmp.data() + total_size);
+            wr2->size = small_size;
+            wr2->entry_type = BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE;
+            wr2->lsn = lsn;
+            wr2->inode = INODE_WITH_POOL(1, 1);
+            wr2->stripe = 0;
+            wr2->version = lsn;
+            wr2->small().offset = (lsn % 32)*4096;
+            wr2->small().len = 4096;
+            wr2->small().location = lsn*4096;
+            memset(wr2->get_ext_bitmap(&heap), 0xff, dsk.clean_entry_bitmap_size);
+            *((uint32_t*)wr2->get_checksum(&heap)) = crc32c(0, buffer_area.data()+wr2->small().location, 4096);
+            wr2->crc32c = wr2->calc_crc32c();
+            total_size += small_size;
+        };
+        for (int i = 0; i < 10; i++)
+            add_small(2 + 2*i); // 2..20
+        for (int i = 0; i < 10; i++)
+            add_small(30 - i); // 21..30
+        for (int i = 0; i < 9; i++)
+            add_small(3 + 2*i); // 3..19
+
+        assert(total_size+heap.get_big_entry_size() <= dsk.meta_block_size);
+        wr1 = (heap_entry_t*)(tmp.data() + total_size);
+        wr1->size = heap.get_big_entry_size();
+        wr1->entry_type = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE;
+        wr1->lsn = 15; // 15 but also compacted
+        wr1->inode = INODE_WITH_POOL(1, 1);
+        wr1->stripe = 0;
+        wr1->version = 1;
+        wr1->set_big_location(&heap, 0x20000);
+        memset(wr1->get_ext_bitmap(&heap), 0xff, dsk.clean_entry_bitmap_size);
+        wr1->crc32c = wr1->calc_crc32c();
+        total_size += wr1->size;
+
+        *(uint16_t*)(tmp.data() + total_size) = dsk.meta_block_size - total_size;
+        *(uint16_t*)(tmp.data() + total_size + 2) = BS_HEAP_FREE_SPACE;
+
+        uint64_t entries_loaded;
+        heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded);
+
+        object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
+        heap_entry_t *obj = heap.read_entry(oid);
+        assert(obj);
+        assert(count_writes(heap, obj) == 22);
+
+        heap.finish_load();
+
+        oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
+        obj = heap.read_entry(oid);
+        assert(obj);
+        assert(count_writes(heap, obj) == 32);
+        uint64_t clsn = 30;
+        bool stable = true;
+        for (auto wr = obj; wr; wr = heap.prev(wr))
+        {
+            assert(wr->lsn == clsn);
+            if (clsn == 20 || clsn == 15)
+            {
+                assert(wr->entry_type == (stable ? BS_HEAP_BIG_WRITE|BS_HEAP_STABLE : BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE));
+                if (stable)
+                    stable = false;
+                else
+                {
+                    clsn--;
+                    stable = true;
+                }
+            }
+            else if (clsn == 1)
+            {
+                assert(wr->entry_type == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE);
+            }
+            else
+            {
+                assert(wr->entry_type == BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE);
+                clsn--;
+            }
+        }
+
+        bool done = heap.recheck_small_writes([&](bool, uint64_t, uint64_t, uint8_t*, std::function<void()> cb) {}, 1);
+        assert(done);
+        heap.finish_recheck();
+        auto mod = heap.get_recheck_modified_blocks();
+        assert(mod.size() == 1);
+
+        oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
+        obj = heap.read_entry(oid);
+        assert(obj);
+        assert(count_writes(heap, obj) == 11);
+    }
+}
+
 // 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[])
@@ -2379,5 +2512,6 @@ int main(int narg, char *args[])
     test_redirect_intent_csums();
     test_explicit_complete();
     test_skip_double_claim();
+    test_postpone_load();
     return 0;
 }