Add 2 tests for intent writes

src/blockstore/blockstore_flush.cpp

@@ -56,6 +56,7 @@ journal_flusher_t::~journal_flusher_t()
 
 journal_flusher_co::~journal_flusher_co()
 {
+    free_buffers();
     if (csum_buf)
     {
         free(csum_buf);
@@ -235,7 +236,7 @@ resume_1:
 #endif
     flusher->active_flushers++;
     // Scan versions to flush
-    read_vec.clear();
+    free_buffers();
     for (auto wr = begin_wr; wr != end_wr; wr = wr->next())
     {
         bs->prepare_read(read_vec, cur_obj, wr, 0, bs->dsk.data_block_size);
@@ -295,18 +296,20 @@ resume_8:
     else if (res == ENOENT || res == EDOM)
     {
         // Abort compaction
-        goto release_oid;
+        flusher->active_flushers--;
+        goto resume_0;
     }
     assert(res == 0);
     // Submit data writes
     copy_count = 0;
     for (i = 0; i < read_vec.size(); i++)
     {
-        if (read_vec[i].copy_flags & COPY_BUF_JOURNAL)
+        if ((read_vec[i].copy_flags & COPY_BUF_JOURNAL) &&
+            !(read_vec[i].copy_flags & COPY_BUF_COALESCED))
         {
             assert(read_vec[i].buf);
             await_sqe(9);
-            data->iov = (struct iovec){ read_vec[i].buf, (size_t)read_vec[i].len };
+            data->iov = (struct iovec){ (bs->dsk.inmemory_journal ? bs->buffer_area + read_vec[i].disk_offset : read_vec[i].buf), (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);
             wait_count++;
@@ -328,7 +331,7 @@ resume_10:
     if (!cur_obj)
     {
         // Abort compaction
-        goto release_oid;
+        goto resume_0;
     }
     calc_block_checksums();
     if (read_to_fill_incomplete)
@@ -341,8 +344,7 @@ resume_24:
         }
     }
     bs->heap->mark_object_compacted(cur_obj, compact_lsn);
-    // Done, free all buffers
+    // Done
-    free_buffers();
 #ifdef BLOCKSTORE_DEBUG
     printf("Compacted %jx:%jx l%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, compact_lsn, copy_count);
 #endif
@@ -360,7 +362,6 @@ resume_15:
         if (!trim_lsn(11))
             return false;
     }
-release_oid:
     if (should_repeat)
     {
         // Flush the same object again
@@ -374,20 +375,23 @@ void journal_flusher_co::iterate_partial_overwrites(std::function<int(int, uint3
 {
     int prev = 0;
     uint32_t prev_begin = 0, prev_end = 0;
-    for (int i = 0; i < read_vec.size() && !(read_vec[i].copy_flags & COPY_BUF_CSUM_FILL); i++)
+    for (int i = 0; i < read_vec.size(); i++)
     {
-        if (read_vec[i].copy_flags != (COPY_BUF_JOURNAL|COPY_BUF_COALESCED))
+        if (!(read_vec[i].copy_flags & COPY_BUF_COALESCED))
         {
             if (read_vec[i].offset > prev_end)
             {
-                i += cb(prev, prev_begin, prev_end);
+                if (prev_end > prev_begin && ((prev_begin % bs->dsk.csum_block_size) || (prev_end % bs->dsk.csum_block_size)))
+                {
+                    i += cb(prev, prev_begin, prev_end);
+                }
                 prev = i;
                 prev_begin = read_vec[i].offset;
             }
             prev_end = read_vec[i].offset + read_vec[i].len;
         }
     }
-    if (prev_end > prev_begin)
+    if (prev_end > prev_begin && ((prev_begin % bs->dsk.csum_block_size) || (prev_end % bs->dsk.csum_block_size)))
     {
         cb(prev, prev_begin, prev_end);
     }
@@ -401,14 +405,12 @@ void journal_flusher_co::iterate_checksum_holes(std::function<void(int, uint32_t
         if ((prev_begin % bs->dsk.csum_block_size) &&
             (prev_begin / bs->dsk.csum_block_size) != (prev_end / bs->dsk.csum_block_size))
         {
-            cb(pos, prev_begin, prev_begin + bs->dsk.csum_block_size - prev_begin%bs->dsk.csum_block_size);
+            cb(pos, prev_begin - prev_begin%bs->dsk.csum_block_size, prev_begin);
             r++;
         }
-        if ((prev_end % bs->dsk.csum_block_size) ||
+        if (prev_end % bs->dsk.csum_block_size)
-            (prev_begin % bs->dsk.csum_block_size) &&
-            (prev_end / bs->dsk.csum_block_size) == (prev_begin / bs->dsk.csum_block_size))
         {
-            cb(i, prev_end - (prev_end % bs->dsk.csum_block_size ? (prev_end % bs->dsk.csum_block_size) : bs->dsk.csum_block_size), prev_end);
+            cb(i, prev_end, prev_end - (prev_end % bs->dsk.csum_block_size) + bs->dsk.csum_block_size);
             r++;
         }
         return r;
@@ -420,17 +422,18 @@ void journal_flusher_co::fill_partial_checksum_blocks()
     iterate_checksum_holes([&](int vec_pos, uint32_t hole_start, uint32_t hole_end)
     {
         read_to_fill_incomplete = true;
-        int out_pos = read_vec.size();
+        bs->prepare_disk_read(read_vec, read_vec.size(), cur_obj, end_wr,
-        bs->prepare_disk_read(read_vec, out_pos, cur_obj, end_wr,
             hole_start - hole_start % bs->dsk.csum_block_size, hole_start - hole_start % bs->dsk.csum_block_size + bs->dsk.csum_block_size,
             hole_start - hole_start % bs->dsk.csum_block_size, hole_start - hole_start % bs->dsk.csum_block_size + bs->dsk.csum_block_size,
             COPY_BUF_CSUM_FILL | (bs->padded_csum_update ? 0 : COPY_BUF_SKIP_CSUM));
-        out_pos--;
+        auto & vec = read_vec[read_vec.size()-1];
+        if (!vec.buf)
+            vec.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, vec.disk_len);
         read_vec.insert(read_vec.begin()+vec_pos, (copy_buffer_t){
             .copy_flags = COPY_BUF_JOURNAL|COPY_BUF_COALESCED,
             .offset = hole_start,
             .len = hole_end-hole_start,
-            .buf = read_vec[out_pos].buf + hole_start - read_vec[out_pos].offset,
+            .buf = vec.buf + hole_start - vec.offset,
         });
     });
 }
@@ -458,7 +461,7 @@ 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_locked_entry(cur_oid, copy_id); // FIXME locks can be removed from flusher
     bool csum_ok = true;
     for (int i = 0; i < read_vec.size(); i++)
     {
@@ -469,9 +472,11 @@ int journal_flusher_co::check_and_punch_checksums()
             while (wr && wr->lsn != vec.wr_lsn)
                 wr = wr->next();
             assert(wr);
+            uint32_t *csums = (uint32_t*)(wr->get_checksums(bs->heap)
+                + (vec.offset/bs->dsk.csum_block_size)*(bs->dsk.data_csum_type & 0xFF)
+                - (((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE) ? 0 : (wr->offset/bs->dsk.csum_block_size)*(bs->dsk.data_csum_type & 0xFF)));
             bs->heap->calc_block_checksums(
-                (uint32_t*)((uint8_t*)wr->get_checksums(bs->heap) + vec.offset/bs->dsk.csum_block_size*(bs->dsk.data_csum_type & 0xFF)),
+                csums, vec.buf, wr->get_int_bitmap(bs->heap), vec.offset, vec.offset+vec.len, false,
-                vec.buf, wr->get_int_bitmap(bs->heap), vec.offset, vec.offset+vec.len, false,
                 [&](uint32_t mismatch_pos, uint32_t expected_csum, uint32_t real_csum)
                 {
                     printf("Checksum mismatch in object %jx:%jx v%ju in %s area at offset 0x%jx: got %08x, expected %08x\n",
@@ -546,7 +551,7 @@ void journal_flusher_co::calc_block_checksums()
     for (auto it = read_vec.begin(); it != read_vec.end(); it++)
     {
         if (it->copy_flags & COPY_BUF_CSUM_FILL)
-            break;
+            continue;
         if (block_done == 0)
         {
             // `read_vec` should contain aligned items, possibly split into pieces
@@ -637,6 +642,8 @@ bool journal_flusher_co::read_buffered(int wait_base)
         {
             await_sqe(0);
             auto & vec = read_vec[i];
+            if (!vec.buf)
+                vec.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, vec.disk_len);
             data->iov = (struct iovec){ vec.buf, (size_t)vec.disk_len };
             wait_count++;
             io_uring_prep_readv(

src/blockstore/blockstore_heap.cpp

@@ -1366,7 +1366,9 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea
     else if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE &&
         (first_wr->flags & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE)
     {
+        // FIXME: All other types of writes should also purge&merge the intent write
         auto second_wr = first_wr->next();
+        second_wr->version = first_wr->version;
         bitmap_set(second_wr->get_int_bitmap(this), first_wr->offset, first_wr->len, dsk->bitmap_granularity);
         if (dsk->csum_block_size)
         {

src/blockstore/blockstore_impl.h

@@ -76,6 +76,7 @@ struct blockstore_op_private_t
 
 class blockstore_impl_t: public blockstore_i
 {
+public:
     blockstore_disk_t dsk;
 
     /******* OPTIONS *******/
@@ -134,10 +135,6 @@ class blockstore_impl_t: public blockstore_i
         return ringloop->get_sqe();
     }
 
-    friend class blockstore_init_meta;
-    friend class journal_flusher_t;
-    friend class journal_flusher_co;
-
     void open_data();
     void open_meta();
     void open_journal();
@@ -159,7 +156,7 @@ class blockstore_impl_t: public blockstore_i
     uint32_t prepare_read_with_bitmaps(std::vector<copy_buffer_t> & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end);
     uint32_t prepare_read_zero(std::vector<copy_buffer_t> & read_vec, uint32_t start, uint32_t end);
     uint32_t prepare_read_simple(std::vector<copy_buffer_t> & read_vec, heap_object_t *obj, heap_write_t *wr, uint32_t start, uint32_t end);
-    void prepare_disk_read(std::vector<copy_buffer_t> & read_vec, int & pos, heap_object_t *obj, heap_write_t *wr,
+    void prepare_disk_read(std::vector<copy_buffer_t> & read_vec, int pos, heap_object_t *obj, heap_write_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<copy_buffer_t> & read_vec, uint32_t item_start, uint32_t item_end,
         std::function<void(int&, uint32_t, uint32_t)> callback);
@@ -186,7 +183,7 @@ class blockstore_impl_t: public blockstore_i
     // List
     void process_list(blockstore_op_t *op);
 
-public:
+/*public:*/
 
     blockstore_impl_t(blockstore_config_t & config, ring_loop_i *ringloop, timerfd_manager_t *tfd, bool mock_mode = false);
     ~blockstore_impl_t();

src/blockstore/blockstore_internal.h

@@ -41,10 +41,10 @@
 // Suspend operation until there are <wait_detail> bytes of free space in the journal on disk
 #define WAIT_COMPACTION 2
 
-#define COPY_BUF_JOURNAL 1
+#define COPY_BUF_JOURNAL    0x01
-#define COPY_BUF_DATA 2
+#define COPY_BUF_DATA       0x02
-#define COPY_BUF_ZERO 4
+#define COPY_BUF_ZERO       0x04
-#define COPY_BUF_CSUM_FILL 8
+#define COPY_BUF_CSUM_FILL  0x08
-#define COPY_BUF_COALESCED 16
+#define COPY_BUF_COALESCED  0x10
-#define COPY_BUF_PADDED 32
+#define COPY_BUF_PADDED     0x20
-#define COPY_BUF_SKIP_CSUM 64
+#define COPY_BUF_SKIP_CSUM  0x40

src/blockstore/blockstore_read.cpp

@@ -195,7 +195,7 @@ uint32_t blockstore_impl_t::prepare_read_simple(std::vector<copy_buffer_t> & rea
         else if (dsk.csum_block_size <= dsk.bitmap_granularity)
         {
             // simple disk read
-            prepare_disk_read(read_vec, pos, obj, wr, start, end, start, end, 0);
+            prepare_disk_read(read_vec, pos++, obj, wr, start, end, start, end, 0);
         }
         else
         {
@@ -221,7 +221,7 @@ uint32_t blockstore_impl_t::prepare_read_simple(std::vector<copy_buffer_t> & rea
                 blk_end == end && blk_start == start)
             {
                 // single block, two partial blocks, or any number of full blocks
-                prepare_disk_read(read_vec, pos, obj, wr, blk_start, blk_end, start, end, skip_csum);
+                prepare_disk_read(read_vec, pos++, obj, wr, blk_start, blk_end, start, end, skip_csum);
             }
             else
             {
@@ -229,18 +229,18 @@ uint32_t blockstore_impl_t::prepare_read_simple(std::vector<copy_buffer_t> & rea
                 uint32_t full_start = (blk_start != start ? blk_start+dsk.csum_block_size : blk_start);
                 uint32_t full_end = (blk_end != end ? blk_end-dsk.csum_block_size : blk_end);
                 if (blk_start != start)
-                    prepare_disk_read(read_vec, pos, obj, wr, blk_start, full_start, start, full_start, skip_csum);
+                    prepare_disk_read(read_vec, pos++, obj, wr, blk_start, full_start, start, full_start, skip_csum);
                 if (full_start > full_end)
-                    prepare_disk_read(read_vec, pos, obj, wr, full_start, full_end, full_start, full_end, skip_csum);
+                    prepare_disk_read(read_vec, pos++, obj, wr, full_start, full_end, full_start, full_end, skip_csum);
                 if (blk_end != end)
-                    prepare_disk_read(read_vec, pos, obj, wr, full_end, blk_end, full_end, end, skip_csum);
+                    prepare_disk_read(read_vec, pos++, obj, wr, full_end, blk_end, full_end, end, skip_csum);
             }
         }
     });
     return res;
 }
 
-void blockstore_impl_t::prepare_disk_read(std::vector<copy_buffer_t> & read_vec, int & pos, heap_object_t *obj, heap_write_t *wr,
+void blockstore_impl_t::prepare_disk_read(std::vector<copy_buffer_t> & read_vec, int pos, heap_object_t *obj, heap_write_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
@@ -256,19 +256,19 @@ void blockstore_impl_t::prepare_disk_read(std::vector<copy_buffer_t> & read_vec,
     if (blk_start != start || blk_end != end)
     {
         vec.copy_flags |= COPY_BUF_PADDED;
-        if (read_vec.size() > pos && (read_vec[pos].copy_flags & ~COPY_BUF_CSUM_FILL) == vec.copy_flags &&
+        if (pos > 0 && read_vec.size() >= pos && (read_vec[pos-1].copy_flags & ~COPY_BUF_CSUM_FILL) == vec.copy_flags &&
-            read_vec[pos].offset >= blk_start && read_vec[pos].offset+read_vec[pos].len <= blk_end)
+            read_vec[pos-1].offset >= blk_start && read_vec[pos-1].offset+read_vec[pos-1].len <= blk_end)
         {
             // This is the same block as the previous one, we can read it only once
             vec.copy_flags |= COPY_BUF_COALESCED;
-            vec.buf = read_vec[pos].buf;
+            vec.buf = read_vec[pos-1].buf;
         }
         else
         {
             vec.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, blk_end-blk_start);
         }
     }
-    read_vec.insert(read_vec.begin() + (pos++), vec);
+    read_vec.insert(read_vec.begin() + pos, vec);
 }
 
 void blockstore_impl_t::find_holes(std::vector<copy_buffer_t> & read_vec,

src/blockstore/blockstore_stable.cpp

@@ -17,6 +17,7 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
     // 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)
     {
         uint32_t modified_block = 0;

src/blockstore/blockstore_write.cpp

@@ -177,13 +177,14 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     {
         // Direct intent-write
         BS_SUBMIT_CHECK_SQES(1);
-        for (auto wr = obj->get_writes(); wr; wr = wr->next())
+        if ((obj->get_writes()->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
         {
-            assert(wr->flags != BS_HEAP_BIG_WRITE);
+            PRIV(op)->location = obj->get_writes()->location;
-            if (wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE))
+        }
-            {
+        else
-                PRIV(op)->location = wr->location;
+        {
-            }
+            assert((obj->get_writes()->next()->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE);
+            PRIV(op)->location = obj->get_writes()->next()->location;
         }
 process_intent:
         uint8_t wr_buf[heap->get_max_write_entry_size()];

src/test/test_blockstore.cpp

@@ -132,6 +132,14 @@ struct bs_test_t
     }
 };
 
+static bool memcheck(uint8_t *buf, uint8_t byte, size_t len)
+{
+    for (size_t i = 0; i < len; i++)
+        if (buf[i] != byte)
+            return false;
+    return true;
+}
+
 static void test_simple()
 {
     printf("\n-- test_simple\n");
@@ -274,10 +282,144 @@ static void test_fsync(bool separate_meta)
     free(op2.buf);
 }
 
+static void test_intent_over_unstable()
+{
+    printf("\n-- test_intent_over_unstable\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.init();
+
+    // Write
+    printf("writing\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 20480;
+    op.len = 4096;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
+    memset(op.buf, 0xaa, 4096);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+    // Write again
+    printf("writing again\n");
+    op.version = 2;
+    op.offset = 28*1024;
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+    free(op.buf);
+}
+
+static void test_padded_csum_intent()
+{
+    printf("\n-- test_padded_csum_intent\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["csum_block_size"] = "16384";
+    test.init();
+
+    // Write
+    printf("writing\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 8192;
+    op.len = 4096;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
+    memset(op.buf, 0xaa, 4096);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+    // Read
+    printf("reading\n");
+    blockstore_op_t op2;
+    op2.opcode = BS_OP_READ;
+    op2.oid = { .inode = 1, .stripe = 0 };
+    op2.version = UINT64_MAX;
+    op2.offset = 0;
+    op2.len = 128*1024;
+    op2.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    test.exec_op(&op2);
+    assert(op2.retval == op2.len);
+    assert(is_zero(op2.buf, 8*1024));
+    assert(memcmp(op2.buf+8*1024, op.buf, 4*1024) == 0);
+    assert(is_zero(op2.buf+12*1024, 116*1024));
+
+    // Write again (intent)
+    printf("writing (intent)\n");
+    op.version = 2;
+    op.offset = 28*1024;
+    memset(op.buf, 0xbb, 4096);
+    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
+    // (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()->next()->next());
+    assert(obj->get_writes()->flags == BS_HEAP_SMALL_WRITE);
+    assert(obj->get_writes()->next()->flags == BS_HEAP_INTENT_WRITE);
+    assert(obj->get_writes()->next()->next()->flags == 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());
+
+    // Trigger & wait compaction
+    test.bs->flusher->request_trim();
+    // FIXME: Не зацикливаться при обломе
+    while (test.bs->heap->get_compact_queue_size())
+        test.ringloop->loop();
+    test.bs->flusher->release_trim();
+
+    // Read again and check
+    printf("reading compacted\n");
+    op2.version = UINT64_MAX;
+    test.exec_op(&op2);
+    assert(op2.retval == op2.len);
+    assert(memcheck(op2.buf, 0, 8*1024));
+    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));
+
+    obj = test.bs->heap->read_entry((object_id){ .inode = 1, .stripe = 0 }, NULL);
+    assert(!obj->get_writes()->next());
+
+    free(op.buf);
+    free(op2.buf);
+}
+
 int main(int narg, char *args[])
 {
     test_simple();
     test_fsync(false);
     test_fsync(true);
+    test_intent_over_unstable();
+    test_padded_csum_intent();
     return 0;
 }