diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index 205dd8d3..9e89bd1c 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -177,7 +177,7 @@ resume_0: } resume_1: should_repeat = false; - cur_obj = bs->heap->lock_and_read_entry(cur_oid, cur_lsn); + cur_obj = bs->heap->lock_and_read_entry(cur_oid, copy_id); if (!cur_obj) { // Object does not exist @@ -185,14 +185,18 @@ resume_1: } cur_version = cur_obj->get_writes()->version; // Find the range to compact - bs->heap->get_compact_range(cur_obj, cur_lsn, &begin_wr, &end_wr); + compact_lsn = bs->heap->get_completed_lsn(); + if (auto cm_it = bs->committing_lsn.find(cur_oid); cm_it != bs->committing_lsn.end()) + { + compact_lsn = cm_it->second; + } + bs->heap->get_compact_range(cur_obj, compact_lsn, &begin_wr, &end_wr); if (!begin_wr) { // Nothing to flush - bs->heap->unlock_entry(cur_oid, cur_lsn); + bs->heap->unlock_entry(cur_oid, copy_id); goto resume_0; } - compact_lsn = begin_wr->lsn; assert(!end_wr->next() && end_wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)); clean_loc = end_wr->location; #ifdef BLOCKSTORE_DEBUG @@ -277,10 +281,10 @@ resume_12: resume_13: if (copy_count && !fsync_batch(false, 11)) return false; - bs->heap->unlock_entry(cur_oid, cur_lsn); + 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 calc_block_checksums(); - bs->heap->compact_object(cur_oid, cur_lsn, new_data_csums); + bs->heap->compact_object(cur_oid, compact_lsn, new_data_csums); // Done, free all buffers free_buffers(); #ifdef BLOCKSTORE_DEBUG @@ -398,7 +402,7 @@ int journal_flusher_co::check_and_punch_checksums() return 0; } // Verify data checksums - cur_obj = bs->heap->read_locked_entry(cur_oid, cur_lsn); + cur_obj = bs->heap->read_locked_entry(cur_oid, copy_id); bool csum_ok = true; for (int i = 0; i < read_vec.size(); i++) { @@ -436,7 +440,7 @@ int journal_flusher_co::check_and_punch_checksums() // Object is deleted, abort compaction return ENOENT; } - bs->heap->get_compact_range(cur_obj, cur_lsn, &begin_wr, &end_wr); + bs->heap->get_compact_range(cur_obj, compact_lsn, &begin_wr, &end_wr); if (!begin_wr || begin_wr->lsn != compact_lsn) { // Object is overwritten, abort compaction @@ -476,7 +480,7 @@ 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, cur_lsn); + 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; diff --git a/src/blockstore/blockstore_flush.h b/src/blockstore/blockstore_flush.h index aec9245a..69cac4f1 100644 --- a/src/blockstore/blockstore_flush.h +++ b/src/blockstore/blockstore_flush.h @@ -49,7 +49,7 @@ class journal_flusher_co std::function simple_callback_r, simple_callback_w; object_id cur_oid; - uint64_t cur_lsn; + uint64_t copy_id; uint64_t compact_lsn; uint64_t min_compact_lsn; uint64_t cur_version; diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index 20caf611..dc9aced9 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -12,6 +12,9 @@ #define BS_HEAP_FREE_MAIN 2 #define FREE_SPACE_BIT 0x8000 +#define HEAP_INFLIGHT_DONE 1 +#define HEAP_INFLIGHT_COMPACTABLE 2 + heap_write_t *heap_write_t::next() { return (next_pos ? (heap_write_t*)((uint8_t*)this + next_pos) : NULL); @@ -584,7 +587,7 @@ skip_object: } if (wr->needs_compact(this->compacted_lsn)) { - compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn }); + tmp_compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn }); } else if (wr->is_compacted(this->compacted_lsn)) { @@ -596,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 - compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn }); + tmp_compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn }); } } } @@ -640,10 +643,15 @@ skip_object: void blockstore_heap_t::finish_load() { - std::sort(compact_queue.begin(), compact_queue.end(), [this](const heap_object_lsn_t & a, const heap_object_lsn_t & b) + std::sort(tmp_compact_queue.begin(), tmp_compact_queue.end(), [this](const heap_object_lsn_t & a, const heap_object_lsn_t & b) { return a.lsn < b.lsn; }); + for (auto & e: tmp_compact_queue) + { + compact_queue.push_back(e.oid); + } + tmp_compact_queue.clear(); } bool blockstore_heap_t::calc_checksums(heap_write_t *wr, uint8_t *data, bool set) @@ -1241,16 +1249,14 @@ int blockstore_heap_t::add_object(object_id oid, heap_write_t *wr, uint32_t *mod new_wr->next_pos = 0; new_wr->size = wr_size; new_wr->lsn = ++next_lsn; + wr->lsn = new_wr->lsn; + push_inflight_lsn(new_wr->lsn, oid, 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); memset(int_bitmap, 0, dsk->clean_entry_bitmap_size); bitmap_set(int_bitmap, wr->offset, wr->len, dsk->bitmap_granularity); } - if (wr->needs_compact(0)) - { - compact_queue.push_back({ .oid = oid, .lsn = new_wr->lsn }); - } new_entry->size = sizeof(heap_object_t); new_entry->crc32c = new_entry->calc_crc32c(); add_used_space(block_num, sizeof(heap_object_t) + wr_size); @@ -1402,6 +1408,8 @@ 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(new_wr->lsn, oid, 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); @@ -1410,11 +1418,6 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea } obj->write_pos = offset - ((uint8_t*)obj - inf.data); obj->crc32c = obj->calc_crc32c(); - // Add to compaction queue - if (new_wr->needs_compact(0)) - { - compact_queue.push_back({ .oid = oid, .lsn = new_wr->lsn }); - } // Change block free space add_used_space(block_num, used_delta); return 0; @@ -1431,7 +1434,7 @@ int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *mod return update_object(block_num, obj, wr, modified_block); } -int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block) +int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *before_compact_lsn, uint64_t *to_compact_lsn) { uint32_t block_num = 0; heap_object_t *obj = read_entry(oid, &block_num); @@ -1445,6 +1448,7 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t heap_write_t *unstable_wr = NULL; heap_write_t *unstable_big_wr = NULL; heap_write_t *wr = obj->get_writes(); + heap_write_t *pre_wr = NULL; if (wr->version < version) { // No such version @@ -1452,7 +1456,12 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t } for (; wr; wr = wr->next()) { - if (!(wr->flags & BS_HEAP_STABLE) && wr->version <= version) + if ((wr->flags & BS_HEAP_STABLE)) + { + pre_wr = wr; + break; + } + else if (wr->version <= version) { unstable_wr = wr; if (!unstable_big_wr && @@ -1472,6 +1481,10 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t { *modified_block = block_num; } + if (before_compact_lsn && pre_wr && pre_wr->needs_compact(0)) + { + *before_compact_lsn = pre_wr->lsn; + } // Save a copy of the object mvcc_save_copy(obj); if (unstable_big_wr && unstable_big_wr->next()) @@ -1494,9 +1507,9 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t to_compact = wr->lsn; } } - if (to_compact) + if (to_compact_lsn) { - compact_queue.push_back({ .oid = oid, .lsn = to_compact }); + *to_compact_lsn = to_compact; } obj->crc32c = obj->calc_crc32c(); return 0; @@ -1577,7 +1590,7 @@ int blockstore_heap_t::get_next_compact(object_id & oid) { while (compact_queue.size()) { - oid = compact_queue.front().oid; + oid = compact_queue.front(); compact_queue.pop_front(); return 0; } @@ -1897,3 +1910,46 @@ void blockstore_heap_t::set_fail_on_warn(bool fail) { fail_on_warn = fail; } + +void blockstore_heap_t::push_inflight_lsn(uint64_t lsn, object_id oid, uint64_t flags) +{ + if (!inflight_lsn.size()) + { + first_inflight_lsn = lsn; + } + else + { + assert(lsn == first_inflight_lsn+inflight_lsn.size()); + } + inflight_lsn.push_back((heap_inflight_lsn_t){ .oid = oid, .flags = flags }); +} + +void blockstore_heap_t::complete_lsn(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) + { + while (inflight_lsn.size() && (inflight_lsn[0].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++; + } + completed_lsn = first_inflight_lsn-1; + } +} + +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 065fd27f..12e6a6df 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -97,6 +97,12 @@ struct __attribute__((__packed__)) heap_block_info_t uint8_t *data = NULL; }; +struct heap_inflight_lsn_t +{ + object_id oid; + uint64_t flags; +}; + struct multilist_alloc_t { const uint32_t count, maxn; @@ -148,6 +154,11 @@ class blockstore_heap_t uint64_t buffer_area_used_space = 0; uint64_t data_used_space = 0; + std::deque inflight_lsn; + uint64_t first_inflight_lsn = 0; + uint64_t completed_lsn = 0; + + std::vector tmp_compact_queue; std::deque recheck_queue; int recheck_in_progress = 0; bool in_recheck = false; @@ -209,7 +220,7 @@ public: int post_write(object_id oid, heap_write_t *wr, uint32_t *modified_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); + int post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *before_compact_lsn, uint64_t *to_compact_lsn); // rollback an unstable object version // return 0 if OK, ENOENT if not exists, EBUSY if already stable int post_rollback(object_id oid, uint64_t version, uint32_t *modified_block); @@ -230,6 +241,12 @@ public: // set a block number for a new object and returns error status: 0, EAGAIN or ENOSPC int get_block_for_new_object(uint32_t & out_block_num); + // inflight write tracking + void push_inflight_lsn(uint64_t lsn, object_id oid, uint64_t flags); + void complete_lsn(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(); bool is_data_used(uint64_t location); diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index 4ccd5ef2..1956168c 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -53,8 +53,10 @@ struct blockstore_op_private_t int pending_ops; int op_state; - // Read + // Read, write uint64_t lsn; + + // Read std::vector read_vec; // Write @@ -63,6 +65,7 @@ struct blockstore_op_private_t // Stabilize/rollback int stab_pos; + std::vector to_compact; // Write struct iovec iov_zerofill[3]; @@ -114,6 +117,7 @@ class blockstore_impl_t: public blockstore_i journal_flusher_t *flusher; int write_iodepth = 0; + std::unordered_map committing_lsn; bool live = false, queue_stall = false; ring_loop_t *ringloop; diff --git a/src/blockstore/blockstore_stable.cpp b/src/blockstore/blockstore_stable.cpp index 7d5b82f7..3e08a325 100644 --- a/src/blockstore/blockstore_stable.cpp +++ b/src/blockstore/blockstore_stable.cpp @@ -16,6 +16,7 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op) assert(!priv->op_state); priv->stab_pos = 0; op->retval = 0; + priv->to_compact.resize(op->len); while (priv->stab_pos < op->len) { io_uring_sqe *sqe = get_sqe(); @@ -27,9 +28,11 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op) priv->wait_for = WAIT_SQE; return 0; } - uint32_t modified_block; + uint32_t modified_block = 0; + uint64_t before_compact = 0; + uint64_t to_compact = 0; int res = op->opcode == BS_OP_STABLE - ? heap->post_stabilize(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block) + ? heap->post_stabilize(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block, &before_compact, &to_compact) : heap->post_rollback(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block); if (res != 0) { @@ -37,8 +40,16 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op) FINISH_OP(op); return 2; } - prepare_meta_block_write(op, modified_block); - priv->pending_ops++; + if (modified_block) + { + if (to_compact) + { + priv->to_compact[priv->stab_pos] = true; + committing_lsn[v[priv->stab_pos].oid] = before_compact; + } + prepare_meta_block_write(op, modified_block); + priv->pending_ops++; + } priv->stab_pos++; } resume_1: @@ -63,6 +74,18 @@ resume_3: return 0; } resume_4: + for (int i = 0; i < op->len; i++) + { + if (priv->to_compact[i]) + { + // Add to compact queue only when metadata writes are finished + heap->add_to_compact_queue(v[i].oid); + auto cm_it = committing_lsn.find(v[i].oid); + assert(cm_it != committing_lsn.end()); + if (cm_it->second <= priv->to_compact[i]) + committing_lsn.erase(cm_it); + } + } // Done. Don't touch op->retval - if anything resulted in ENOENT, return it as is FINISH_OP(op); return 2; diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index db7e633e..a222497f 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -25,13 +25,10 @@ void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, int retval) found = true; } else if (found && other_op->oid == op->oid && - (other_op->opcode == BS_OP_WRITE || other_op->opcode == BS_OP_WRITE_STABLE)) + (other_op->opcode == BS_OP_WRITE || other_op->opcode == BS_OP_WRITE_STABLE) && + !PRIV(other_op)->op_state) { // Mark operations to cancel them - if (PRIV(other_op)->op_state != 0 && PRIV(other_op)->op_state != 100) - { - write_iodepth--; - } PRIV(other_op)->op_state = 100; other_op->retval = retval; } @@ -103,7 +100,7 @@ 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() && !flusher->get_active()) + if (!heap->get_compact_queue_size() && !write_iodepth && !flusher->get_active()) { // no space cancel_all_writes(op, -ENOSPC); @@ -179,7 +176,7 @@ 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() && !flusher->get_active()) + if (!heap->get_compact_queue_size() && !write_iodepth && !flusher->get_active()) { // no space cancel_all_writes(op, -ENOSPC); @@ -191,6 +188,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) return 0; } assert(res == 0); + PRIV(op)->lsn = wr->lsn; prepare_meta_block_write(op, modified_block); unsynced_small_write_count++; PRIV(op)->op_state = 9; @@ -225,7 +223,7 @@ 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() && !flusher->get_active()) + if (!heap->get_compact_queue_size() && !write_iodepth && !flusher->get_active()) { // no space cancel_all_writes(op, -ENOSPC); @@ -237,6 +235,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) return 0; } assert(res == 0); + PRIV(op)->lsn = wr->lsn; heap->use_buffer_area(op->oid.inode, loc, op->len); prepare_meta_block_write(op, modified_block); if (op->len > 0) @@ -317,6 +316,7 @@ resume_4: return 1; } assert(res == 0); + PRIV(op)->lsn = wr->lsn; prepare_meta_block_write(op, modified_block); PRIV(op)->op_state = 5; return 1; @@ -361,6 +361,7 @@ resume_6: resume_8: // Acknowledge write op->retval = op->len; + heap->complete_lsn(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 5154dd6c..a0d9c76d 100644 --- a/src/test/test_heap.cpp +++ b/src/test/test_heap.cpp @@ -434,19 +434,25 @@ void test_compact(bool csum, bool stable) res = heap.get_next_compact(compact_oid); assert(res == ENOENT); - res = heap.post_stabilize({ .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }, 3, NULL); + uint64_t to_lsn = 0, before_lsn = 0; + res = heap.post_stabilize({ .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }, 3, NULL, &before_lsn, &to_lsn); assert(res == ENOENT); - res = heap.post_stabilize(oid, 5, NULL); + res = heap.post_stabilize(oid, 5, NULL, &before_lsn, &to_lsn); assert(res == ENOENT); - res = heap.post_stabilize(oid, 1, &mblock); + res = heap.post_stabilize(oid, 1, &mblock, &before_lsn, &to_lsn); assert(res == 0); - res = heap.post_stabilize(oid, 3, &mblock); + assert(before_lsn == 0); + assert(to_lsn == 0); + res = heap.post_stabilize(oid, 3, &mblock, &before_lsn, &to_lsn); assert(res == 0); assert(mblock == 0); + assert(before_lsn == 0); + assert(to_lsn == 2); assert(check_used_space(heap, dsk, 0)); assert(heap.get_meta_block_used_space(0) == 2*old_size + wr_size); } + heap.add_to_compact_queue(oid); assert(heap.get_compact_queue_size() == 1); res = heap.get_next_compact(compact_oid); assert(res == 0); @@ -762,7 +768,7 @@ void test_full_overwrite(bool stable) if (!stable) { - res = heap.post_stabilize(oid, 4, NULL); + res = heap.post_stabilize(oid, 4, NULL, NULL, NULL); assert(res == 0); }