Experimental INTENT_WRITE write mode with WA=2 instead of 3

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<void(uint64_t, uint64_t, uint8_t*, std::function<void()>)> read_buffer, int queue_depth)
+bool blockstore_heap_t::recheck_small_writes(std::function<void(bool is_data, uint64_t offset, uint64_t len, uint8_t* buf, std::function<void()>)> 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::function<void(uint64_t, uint64
         {
             if (wr->needs_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::function<void(uint64_t, uint64
         auto cb = std::move(recheck_cb);
         recheck_queue_depth = 0;
         if (cb)
-            cb(0, 0, NULL, NULL);
+        {
+            cb(false, 0, 0, NULL, NULL);
+        }
         return true;
     }
     return false;
@@ -941,8 +960,9 @@ void blockstore_heap_t::get_compact_range(heap_object_t *obj, uint64_t max_lsn,
         if (wr->is_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)
     {

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<object_id> recheck_queue;
     int recheck_in_progress = 0;
     bool in_recheck = false;
-    std::function<void(uint64_t, uint64_t, uint8_t*, std::function<void()>)> recheck_cb;
+    std::function<void(bool is_data, uint64_t offset, uint64_t len, uint8_t* buf, std::function<void()>)> 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<void(uint64_t, uint64_t, uint8_t*, std::function<void()>)> read_buffer, int queue_depth);
+    bool recheck_small_writes(std::function<void(bool is_data, uint64_t offset, uint64_t len, uint8_t* buf, std::function<void()>)> 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();

src/blockstore/blockstore_init.cpp

@@ -250,37 +250,34 @@ resume_4:
             return 1;
         }
     }
-    if (!bs->dsk.inmemory_journal)
+    // asynchronous recheck
-    {
-        // asynchronous recheck
-        bs->heap->recheck_small_writes([this](uint64_t offset, uint64_t len, uint8_t *buf, std::function<void()> 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);
 resume_6:
-        wait_state = 6;
+    wait_state = 6;
-        return 1;
+    bs->heap->recheck_small_writes([this](bool is_data, uint64_t offset, uint64_t len, uint8_t *buf, std::function<void()> 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;

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<copy_buffer_t> & 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;
     }

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)
+        assert(PRIV(op)->pending_ops);
-        {
+        PRIV(op)->op_state = 5;
-            PRIV(op)->op_state = 6;
+        write_iodepth++;
-            write_iodepth++;
-            return continue_write(op);
-        }
-        else
-        {
-            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)

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<void()> cb)
+            bool done = heap.recheck_small_writes([&](bool is_data, uint64_t offset, uint64_t len, uint8_t *buf, std::function<void()> cb)
             {
                 calls++;
                 if (len)
                 {
+                    assert(!is_data);
                     assert(len == 4096);
                     assert(offset == 16384 || offset == 20480);
                     assert(cb);