diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index ddcf4775..03abd160 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -24,6 +24,11 @@ journal_flusher_t::journal_flusher_t(blockstore_impl_t *bs) { co[i].co_id = i; co[i].bs = bs; + if (bs->dsk.csum_block_size > bs->dsk.bitmap_granularity) + { + co[i].new_csums = (uint8_t*)malloc_or_die(bs->dsk.data_block_size / bs->dsk.csum_block_size * (bs->dsk.data_csum_type & 0xFF)); + co[i].new_bmp = (uint8_t*)malloc_or_die(bs->dsk.clean_entry_bitmap_size); + } co[i].flusher = this; } } @@ -54,6 +59,16 @@ journal_flusher_t::~journal_flusher_t() journal_flusher_co::~journal_flusher_co() { + if (new_csums) + { + free(new_csums); + new_csums = NULL; + } + if (new_bmp) + { + free(new_bmp); + new_bmp = NULL; + } free_buffers(); } @@ -151,14 +166,6 @@ bool journal_flusher_co::loop() else if (wait_state == 14) goto resume_14; 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; - else if (wait_state == 22) goto resume_22; - else if (wait_state == 23) goto resume_23; - else if (wait_state == 24) goto resume_24; resume_0: wait_state = 0; wait_count = 0; @@ -168,10 +175,10 @@ resume_0: (!bs->dsk.disable_journal_fsync || !bs->dsk.disable_meta_fsync)) { flusher->active_flushers++; -resume_21: -resume_22: - res = fsync_buffer(21); - if (!res) +resume_14: +resume_15: +resume_16: + if (!fsync_buffer(14)) { return false; } @@ -180,18 +187,6 @@ resume_22: } if (res == ENOENT) { - if (co_id == 0 && flusher->force_start > 0) - { - flusher->active_flushers++; -resume_16: -resume_17: -resume_18: -resume_19: -resume_20: - if (!trim_lsn(16)) - return false; - flusher->active_flushers--; - } cur_oid = {}; wait_state = 0; return true; @@ -206,35 +201,37 @@ resume_20: } } resume_1: + wait_state = 1; should_repeat = false; - cur_obj = bs->heap->lock_and_read_entry(cur_oid, copy_id); + cur_obj = bs->heap->lock_and_read_entry(cur_oid); if (!cur_obj) { // Object does not exist goto resume_0; } - cur_version = cur_obj->get_writes()->version; - // Find the range to compact - compact_lsn = bs->heap->get_fsynced_lsn(); - 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, copy_id); - goto resume_0; - } - assert(!end_wr->next() && end_wr->entry_type == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)); - clean_loc = end_wr->big_location(bs->heap); - if (bs->log_level > 10) - printf("Compacting %jx:%jx l%ju .. l%ju (last l%ju)\n", cur_oid.inode, cur_oid.stripe, end_wr->lsn, begin_wr->lsn, compact_lsn); - flusher->active_flushers++; // Scan versions to flush free_buffers(); copy_count = 0; - for (auto wr = begin_wr; wr != end_wr; wr = wr->next()) + fsynced_lsn = bs->heap->get_fsynced_lsn(); + compact_info = bs->heap->iterate_compaction(cur_obj, fsynced_lsn, flusher->force_start, [&](heap_entry_t *wr) { - bs->prepare_read(read_vec, cur_obj, wr, 0, bs->dsk.data_block_size); - copy_count++; + if (wr->type() == BS_HEAP_SMALL_WRITE || + wr->type() == BS_HEAP_INTENT_WRITE && bs->dsk.csum_block_size > bs->dsk.bitmap_granularity) + { + bs->prepare_read(read_vec, cur_obj, wr, 0, bs->dsk.data_block_size); + copy_count++; + } + }); + if (!compact_info.compact_lsn) + { + // Flushing is aborted + bs->heap->unlock_entry(cur_oid); + goto resume_0; + } + flusher->active_flushers++; + if (bs->log_level > 10) + { + printf("Compacting %jx:%jx l%ju .. l%ju\n", cur_oid.inode, cur_oid.stripe, compact_info.clean_lsn, compact_info.compact_lsn); } overwrite_start = overwrite_end = 0; if (read_vec.size() > 0) @@ -254,7 +251,6 @@ resume_1: } // Read buffered data cur_obj = NULL; - begin_wr = end_wr = NULL; resume_2: resume_3: if (!read_buffered(2)) @@ -272,29 +268,45 @@ resume_3: flusher->wanting_meta_fsync--; } res = check_and_punch_checksums(); - if (res == EBUSY) - { -resume_4: -resume_5: - if (!write_meta_block(4)) - { - return false; - } -resume_6: -resume_7: -resume_8: - if (!fsync_meta(6)) - { - return false; - } - res = 0; - } - else if (res == ENOENT || res == EDOM) + if (res == ENOENT || res == EDOM) { // Abort compaction flusher->active_flushers--; goto resume_0; } + if (res == EBUSY) + { +resume_4: + modified_block = UINT32_MAX; + res = bs->heap->add_punch_holes(cur_obj, compact_info.clean_lsn, compact_info.clean_version, new_bmp, new_csums, &modified_block); + if (res == ENOENT) + { + // Abort compaction + flusher->active_flushers--; + goto resume_0; + } + if (res == EAGAIN) + { + // Retry, block is busy + wait_state = 4; + return false; + } + assert(res == 0); +resume_5: +resume_6: + if (!write_meta_block(5)) + { + return false; + } +resume_7: +resume_8: +resume_9: + if (!fsync_meta(7)) + { + return false; + } + res = 0; + } assert(res == 0); // Submit data writes for (i = 0; i < read_vec.size(); i++) @@ -304,26 +316,26 @@ resume_8: (read_vec[i].copy_flags & COPY_BUF_PADDED)) // FIXME Shit, simplify these flags { assert(read_vec[i].buf); - await_sqe(9); + await_sqe(10); data->iov = (struct iovec){ read_vec[i].buf + (read_vec[i].copy_flags & COPY_BUF_PADDED ? read_vec[i].offset - read_vec[i].disk_offset : 0), (size_t)read_vec[i].len }; data->callback = simple_callback_w; - io_uring_prep_writev(sqe, bs->dsk.data_fd, &data->iov, 1, bs->dsk.data_offset + clean_loc + read_vec[i].offset); + io_uring_prep_writev(sqe, bs->dsk.data_fd, &data->iov, 1, bs->dsk.data_offset + compact_info.clean_loc + read_vec[i].offset); wait_count++; } } -resume_10: +resume_11: if (wait_count > 0) { - wait_state = 10; + wait_state = 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); + bs->heap->unlock_entry(cur_oid); // 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, &modified_block); + cur_obj = bs->heap->read_entry(cur_oid); if (!cur_obj) { // Abort compaction @@ -334,40 +346,20 @@ resume_10: // Abort compaction goto resume_0; } - if (read_to_fill_incomplete) - { -resume_23: -resume_24: - if (!write_meta_block(23)) - { - return false; - } - // Recheck the object because it could be invalidated again - cur_obj = bs->heap->read_entry(cur_oid, &modified_block); - if (!cur_obj) - { - // Abort compaction - goto resume_0; - } - } - bs->heap->mark_object_compacted(cur_obj, compact_lsn); - // Done - if (bs->log_level > 10) - printf("Compacted %jx:%jx l%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, compact_lsn, copy_count); - flusher->compact_counter++; - flusher->active_flushers--; - // Advance compacted_lsn every objects - if (co_id == 0 && !((++flusher->advance_lsn_counter) % bs->journal_trim_interval)) - { - flusher->advance_lsn_counter = 0; -resume_11: + bs->heap->add_compact(cur_obj, compact_info.compact_lsn, &modified_block, new_csums); resume_12: resume_13: -resume_14: -resume_15: - if (!trim_lsn(11)) - return false; + if (!write_meta_block(12)) + { + return false; } + // Done + if (bs->log_level > 10) + { + printf("Compacted %jx:%jx l%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, compact_info.compact_lsn, copy_count); + } + flusher->compact_counter++; + flusher->active_flushers--; if (should_repeat) { // Flush the same object again @@ -409,11 +401,11 @@ void journal_flusher_co::fill_partial_checksum_blocks() .copy_flags = COPY_BUF_DATA | copy_flags, .offset = blk_begin, .len = blk_end - blk_begin, - .disk_loc = end_wr->big_location(bs->heap), + .disk_loc = compact_info.clean_loc, .disk_offset = blk_begin, .disk_len = blk_end - blk_begin, .buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, blk_end - blk_begin), - .wr_lsn = end_wr->lsn, + .wr_lsn = compact_info.clean_lsn, }); } auto & vec = read_vec[read_vec.size()-1]; @@ -451,16 +443,18 @@ int journal_flusher_co::check_and_punch_checksums() return 0; } // Verify data checksums - cur_obj = bs->heap->read_locked_entry(cur_oid, copy_id); + cur_obj = bs->heap->read_entry(cur_oid); bool csum_ok = true; for (int i = 0; i < read_vec.size(); i++) { auto & vec = read_vec[i]; if (!(vec.copy_flags & (COPY_BUF_COALESCED|COPY_BUF_ZERO|COPY_BUF_SKIP_CSUM))) { - heap_write_t *wr = cur_obj->get_writes(); + heap_entry_t *wr = cur_obj; while (wr && wr->lsn != vec.wr_lsn) - wr = wr->next(); + { + wr = bs->heap->prev(wr); + } assert(wr); uint32_t *csums = (uint32_t*)(wr->get_checksums(bs->heap) + (vec.disk_offset/bs->dsk.csum_block_size)*(bs->dsk.data_csum_type & 0xFF) @@ -489,20 +483,30 @@ int journal_flusher_co::check_and_punch_checksums() // Nothing to do return 0; } - cur_obj = bs->heap->read_entry(cur_oid, &modified_block); + cur_obj = bs->heap->read_entry(cur_oid); if (!cur_obj) { // Object is deleted, abort compaction return ENOENT; } - bs->heap->get_compact_range(cur_obj, compact_lsn, &begin_wr, &end_wr); - if (!begin_wr) + heap_entry_t *clean_wr = NULL; + for (auto wr = cur_obj; wr; wr = bs->heap->prev(wr)) { - // Object is overwritten, abort compaction - return ENOENT; + if (wr->is_overwrite() && wr->lsn > compact_info.clean_lsn && + wr->lsn <= fsynced_lsn) + { + // Object is overwritten, abort compaction + return ENOENT; + } + if (wr->lsn == compact_info.clean_lsn) + { + clean_wr = wr; + break; + } } - uint8_t *bmp = end_wr->get_int_bitmap(bs->heap); - uint8_t *csums = end_wr->get_checksums(bs->heap); + assert(clean_wr); + memcpy(new_bmp, clean_wr->get_int_bitmap(bs->heap), bs->dsk.clean_entry_bitmap_size); + memcpy(new_csums, clean_wr->get_checksums(bs->heap), bs->dsk.data_block_size/bs->dsk.csum_block_size * (bs->dsk.data_csum_type & 0xFF)); // Clear bits for (auto & vec: read_vec) { @@ -513,7 +517,7 @@ int journal_flusher_co::check_and_punch_checksums() if (!(vec.copy_flags & COPY_BUF_COALESCED) && ((vec.offset % bs->dsk.csum_block_size) || (vec.len % bs->dsk.csum_block_size))) { - bitmap_clear(bmp, vec.offset, vec.len, bs->dsk.bitmap_granularity); + bitmap_clear(new_bmp, vec.offset, vec.len, bs->dsk.bitmap_granularity); } } // Update partial block checksums @@ -522,17 +526,10 @@ int journal_flusher_co::check_and_punch_checksums() if (vec.copy_flags & COPY_BUF_CSUM_FILL) { uint32_t csum_off = vec.offset/bs->dsk.csum_block_size * (bs->dsk.data_csum_type & 0xFF); - bs->heap->calc_block_checksums((uint32_t*)(csums+csum_off), vec.buf, bmp, vec.offset, vec.offset+vec.len, true, NULL); + bs->heap->calc_block_checksums((uint32_t*)(new_csums+csum_off), vec.buf, new_bmp, vec.offset, vec.offset+vec.len, true, NULL); } } - cur_obj->crc32c = cur_obj->calc_crc32c(); - if (res == ENOENT) - { - // Object is deleted, abort compaction - return ENOENT; - } - // Modified, we should write the block to disk - assert(!res); + // Modified, we should add_punch_holes and then write the block to disk return EBUSY; } @@ -542,20 +539,30 @@ bool journal_flusher_co::calc_block_checksums() { return true; } - bs->heap->get_compact_range(cur_obj, compact_lsn, &begin_wr, &end_wr); - if (!begin_wr) + heap_entry_t *clean_wr = NULL; + for (auto wr = cur_obj; wr; wr = bs->heap->prev(wr)) { - // Object is overwritten, abort compaction - return false; + if (wr->is_overwrite() && wr->lsn > compact_info.clean_lsn && + wr->lsn <= fsynced_lsn) + { + // Object is overwritten, abort compaction + return false; + } + if (wr->lsn == compact_info.clean_lsn) + { + clean_wr = wr; + break; + } } - uint8_t *bmp = end_wr->get_int_bitmap(bs->heap); - uint8_t *csums = end_wr->get_checksums(bs->heap); + assert(clean_wr); + memcpy(new_bmp, clean_wr->get_int_bitmap(bs->heap), bs->dsk.clean_entry_bitmap_size); + memcpy(new_csums, clean_wr->get_checksums(bs->heap), bs->dsk.data_block_size/bs->dsk.csum_block_size * (bs->dsk.data_csum_type & 0xFF)); // Set bits for (auto & vec: read_vec) { if (!(vec.copy_flags & (COPY_BUF_COALESCED|COPY_BUF_CSUM_FILL))) { - bitmap_set(bmp, vec.offset, vec.len, bs->dsk.bitmap_granularity); + bitmap_set(new_bmp, vec.offset, vec.len, bs->dsk.bitmap_granularity); } } // Update block checksums @@ -576,7 +583,7 @@ bool journal_flusher_co::calc_block_checksums() assert(!(end % bs->dsk.csum_block_size)); uint32_t csum_off = start/bs->dsk.csum_block_size * (bs->dsk.data_csum_type & 0xFF); bs->heap->calc_block_checksums( - (uint32_t*)(csums+csum_off), bmp, start, end, + (uint32_t*)(new_csums+csum_off), new_bmp, start, end, [&](uint32_t start, uint32_t & len) { // O(n^2) search, may be fixed later :-p @@ -593,7 +600,6 @@ bool journal_flusher_co::calc_block_checksums() }, true, NULL ); } - cur_obj->crc32c = cur_obj->calc_crc32c(); return true; } @@ -603,13 +609,15 @@ bool journal_flusher_co::write_meta_block(int wait_base) goto resume_0; else if (wait_state == wait_base+1) goto resume_1; - await_sqe(0); - data->iov = (struct iovec){ bs->heap->get_meta_block(modified_block), (size_t)bs->dsk.meta_block_size }; - data->callback = simple_callback_w; - io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset + (modified_block+1)*bs->dsk.meta_block_size); - wait_count++; +resume_0: + if (bs->ringloop->space_left() < 1) + { + wait_state = wait_base+0; + return 0; + } + bs->prepare_meta_block_write(modified_block); resume_1: - if (wait_count > 0) + if (bs->meta_block_is_pending(modified_block)) { wait_state = wait_base+1; return false; @@ -664,11 +672,11 @@ bool journal_flusher_co::fsync_meta(int wait_base) if (wait_state == wait_base) goto resume_0; else if (wait_state == wait_base+1) goto resume_1; else if (wait_state == wait_base+2) goto resume_2; -resume_0: if (bs->dsk.disable_meta_fsync) { return true; } +resume_0: if (flusher->wanting_meta_fsync || flusher->fsyncing_meta > 0) { wait_state = wait_base; @@ -693,69 +701,31 @@ resume_2: return true; } -int journal_flusher_co::fsync_buffer(int wait_base) -{ - if (wait_state == wait_base) goto resume_0; - else if (wait_state == wait_base+1) goto resume_1; - if (bs->dsk.disable_journal_fsync && bs->dsk.disable_meta_fsync && bs->dsk.disable_data_fsync || !bs->unsynced_big_write_count && !bs->unsynced_small_write_count) - { - return 1; - } - if (flusher->syncing_buffer) - { - return 0; - } - flusher->active_flushers++; - flusher->syncing_buffer++; -resume_0: - assert(!wait_count); - compact_lsn = bs->heap->get_completed_lsn(); - if (!bs->submit_fsyncs(wait_count)) - { - wait_state = wait_base+0; - return 0; - } -resume_1: - if (wait_count > 0) - { - wait_state = wait_base+1; - return 0; - } - bs->heap->mark_lsn_fsynced(compact_lsn); - flusher->active_flushers--; - flusher->syncing_buffer--; - return 2; -} - -bool journal_flusher_co::trim_lsn(int wait_base) +bool journal_flusher_co::fsync_buffer(int wait_base) { if (wait_state == wait_base) goto resume_0; 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; - else if (wait_state == wait_base+4) goto resume_4; - compact_lsn = bs->heap->get_compacted_lsn(); - if (((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn == compact_lsn) + if (bs->dsk.disable_journal_fsync && bs->dsk.disable_meta_fsync && bs->dsk.disable_data_fsync || + !bs->unsynced_big_write_count && !bs->unsynced_small_write_count) { return true; } - flusher->active_flushers++; - assert(!wait_count); - if (!bs->dsk.disable_meta_fsync) +resume_0: + if (flusher->syncing_buffer) { - await_sqe(0); - data->iov = { 0 }; - data->callback = simple_callback_w; - io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC); - wait_count++; + wait_state = wait_base+0; + return false; } - if (!bs->dsk.disable_data_fsync && bs->dsk.data_fd != bs->dsk.meta_fd) + flusher->active_flushers++; + flusher->syncing_buffer++; +resume_1: + assert(!wait_count); + fsynced_lsn = bs->heap->get_completed_lsn(); + if (!bs->submit_fsyncs(wait_count)) { - await_sqe(1); - data->iov = { 0 }; - data->callback = simple_callback_w; - io_uring_prep_fsync(sqe, bs->dsk.data_fd, IORING_FSYNC_DATASYNC); - wait_count++; + wait_state = wait_base+1; + return false; } resume_2: if (wait_count > 0) @@ -763,23 +733,8 @@ resume_2: wait_state = wait_base+2; return false; } - ((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn = compact_lsn; - ((blockstore_meta_header_v3_t*)bs->meta_superblock)->set_crc32c(); - await_sqe(3); - data->iov = (struct iovec){ bs->meta_superblock, (size_t)bs->dsk.meta_block_size }; - data->callback = simple_callback_w; - io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset); - // Update superblock with datasync - sqe->rw_flags = RWF_DSYNC; - wait_count++; -resume_4: - if (wait_count > 0) - { - wait_state = wait_base+4; - return false; - } - bs->heap->mark_lsn_trimmed(compact_lsn); - flusher->compact_counter++; + bs->heap->mark_lsn_fsynced(fsynced_lsn); flusher->active_flushers--; + flusher->syncing_buffer--; return true; } diff --git a/src/blockstore/blockstore_flush.h b/src/blockstore/blockstore_flush.h index 3ed9d2ff..82e3ed66 100644 --- a/src/blockstore/blockstore_flush.h +++ b/src/blockstore/blockstore_flush.h @@ -36,26 +36,23 @@ class journal_flusher_co int wait_state, wait_count; struct io_uring_sqe *sqe; struct ring_data_t *data; + uint8_t *new_csums = NULL; + uint8_t *new_bmp = NULL; std::function simple_callback_r, simple_callback_w; object_id cur_oid; - uint64_t copy_id; - uint64_t compact_lsn; - uint64_t cur_version; - heap_object_t *cur_obj; - heap_write_t *begin_wr, *end_wr; + heap_entry_t *cur_obj; + uint64_t fsynced_lsn; + heap_compact_t compact_info; uint32_t modified_block; bool should_repeat; std::vector read_vec; uint32_t overwrite_start, overwrite_end; - uint32_t big_start, big_end; int i, res; bool read_to_fill_incomplete; int copy_count; - uint64_t clean_loc; - flusher_meta_write_t meta_old, meta_new; bool do_repeat = false; friend class journal_flusher_t; @@ -68,8 +65,7 @@ class journal_flusher_co bool write_meta_block(int wait_base); bool read_buffered(int wait_base); bool fsync_meta(int wait_base); - int fsync_buffer(int wait_base); - bool trim_lsn(int wait_base); + bool fsync_buffer(int wait_base); public: journal_flusher_co(); ~journal_flusher_co(); diff --git a/src/blockstore/blockstore_impl.cpp b/src/blockstore/blockstore_impl.cpp index 58bdb2fc..c05d1fc2 100644 --- a/src/blockstore/blockstore_impl.cpp +++ b/src/blockstore/blockstore_impl.cpp @@ -197,6 +197,12 @@ void blockstore_impl_t::loop() { throw std::runtime_error(std::string("io_uring_submit: ") + strerror(-ret)); } + for (auto & block_num: pending_modified_blocks) + { + heap->start_block_write(block_num); + modified_blocks.insert(block_num); + } + pending_modified_blocks.clear(); if ((initial_ring_space - ringloop->space_left()) > 0) { live = true; diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index 67a19d0f..fa2c1113 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -53,8 +53,8 @@ struct blockstore_op_private_t int pending_ops; int op_state; - // Read, write, sync, stabilize - uint64_t lsn; + // Write, sync, stabilize + uint32_t modified_block, modified_block2; // Read std::vector read_vec; @@ -66,9 +66,6 @@ struct blockstore_op_private_t // Stabilize, rollback int stab_pos; - // Stabilize - uint64_t to_lsn; - // Write struct iovec iov_zerofill[3]; timespec tv_begin; @@ -119,6 +116,9 @@ public: int unsynced_queued_ops = 0; uint8_t *zero_object = NULL; + std::vector pending_modified_blocks; + robin_hood::unordered_flat_set modified_blocks; + journal_flusher_t *flusher; int write_iodepth = 0; int inflight_big = 0; @@ -153,11 +153,11 @@ public: // Read int dequeue_read(blockstore_op_t *op); int fulfill_read(blockstore_op_t *op); - uint32_t prepare_read(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end); - uint32_t prepare_read_with_bitmaps(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end); + uint32_t prepare_read(std::vector & read_vec, heap_entry_t *obj, heap_entry_t *wr, uint32_t start, uint32_t end); + uint32_t prepare_read_with_bitmaps(std::vector & read_vec, heap_entry_t *obj, heap_entry_t *wr, uint32_t start, uint32_t end); uint32_t prepare_read_zero(std::vector & read_vec, uint32_t start, uint32_t end); - uint32_t prepare_read_simple(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end); - void prepare_disk_read(std::vector & read_vec, int pos, heap_object_t *obj, heap_write_t *wr, + uint32_t prepare_read_simple(std::vector & read_vec, heap_entry_t *obj, heap_entry_t *wr, uint32_t start, uint32_t end); + void prepare_disk_read(std::vector & read_vec, int pos, heap_entry_t *obj, heap_entry_t *wr, uint32_t blk_start, uint32_t blk_end, uint32_t start, uint32_t end, uint32_t copy_flags); void find_holes(std::vector & read_vec, uint32_t item_start, uint32_t item_end, std::function callback); @@ -167,9 +167,10 @@ public: // Write bool enqueue_write(blockstore_op_t *op); - void prepare_meta_block_write(blockstore_op_t *op, uint64_t modified_block, io_uring_sqe *sqe = NULL); + void prepare_meta_block_write(uint32_t modified_block); + bool meta_block_is_pending(uint32_t modified_block); + bool intent_write_allowed(blockstore_op_t *op, heap_entry_t *obj); int dequeue_write(blockstore_op_t *op); - int make_big_write(blockstore_op_t *op, uint32_t offset, uint32_t len, uint32_t *modified_block, uint32_t *moved_from_block); int continue_write(blockstore_op_t *op); void handle_write_event(ring_data_t *data, blockstore_op_t *op); diff --git a/src/blockstore/blockstore_init.cpp b/src/blockstore/blockstore_init.cpp index 4a8b3e80..8a39829d 100644 --- a/src/blockstore/blockstore_init.cpp +++ b/src/blockstore/blockstore_init.cpp @@ -79,7 +79,6 @@ resume_1: hdr->meta_block_size = bs->dsk.meta_block_size; hdr->data_block_size = bs->dsk.data_block_size; hdr->bitmap_granularity = bs->dsk.bitmap_granularity; - hdr->compacted_lsn = 0; if (bs->dsk.meta_format >= BLOCKSTORE_META_FORMAT_V2) { hdr->data_csum_type = bs->dsk.data_csum_type; @@ -156,7 +155,6 @@ resume_1: bs->dsk.check_lengths(); } bs->init(); - bs->heap->set_compacted_lsn(((blockstore_meta_header_v3_t *)bs->meta_superblock)->compacted_lsn); if (bs->dsk.inmemory_journal) { // Read buffer area @@ -224,7 +222,11 @@ resume_4: if (bufs[i].state == INIT_META_READ_DONE) { // Handle result - entries_loaded += bs->heap->load_blocks(bufs[i].offset-bs->dsk.meta_block_size, bufs[i].size, bufs[i].buf); + uint64_t loaded = 0; + int r = bs->heap->load_blocks(bufs[i].offset-bs->dsk.meta_block_size, bufs[i].size, bufs[i].buf, loaded); + if (r != 0) + exit(1); + entries_loaded += loaded; bufs[i].state = 0; bs->ringloop->wakeup(); } diff --git a/src/blockstore/blockstore_read.cpp b/src/blockstore/blockstore_read.cpp index c72efb12..9204c063 100644 --- a/src/blockstore/blockstore_read.cpp +++ b/src/blockstore/blockstore_read.cpp @@ -7,7 +7,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op) { - heap_object_t *obj = heap->lock_and_read_entry(op->oid, PRIV(op)->lsn); + heap_entry_t *obj = heap->lock_and_read_entry(op->oid); if (!obj) { op->version = 0; @@ -20,11 +20,11 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op) auto & rv = PRIV(op)->read_vec; uint64_t result_version = 0; bool found = false; - for (auto wr = obj->get_writes(); wr; wr = wr->next()) + heap->iterate_with_stable(obj, obj->lsn, [&](heap_entry_t *wr, bool stable) { if (op->version < wr->version) { - continue; + return true; } if (!found) { @@ -37,16 +37,16 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op) } fulfilled += prepare_read(PRIV(op)->read_vec, obj, wr, op->offset, op->offset+op->len); if (fulfilled == op->len || - wr->type() == BS_HEAP_BIG_WRITE || - wr->type() == BS_HEAP_TOMBSTONE) + wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_DELETE) { - break; + return false; } - } + return true; + }); if (!found) { // May happen if there are entries but all of them are > requested version - heap->unlock_entry(op->oid, PRIV(op)->lsn); + heap->unlock_entry(op->oid); op->version = 0; op->retval = -ENOENT; FINISH_OP(op); @@ -56,7 +56,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op) if (!fulfill_read(op)) { // Need to wait. undo added requests, unlock lsn - heap->unlock_entry(op->oid, PRIV(op)->lsn); + heap->unlock_entry(op->oid); free_read_buffers(rv); rv.clear(); return 0; @@ -65,7 +65,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op) if (!PRIV(op)->pending_ops) { // everything is fulfilled from memory - heap->unlock_entry(op->oid, PRIV(op)->lsn); + heap->unlock_entry(op->oid); op->retval = op->len; free_read_buffers(rv); FINISH_OP(op); @@ -108,20 +108,20 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *op) return 1; } -uint32_t blockstore_impl_t::prepare_read(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end) +uint32_t blockstore_impl_t::prepare_read(std::vector & read_vec, heap_entry_t *obj, heap_entry_t *wr, uint32_t start, uint32_t end) { if (wr->type() == BS_HEAP_BIG_WRITE) { return prepare_read_with_bitmaps(read_vec, obj, wr, start, end); } - if (wr->type() == BS_HEAP_TOMBSTONE) + if (wr->type() == BS_HEAP_DELETE) { return prepare_read_zero(read_vec, start, end); } return prepare_read_simple(read_vec, obj, wr, start, end); } -uint32_t blockstore_impl_t::prepare_read_with_bitmaps(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end) +uint32_t blockstore_impl_t::prepare_read_with_bitmaps(std::vector & read_vec, heap_entry_t *obj, heap_entry_t *wr, uint32_t start, uint32_t end) { // BIG_WRITEs contain a bitmap and we have to handle its holes uint32_t res = 0; @@ -166,7 +166,7 @@ uint32_t blockstore_impl_t::prepare_read_zero(std::vector & read_ return res; } -uint32_t blockstore_impl_t::prepare_read_simple(std::vector & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end) +uint32_t blockstore_impl_t::prepare_read_simple(std::vector & read_vec, heap_entry_t *obj, heap_entry_t *wr, uint32_t start, uint32_t end) { uint32_t res = 0; if (wr->type() == BS_HEAP_SMALL_WRITE || wr->type() == BS_HEAP_INTENT_WRITE) @@ -212,10 +212,14 @@ uint32_t blockstore_impl_t::prepare_read_simple(std::vector & rea uint32_t skip_csum = 0; if (!perfect_csum_update && wr->type() == BS_HEAP_BIG_WRITE) { - for (auto owr = obj->get_writes(); owr && owr != wr; owr = owr->next()) + for (auto owr = obj; owr && owr != wr; owr = heap->prev(owr)) + { if ((owr->type() == BS_HEAP_INTENT_WRITE || owr->type() == BS_HEAP_SMALL_WRITE) && owr->small().offset < blk_end && owr->small().offset+owr->small().len > blk_start) + { skip_csum = COPY_BUF_SKIP_CSUM; + } + } } if ((blk_end-1)/dsk.csum_block_size == blk_start/dsk.csum_block_size || blk_end/dsk.csum_block_size == blk_start/dsk.csum_block_size+1 && blk_end != end && blk_start != start || @@ -243,18 +247,34 @@ uint32_t blockstore_impl_t::prepare_read_simple(std::vector & rea return res; } -void blockstore_impl_t::prepare_disk_read(std::vector & read_vec, int pos, heap_object_t *obj, heap_write_t *wr, +void blockstore_impl_t::prepare_disk_read(std::vector & read_vec, int pos, heap_entry_t *obj, heap_entry_t *wr, uint32_t blk_start, uint32_t blk_end, uint32_t start, uint32_t end, uint32_t copy_flags) { // Only one INTENT_WRITE is allowed at a time - assert(wr->type() != BS_HEAP_INTENT_WRITE || wr->next()->type() == BS_HEAP_BIG_WRITE); + uint64_t loc = 0; + if (wr->type() == BS_HEAP_INTENT_WRITE) + { + heap_entry_t *big_wr = wr; + while (big_wr && big_wr->type() == BS_HEAP_INTENT_WRITE) + { + big_wr = heap->prev(big_wr); + } + assert(big_wr); + loc = big_wr->big_location(heap); + } + else if (wr->type() == BS_HEAP_SMALL_WRITE) + { + loc = wr->small().location-wr->small().offset; + } + else /*if (wr->type() == BS_HEAP_BIG_WRITE)*/ + { + loc = wr->big_location(heap); + } copy_buffer_t vec = { .copy_flags = (wr->type() == BS_HEAP_SMALL_WRITE ? COPY_BUF_JOURNAL : COPY_BUF_DATA) | copy_flags, .offset = start, .len = end-start, - .disk_loc = (wr->type() == BS_HEAP_INTENT_WRITE ? wr->next()->big_location(heap) - : (wr->type() == BS_HEAP_SMALL_WRITE ? wr->small().location-wr->small().offset - : wr->big_location(heap))), + .disk_loc = loc, .disk_offset = blk_start, .disk_len = blk_end - blk_start, .wr_lsn = wr->lsn, @@ -349,7 +369,7 @@ void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op op->retval = -EDOM; else if (op->retval == 0) op->retval = op->len; - heap->unlock_entry(op->oid, PRIV(op)->lsn); + heap->unlock_entry(op->oid); free_read_buffers(PRIV(op)->read_vec); FINISH_OP(op); } @@ -357,7 +377,7 @@ void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op bool blockstore_impl_t::verify_read_checksums(blockstore_op_t *op) { - heap_object_t *obj = heap->read_locked_entry(op->oid, PRIV(op)->lsn); + heap_entry_t *obj = heap->read_entry(op->oid); auto & rv = PRIV(op)->read_vec; for (auto & vec: rv) { @@ -367,9 +387,9 @@ bool blockstore_impl_t::verify_read_checksums(blockstore_op_t *op) memcpy(op->buf + vec.offset - op->offset, vec.buf + vec.offset - vec.disk_offset, vec.len); if (vec.copy_flags & (COPY_BUF_COALESCED|COPY_BUF_SKIP_CSUM)) continue; - heap_write_t *wr = obj->get_writes(); + heap_entry_t *wr = obj; while (wr && wr->lsn != vec.wr_lsn) - wr = wr->next(); + wr = heap->prev(wr); assert(wr); uint8_t *buf = vec.buf ? vec.buf : (op->buf + vec.offset - op->offset); uint32_t *csums = (uint32_t*)(wr->get_checksums(heap) @@ -394,23 +414,29 @@ bool blockstore_impl_t::verify_read_checksums(blockstore_op_t *op) int blockstore_impl_t::read_bitmap(object_id oid, uint64_t target_version, void *bitmap, uint64_t *result_version) { - heap_object_t *obj = heap->read_entry(oid, NULL); + heap_entry_t *obj = heap->read_entry(oid); if (obj) { - for (auto wr = obj->get_writes(); wr; wr = wr->next()) + bool found = false; + heap->iterate_with_stable(obj, obj->lsn, [&](heap_entry_t *wr, bool stable) { - if (target_version < wr->version) + if (target_version >= wr->version) { - continue; - } - if (result_version) - { - *result_version = wr->version; - } - if (bitmap) - { - memcpy(bitmap, wr->get_ext_bitmap(heap), dsk.clean_entry_bitmap_size); + found = true; + if (result_version) + { + *result_version = wr->version; + } + if (bitmap) + { + memcpy(bitmap, wr->get_ext_bitmap(heap), dsk.clean_entry_bitmap_size); + } + return false; } + return true; + }); + if (found) + { return 0; } } diff --git a/src/blockstore/blockstore_stable.cpp b/src/blockstore/blockstore_stable.cpp index 010d7360..f3253ec4 100644 --- a/src/blockstore/blockstore_stable.cpp +++ b/src/blockstore/blockstore_stable.cpp @@ -13,73 +13,76 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op) else if (priv->op_state == 2) goto resume_2; else if (priv->op_state == 3) goto resume_3; else if (priv->op_state == 4) goto resume_4; + else if (priv->op_state == 5) goto resume_5; assert(!priv->op_state); - // Modify in-memory state and assign contiguous LSNs - priv->stab_pos = 0; - priv->lsn = priv->to_lsn = 0; op->retval = 0; - while (priv->stab_pos < op->len) + priv->modified_block = priv->modified_block2 = UINT32_MAX; + for (priv->stab_pos = 0; priv->stab_pos < op->len; priv->stab_pos++) { - uint32_t modified_block = 0; - uint64_t new_lsn = 0; - uint64_t new_to_lsn = 0; - int res = op->opcode == BS_OP_STABLE - ? heap->post_stabilize(v[priv->stab_pos].oid, v[priv->stab_pos].version, &modified_block, &new_lsn, &new_to_lsn) - : heap->post_rollback(v[priv->stab_pos].oid, v[priv->stab_pos].version, &new_lsn, &modified_block); - if (res != 0) { - assert(res == ENOENT || res == EBUSY); - op->retval = -res; - } - if (new_lsn) - { - assert(priv->lsn == 0 || priv->to_lsn == new_lsn-1); - if (!priv->lsn) - priv->lsn = new_lsn; - priv->to_lsn = op->opcode == BS_OP_STABLE ? new_to_lsn : new_lsn; - } - priv->stab_pos++; - } - // Submit metadata writes - priv->stab_pos = 0; -resume_1: - priv->op_state = 1; - while (priv->stab_pos < op->len) - { - uint32_t block_num = 0; - heap_object_t *obj = heap->read_entry(v[priv->stab_pos].oid, &block_num); - if (obj) - { - io_uring_sqe *sqe = get_sqe(); - if (!sqe) + auto obj = heap->read_entry(v[priv->stab_pos].oid); + if (!obj) { - if (priv->pending_ops > 0) - return 1; - priv->wait_detail = 1; - priv->wait_for = WAIT_SQE; + op->retval = -ENOENT; + FINISH_OP(op); + return 2; + } + int res = op->opcode == BS_OP_STABLE + ? heap->add_commit(obj, v[priv->stab_pos].version, &priv->modified_block2) + : heap->add_rollback(obj, v[priv->stab_pos].version, &priv->modified_block2); + if (res == EBUSY) + { + op->retval = -EBUSY; + FINISH_OP(op); + return 2; + } + if (res == ENOSPC) + { + if (!heap->get_inflight_queue_size()) + { + // no space + op->retval = -ENOSPC; + FINISH_OP(op); + return 2; + } + if (priv->modified_block2 != UINT32_MAX) + { + priv->stab_pos--; + goto resume_1; + } + priv->wait_for = WAIT_COMPACTION; + priv->wait_detail = flusher->get_compact_counter(); + flusher->request_trim(); return 0; } - prepare_meta_block_write(op, block_num, sqe); + assert(res == 0); + } + if (priv->modified_block != UINT32_MAX && priv->modified_block2 != priv->modified_block) + { +resume_1: + BS_SUBMIT_CHECK_SQES(1); + prepare_meta_block_write(priv->modified_block); +resume_2: + if (meta_block_is_pending(priv->modified_block)) + { + priv->op_state = 2; + return 1; + } + } + priv->modified_block = priv->modified_block2; + if (priv->stab_pos == op->len-1 && priv->modified_block2 != UINT32_MAX) + { + priv->modified_block2 = UINT32_MAX; + goto resume_1; } - priv->stab_pos++; - } - if (priv->pending_ops > 0) - { - priv->op_state = 1; - return 1; - } - // Mark writes as completed to allow compaction - for (uint64_t lsn = priv->lsn; lsn <= priv->to_lsn; lsn++) - { - heap->mark_lsn_completed(lsn); } unsynced_meta_write_count++; // Fsync, just because our semantics imply that commit (stabilize) is immediately fsynced - priv->op_state = 2; -resume_2: + priv->op_state = 3; resume_3: resume_4: - int res = do_sync(op, 2); +resume_5: + int res = do_sync(op, 3); if (res != 2) { return res; diff --git a/src/blockstore/blockstore_sync.cpp b/src/blockstore/blockstore_sync.cpp index fc136d5b..99a9894d 100644 --- a/src/blockstore/blockstore_sync.cpp +++ b/src/blockstore/blockstore_sync.cpp @@ -96,7 +96,7 @@ int blockstore_impl_t::do_sync(blockstore_op_t *op, int base_state) unsynced_big_write_count = unsynced_small_write_count = unsynced_meta_write_count = 0; return 2; } - PRIV(op)->lsn = heap->get_completed_lsn(); + PRIV(op)->modified_block = heap->get_completed_lsn(); if (!submit_fsyncs(PRIV(op)->pending_ops)) { PRIV(op)->wait_detail = 1; @@ -110,6 +110,6 @@ resume_1: return 1; } resume_2: - heap->mark_lsn_fsynced(PRIV(op)->lsn); + heap->mark_lsn_fsynced(PRIV(op)->modified_block); return 2; } diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index 87f62408..44b88b61 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -11,23 +11,108 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op) return true; } -void blockstore_impl_t::prepare_meta_block_write(blockstore_op_t *op, uint64_t modified_block, io_uring_sqe *sqe) +void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block) { - if (!sqe) + //assert(modified_blocks.find(modified_block) == modified_blocks.end()); + for (auto & block_num: pending_modified_blocks) { - sqe = get_sqe(); - assert(sqe != NULL); + if (block_num == modified_block) + return; } + io_uring_sqe *sqe = get_sqe(); + assert(sqe != NULL); // FIXME + pending_modified_blocks.push_back(modified_block); ring_data_t *data = ((ring_data_t*)sqe->user_data); data->iov = (struct iovec){ heap->get_meta_block(modified_block), (size_t)dsk.meta_block_size }; - data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); }; - PRIV(op)->pending_ops++; + data->callback = [this, modified_block](ring_data_t *data) + { + live = true; + if (data->res != data->iov.iov_len) + { + // FIXME: our state becomes corrupted after a write error. maybe do something better than just die + disk_error_abort("data write", data->res, data->iov.iov_len); + } + modified_blocks.erase(modified_block); + heap->complete_block_write(modified_block); + ringloop->wakeup(); + }; io_uring_prep_writev( sqe, dsk.meta_fd, &data->iov, 1, dsk.meta_offset + (modified_block+1)*dsk.meta_block_size ); } -// First step of the write algorithm: dequeue operation and submit initial write(s) +bool blockstore_impl_t::meta_block_is_pending(uint32_t modified_block) +{ + auto mb_it = modified_blocks.find(modified_block); + if (mb_it != modified_blocks.end()) + return true; + for (auto & block_num: pending_modified_blocks) + { + if (block_num == modified_block) + return true; + } + return false; +} + +bool blockstore_impl_t::intent_write_allowed(blockstore_op_t *op, heap_entry_t *obj) +{ + // Parallel writes to the same object are forbidden so "one intent at a time" is fulfilled automatically + // Intent writes are disabled when metadata fsync is enabled + if (!dsk.disable_meta_fsync) + { + return false; + } + // Intent writes are only for replication + if (op->opcode != BS_OP_WRITE_STABLE) + { + return false; + } + // Operation size should be less than or equal to atomic write size + if (!op->len || op->len > dsk.atomic_write_size) + { + return false; + } + // Intent-writes are disabled if "absolutely correct during compaction" checksum validation algorithm is enabled + // We could also do RMW here when perfect_csum_update is enabled, but it's unclear if we need it + if (perfect_csum_update && dsk.csum_block_size > dsk.bitmap_granularity && + ((op->offset % dsk.csum_block_size) || (op->len % dsk.csum_block_size))) + { + return false; + } + bool ok = true, has_intent = false; + heap->iterate_with_stable(obj, obj->lsn, [&](heap_entry_t *wr, bool stable) + { + // Intent writes are not allowed over buffered writes + if (wr->type() == BS_HEAP_SMALL_WRITE) + { + ok = false; + return false; + } + // Intent writes are not allowed over unstable writes + if (!stable) + { + ok = false; + return false; + } + // One intent-write is allowed even with fsyncs because BIG_WRITE is always counted as fsynced + if (dsk.disable_data_fsync && wr->type() == BS_HEAP_INTENT_WRITE) + { + if (has_intent) + { + ok = false; + return false; + } + has_intent = true; + } + if (wr->type() == BS_HEAP_BIG_WRITE) + { + return false; + } + return true; + }); + return ok; +} + int blockstore_impl_t::dequeue_write(blockstore_op_t *op) { if (PRIV(op)->op_state) @@ -38,13 +123,13 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) { return 0; } + PRIV(op)->modified_block = UINT32_MAX; PRIV(op)->is_big = false; - uint32_t modified_block = UINT32_MAX, moved_from_block = UINT32_MAX; - heap_object_t *obj = heap->read_entry(op->oid, &modified_block); + heap_entry_t *obj = heap->read_entry(op->oid); if (op->opcode == BS_OP_DELETE) { // Delete - if (!obj) + if (!obj || obj->type() == BS_HEAP_DELETE) { // Already deleted op->retval = 0; @@ -52,24 +137,23 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) return 2; } BS_SUBMIT_CHECK_SQES(1); - int res = heap->post_delete(modified_block, obj, &PRIV(op)->lsn); + int res = heap->add_delete(obj, &PRIV(op)->modified_block); assert(res == 0); - prepare_meta_block_write(op, modified_block); + prepare_meta_block_write(PRIV(op)->modified_block); + PRIV(op)->pending_ops++; PRIV(op)->op_state = 5; write_iodepth++; } + // FIXME: Add 'big_intent' write mode // FIXME: Allow to do initial writes as buffered, not redirected // FIXME: Allow to do direct writes over holes - else if (!obj || obj->get_writes()->type() == BS_HEAP_TOMBSTONE || - op->offset == 0 && op->len == dsk.data_block_size) + else if (!obj || obj->type() == BS_HEAP_DELETE || op->offset == 0 && op->len == dsk.data_block_size) { // Big (redirect) write BS_SUBMIT_CHECK_SQES(1); PRIV(op)->is_big = true; - uint32_t tmp_block; uint64_t loc = heap->find_free_data(); - if (loc == UINT64_MAX || - !obj && heap->get_block_for_new_object(tmp_block) != 0) + if (loc == UINT64_MAX) { if (!heap->get_inflight_queue_size()) { @@ -91,17 +175,6 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) ); #endif heap->use_data(op->oid.inode, PRIV(op)->location); - if (!dsk.disable_data_fsync && dsk.disable_meta_fsync) - { - // Do big_write as an INTENT to avoid data fsync - int res = make_big_write(op, 0, 0, &modified_block, &moved_from_block); - assert(res == 0); - if (moved_from_block != UINT32_MAX) - prepare_meta_block_write(op, moved_from_block); - obj = heap->read_entry(op->oid, &modified_block); - heap->mark_lsn_completed(PRIV(op)->lsn); - goto process_intent; - } io_uring_sqe *sqe = get_sqe(); ring_data_t *data = ((ring_data_t*)sqe->user_data); uint64_t stripe_offset = (op->offset % dsk.bitmap_granularity); @@ -128,46 +201,19 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) write_iodepth++; inflight_big++; } - // Only one INTENT_WRITE is allowed at a time, but in fact, - // parallel writes to the same object are forbidden anyway - else if (op->len > 0 && op->len <= dsk.atomic_write_size && - // Intent-writes are disabled if "absolutely correct during compaction" checksum validation algorithm is enabled - // We could also do RMW here when perfect_csum_update is enabled, but it's unclear if we need it - (!perfect_csum_update || dsk.csum_block_size <= dsk.bitmap_granularity || - !(op->offset % dsk.csum_block_size) && - !(op->len % dsk.csum_block_size) && - (obj->get_writes()->entry_type != (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) || - obj->get_writes()->can_be_collapsed(heap))) && - // One intent-write is allowed even with fsyncs because BIG_WRITE is always counted as fsynced - dsk.disable_meta_fsync && - (op->opcode == BS_OP_WRITE_STABLE && - (obj->get_writes()->entry_type == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) || - obj->get_writes()->entry_type == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) && dsk.disable_data_fsync) || - op->opcode == BS_OP_WRITE && obj->get_writes()->entry_type == BS_HEAP_BIG_WRITE)) + else if (intent_write_allowed(op, obj)) { // Direct intent-write BS_SUBMIT_CHECK_SQES(1); - if (obj->get_writes()->type() == BS_HEAP_BIG_WRITE) + auto wr = obj; + while (wr && (wr->type() == BS_HEAP_INTENT_WRITE || wr->type() == BS_HEAP_COMMIT || wr->type() == BS_HEAP_ROLLBACK)) { - PRIV(op)->location = obj->get_writes()->big_location(heap); + wr = heap->prev(wr); } - else - { - assert(obj->get_writes()->next()->type() == BS_HEAP_BIG_WRITE); - PRIV(op)->location = obj->get_writes()->next()->big_location(heap); - } -process_intent: - uint8_t wr_buf[heap->get_max_write_entry_size()]; - heap_write_t *wr = (heap_write_t*)wr_buf; - wr->version = op->version; - wr->entry_type = BS_HEAP_INTENT_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0); - wr->small().offset = op->offset; - wr->small().len = op->len; - wr->small().location = 0; - if (op->bitmap) - memcpy(wr->get_ext_bitmap(heap), op->bitmap, dsk.clean_entry_bitmap_size); - heap->calc_checksums(wr, (uint8_t*)op->buf, true); - int res = heap->post_write(modified_block, op->oid, obj, wr, &moved_from_block); + assert(wr->type() == BS_HEAP_BIG_WRITE); + PRIV(op)->location = wr->big_location(heap); + int res = heap->add_small_write(op->oid, obj, (BS_HEAP_INTENT_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)), + op->version, op->offset, op->len, 0, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block); if (res == EAGAIN) { assert(heap->get_inflight_queue_size()); @@ -184,10 +230,8 @@ process_intent: return 2; } assert(res == 0); - PRIV(op)->lsn = wr->lsn; - if (moved_from_block != UINT32_MAX) - prepare_meta_block_write(op, moved_from_block); - prepare_meta_block_write(op, modified_block); + prepare_meta_block_write(PRIV(op)->modified_block); + PRIV(op)->pending_ops++; PRIV(op)->op_state = 9; write_iodepth++; } @@ -205,18 +249,8 @@ process_intent: } // There is sufficient space. Check SQE(s) BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0)); - uint8_t wr_buf[heap->get_max_write_entry_size()]; - heap_write_t *wr = (heap_write_t*)wr_buf; - wr->version = op->version; - wr->entry_type = BS_HEAP_SMALL_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0); - wr->small().offset = op->offset; - wr->small().len = op->len; - wr->small().location = loc; - PRIV(op)->location = loc; - if (op->bitmap) - memcpy(wr->get_ext_bitmap(heap), op->bitmap, dsk.clean_entry_bitmap_size); - heap->calc_checksums(wr, (uint8_t*)op->buf, true); - int res = heap->post_write(modified_block, op->oid, obj, wr, &moved_from_block); + int res = heap->add_small_write(op->oid, obj, (BS_HEAP_SMALL_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)), + op->version, op->offset, op->len, loc, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block); if (res == EAGAIN) { assert(heap->get_inflight_queue_size()); @@ -233,12 +267,10 @@ process_intent: return 2; } assert(res == 0); - PRIV(op)->lsn = wr->lsn; if (op->len) heap->use_buffer_area(op->oid.inode, loc, op->len); - if (moved_from_block != UINT32_MAX) - prepare_meta_block_write(op, moved_from_block); - prepare_meta_block_write(op, modified_block); + prepare_meta_block_write(PRIV(op)->modified_block); + PRIV(op)->pending_ops++; if (op->len > 0) { // Prepare buffered data write @@ -263,26 +295,6 @@ process_intent: return 1; } -int blockstore_impl_t::make_big_write(blockstore_op_t *op, uint32_t offset, uint32_t len, uint32_t *modified_block, uint32_t *moved_from_block) -{ - uint8_t wr_buf[heap->get_max_write_entry_size()]; - heap_write_t *wr = (heap_write_t*)wr_buf; - wr->entry_type = BS_HEAP_BIG_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0); - wr->version = op->version; - wr->set_big_location(heap, PRIV(op)->location); - if (op->bitmap) - memcpy(wr->get_ext_bitmap(heap), op->bitmap, dsk.clean_entry_bitmap_size); - memset(wr->get_int_bitmap(heap), 0, dsk.clean_entry_bitmap_size); - bitmap_set(wr->get_int_bitmap(heap), offset, len, dsk.bitmap_granularity); - heap->calc_checksums(wr, (uint8_t*)op->buf, true, offset, len); - int res = heap->post_write(op->oid, wr, modified_block, moved_from_block); - if (res != 0) - return res; - assert(res == 0); - PRIV(op)->lsn = wr->lsn; - return 0; -} - int blockstore_impl_t::continue_write(blockstore_op_t *op) { int op_state = PRIV(op)->op_state; @@ -305,6 +317,12 @@ again: { // In progress assert(op_state < 10); + if (PRIV(op)->modified_block != UINT32_MAX && + !meta_block_is_pending(PRIV(op)->modified_block)) + { + PRIV(op)->pending_ops--; + PRIV(op)->modified_block = UINT32_MAX; + } if (PRIV(op)->pending_ops > 0) return 1; op_state++; @@ -313,7 +331,6 @@ again: resume_2: // We must fsync all big writes to avoid complex write workflows // It's OK for all HDDs and for server SSDs, but slightly worse for desktop SSDs - // The other way is to add another type of MVCC to blockstore_heap: "forward" MVCC :) inflight_big--; if (!dsk.disable_data_fsync) { @@ -349,8 +366,9 @@ resume_12: } resume_4: { - uint32_t modified_block = UINT32_MAX, moved_from_block = UINT32_MAX; - int res = make_big_write(op, op->offset, op->len, &modified_block, &moved_from_block); + auto obj = heap->read_entry(op->oid); + int res = heap->add_big_write(op->oid, obj, (op->opcode == BS_OP_WRITE_STABLE), op->version, + op->offset, op->len, PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block); if (res == EAGAIN) { assert(heap->get_inflight_queue_size()); @@ -367,9 +385,9 @@ resume_4: FINISH_OP(op); return 2; } - if (moved_from_block != UINT32_MAX) - prepare_meta_block_write(op, moved_from_block); - prepare_meta_block_write(op, modified_block); + assert(res == 0); + prepare_meta_block_write(PRIV(op)->modified_block); + PRIV(op)->pending_ops++; PRIV(op)->op_state = 5; return 1; } @@ -411,7 +429,6 @@ resume_8: printf("Ack write %jx:%jx v%ju\n", op->oid.inode, op->oid.stripe, op->version); #endif op->retval = op->len; - heap->mark_lsn_completed(PRIV(op)->lsn); if (PRIV(op)->is_big) unsynced_big_write_count++; else diff --git a/src/disk_tool/disk_simple_offsets.cpp b/src/disk_tool/disk_simple_offsets.cpp index f174c9dd..a51493f1 100644 --- a/src/disk_tool/disk_simple_offsets.cpp +++ b/src/disk_tool/disk_simple_offsets.cpp @@ -132,16 +132,13 @@ void disk_tool_simple_offsets(json11::Json cfg, bool json_output) uint64_t meta_size; if (meta_format == BLOCKSTORE_META_FORMAT_HEAP) { - uint32_t min_object_size = sizeof(heap_object_t)+sizeof(heap_write_t)+data_csum_size+2*clean_entry_bitmap_size; - uint32_t meta_block_target_free_space = cfg["meta_block_target_free_space"].uint64_value(); - if (!meta_block_target_free_space || meta_block_target_free_space > device_block_size-min_object_size) - meta_block_target_free_space = 800; + uint32_t min_object_size = sizeof(heap_big_write_t) + data_csum_size + 2*clean_entry_bitmap_size; double meta_reserve = cfg["meta_reserve"].number_value(); if (!meta_reserve) meta_reserve = 1.5; else if (meta_reserve < 1) meta_reserve = 1; - uint32_t entries_per_block = (device_block_size-meta_block_target_free_space) / min_object_size; + uint32_t entries_per_block = device_block_size / min_object_size; meta_size = device_block_size * (uint64_t)((object_count+entries_per_block-1) / entries_per_block * meta_reserve); } else if (meta_format == BLOCKSTORE_META_FORMAT_V2) diff --git a/src/disk_tool/disk_tool.h b/src/disk_tool/disk_tool.h index c62b55ed..21922293 100644 --- a/src/disk_tool/disk_tool.h +++ b/src/disk_tool/disk_tool.h @@ -100,15 +100,15 @@ struct disk_tool_t int process_journal(std::function block_fn, bool do_open = true); int process_journal_block(void *buf, std::function iter_fn); int process_meta(std::function hdr_fn, - std::function obj_fn, + std::function obj_fn, std::function record_fn, bool with_data, bool do_open); int dump_meta(); void dump_meta_header(blockstore_meta_header_v3_t *hdr); void dump_meta_entry(uint64_t block_num, clean_disk_entry *entry, uint8_t *bitmap); - void dump_heap_entry_as_old(blockstore_heap_t *heap, heap_object_t *obj); - void dump_heap_entry(blockstore_heap_t *heap, heap_object_t *obj); + void dump_heap_entry_as_old(blockstore_heap_t *heap, heap_entry_t *obj); + void dump_heap_entry(blockstore_heap_t *heap, heap_entry_t *obj); int dump_load_check_superblock(const std::string & device); diff --git a/src/disk_tool/disk_tool_discard.cpp b/src/disk_tool/disk_tool_discard.cpp index 86d63857..66798542 100644 --- a/src/disk_tool/disk_tool_discard.cpp +++ b/src/disk_tool/disk_tool_discard.cpp @@ -55,9 +55,9 @@ int disk_tool_t::trim_data(std::string device) data_alloc = new allocator_t(dsk.block_count); r = process_meta( [this](blockstore_meta_header_v3_t *hdr) {}, - [this](blockstore_heap_t *heap, heap_object_t *obj, uint32_t meta_block_num) + [this](blockstore_heap_t *heap, heap_entry_t *obj, uint32_t meta_block_num) { - for (auto wr = obj->get_writes(); wr; wr = wr->next()) + for (auto wr = obj; wr; wr = heap->prev(wr)) { if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) { diff --git a/src/disk_tool/disk_tool_meta.cpp b/src/disk_tool/disk_tool_meta.cpp index 80e04574..7bee467c 100644 --- a/src/disk_tool/disk_tool_meta.cpp +++ b/src/disk_tool/disk_tool_meta.cpp @@ -10,7 +10,7 @@ #define FREE_SPACE_BIT 0x8000 int disk_tool_t::process_meta(std::function hdr_fn, - std::function obj_fn, + std::function obj_fn, std::function record_fn, bool with_data, bool do_open) { @@ -108,7 +108,7 @@ close_error: { uint64_t read_len = buf_size < dsk.meta_area_size-meta_pos ? buf_size : dsk.meta_area_size-meta_pos; read_blocking(dsk.meta_fd, data, read_len); - heap->read_blocks(meta_pos-dsk.meta_block_size, read_len, data, [&](heap_object_t *obj) + heap->read_blocks(meta_pos-dsk.meta_block_size, read_len, data, [&](heap_entry_t *obj) { obj_fn(heap, obj, ((uint8_t*)obj-data+meta_pos)/dsk.meta_block_size); }, [](uint32_t, uint32_t, uint8_t*){}); @@ -281,7 +281,7 @@ int disk_tool_t::dump_meta() } dump_meta_header(hdr); }, - [this](blockstore_heap_t *heap, heap_object_t *obj, uint32_t meta_block_num) + [this](blockstore_heap_t *heap, heap_entry_t *obj, uint32_t meta_block_num) { if (dump_as_old) dump_heap_entry_as_old(heap, obj); @@ -334,11 +334,10 @@ void disk_tool_t::dump_meta_header(blockstore_meta_header_v3_t *hdr) first_entry = true; } -void disk_tool_t::dump_heap_entry_as_old(blockstore_heap_t *heap, heap_object_t *obj) +void disk_tool_t::dump_heap_entry_as_old(blockstore_heap_t *heap, heap_entry_t *obj) { - heap_write_t *wr = NULL; - for (wr = obj->get_writes(); wr && wr->entry_type != (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) && - wr->entry_type != (BS_HEAP_TOMBSTONE|BS_HEAP_STABLE); wr = wr->next()) + heap_entry_t *wr = NULL; + for (wr = obj; wr && !wr->is_overwrite(); wr = heap->prev(wr)) { } if (!wr || wr->entry_type != (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)) @@ -365,7 +364,7 @@ void disk_tool_t::dump_heap_entry_as_old(blockstore_heap_t *heap, heap_object_t printf("%02x", bitmap[i]); } uint8_t *csums = wr->get_checksums(heap); - uint32_t csum_size = wr->get_csum_size(heap); + uint32_t csum_size = heap->get_csum_size(wr); if (csums) { printf("\",\"block_csums\":\""); @@ -382,83 +381,76 @@ void disk_tool_t::dump_heap_entry_as_old(blockstore_heap_t *heap, heap_object_t first_entry = false; } -void disk_tool_t::dump_heap_entry(blockstore_heap_t *heap, heap_object_t *obj) +void disk_tool_t::dump_heap_entry(blockstore_heap_t *heap, heap_entry_t *wr) { + auto t = wr->type(); printf( -#define ENTRY_FMT "{\"pool\":%u,\"inode\":\"0x%jx\",\"stripe\":\"0x%jx\",\"writes\":[" +#define ENTRY_FMT "{\"pool\":%u,\"inode\":\"0x%jx\",\"stripe\":\"0x%jx\",\"lsn\":%ju,\"version\":%ju,\"type\":\"%s\",\"stable\":%s" (first_entry ? ENTRY_FMT : (",\n" ENTRY_FMT)), #undef ENTRY_FMT - INODE_POOL(obj->inode), INODE_NO_POOL(obj->inode), obj->stripe + INODE_POOL(wr->inode), INODE_NO_POOL(wr->inode), wr->stripe, + wr->lsn, wr->version, + t == BS_HEAP_BIG_WRITE ? "big" : ( + t == BS_HEAP_SMALL_WRITE ? "small" : ( + t == BS_HEAP_INTENT_WRITE ? "intent" : ( + t == BS_HEAP_DELETE ? "delete" : ( + t == BS_HEAP_COMMIT ? "commit" : ( + t == BS_HEAP_ROLLBACK ? "rollback" : ( + "unknown")))))), + (wr->entry_type & BS_HEAP_STABLE) ? "true" : "false" ); - heap_write_t *wr = NULL; - bool first_wr = true; - for (wr = obj->get_writes(); wr; wr = wr->next()) + if (t == BS_HEAP_BIG_WRITE) { - printf( -#define ENTRY_FMT "{\"lsn\":%ju,\"version\":%ju,\"type\":\"%s\",\"stable\":%s" - (first_wr ? ENTRY_FMT : ("," ENTRY_FMT)), -#undef ENTRY_FMT - wr->lsn, wr->version, (wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE ? "small" : ( - (wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE ? "big" : ( - (wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE ? "intent" : ( - (wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_TOMBSTONE ? "tombstone" : "unknown"))), - (wr->entry_type & BS_HEAP_STABLE) ? "true" : "false" - ); - if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) - { - printf(",\"location\":%ju", wr->big_location(heap)); - } - else if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE) - { - printf(",\"offset\":%u,\"len\":%u", wr->small().offset, wr->small().len); - } - else if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE) - { - if (!dump_with_data) - { - printf(",\"offset\":%u,\"len\":%u,\"location\":%ju", wr->small().offset, wr->small().len, wr->small().location); - } - else - { - printf(",\"data\":\""); - for (uint32_t i = 0; i < wr->small().len; i++) - printf("%02x", buffer_area[wr->small().location + i]); - printf("\""); - } - } - uint8_t* bitmap = wr->get_int_bitmap(heap); - if (bitmap) - { - printf(",\"bitmap\":\""); - for (uint64_t i = 0; i < dsk.clean_entry_bitmap_size; i++) - printf("%02x", bitmap[i]); - printf("\""); - } - bitmap = wr->get_ext_bitmap(heap); - if (bitmap) - { - printf(",\"ext_bitmap\":\""); - for (uint64_t i = 0; i < dsk.clean_entry_bitmap_size; i++) - printf("%02x", bitmap[i]); - printf("\""); - } - uint8_t *csums = wr->get_checksums(heap); - if (csums) - { - printf(",\"block_csums\":\""); - uint32_t csum_size = wr->get_csum_size(heap); - for (uint32_t i = 0; i < csum_size; i++) - printf("%02x", csums[i]); - printf("\""); - } - if (wr->get_checksum(heap)) - { - printf(",\"data_crc32c\":\"%08x\"", *wr->get_checksum(heap)); - } - printf("}"); - first_wr = false; + printf(",\"location\":%ju", wr->big_location(heap)); } - printf("]}"); + else if (t == BS_HEAP_INTENT_WRITE) + { + printf(",\"offset\":%u,\"len\":%u", wr->small().offset, wr->small().len); + } + else if (t == BS_HEAP_SMALL_WRITE) + { + if (!dump_with_data) + { + printf(",\"offset\":%u,\"len\":%u,\"location\":%ju", wr->small().offset, wr->small().len, wr->small().location); + } + else + { + printf(",\"data\":\""); + for (uint32_t i = 0; i < wr->small().len; i++) + printf("%02x", buffer_area[wr->small().location + i]); + printf("\""); + } + } + uint8_t* bitmap = wr->get_int_bitmap(heap); + if (bitmap) + { + printf(",\"bitmap\":\""); + for (uint64_t i = 0; i < dsk.clean_entry_bitmap_size; i++) + printf("%02x", bitmap[i]); + printf("\""); + } + bitmap = wr->get_ext_bitmap(heap); + if (bitmap) + { + printf(",\"ext_bitmap\":\""); + for (uint64_t i = 0; i < dsk.clean_entry_bitmap_size; i++) + printf("%02x", bitmap[i]); + printf("\""); + } + uint8_t *csums = wr->get_checksums(heap); + if (csums) + { + printf(",\"block_csums\":\""); + uint32_t csum_size = heap->get_csum_size(wr); + for (uint32_t i = 0; i < csum_size; i++) + printf("%02x", csums[i]); + printf("\""); + } + if (wr->get_checksum(heap)) + { + printf(",\"data_crc32c\":\"%08x\"", *wr->get_checksum(heap)); + } + printf("}"); first_entry = false; } @@ -603,101 +595,127 @@ int disk_tool_t::write_json_heap(json11::Json meta, json11::Json journal) } uint64_t total_used_space = 0; uint32_t used_space = 0; + uint64_t meta_offset = 0; + // FIXME: Rather ugly. Remove the dependency on dsk from heap? + blockstore_disk_t dsk; + dsk.bitmap_granularity = new_meta_hdr->bitmap_granularity; + dsk.block_count = 16; + dsk.data_block_size = new_meta_hdr->data_block_size; + dsk.clean_entry_bitmap_size = new_clean_entry_bitmap_size; + dsk.csum_block_size = new_meta_hdr->csum_block_size; + dsk.data_csum_type = new_meta_hdr->data_csum_type; + dsk.journal_len = 4096; + dsk.meta_area_size = new_meta_len; + dsk.meta_block_size = new_meta_hdr->meta_block_size; + blockstore_heap_t heap(&dsk, NULL, 0); + heap_entry_t *wr = NULL; + auto get_wr = [&](uint32_t entry_size) + { + if (used_space > new_meta_hdr->meta_block_size-entry_size) + { + if (used_space < new_meta_hdr->meta_block_size-2) + { + *((uint16_t*)(new_meta_buf + meta_offset + used_space)) = FREE_SPACE_BIT | (uint16_t)(new_meta_hdr->meta_block_size-used_space); + } + meta_offset += new_meta_hdr->meta_block_size; + used_space = 0; + if (meta_offset >= new_meta_len) + { + fprintf(stderr, "Metadata doesn't fit into the new area (total used space: %ju)\n", total_used_space); + return (heap_entry_t*)NULL; + } + } + auto wr = (heap_entry_t*)(new_meta_buf + meta_offset + used_space); + used_space += entry_size; + return wr; + }; // FIXME: Use a streaming json parser if (meta["version"] == "3.0") { // New format - std::vector object_buf; - new_heap = new blockstore_heap_t(&dsk, new_journal_buf, 0); for (const auto & meta_entry: meta["entries"].array_items()) { - bool invalid = false; object_id oid = { .inode = (sscanf_json(NULL, meta_entry["pool"]) << (64-POOL_ID_BITS)) | sscanf_json(NULL, meta_entry["inode"]), .stripe = sscanf_json(NULL, meta_entry["stripe"]), }; - object_buf.clear(); - object_buf.resize(sizeof(heap_object_t)); - heap_object_t *obj = (heap_object_t*)object_buf.data(); - obj->size = sizeof(heap_object_t); - obj->write_pos = meta_entry["writes"].array_items().size() ? sizeof(heap_object_t) : 0; - obj->entry_type = BS_HEAP_OBJECT; - obj->inode = oid.inode; - obj->stripe = oid.stripe; - size_t pos = sizeof(heap_object_t); - heap_write_t *last_wr = NULL; - for (auto & write_entry: meta_entry["writes"].array_items()) + uint32_t wr_type = 0; + if (meta_entry["type"] == "small") + wr_type = BS_HEAP_SMALL_WRITE; + else if (meta_entry["type"] == "intent") + wr_type = BS_HEAP_INTENT_WRITE; + else if (meta_entry["type"] == "big") + wr_type = BS_HEAP_BIG_WRITE; + else if (meta_entry["type"] == "delete") + wr_type = BS_HEAP_DELETE; + else if (meta_entry["type"] == "commit") + wr_type = BS_HEAP_COMMIT; + else if (meta_entry["type"] == "rollback") + wr_type = BS_HEAP_ROLLBACK; + else { - object_buf.resize(object_buf.size() + new_heap->get_max_write_entry_size()); - heap_write_t *wr = (heap_write_t*)(object_buf.data() + pos); - last_wr = wr; - uint8_t wr_type = 0; - if (write_entry["type"] == "small") - wr_type = BS_HEAP_SMALL_WRITE; - else if (write_entry["type"] == "intent") - wr_type = BS_HEAP_INTENT_WRITE; - else if (write_entry["type"] == "big") - wr_type = BS_HEAP_BIG_WRITE; - else if (write_entry["type"] == "tombstone") - wr_type = BS_HEAP_TOMBSTONE; - else + fprintf(stderr, "Write entry in %s has invalid type: %s, skipping\n", meta_entry.dump().c_str(), meta_entry["type"].dump().c_str()); +close_err0: + free(new_meta_buf); + new_meta_buf = NULL; + return 1; + } + uint64_t wr_offset = meta_entry["offset"].uint64_value(); + uint64_t wr_len = meta_entry["len"].uint64_value(); + uint32_t wr_size = (wr_type == BS_HEAP_SMALL_WRITE || wr_type == BS_HEAP_INTENT_WRITE + ? heap.get_small_entry_size(wr_offset, wr_len) + : (wr_type == BS_HEAP_BIG_WRITE ? heap.get_big_entry_size() : heap.get_simple_entry_size())); + if (!(wr = get_wr(wr_size))) + goto close_err0; + wr->inode = oid.inode; + wr->stripe = oid.stripe; + wr->entry_type = wr_type | (meta_entry["stable"].bool_value() ? BS_HEAP_STABLE : 0); + wr->lsn = meta_entry["lsn"].uint64_value(); + wr->version = meta_entry["version"].uint64_value(); + wr->size = wr->get_size(&heap); + if (wr_type == BS_HEAP_SMALL_WRITE || wr_type == BS_HEAP_INTENT_WRITE) + { + wr->small().offset = wr_offset; + wr->small().len = wr_len; + wr->small().location = meta_entry["location"].uint64_value(); + if (wr_type == BS_HEAP_SMALL_WRITE && meta_entry["data"].is_string() && wr->small().len > 0) { - fprintf(stderr, "Write entry in %s has invalid type: %s, skipping object\n", meta_entry.dump().c_str(), write_entry["type"].dump().c_str()); - invalid = true; - break; - } - wr->entry_type = wr_type | (write_entry["stable"].bool_value() ? BS_HEAP_STABLE : 0); - wr->lsn = write_entry["lsn"].uint64_value(); - wr->version = write_entry["version"].uint64_value(); - wr->size = wr->get_size(new_heap); - wr->next_pos = wr->size; - if (wr_type == BS_HEAP_SMALL_WRITE || wr_type == BS_HEAP_INTENT_WRITE) - { - wr->small().offset = write_entry["offset"].uint64_value(); - wr->small().len = write_entry["len"].uint64_value(); - wr->small().location = write_entry["location"].uint64_value(); - if (wr_type == BS_HEAP_SMALL_WRITE && write_entry["data"].is_string() && wr->small().len > 0) + if (!new_journal_buf) { - if (!new_journal_buf) - { - fprintf(stderr, "Loading small write data requires overwriting buffer area\n"); - free_new_meta(); - return 1; - } - wr->small().location = new_heap->find_free_buffer_area(wr->small().len); - fromhexstr(write_entry["data"].string_value(), wr->small().len, new_journal_buf + wr->small().location); + fprintf(stderr, "Loading small write data requires overwriting buffer area\n"); + free_new_meta(); + return 1; } - } - else if (wr_type == BS_HEAP_BIG_WRITE) - { - uint64_t loc = write_entry["location"].uint64_value(); - assert(!(loc % dsk.data_block_size)); - assert((loc / dsk.data_block_size) < 0xFFFF0000); - wr->set_big_location(new_heap, loc); - } - if (write_entry["bitmap"].is_string() && wr->get_int_bitmap(new_heap)) - { - fromhexstr(write_entry["bitmap"].string_value(), new_clean_entry_bitmap_size, wr->get_int_bitmap(new_heap)); - } - if (write_entry["ext_bitmap"].is_string() && wr->get_ext_bitmap(new_heap)) - { - fromhexstr(write_entry["ext_bitmap"].string_value(), new_clean_entry_bitmap_size, wr->get_ext_bitmap(new_heap)); - } - if (write_entry["block_csums"].is_string() && wr->get_checksums(new_heap)) - { - fromhexstr(write_entry["block_csums"].string_value(), wr->get_csum_size(new_heap), wr->get_ext_bitmap(new_heap)); - } - if (write_entry["data_crc32c"].is_string() && wr->get_checksum(new_heap)) - { - *wr->get_checksum(new_heap) = sscanf_json("%jx", write_entry["data_crc32c"]); + wr->small().location = heap.find_free_buffer_area(wr->small().len); + fromhexstr(meta_entry["data"].string_value(), wr->small().len, new_journal_buf + wr->small().location); } } - if (invalid) + else if (wr_type == BS_HEAP_BIG_WRITE) { - continue; + uint64_t loc = meta_entry["location"].uint64_value(); + assert(!(loc % dsk.data_block_size)); + assert((loc / dsk.data_block_size) < 0xFFFF0000); + wr->set_big_location(&heap, loc); } - last_wr->next_pos = 0; - new_heap->copy_object(obj, NULL); + if (meta_entry["bitmap"].is_string() && wr->get_int_bitmap(&heap)) + { + fromhexstr(meta_entry["bitmap"].string_value(), new_clean_entry_bitmap_size, wr->get_int_bitmap(&heap)); + } + if (meta_entry["ext_bitmap"].is_string() && wr->get_ext_bitmap(&heap)) + { + fromhexstr(meta_entry["ext_bitmap"].string_value(), new_clean_entry_bitmap_size, wr->get_ext_bitmap(&heap)); + } + if (meta_entry["block_csums"].is_string() && wr->get_checksums(&heap)) + { + fromhexstr(meta_entry["block_csums"].string_value(), heap.get_csum_size(wr), wr->get_ext_bitmap(&heap)); + } + if (meta_entry["data_crc32c"].is_string() && wr->get_checksum(&heap)) + { + *wr->get_checksum(&heap) = sscanf_json("%jx", meta_entry["data_crc32c"]); + } + wr->crc32c = wr->calc_crc32c(); + + assert((uint8_t*)wr + wr->size == new_meta_buf + meta_offset + used_space); } } else @@ -718,82 +736,17 @@ close_err: journal = json11::Json(); // Convert old format to the new format uint64_t next_lsn = 0; - uint64_t meta_offset = 0; - const uint32_t space_per_object = sizeof(heap_object_t) + sizeof(heap_write_t) + - new_clean_entry_bitmap_size*2 + new_data_csum_size; uint64_t buffer_pos = 0; - // FIXME: Rather ugly. Remove the dependency on dsk from heap? - blockstore_disk_t dsk; - dsk.bitmap_granularity = new_meta_hdr->bitmap_granularity; - dsk.block_count = 16; - dsk.data_block_size = new_meta_hdr->data_block_size; - dsk.clean_entry_bitmap_size = new_clean_entry_bitmap_size; - dsk.csum_block_size = new_meta_hdr->csum_block_size; - dsk.data_csum_type = new_meta_hdr->data_csum_type; - dsk.journal_len = 4096; - dsk.meta_area_size = new_meta_len; - dsk.meta_block_size = new_meta_hdr->meta_block_size; - dsk.meta_block_target_free_space = 800; - blockstore_heap_t heap(&dsk, NULL, 0); for (const auto & meta_entry: meta["entries"].array_items()) { object_id oid = { .inode = (sscanf_json(NULL, meta_entry["pool"]) << (64-POOL_ID_BITS)) | sscanf_json(NULL, meta_entry["inode"]), .stripe = sscanf_json(NULL, meta_entry["stripe"]), }; - uint32_t space_for_this = space_per_object; - auto j_it = journal_by_object.find(oid); - if (j_it != journal_by_object.end()) - { - for (auto & rec: j_it->second) - { - if (rec["type"] == "small_write" || rec["type"] == "small_write_instant") - { - uint64_t off = rec["offset"].uint64_value(); - uint64_t len = rec["len"].uint64_value(); - if (off+len > new_meta_hdr->data_block_size) - { - fprintf(stderr, "Journal entry has too large offset or length: %s\n", json11::Json(rec).dump().c_str()); - goto close_err; - } - space_for_this += sizeof(heap_write_t) + new_clean_entry_bitmap_size + - ((off+len+new_meta_hdr->csum_block_size-1)/new_meta_hdr->csum_block_size - off/new_meta_hdr->csum_block_size) * (new_meta_hdr->data_csum_type & 0xFF); - } - else /*if (rec["type"] == "big_write" || rec["type"] == "big_write_instant")*/ - { - space_for_this += sizeof(heap_write_t) + 2*new_clean_entry_bitmap_size + new_data_csum_size; - } - } - } - if (space_for_this > new_meta_hdr->meta_block_size) - { - fprintf(stderr, "Object doesn't fit in a single metadata block. Object meta: %s, object journal: %s\n", - meta_entry.dump().c_str(), json11::Json(j_it->second).dump().c_str()); + if (!(wr = get_wr(heap.get_big_entry_size()))) goto close_err; - } - if (used_space + space_for_this > new_meta_hdr->meta_block_size-dsk.meta_block_target_free_space) - { - if (used_space < new_meta_hdr->meta_block_size-2) - { - *((uint16_t*)(new_meta_buf + meta_offset + used_space)) = FREE_SPACE_BIT | (uint16_t)(new_meta_hdr->meta_block_size-used_space); - } - meta_offset += new_meta_hdr->meta_block_size; - used_space = 0; - if (meta_offset >= new_meta_len) - { - fprintf(stderr, "Metadata doesn't fit into the new area (total used space: %ju, minimum free space in block: %u/%u)\n", - total_used_space, dsk.meta_block_target_free_space, new_meta_hdr->meta_block_size); - goto close_err; - } - } - heap_object_t *obj = (heap_object_t*)(new_meta_buf + meta_offset + used_space); - obj->size = sizeof(heap_object_t); - obj->write_pos = sizeof(heap_object_t); - obj->entry_type = BS_HEAP_OBJECT; - obj->inode = oid.inode; - obj->stripe = oid.stripe; - heap_write_t *wr = obj->get_writes(); - wr->next_pos = 0; + wr->inode = oid.inode; + wr->stripe = oid.stripe; wr->entry_type = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE; wr->lsn = ++next_lsn; wr->version = sscanf_json(NULL, meta_entry["version"]); @@ -803,17 +756,24 @@ close_err: fromhexstr(meta_entry["ext_bitmap"].string_value(), new_clean_entry_bitmap_size, wr->get_ext_bitmap(&heap)); if (new_meta_hdr->data_csum_type != 0) fromhexstr(meta_entry["data_csum"].string_value(), new_data_csum_size, wr->get_checksums(&heap)); + wr->crc32c = wr->calc_crc32c(); + assert((uint8_t*)wr + wr->size == new_meta_buf + meta_offset + used_space); + auto j_it = journal_by_object.find(oid); if (j_it != journal_by_object.end()) { for (auto & rec: j_it->second) { - wr->next_pos = wr->get_size(&heap); - wr = wr->next(); - wr->next_pos = 0; - wr->lsn = ++next_lsn; - wr->version = rec["ver"].uint64_value(); uint64_t wr_offset = rec["offset"].uint64_value(); uint64_t wr_len = rec["len"].uint64_value(); + if (!(wr = get_wr(rec["type"] == "small_write" || rec["type"] == "small_write_instant" + ? heap.get_small_entry_size(wr_offset, wr_len) : heap.get_big_entry_size()))) + { + goto close_err; + } + wr->inode = oid.inode; + wr->stripe = oid.stripe; + wr->lsn = ++next_lsn; + wr->version = rec["ver"].uint64_value(); if (rec["type"] == "small_write" || rec["type"] == "small_write_instant") { if (wr_len > 0 && !rec["data"].is_string()) @@ -862,12 +822,10 @@ close_err: assert(0); } wr->size = wr->get_size(&heap); + wr->crc32c = wr->calc_crc32c(); + assert((uint8_t*)wr + wr->size == new_meta_buf + meta_offset + used_space); } } - obj->crc32c = obj->calc_crc32c(); - assert(((uint8_t*)wr + wr->size - (uint8_t*)obj) == space_for_this); - used_space += space_for_this; - total_used_space += space_for_this; } if (used_space > 0 && used_space < new_meta_hdr->meta_block_size-2) { diff --git a/src/disk_tool/disk_tool_resize.cpp b/src/disk_tool/disk_tool_resize.cpp index 83c9b452..526e8157 100644 --- a/src/disk_tool/disk_tool_resize.cpp +++ b/src/disk_tool/disk_tool_resize.cpp @@ -37,9 +37,9 @@ int disk_tool_t::raw_resize() { resize_init(hdr); }, - [this](blockstore_heap_t *heap, heap_object_t *obj, uint32_t meta_block_num) + [this](blockstore_heap_t *heap, heap_entry_t *obj, uint32_t meta_block_num) { - for (auto wr = obj->get_writes(); wr; wr = wr->next()) + for (auto wr = obj; wr; wr = heap->prev(wr)) { if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) { @@ -542,7 +542,7 @@ int disk_tool_t::resize_rebuild_meta() memset(new_meta_buf, 0, new_meta_len); new_meta_hdr = (blockstore_meta_header_v3_t *)new_meta_buf; } - std::vector writes; + std::vector writes; int r = process_meta( [&](blockstore_meta_header_v3_t *hdr) { @@ -563,102 +563,91 @@ int disk_tool_t::resize_rebuild_meta() build_journal_start(); } }, - [&](blockstore_heap_t *heap, heap_object_t *obj, uint32_t meta_block_num) + [&](blockstore_heap_t *heap, heap_entry_t *wr, uint32_t meta_block_num) { - for (auto wr = obj->get_writes(); wr; wr = wr->next()) + if (wr->type() == BS_HEAP_BIG_WRITE) { - if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) + uint64_t block_num = wr->big().block_num; + auto remap_it = data_remap.find(block_num); + if (remap_it != data_remap.end()) + block_num = remap_it->second; + if (block_num < free_first || block_num >= total_blocks-free_last) { - uint64_t block_num = wr->big().block_num; - auto remap_it = data_remap.find(block_num); - if (remap_it != data_remap.end()) - block_num = remap_it->second; - if (block_num < free_first || block_num >= total_blocks-free_last) - { - fprintf(stderr, "BUG: remapped block %ju not in range %ju..%ju\n", block_num, free_first, total_blocks-free_last); - exit(1); - } - block_num += data_idx_diff; - wr->big().block_num = block_num; - } - else if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE) - { - if (new_heap && wr->small().len > 0) - { - if (new_journal_ptr-new_journal_buf+wr->small().len > new_journal_len) - { - fprintf(stderr, "Small write data doesn't fit into the new buffer area\n"); - exit(1); - } - memcpy(new_journal_ptr, buffer_area+wr->small().location, wr->small().len); - wr->small().location = new_journal_ptr-new_journal_buf; - new_journal_ptr += wr->small().len; - } - } - else if (!new_heap) - { - fprintf(stderr, "Object %jx:%jx can't be converted to the old format because it contains %s\n", - obj->inode, obj->stripe, (wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_TOMBSTONE - ? "a tombstone" : ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE ? "an intent_write entry" : "an unknown entry")); + fprintf(stderr, "BUG: remapped block %ju not in range %ju..%ju\n", block_num, free_first, total_blocks-free_last); exit(1); } + block_num += data_idx_diff; + wr->big().block_num = block_num; + } + else if (wr->type() == BS_HEAP_SMALL_WRITE) + { + if (new_heap && wr->small().len > 0) + { + if (new_journal_ptr-new_journal_buf+wr->small().len > new_journal_len) + { + fprintf(stderr, "Small write data doesn't fit into the new buffer area\n"); + exit(1); + } + memcpy(new_journal_ptr, buffer_area+wr->small().location, wr->small().len); + wr->small().location = new_journal_ptr-new_journal_buf; + new_journal_ptr += wr->small().len; + } + } + // FIXME skip BS_HEAP_DELETE + else if (!new_heap) + { + fprintf(stderr, "Object %jx:%jx can't be converted to the old format because it contains %s\n", + wr->inode, wr->stripe, ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE ? "an intent_write entry" : "an unknown entry")); + exit(1); } if (new_heap) { // New -> New - new_heap->copy_object(obj, NULL); + //new_heap->copy_object(obj, NULL); } else { // Fill journal - writes.clear(); - for (auto wr = obj->get_writes(); wr; wr = wr->next()) + // It should be done in order + assert((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE || wr->entry_type == BS_HEAP_BIG_WRITE); + uint32_t je_size = ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE + ? sizeof(journal_entry_small_write) + dsk.dirty_dyn_size(wr->small().offset, wr->small().len) + : sizeof(journal_entry_big_write) + dsk.dirty_dyn_size(0, dsk.data_block_size)); + choose_journal_block(je_size); + journal_entry *je = (journal_entry*)(new_journal_ptr + new_journal_in_pos); + je->magic = JOURNAL_MAGIC; + je->type = (wr->entry_type & BS_HEAP_STABLE) ? JE_SMALL_WRITE_INSTANT : JE_SMALL_WRITE; + je->size = je_size; + je->crc32_prev = new_crc32_prev; + je->small_write.oid = (object_id){ .inode = wr->inode, .stripe = wr->stripe }; + je->small_write.version = wr->version; + if (wr->type() == BS_HEAP_SMALL_WRITE) { - writes.push_back(wr); + je->small_write.offset = wr->small().offset; + je->small_write.len = wr->small().len; + je->small_write.data_offset = new_journal_data-new_journal_buf; + if (je->small_write.data_offset + je->small_write.len > new_journal_len) + { + fprintf(stderr, "Error: live entries don't fit to the new journal\n"); + exit(1); + } + memcpy(new_journal_data, buffer_area+wr->small().location, je->small_write.len); + new_journal_data += je->small_write.len; + if (dsk.data_csum_type == 0 && wr->get_checksum(heap)) + je->small_write.crc32_data = *wr->get_checksum(heap); } - for (ssize_t i = writes.size()-2; i >= 0; i--) + else { - auto wr = writes[i]; - assert((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE || wr->entry_type == BS_HEAP_BIG_WRITE); - uint32_t je_size = ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE - ? sizeof(journal_entry_small_write) + dsk.dirty_dyn_size(wr->small().offset, wr->small().len) - : sizeof(journal_entry_big_write) + dsk.dirty_dyn_size(0, dsk.data_block_size)); - choose_journal_block(je_size); - journal_entry *je = (journal_entry*)(new_journal_ptr + new_journal_in_pos); - je->magic = JOURNAL_MAGIC; - je->type = (wr->entry_type & BS_HEAP_STABLE) ? JE_SMALL_WRITE_INSTANT : JE_SMALL_WRITE; - je->size = je_size; - je->crc32_prev = new_crc32_prev; - je->small_write.oid = (object_id){ .inode = obj->inode, .stripe = obj->stripe }; - je->small_write.version = wr->version; - if (wr->type() == BS_HEAP_SMALL_WRITE) - { - je->small_write.offset = wr->small().offset; - je->small_write.len = wr->small().len; - je->small_write.data_offset = new_journal_data-new_journal_buf; - if (je->small_write.data_offset + je->small_write.len > new_journal_len) - { - fprintf(stderr, "Error: live entries don't fit to the new journal\n"); - exit(1); - } - memcpy(new_journal_data, buffer_area+wr->small().location, je->small_write.len); - new_journal_data += je->small_write.len; - if (dsk.data_csum_type == 0 && wr->get_checksum(heap)) - je->small_write.crc32_data = *wr->get_checksum(heap); - } - else - { - je->big_write.location = wr->big_location(heap); - } - memcpy((uint8_t*)je + je->size, wr->get_ext_bitmap(heap), new_clean_entry_bitmap_size); - if (dsk.data_csum_type != 0 && wr->get_checksums(heap)) - { - memcpy((uint8_t*)je + je->size + new_clean_entry_bitmap_size, wr->get_checksums(heap), wr->get_csum_size(heap)); - } - je->crc32 = je_crc32(je); - new_journal_in_pos += je->size; - new_crc32_prev = je->crc32; + je->big_write.location = wr->big_location(heap); } + memcpy((uint8_t*)je + je->size, wr->get_ext_bitmap(heap), new_clean_entry_bitmap_size); + if (dsk.data_csum_type != 0 && wr->get_checksums(heap)) + { + memcpy((uint8_t*)je + je->size + new_clean_entry_bitmap_size, wr->get_checksums(heap), heap->get_csum_size(wr)); + } + je->crc32 = je_crc32(je); + new_journal_in_pos += je->size; + new_crc32_prev = je->crc32; // New -> Old if (writes[writes.size()-1]->entry_type == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) { @@ -667,7 +656,7 @@ int disk_tool_t::resize_rebuild_meta() clean_disk_entry *new_entry = (clean_disk_entry*)(new_meta_buf + dsk.meta_block_size + dsk.meta_block_size*(block_num / new_entries_per_block) + new_clean_entry_size*(block_num % new_entries_per_block)); - new_entry->oid = (object_id){ .inode = obj->inode, .stripe = obj->stripe }; + new_entry->oid = (object_id){ .inode = wr->inode, .stripe = wr->stripe }; new_entry->version = big_wr->version; memcpy(new_entry->bitmap, big_wr->get_ext_bitmap(heap), new_clean_entry_bitmap_size); memcpy(new_entry->bitmap + new_clean_entry_bitmap_size, big_wr->get_int_bitmap(heap), new_clean_entry_bitmap_size); @@ -691,11 +680,12 @@ int disk_tool_t::resize_rebuild_meta() if (new_heap) { // Old -> New - uint8_t wr_buf[new_heap->get_max_write_entry_size()]; - heap_write_t *wr = (heap_write_t*)wr_buf; + heap_entry_t *wr = NULL; wr->entry_type = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE; + wr->inode = entry->oid.inode; + wr->stripe = entry->oid.stripe; + wr->version = entry->version; wr->big().block_num = block_num; - wr->next_pos = 0; wr->size = wr->get_size(new_heap); if (bitmap) { @@ -703,7 +693,7 @@ int disk_tool_t::resize_rebuild_meta() memcpy(wr->get_int_bitmap(new_heap), bitmap+new_clean_entry_bitmap_size, new_clean_entry_bitmap_size); memcpy(wr->get_checksums(new_heap), bitmap+2*new_clean_entry_bitmap_size, new_data_csum_size); } - new_heap->post_write(entry->oid, wr, NULL, NULL); + // FIXME add } else { diff --git a/src/test/ringloop_mock.cpp b/src/test/ringloop_mock.cpp index 336d1a38..fc718bb0 100644 --- a/src/test/ringloop_mock.cpp +++ b/src/test/ringloop_mock.cpp @@ -171,6 +171,7 @@ disk_mock_t::disk_mock_t(size_t size, bool buffered) this->size = size; this->data = (uint8_t*)malloc_or_die(size); this->buffered = buffered; + memset(this->data, 0, size); } disk_mock_t::~disk_mock_t() @@ -238,16 +239,18 @@ void disk_mock_t::clear(size_t offset, size_t len) void disk_mock_t::discard_buffers(bool all, uint32_t seed) { - if (trace) - printf("disk: discard buffers all=%d seed=%u\n", all, seed); if (all) { + if (trace) + printf("disk: discard all buffers (%zu)\n", buffers.size()); for (auto & b: buffers) free(b.second.iov_base); buffers.clear(); } else { + if (trace) + printf("disk: discard random buffers seed=%u\n", seed); std::mt19937 rnd(seed); for (auto it = buffers.begin(); it != buffers.end(); ) { diff --git a/src/test/test_blockstore.cpp b/src/test/test_blockstore.cpp index ef4836bb..3faa975b 100644 --- a/src/test/test_blockstore.cpp +++ b/src/test/test_blockstore.cpp @@ -240,12 +240,14 @@ static void test_fsync(bool separate_meta) if (separate_meta) { test.config["meta_device"] = "./test_meta.bin"; - test.config["disable_meta_fsync"] = "1"; + test.config["disable_meta_fsync"] = "0"; test.config["meta_device_size"] = "33554432"; test.config["meta_device_sect"] = "4096"; test.config["data_offset"] = "0"; } test.init(); + if (test.meta_disk) + test.meta_disk->trace = 1; // Write printf("writing\n"); @@ -263,7 +265,10 @@ static void test_fsync(bool separate_meta) // Destroy and restart without sync printf("destroying\n"); test.destroy_bs(); - test.data_disk->discard_buffers(true, 0); + if (separate_meta) + test.meta_disk->discard_buffers(true, 0); + else + test.data_disk->discard_buffers(true, 0); test.init(); // Check ENOENT @@ -292,7 +297,10 @@ static void test_fsync(bool separate_meta) // Discard and restart again printf("destroying again\n"); test.destroy_bs(); - test.data_disk->discard_buffers(true, 0); + if (separate_meta) + test.meta_disk->discard_buffers(true, 0); + else + test.data_disk->discard_buffers(true, 0); test.init(); // Check that it's present now @@ -353,7 +361,7 @@ static void test_padded_csum_intent(bool perfect) // Write printf("writing\n"); blockstore_op_t op; - op.opcode = BS_OP_WRITE; + op.opcode = BS_OP_WRITE_STABLE; op.oid = { .inode = 1, .stripe = 0 }; op.version = 1; op.offset = 8192; @@ -386,36 +394,16 @@ static void test_padded_csum_intent(bool perfect) test.exec_op(&op); assert(op.retval == op.len); - // Write again (small because uncompactable) - printf("writing (small)\n"); - op.version = 3; - op.offset = 60*1024; - memset(op.buf, 0xcc, 4096); - test.exec_op(&op); - assert(op.retval == op.len); - - // Check that these are really big+intent+small writes + // Check that these are really big+intent writes // (intent is not collapsible because of csum_block_size > bitmap_granularity) - heap_object_t *obj = test.bs->heap->read_entry((object_id){ .inode = 1, .stripe = 0 }, NULL); - assert(obj); - assert(obj->get_writes()->next()); - assert(obj->get_writes()->next()->next()); - assert(!obj->get_writes()->next()->next()->next()); - assert(obj->get_writes()->entry_type == BS_HEAP_SMALL_WRITE); - assert(obj->get_writes()->next()->entry_type == (perfect ? BS_HEAP_SMALL_WRITE : BS_HEAP_INTENT_WRITE)); - assert(obj->get_writes()->next()->next()->entry_type == BS_HEAP_BIG_WRITE); - - // Commit - printf("commit version 3\n"); - op.opcode = BS_OP_STABLE; - op.len = 1; - *((obj_ver_id*)op.buf) = { - .oid = { .inode = 1, .stripe = 0 }, - .version = 3, - }; - test.exec_op(&op); - assert(op.retval == 0); - assert(test.bs->heap->get_compact_queue_size()); + heap_entry_t *obj = test.bs->heap->read_entry((object_id){ .inode = 1, .stripe = 0 }); + auto wr = obj; + assert(wr); + assert(wr->entry_type == (perfect ? BS_HEAP_SMALL_WRITE : BS_HEAP_INTENT_WRITE) | BS_HEAP_STABLE); + wr = test.bs->heap->prev(wr); + assert(wr); + assert(wr->entry_type == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE); + assert(!test.bs->heap->prev(wr)); // Trigger & wait compaction test.bs->flusher->request_trim(); @@ -436,12 +424,10 @@ static void test_padded_csum_intent(bool perfect) assert(memcheck(op2.buf+8*1024, 0xaa, 4*1024)); assert(memcheck(op2.buf+12*1024, 0, 16*1024)); assert(memcheck(op2.buf+28*1024, 0xbb, 4*1024)); - assert(memcheck(op2.buf+32*1024, 0, 28*1024)); - assert(memcheck(op2.buf+60*1024, 0xcc, 4*1024)); - assert(memcheck(op2.buf+64*1024, 0, 64*1024)); + assert(memcheck(op2.buf+32*1024, 0, 96*1024)); - obj = test.bs->heap->read_entry((object_id){ .inode = 1, .stripe = 0 }, NULL); - assert(!obj->get_writes()->next()); + obj = test.bs->heap->read_entry((object_id){ .inode = 1, .stripe = 0 }); + assert(!test.bs->heap->prev(obj)); free(op.buf); free(op2.buf);