diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index 88c6dca4..8aa803ae 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -68,14 +68,14 @@ int journal_flusher_t::get_active() return active_flushers; } -uint64_t journal_flusher_t::get_counter() +uint64_t journal_flusher_t::get_compact_counter() { return compact_counter; } bool journal_flusher_t::is_active() { - return active_flushers > 0 || bs->heap->get_compact_queue_size() > (force_start > 0 ? 0 : bs->flusher_start_threshold); + return active_flushers > 0 || force_start > 0 || bs->heap->get_compact_queue_size() > bs->flusher_start_threshold; } void journal_flusher_t::request_trim() @@ -158,10 +158,23 @@ bool journal_flusher_co::loop() else if (wait_state == 15) goto resume_15; else if (wait_state == 16) goto resume_16; else if (wait_state == 17) goto resume_17; + else if (wait_state == 18) goto resume_18; + else if (wait_state == 19) goto resume_19; + else if (wait_state == 20) goto resume_20; + else if (wait_state == 21) goto resume_21; resume_0: res = bs->heap->get_next_compact(cur_oid); if (res == ENOENT) { + if (co_id == 0 && flusher->force_start > 0) + { +resume_18: +resume_19: +resume_20: +resume_21: + if (!trim_lsn(18)) + return false; + } cur_oid = {}; wait_state = 0; return true; @@ -203,7 +216,6 @@ resume_1: read_vec.clear(); for (auto wr = begin_wr; wr != end_wr; wr = wr->next()) { - min_compact_lsn = wr->lsn; bs->prepare_read(read_vec, cur_obj, wr, 0, bs->dsk.data_block_size); } overwrite_start = overwrite_end = 0; @@ -277,19 +289,30 @@ resume_12: resume_13: if (copy_count && !fsync_batch(false, 11)) return false; + // Lock is only needed to prevent freeing the big_write because we overwrite it... 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 + // Mark the object compacted, but don't free and remove small_writes + // We'll free and remove them only when trimming + // The only thing we modify here are big_write block checksums if >4k block is used + cur_obj = bs->heap->read_entry(cur_oid, NULL); + if (!cur_obj) + { + // Abort compaction + goto release_oid; + } calc_block_checksums(); - bs->heap->compact_object(cur_oid, compact_lsn, new_data_csums); + bs->heap->mark_object_compacted(cur_obj, compact_lsn); // Done, free all buffers free_buffers(); #ifdef BLOCKSTORE_DEBUG printf("Compacted %jx:%jx v%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, cur_version, copy_count); #endif + flusher->compact_counter++; + flusher->active_flushers--; // Advance compacted_lsn every objects - bs->heap->set_compacted_lsn(min_compact_lsn); - if (bs->journal_trim_interval && !((++flusher->advance_lsn_counter) % bs->journal_trim_interval)) + if (co_id == 0 && !((++flusher->advance_lsn_counter) % bs->journal_trim_interval)) { + flusher->advance_lsn_counter = 0; resume_14: resume_15: resume_16: @@ -298,14 +321,12 @@ resume_17: return false; } release_oid: - flusher->active_flushers--; if (should_repeat) { // Flush the same object again goto resume_1; } // All done - flusher->compact_counter++; wait_state = 0; goto resume_0; } @@ -479,7 +500,6 @@ 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, copy_id); uint64_t block_offset = 0; uint32_t block_done = 0; uint32_t block_csum = 0; @@ -667,6 +687,12 @@ bool journal_flusher_co::trim_lsn(int wait_base) else if (wait_state == wait_base+1) goto resume_1; else if (wait_state == wait_base+2) goto resume_2; else if (wait_state == wait_base+3) goto resume_3; + compact_lsn = bs->heap->get_compacted_lsn(); + if (((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn == compact_lsn) + { + return true; + } + flusher->active_flushers++; if (!bs->disable_meta_fsync) { await_sqe(0); @@ -681,7 +707,7 @@ resume_1: return false; } } - ((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn = bs->heap->get_compacted_lsn(); + ((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn = compact_lsn; ((blockstore_meta_header_v3_t*)bs->meta_superblock)->set_crc32c(); await_sqe(2); data->iov = (struct iovec){ bs->meta_superblock, (size_t)bs->dsk.meta_block_size }; @@ -694,6 +720,8 @@ resume_3: wait_state = wait_base+3; return false; } - flusher->advance_lsn_counter = 0; + bs->heap->mark_lsn_trimmed(compact_lsn); + flusher->compact_counter++; + flusher->active_flushers--; return true; } diff --git a/src/blockstore/blockstore_flush.h b/src/blockstore/blockstore_flush.h index 4b18a99a..cee23b5d 100644 --- a/src/blockstore/blockstore_flush.h +++ b/src/blockstore/blockstore_flush.h @@ -51,7 +51,6 @@ class journal_flusher_co object_id cur_oid; uint64_t copy_id; uint64_t compact_lsn; - uint64_t min_compact_lsn; uint64_t cur_version; heap_object_t *cur_obj; heap_write_t *begin_wr, *end_wr; @@ -108,7 +107,7 @@ public: ~journal_flusher_t(); void loop(); int get_active(); - uint64_t get_counter(); + uint64_t get_compact_counter(); bool is_active(); void request_trim(); void release_trim(); diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index a1ab6ae1..ec1f7aa8 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -61,7 +61,7 @@ bool heap_write_t::needs_recheck(blockstore_heap_t *heap) bool heap_write_t::needs_compact(uint64_t compacted_lsn) { - return lsn > compacted_lsn && (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) || flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)); + return lsn > compacted_lsn && flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); } bool heap_write_t::is_compacted(uint64_t compacted_lsn) @@ -585,9 +585,9 @@ skip_object: // Mark data block as used use_data(obj->inode, wr->location); } - if (wr->needs_compact(this->compacted_lsn)) + if (wr->lsn > this->compacted_lsn) { - tmp_compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn }); + tmp_compact_queue.push_back((tmp_compact_item_t){ .oid = oid, .lsn = wr->lsn, .compact = wr->needs_compact(0) }); } else if (wr->is_compacted(this->compacted_lsn)) { @@ -599,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 - tmp_compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn }); + tmp_compact_queue.push_back((tmp_compact_item_t){ .oid = oid, .lsn = wr->lsn, .compact = wr->needs_compact(0) }); } } } @@ -643,13 +643,28 @@ skip_object: void blockstore_heap_t::finish_load() { - std::sort(tmp_compact_queue.begin(), tmp_compact_queue.end(), [this](const heap_object_lsn_t & a, const heap_object_lsn_t & b) + completed_lsn = first_inflight_lsn = next_lsn+1; + if (!tmp_compact_queue.size()) + { + return; + } + std::sort(tmp_compact_queue.begin(), tmp_compact_queue.end(), [this](const tmp_compact_item_t & a, const tmp_compact_item_t & b) { return a.lsn < b.lsn; }); + first_inflight_lsn = tmp_compact_queue[0].lsn; + if (compacted_lsn < tmp_compact_queue[0].lsn-1) + { + compacted_lsn = tmp_compact_queue[0].lsn-1; + } for (auto & e: tmp_compact_queue) { - compact_queue.push_back(e.oid); + push_inflight_lsn(e.oid, e.lsn, HEAP_INFLIGHT_DONE | (e.compact ? HEAP_INFLIGHT_COMPACTABLE : 0)); + } + while (compacted_lsn+1-first_inflight_lsn < inflight_lsn.size() && + !(inflight_lsn[compacted_lsn+1-first_inflight_lsn].flags & HEAP_INFLIGHT_COMPACTABLE)) + { + compacted_lsn++; } tmp_compact_queue.clear(); } @@ -758,7 +773,7 @@ bool blockstore_heap_t::recheck_small_writes(std::functionlen); if (log_level > 5) { - fprintf(stderr, "Notice: rechecking %u bytes at %ju in %s area\n", wr->len, loc, is_intent ? "data" : "buffer"); + fprintf(stderr, "Notice: rechecking %u bytes at %ju in %s area (lsn %lu)\n", wr->len, loc, is_intent ? "data" : "buffer", wr->lsn); } recheck_cb(is_intent, loc, wr->len, buf, [this, oid, lsn = wr->lsn, buf]() { @@ -1264,7 +1279,7 @@ int blockstore_heap_t::add_object(object_id oid, heap_write_t *wr, uint32_t *mod new_wr->size = wr_size; new_wr->lsn = ++next_lsn; wr->lsn = new_wr->lsn; - push_inflight_lsn(oid, new_wr); + push_inflight_lsn(oid, new_wr->lsn, new_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); @@ -1430,7 +1445,7 @@ 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(oid, new_wr); + push_inflight_lsn(oid, new_wr->lsn, new_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); @@ -1528,7 +1543,7 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t { wr->flags |= BS_HEAP_STABLE; wr->lsn = last_lsn--; - push_inflight_lsn(oid, wr); + push_inflight_lsn(oid, wr->lsn, wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0); } } obj->crc32c = obj->calc_crc32c(); @@ -1608,10 +1623,24 @@ int blockstore_heap_t::post_delete(object_id oid, uint32_t *modified_block) int blockstore_heap_t::get_next_compact(object_id & oid) { - while (compact_queue.size()) + if (next_compact_lsn < first_inflight_lsn) { - oid = compact_queue.front(); - compact_queue.pop_front(); + next_compact_lsn = first_inflight_lsn; + } + while (next_compact_lsn-first_inflight_lsn < inflight_lsn.size()) + { + auto & item = inflight_lsn[next_compact_lsn-first_inflight_lsn]; + if (!(item.flags & HEAP_INFLIGHT_COMPACTABLE)) + { + next_compact_lsn++; + continue; + } + if (!(item.flags & HEAP_INFLIGHT_DONE)) + { + break; + } + next_compact_lsn++; + oid = item.oid; return 0; } return ENOENT; @@ -1918,7 +1947,12 @@ uint32_t blockstore_heap_t::get_meta_nearfull_blocks() uint32_t blockstore_heap_t::get_compact_queue_size() { - return compact_queue.size(); + return to_compact_count; +} + +uint32_t blockstore_heap_t::get_inflight_queue_size() +{ + return inflight_lsn.size(); } uint32_t blockstore_heap_t::get_max_write_entry_size() @@ -1931,39 +1965,95 @@ void blockstore_heap_t::set_fail_on_warn(bool fail) fail_on_warn = fail; } -void blockstore_heap_t::push_inflight_lsn(object_id oid, heap_write_t *wr) +void blockstore_heap_t::push_inflight_lsn(object_id oid, uint64_t lsn, uint64_t flags) { uint64_t next_inf = first_inflight_lsn + inflight_lsn.size(); - uint64_t flags = wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0; - if (wr->lsn == next_inf) + if (flags & HEAP_INFLIGHT_COMPACTABLE) + { + to_compact_count++; + } + if (lsn == next_inf) { inflight_lsn.push_back((heap_inflight_lsn_t){ .oid = oid, .flags = flags }); } else { - if (wr->lsn > next_inf) - inflight_lsn.resize(wr->lsn-first_inflight_lsn+1); - inflight_lsn[wr->lsn-first_inflight_lsn] = (heap_inflight_lsn_t){ .oid = oid, .flags = flags }; + if (lsn > next_inf) + { + inflight_lsn.resize(lsn-first_inflight_lsn+1, (heap_inflight_lsn_t){ .flags = HEAP_INFLIGHT_DONE }); + } + inflight_lsn[lsn-first_inflight_lsn] = (heap_inflight_lsn_t){ .oid = oid, .flags = flags }; } } -void blockstore_heap_t::complete_lsn(uint64_t lsn) +void blockstore_heap_t::mark_lsn_completed(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) + auto & item = inflight_lsn[lsn - first_inflight_lsn]; + assert(!(item.flags & HEAP_INFLIGHT_DONE)); + item.flags |= HEAP_INFLIGHT_DONE; + if (lsn == compacted_lsn+1 && !(item.flags & HEAP_INFLIGHT_COMPACTABLE)) { - while (inflight_lsn.size() && (inflight_lsn[0].flags & HEAP_INFLIGHT_DONE)) + assert(compacted_lsn+1 >= first_inflight_lsn); + while (compacted_lsn+1-first_inflight_lsn < inflight_lsn.size() && + (inflight_lsn[compacted_lsn+1-first_inflight_lsn].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++; + compacted_lsn++; } - completed_lsn = first_inflight_lsn-1; + assert(inflight_lsn[compacted_lsn-first_inflight_lsn].flags == HEAP_INFLIGHT_DONE); + } + if (lsn > completed_lsn) + { + assert(completed_lsn+1 >= first_inflight_lsn); + while (completed_lsn+1-first_inflight_lsn < inflight_lsn.size() && + (inflight_lsn[completed_lsn+1-first_inflight_lsn].flags & HEAP_INFLIGHT_DONE)) + { + completed_lsn++; + } + } +} + +void blockstore_heap_t::mark_lsn_compacted(uint64_t lsn) +{ + assert(lsn >= first_inflight_lsn && lsn < first_inflight_lsn+inflight_lsn.size()); + auto & item = inflight_lsn[lsn - first_inflight_lsn]; + assert(item.flags & HEAP_INFLIGHT_DONE); + if (!(item.flags & HEAP_INFLIGHT_COMPACTABLE)) + return; + item.flags -= HEAP_INFLIGHT_COMPACTABLE; + to_compact_count--; + if (lsn == compacted_lsn+1) + { + assert(completed_lsn+1 >= first_inflight_lsn); + while (compacted_lsn+1-first_inflight_lsn < inflight_lsn.size() && + (inflight_lsn[compacted_lsn+1-first_inflight_lsn].flags == HEAP_INFLIGHT_DONE)) + { + compacted_lsn++; + } + } +} + +void blockstore_heap_t::mark_object_compacted(heap_object_t *obj, uint64_t max_lsn) +{ + for (auto wr = obj->get_writes(); wr; wr = wr->next()) + { + if (wr->is_compacted(max_lsn)) + { + mark_lsn_compacted(wr->lsn); + } + } +} + +void blockstore_heap_t::mark_lsn_trimmed(uint64_t lsn) +{ + assert(lsn >= first_inflight_lsn && lsn < first_inflight_lsn+inflight_lsn.size()); + while (first_inflight_lsn <= lsn && inflight_lsn.size() > 0) + { + assert(inflight_lsn[0].flags == HEAP_INFLIGHT_DONE); + // FIXME don't touch unflushable lsns + compact_object(inflight_lsn[0].oid, lsn, NULL); // FIXME from prev trimmed lsn + inflight_lsn.pop_front(); + first_inflight_lsn++; } } @@ -1971,8 +2061,3 @@ 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); -} diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index 33a45001..1a3fe64f 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -84,6 +84,13 @@ inline bool operator < (const heap_object_lsn_t & a, const heap_object_lsn_t & b return a.oid < b.oid || a.oid == b.oid && a.lsn < b.lsn; } +struct tmp_compact_item_t +{ + object_id oid; + uint64_t lsn; + bool compact; +}; + struct heap_object_mvcc_t { uint32_t readers = 0; @@ -136,11 +143,9 @@ class blockstore_heap_t const int meta_alloc_buckets = 4; uint64_t next_lsn = 0; - uint64_t compacted_lsn = 0; std::map pool_shard_settings; // PG => inode => stripe => block number std::map>> block_index; - std::deque compact_queue; std::vector block_info; allocator_t *data_alloc = NULL; allocator_t *meta_allocs[4] = {}; @@ -154,11 +159,15 @@ class blockstore_heap_t uint64_t buffer_area_used_space = 0; uint64_t data_used_space = 0; + // LSN queue: inflight (writing) -> completed [-> fsynced] -> compactable -> compacted [-> fsynced] -> trimmed and removed std::deque inflight_lsn; - uint64_t first_inflight_lsn = 1; + uint32_t to_compact_count = 0; + uint64_t first_inflight_lsn = 0; uint64_t completed_lsn = 0; + uint64_t compacted_lsn = 0; + uint64_t next_compact_lsn = 0; - std::vector tmp_compact_queue; + std::vector tmp_compact_queue; std::deque recheck_queue; int recheck_in_progress = 0; bool in_recheck = false; @@ -180,7 +189,7 @@ class blockstore_heap_t void erase_block_index(inode_t inode, uint64_t stripe); void free_object_space(inode_t inode, heap_write_t *from, heap_write_t *to, int mode = 0); void add_used_space(uint32_t block_num, int32_t used_delta); - void push_inflight_lsn(object_id oid, heap_write_t *wr); + void push_inflight_lsn(object_id oid, uint64_t lsn, uint64_t flags); public: blockstore_heap_t(blockstore_disk_t *dsk, uint8_t *buffer_area, int log_level = 0); @@ -243,9 +252,11 @@ public: int get_block_for_new_object(uint32_t & out_block_num); // inflight write tracking - void complete_lsn(uint64_t lsn); + void mark_lsn_completed(uint64_t lsn); + void mark_lsn_compacted(uint64_t lsn); + void mark_object_compacted(heap_object_t *obj, uint64_t max_lsn); + void mark_lsn_trimmed(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(); @@ -269,6 +280,7 @@ public: uint64_t get_meta_total_space(); uint64_t get_meta_used_space(); uint32_t get_meta_nearfull_blocks(); + uint32_t get_inflight_queue_size(); uint32_t get_compact_queue_size(); // get maximum size for a temporary heap_write_t buffer diff --git a/src/blockstore/blockstore_impl.cpp b/src/blockstore/blockstore_impl.cpp index 197000eb..bfde937b 100644 --- a/src/blockstore/blockstore_impl.cpp +++ b/src/blockstore/blockstore_impl.cpp @@ -244,7 +244,7 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op) } else if (PRIV(op)->wait_for == WAIT_COMPACTION) { - if (flusher->get_counter() <= PRIV(op)->wait_detail) + if (flusher->get_compact_counter() <= PRIV(op)->wait_detail) { // do not submit #ifdef BLOCKSTORE_DEBUG diff --git a/src/blockstore/blockstore_init.cpp b/src/blockstore/blockstore_init.cpp index 6acf8ad0..2c9dadf7 100644 --- a/src/blockstore/blockstore_init.cpp +++ b/src/blockstore/blockstore_init.cpp @@ -117,7 +117,7 @@ resume_1: } else { - blockstore_meta_header_v2_t *hdr = (blockstore_meta_header_v2_t *)bs->meta_superblock; + blockstore_meta_header_v3_t *hdr = (blockstore_meta_header_v3_t *)bs->meta_superblock; if (hdr->zero != 0 || hdr->magic != BLOCKSTORE_META_MAGIC_V1 || hdr->version < BLOCKSTORE_META_FORMAT_V1) { printf( @@ -153,6 +153,7 @@ resume_1: ); exit(1); } + bs->heap->set_compacted_lsn(hdr->compacted_lsn); } if (bs->dsk.inmemory_journal) { diff --git a/src/blockstore/blockstore_stable.cpp b/src/blockstore/blockstore_stable.cpp index a0038232..0d506b6b 100644 --- a/src/blockstore/blockstore_stable.cpp +++ b/src/blockstore/blockstore_stable.cpp @@ -86,7 +86,7 @@ resume_4: // FIXME: Also mark as fsynced for (uint64_t lsn = priv->lsn; lsn <= priv->to_lsn; lsn++) { - heap->complete_lsn(lsn); + heap->mark_lsn_completed(lsn); } // Done. Don't touch op->retval - if anything resulted in ENOENT, return it as is FINISH_OP(op); diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index 381f93cd..f149da0f 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -103,14 +103,14 @@ 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() && !write_iodepth && !flusher->get_active()) + if (!heap->get_inflight_queue_size()) { // no space cancel_all_writes(op, -ENOSPC); return 2; } PRIV(op)->wait_for = WAIT_COMPACTION; - PRIV(op)->wait_detail = flusher->get_counter(); + PRIV(op)->wait_detail = flusher->get_compact_counter(); flusher->request_trim(); return 0; } @@ -179,14 +179,14 @@ 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() && !write_iodepth && !flusher->get_active()) + if (!heap->get_inflight_queue_size()) { // no space cancel_all_writes(op, -ENOSPC); return 2; } PRIV(op)->wait_for = WAIT_COMPACTION; - PRIV(op)->wait_detail = flusher->get_counter(); + PRIV(op)->wait_detail = flusher->get_compact_counter(); flusher->request_trim(); return 0; } @@ -205,7 +205,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) if (loc == UINT64_MAX) { PRIV(op)->wait_for = WAIT_COMPACTION; - PRIV(op)->wait_detail = flusher->get_counter(); + PRIV(op)->wait_detail = flusher->get_compact_counter(); flusher->request_trim(); return 0; } @@ -226,14 +226,14 @@ 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() && !write_iodepth && !flusher->get_active()) + if (!heap->get_inflight_queue_size()) { // no space cancel_all_writes(op, -ENOSPC); return 2; } PRIV(op)->wait_for = WAIT_COMPACTION; - PRIV(op)->wait_detail = flusher->get_counter(); + PRIV(op)->wait_detail = flusher->get_compact_counter(); flusher->request_trim(); return 0; } @@ -315,7 +315,7 @@ resume_4: if (res == ENOSPC) { PRIV(op)->wait_for = WAIT_COMPACTION; - PRIV(op)->wait_detail = flusher->get_counter(); + PRIV(op)->wait_detail = flusher->get_compact_counter(); return 1; } assert(res == 0); @@ -364,7 +364,7 @@ resume_6: resume_8: // Acknowledge write op->retval = op->len; - heap->complete_lsn(PRIV(op)->lsn); + heap->mark_lsn_completed(PRIV(op)->lsn); write_iodepth--; FINISH_OP(op); return 2; diff --git a/src/test/test_heap.cpp b/src/test/test_heap.cpp index ba7ea966..0daa3ab7 100644 --- a/src/test/test_heap.cpp +++ b/src/test/test_heap.cpp @@ -403,6 +403,7 @@ void test_compact(bool csum, bool stable) _test_small_write(heap, dsk, 1, 0, 3, 8192, 4096, 16384, stable); obj = heap.read_entry(oid, NULL); uint64_t wr_size = obj->get_writes()->get_size(&heap); + assert(obj->get_writes()->lsn == 2); assert(check_used_space(heap, dsk, 0)); assert(heap.get_meta_block_used_space(0) == old_size + wr_size); @@ -412,6 +413,10 @@ void test_compact(bool csum, bool stable) assert(obj); assert(count_writes(obj) == 2); + heap.mark_lsn_completed(1); + heap.mark_lsn_completed(2); + heap.mark_lsn_completed(3); + uint32_t mblock; object_id compact_oid = {}; if (!stable) @@ -435,9 +440,9 @@ void test_compact(bool csum, bool stable) assert(new_to_lsn == 4); assert(check_used_space(heap, dsk, 0)); assert(heap.get_meta_block_used_space(0) == 2*old_size + wr_size); + heap.mark_lsn_completed(4); } - heap.add_to_compact_queue(oid); assert(heap.get_compact_queue_size() == 1); res = heap.get_next_compact(compact_oid); assert(res == 0); @@ -447,15 +452,18 @@ void test_compact(bool csum, bool stable) obj = heap.read_entry(oid, NULL); assert(obj); assert(count_writes(obj) == 2); - heap.get_compact_range(obj, UINT64_MAX, &compact_begin, &compact_end); + heap.get_compact_range(obj, 4, &compact_begin, &compact_end); assert(compact_begin == obj->get_writes()); assert(compact_end == obj->get_writes()->next()); + heap.mark_object_compacted(obj, 4); + assert(heap.get_compacted_lsn() == (stable ? 3 : 4)); + assert(heap.get_compact_queue_size() == 0); + res = heap.compact_object((object_id){ .inode = INODE_WITH_POOL(1, 3), .stripe = 0 }, compact_begin->lsn, NULL); assert(res == ENOENT); - res = heap.compact_object(compact_oid, compact_begin->lsn, NULL); - assert(res == 0); + heap.mark_lsn_trimmed((stable ? 3 : 4)); assert(check_used_space(heap, dsk, 0)); assert(heap.get_meta_block_used_space(0) == 2*old_size);