diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index d3f4c3c9..20caf611 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -33,7 +33,11 @@ uint32_t heap_write_t::get_csum_size(blockstore_heap_t *heap) { if (!heap->dsk->csum_block_size) { - return ((flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE ? 4 : 0); + return ((flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE || (flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE ? 4 : 0); + } + if ((flags & BS_HEAP_TYPE) == BS_HEAP_TOMBSTONE) + { + return 0; } if ((flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) { @@ -48,28 +52,32 @@ uint32_t heap_write_t::get_csum_size(blockstore_heap_t *heap) bool heap_write_t::needs_recheck(blockstore_heap_t *heap) { - return len > 0 && lsn >= heap->compacted_lsn && (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) || flags == BS_HEAP_SMALL_WRITE); + return len > 0 && lsn >= heap->compacted_lsn && (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) + || flags == BS_HEAP_SMALL_WRITE || flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)); } bool heap_write_t::needs_compact(uint64_t compacted_lsn) { - return lsn > compacted_lsn && flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); + return lsn > compacted_lsn && (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) || flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)); } bool heap_write_t::is_compacted(uint64_t compacted_lsn) { - return lsn <= compacted_lsn && flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE); + return lsn <= compacted_lsn && (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) || flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)); } bool heap_write_t::can_be_collapsed(blockstore_heap_t *heap) { - return !heap->dsk->csum_block_size || heap->dsk->csum_block_size == heap->dsk->bitmap_granularity || + return flags == BS_HEAP_INTENT_WRITE || + !heap->dsk->csum_block_size || heap->dsk->csum_block_size == heap->dsk->bitmap_granularity || !(offset % heap->dsk->csum_block_size) && !(len % heap->dsk->csum_block_size); } bool heap_write_t::is_allowed_before_compacted(uint64_t compacted_lsn, bool is_last_entry) { - return lsn <= compacted_lsn && flags == ((is_last_entry ? BS_HEAP_BIG_WRITE : BS_HEAP_SMALL_WRITE) | BS_HEAP_STABLE); + return lsn <= compacted_lsn && (is_last_entry + ? (flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)) + : (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) || flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE))); } uint8_t *heap_write_t::get_ext_bitmap(blockstore_heap_t *heap) @@ -90,7 +98,8 @@ uint8_t *heap_write_t::get_checksums(blockstore_heap_t *heap) { if (!heap->dsk->csum_block_size || !len) return NULL; - if ((flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE) + if ((flags & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE || + (flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE) return ((uint8_t*)this + sizeof(heap_write_t) + heap->dsk->clean_entry_bitmap_size); if ((flags & BS_HEAP_TYPE) != BS_HEAP_BIG_WRITE) return NULL; @@ -99,7 +108,8 @@ uint8_t *heap_write_t::get_checksums(blockstore_heap_t *heap) uint32_t *heap_write_t::get_checksum(blockstore_heap_t *heap) { - if (heap->dsk->csum_block_size || (flags & BS_HEAP_TYPE) != BS_HEAP_SMALL_WRITE || !len) + if (heap->dsk->csum_block_size || !len || + (flags & BS_HEAP_TYPE) != BS_HEAP_SMALL_WRITE && (flags & BS_HEAP_TYPE) != BS_HEAP_INTENT_WRITE) return NULL; return (uint32_t*)((uint8_t*)this + sizeof(heap_write_t) + heap->dsk->clean_entry_bitmap_size); } @@ -525,7 +535,7 @@ skip_object: { if (wr->needs_recheck(this)) { - if (!buffer_area) + if (!buffer_area || (wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE) { to_recheck = true; } @@ -705,13 +715,8 @@ bool blockstore_heap_t::calc_block_checksums(uint32_t *block_csums, uint8_t *dat return res; } -bool blockstore_heap_t::recheck_small_writes(std::function)> read_buffer, int queue_depth) +bool blockstore_heap_t::recheck_small_writes(std::function)> read_buffer, int queue_depth) { - if (buffer_area) - { - // Already checked - return true; - } if (in_recheck) { // Recheck already entered @@ -733,9 +738,21 @@ bool blockstore_heap_t::recheck_small_writes(std::functionneeds_recheck(this)) { + bool is_intent = (wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)); + uint64_t loc = wr->location; + if (is_intent) + { + auto next_wr = wr->next(); + assert(next_wr && next_wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)); + loc = wr->offset + next_wr->location; + } recheck_in_progress++; uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, wr->len); - recheck_cb(wr->location, wr->len, buf, [this, oid, lsn = wr->lsn, buf]() + if (log_level > 5) + { + fprintf(stderr, "Notice: rechecking %u bytes at %ju in %s area\n", wr->len, loc, is_intent ? "data" : "buffer"); + } + recheck_cb(is_intent, loc, wr->len, buf, [this, oid, lsn = wr->lsn, buf]() { uint32_t block_num = 0; heap_object_t *obj = read_entry(oid, &block_num); @@ -784,7 +801,9 @@ bool blockstore_heap_t::recheck_small_writes(std::functionis_compacted(max_lsn)) { *begin_wr = wr; + *end_wr = wr; } - if (*begin_wr) + else if (*begin_wr) { bool is_last = !wr->next(); if (is_last) @@ -976,7 +996,7 @@ uint32_t blockstore_heap_t::compact_object_to(heap_object_t *obj, uint64_t compa big_wr = wr; } // all subsequent small write entries must also be compacted - assert(wr->is_allowed_before_compacted(compact_lsn, is_last)); + assert(compacted_wr_count == 1 || wr->is_allowed_before_compacted(compact_lsn, is_last)); if (!new_csums && !wr->can_be_collapsed(this)) { skip = true; @@ -1331,8 +1351,6 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea return EINVAL; } } - const uint32_t offset = find_block_space(block_num, wr_size); - assert(offset != UINT32_MAX); if (modified_block) { *modified_block = block_num; @@ -1357,17 +1375,32 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea mvcc_buffer_refs[wr->location]++; } } + const uint32_t offset = find_block_space(block_num, wr_size); + assert(offset != UINT32_MAX); + memcpy(inf.data + offset, wr, wr_size); + heap_write_t *new_wr = (heap_write_t*)(inf.data + offset); int32_t used_delta = wr_size; if (is_overwrite) { free_object_space(obj->inode, obj->get_writes(), NULL); // Free old write entries used_delta -= free_writes(obj->get_writes(), NULL); + new_wr->next_pos = 0; + } + else if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE && + (obj->get_writes()->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE) + { + assert(wr->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)); + auto second_wr = obj->get_writes()->next(); + free_object_space(obj->inode, obj->get_writes(), second_wr); + used_delta -= free_writes(obj->get_writes(), second_wr); + new_wr->next_pos = (uint8_t*)second_wr - (uint8_t*)new_wr; + } + else + { + new_wr->next_pos = ((uint8_t*)obj + obj->write_pos) - (uint8_t*)new_wr; } - memcpy(inf.data + offset, wr, wr_size); - heap_write_t *new_wr = (heap_write_t*)(inf.data + offset); new_wr->size = wr_size; - new_wr->next_pos = (is_overwrite ? 0 : ((uint8_t*)obj + obj->write_pos) - (uint8_t*)new_wr); new_wr->lsn = ++next_lsn; if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) { diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index e0025d18..065fd27f 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -19,11 +19,12 @@ struct pool_shard_settings_t uint32_t pg_stripe_size; }; -#define BS_HEAP_TYPE 3 +#define BS_HEAP_TYPE 7 #define BS_HEAP_SMALL_WRITE 1 #define BS_HEAP_BIG_WRITE 2 #define BS_HEAP_TOMBSTONE 3 -#define BS_HEAP_STABLE 4 +#define BS_HEAP_INTENT_WRITE 4 +#define BS_HEAP_STABLE 8 class blockstore_heap_t; @@ -150,7 +151,7 @@ class blockstore_heap_t std::deque recheck_queue; int recheck_in_progress = 0; bool in_recheck = false; - std::function)> recheck_cb; + std::function)> recheck_cb; int recheck_queue_depth = 0; const uint32_t max_write_entry_size; @@ -180,7 +181,7 @@ public: // finish loading void finish_load(); // recheck small write data after reading the database from disk - bool recheck_small_writes(std::function)> read_buffer, int queue_depth); + bool recheck_small_writes(std::function)> read_buffer, int queue_depth); // initialize metadata area (fill it with empty data) // returns 0 when done, EAGAIN when the caller has to wait more int initialize(); diff --git a/src/blockstore/blockstore_init.cpp b/src/blockstore/blockstore_init.cpp index 52158859..6acf8ad0 100644 --- a/src/blockstore/blockstore_init.cpp +++ b/src/blockstore/blockstore_init.cpp @@ -250,37 +250,34 @@ resume_4: return 1; } } - if (!bs->dsk.inmemory_journal) - { - // asynchronous recheck - bs->heap->recheck_small_writes([this](uint64_t offset, uint64_t len, uint8_t *buf, std::function cb) - { - if (!buf) - { - wait_state = 7; - bs->ringloop->wakeup(); - return; - } - GET_SQE(); - data->iov = (iovec){ buf, len }; - data->callback = [this, offset, cb](ring_data_t *data) - { - if (data->res < 0) - { - fprintf(stderr, "Buffer area read failed at offset %ju: %d\n", offset, data->res); - exit(1); - } - cb(); - }; - io_uring_prep_readv(sqe, bs->dsk.journal_fd, &data->iov, 1, bs->dsk.journal_offset + offset); - bs->ringloop->submit(); - }, bs->meta_write_recheck_parallelism); + // asynchronous recheck resume_6: - wait_state = 6; - return 1; + wait_state = 6; + bs->heap->recheck_small_writes([this](bool is_data, uint64_t offset, uint64_t len, uint8_t *buf, std::function cb) + { + if (!buf) + { + wait_state = 7; + bs->ringloop->wakeup(); + return; + } + GET_SQE(); + data->iov = (iovec){ buf, len }; + data->callback = [this, offset, cb](ring_data_t *data) + { + if (data->res < 0) + { + fprintf(stderr, "Buffer area read failed at offset %ju: %d\n", offset, data->res); + exit(1); + } + cb(); + }; + io_uring_prep_readv(sqe, (is_data ? bs->dsk.data_fd : bs->dsk.journal_fd), &data->iov, 1, + (is_data ? bs->dsk.data_offset : bs->dsk.journal_offset) + offset); + bs->ringloop->submit(); + }, bs->meta_write_recheck_parallelism); + return 1; resume_7: - ; - } free(metadata_buffer); metadata_buffer = NULL; return 0; diff --git a/src/blockstore/blockstore_read.cpp b/src/blockstore/blockstore_read.cpp index 6c8eedf7..aa363fad 100644 --- a/src/blockstore/blockstore_read.cpp +++ b/src/blockstore/blockstore_read.cpp @@ -109,7 +109,7 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *op) 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) { - if (wr->offset >= end || wr->offset+wr->len <= start) + if (wr->offset >= end || wr->offset+wr->len <= start || (wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE) { return 0; } diff --git a/src/blockstore/blockstore_write.cpp b/src/blockstore/blockstore_write.cpp index 2ec340b0..db7e633e 100644 --- a/src/blockstore/blockstore_write.cpp +++ b/src/blockstore/blockstore_write.cpp @@ -29,7 +29,9 @@ void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, int retval) { // 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; } @@ -91,7 +93,8 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) } // FIXME: Allow to do initial writes as buffered, not redirected // FIXME: Allow to do direct writes over holes - else if (!obj || op->offset == 0 && op->len == dsk.data_block_size) + else if (!obj || (obj->get_writes()->flags & BS_HEAP_TYPE) == BS_HEAP_TOMBSTONE || + op->offset == 0 && op->len == dsk.data_block_size) { // Big (redirect) write PRIV(op)->is_big = true; @@ -144,6 +147,55 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) PRIV(op)->op_state = 1; write_iodepth++; } + // Only one INTENT_WRITE is allowed at a time, but in fact, + // parallel writes to the same object are forbidden anyway + else if (disable_data_fsync && + op->opcode == BS_OP_WRITE_STABLE && + op->len > 0 && op->len <= dsk.bitmap_granularity /* FIXME atomic_write_size */ && + (obj->get_writes()->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) || + obj->get_writes()->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE))) + { + // Direct intent-write + BS_SUBMIT_CHECK_SQES(1); + for (auto wr = obj->get_writes(); wr; wr = wr->next()) + { + assert(wr->flags != BS_HEAP_BIG_WRITE); + if (wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)) + { + PRIV(op)->location = wr->location; + } + } + uint8_t wr_buf[heap->get_max_write_entry_size()]; + heap_write_t *wr = (heap_write_t*)wr_buf; + wr->version = op->version; + wr->offset = op->offset; + wr->len = op->len; + wr->location = 0; + wr->flags = BS_HEAP_INTENT_WRITE | BS_HEAP_STABLE; + 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); + uint32_t modified_block; + int res = heap->post_write(op->oid, wr, &modified_block); + if (res == ENOSPC) + { + if (!heap->get_compact_queue_size() && !flusher->get_active()) + { + // no space + cancel_all_writes(op, -ENOSPC); + return 2; + } + PRIV(op)->wait_for = WAIT_COMPACTION; + PRIV(op)->wait_detail = flusher->get_counter(); + flusher->request_trim(); + return 0; + } + assert(res == 0); + prepare_meta_block_write(op, modified_block); + unsynced_small_write_count++; + PRIV(op)->op_state = 9; + write_iodepth++; + } else { // Small (buffered) overwrite @@ -170,7 +222,6 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) memcpy(wr->get_ext_bitmap(heap), op->bitmap, dsk.clean_entry_bitmap_size); heap->calc_checksums(wr, (uint8_t*)op->buf, true); uint32_t modified_block; - heap->use_buffer_area(op->oid.inode, loc, op->len); int res = heap->post_write(op->oid, wr, &modified_block); if (res == ENOSPC) { @@ -186,6 +237,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) return 0; } assert(res == 0); + heap->use_buffer_area(op->oid.inode, loc, op->len); prepare_meta_block_write(op, modified_block); if (op->len > 0) { @@ -203,17 +255,9 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op) // Zero-length overwrite. Allowed to bump object version in EC placement groups without actually writing data } unsynced_small_write_count++; - if (!PRIV(op)->pending_ops) - { - PRIV(op)->op_state = 6; - write_iodepth++; - return continue_write(op); - } - else - { - PRIV(op)->op_state = 5; - write_iodepth++; - } + assert(PRIV(op)->pending_ops); + PRIV(op)->op_state = 5; + write_iodepth++; } return 1; } @@ -229,6 +273,8 @@ int blockstore_impl_t::continue_write(blockstore_op_t *op) goto resume_6; else if (op_state == 8) goto resume_8; + else if (op_state == 10) + goto resume_10; else { // In progress @@ -318,6 +364,26 @@ resume_8: write_iodepth--; FINISH_OP(op); return 2; +resume_10: + // Direct intent-write + heap_object_t *obj = heap->read_entry(op->oid, NULL); + uint64_t loc = UINT64_MAX; + for (auto wr = obj->get_writes(); wr; wr = wr->next()) + { + if (wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)) + loc = wr->location; + } + if (loc != PRIV(op)->location) + { + goto resume_8; + } + BS_SUBMIT_GET_SQE(sqe, data); + data->iov = (struct iovec){ op->buf, op->len }; + data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); }; + io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + loc + op->offset); + PRIV(op)->pending_ops++; + PRIV(op)->op_state = 7; + return 1; } void blockstore_impl_t::handle_write_event(ring_data_t *data, blockstore_op_t *op) diff --git a/src/test/test_heap.cpp b/src/test/test_heap.cpp index 15b61ada..5154dd6c 100644 --- a/src/test/test_heap.cpp +++ b/src/test/test_heap.cpp @@ -569,11 +569,12 @@ void test_recheck(bool async, bool csum) if (async) { int calls = 0; - bool done = heap.recheck_small_writes([&](uint64_t offset, uint64_t len, uint8_t *buf, std::function cb) + bool done = heap.recheck_small_writes([&](bool is_data, uint64_t offset, uint64_t len, uint8_t *buf, std::function cb) { calls++; if (len) { + assert(!is_data); assert(len == 4096); assert(offset == 16384 || offset == 20480); assert(cb);