diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index d25d67dd..f0d7fffa 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/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 - std::sort(loaded_list_items.begin(), loaded_list_items.end(), [this](const heap_list_item_t* a, const heap_list_item_t* b) + // Sort "postponed" items and load in batches + 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); - if (old_head && !old_head->entry.is_before(&li->entry)) + inode_map_get(inode_idx, li_it, old_head, wr->stripe); + 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 prev_li = old_head->prev; + auto li = v[i]; 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
  • between and + 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; - 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); + next_li = 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(); diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index 21852762..e89c39b2 100644 --- a/src/blockstore/blockstore_heap.h +++ b/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_compactable(); - bool is_before(heap_entry_t *other); + bool is_overwrite() const; + bool is_compactable() const; + 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 loaded_list_items; + std::vector postponed_items; std::set recheck_modified_blocks; std::deque 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, diff --git a/src/test/test_heap.cpp b/src/test/test_heap.cpp index 0ba57394..8d978c0c 100644 --- a/src/test/test_heap.cpp +++ b/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 tmp(dsk.meta_block_size*10); + std::vector 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 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; }