Fix disk_tool for lsmeta

src/blockstore/blockstore_heap.cpp

@@ -260,8 +260,9 @@ void blockstore_heap_t::start_load(uint64_t completed_lsn)
     this->completed_lsn = completed_lsn;
 }
 
-int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uint8_t *buf,
+int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uint8_t *buf, bool allow_corrupted,
-    std::function<void(uint32_t block_num, heap_entry_t* wr)> handle_write, std::function<void(uint32_t, uint32_t, uint8_t*)> handle_block)
+    std::function<void(uint32_t block_num, heap_entry_t* wr)> handle_write,
+    std::function<void(uint32_t, uint32_t, uint8_t*)> handle_block)
 {
     for (uint64_t buf_offset = 0; buf_offset < disk_size; buf_offset += dsk->meta_block_size)
     {
@@ -284,9 +285,18 @@ int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uin
             }
             if ((wr->size & ~FREE_SPACE_BIT) > dsk->meta_block_size-block_offset)
             {
-                fprintf(stderr, "Error: entry is too large in metadata block %u at %u (%u > max %u bytes). Metadata is corrupted, aborting\n",
+                fprintf(stderr, "Error: entry is too large in metadata block %u at %u (%u > max %u bytes). ",
                     block_num, block_offset, (wr->size & ~FREE_SPACE_BIT), dsk->meta_block_size-block_offset);
-                return EDOM;
+                if (allow_corrupted)
+                {
+                    fprintf(stderr, "Metadata block is corrupted, skipping\n");
+                    break;
+                }
+                else
+                {
+                    fprintf(stderr, "Metadata is corrupted, aborting\n");
+                    return EDOM;
+                }
             }
             if (wr->size & FREE_SPACE_BIT)
             {
@@ -299,15 +309,26 @@ int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uin
                 (wr->entry_type & BS_HEAP_TYPE) > BS_HEAP_ROLLBACK ||
                 (wr->entry_type & ~(BS_HEAP_TYPE|BS_HEAP_STABLE)))
             {
-                fprintf(stderr, "Error: entry has unknown type %u in metadata block %u at %u. Metadata is corrupted, aborting\n",
+                fprintf(stderr, "Error: entry has unknown type %u in metadata block %u at %u. ",
                     wr->entry_type, block_num, block_offset);
-                return EDOM;
+corrupted_object:
+                if (allow_corrupted)
+                {
+                    fprintf(stderr, "Entry is corrupted, skipping\n");
+                    block_offset += wr->size;
+                    continue;
+                }
+                else
+                {
+                    fprintf(stderr, "Metadata is corrupted, aborting\n");
+                    return EDOM;
+                }
             }
             if (wr->size != wr->get_size(this))
             {
                 fprintf(stderr, "Error: entry %jx:%jx v%ju has invalid size in metadata block %u at %u (%u != expected %u bytes). Metadata is corrupted, aborting\n",
                     wr->inode, wr->stripe, wr->version, block_num, block_offset, wr->size, wr->get_size(this));
-                return EDOM;
+                goto corrupted_object;
             }
             // Verify crc
             uint32_t expected_crc32c = wr->calc_crc32c();
@@ -316,7 +337,7 @@ int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uin
                 fprintf(stderr, "Error: entry %jx:%jx v%ju in metadata block %u at %u is corrupt (crc32c mismatch: expected %08x, got %08x). Metadata is corrupted, aborting\n",
                     wr->inode, wr->stripe, wr->version,
                     block_num, block_offset, expected_crc32c, wr->crc32c);
-                return EDOM;
+                goto corrupted_object;
             }
             handle_write(block_num, wr);
             block_offset += wr->size;
@@ -326,10 +347,10 @@ int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uin
     return 0;
 }
 
-int blockstore_heap_t::load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf, uint64_t &entries_loaded)
+int blockstore_heap_t::load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf, bool allow_corrupted, uint64_t &entries_loaded)
 {
     entries_loaded = 0;
-    return read_blocks(disk_offset, size, buf, [&](uint32_t block_num, heap_entry_t *wr_orig)
+    return read_blocks(disk_offset, size, buf, allow_corrupted, [&](uint32_t block_num, heap_entry_t *wr_orig)
     {
         heap_list_item_t *li = (heap_list_item_t*)malloc_or_die(wr_orig->size + sizeof(heap_list_item_t) - sizeof(heap_entry_t));
         li->block_num = block_num;
@@ -1562,6 +1583,21 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
     return res;
 }
 
+void blockstore_heap_t::iterate_objects(std::function<void(heap_entry_t*, uint32_t block_num)> cb)
+{
+    for (auto & pgp: block_index)
+    {
+        for (auto & ip: pgp.second)
+        {
+            for (auto & op: ip.second)
+            {
+                auto li = op.second.ptr;
+                cb(&li->entry, li->block_num);
+            }
+        }
+    }
+}
+
 int blockstore_heap_t::list_objects(uint32_t pg_num, object_id min_oid, object_id max_oid,
     obj_ver_id **result_list, size_t *stable_count, size_t *unstable_count)
 {

src/blockstore/blockstore_heap.h

@@ -213,10 +213,11 @@ public:
     ~blockstore_heap_t();
     void start_load(uint64_t completed_lsn);
     // load data from the disk, returns EDOM on corruption
-    int read_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf,
+    int read_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf, bool allow_corrupted,
         std::function<void(uint32_t block_num, heap_entry_t* wr)> handle_write,
         std::function<void(uint32_t, uint32_t, uint8_t*)> handle_block);
-    int load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf, uint64_t &entries_loaded);
+    int load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf,
+        bool allow_corrupted, uint64_t &entries_loaded);
     // finish loading
     void finish_load();
     // recheck small write data after reading the database from disk
@@ -272,6 +273,8 @@ public:
     // iterate compactable entries
     heap_compact_t iterate_compaction(heap_entry_t *obj, uint64_t fsynced_lsn, bool under_pressure,
         std::function<void(heap_entry_t*)> small_wr_cb);
+    // iterate all objects
+    void iterate_objects(std::function<void(heap_entry_t*, uint32_t block_num)> cb);
     // retrieve object listing from a PG
     int list_objects(uint32_t pg_num, object_id min_oid, object_id max_oid,
         obj_ver_id **result_list, size_t *stable_count, size_t *unstable_count);

src/blockstore/blockstore_init.cpp

@@ -224,7 +224,7 @@ resume_4:
         {
             // Handle result
             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);
+            int r = bs->heap->load_blocks(bufs[i].offset-bs->dsk.meta_block_size, bufs[i].size, bufs[i].buf, false, loaded);
             if (r != 0)
                 exit(1);
             entries_loaded += loaded;

src/disk_tool/disk_tool.h

@@ -76,8 +76,6 @@ struct disk_tool_t
     journal_entry_start je_start;
     uint8_t *new_journal_buf = NULL, *new_meta_buf = NULL, *new_journal_ptr = NULL, *new_journal_data = NULL;
     blockstore_meta_header_v3_t *new_meta_hdr = NULL;
-    blockstore_disk_t new_dsk;
-    blockstore_heap_t *new_heap = NULL;
     uint64_t new_journal_in_pos;
     int64_t data_idx_diff;
     uint64_t total_blocks, free_first, free_last;
@@ -132,6 +130,10 @@ struct disk_tool_t
     void choose_journal_block(uint32_t je_size);
     int resize_rebuild_journal();
     int resize_write_new_journal();
+    void remap_big_write(heap_entry_t *wr);
+    void remap_small_write(heap_entry_t *wr);
+    void fill_old_clean_entry(blockstore_heap_t *heap, heap_entry_t *big_wr);
+    void fill_old_journal_entry(blockstore_heap_t *heap, heap_entry_t *wr);
     int resize_rebuild_meta();
     int resize_write_new_meta();
     void free_new_meta();

src/disk_tool/disk_tool_discard.cpp

@@ -59,7 +59,8 @@ int disk_tool_t::trim_data(std::string device)
         {
             for (auto wr = obj; wr; wr = heap->prev(wr))
             {
-                if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
+                if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE ||
+                    (wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_INTENT)
                 {
                     data_alloc->set(wr->big_location(heap) / dsk.data_block_size, true);
                 }

src/disk_tool/disk_tool_meta.cpp

@@ -104,16 +104,18 @@ close_error:
         hdr = NULL;
         meta_pos = dsk.meta_block_size;
         lseek64(dsk.meta_fd, dsk.meta_offset+meta_pos, 0);
+        uint64_t entries_loaded = 0;
         while (meta_pos < dsk.meta_area_size)
         {
             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, [&](uint32_t block_num, heap_entry_t *obj)
+            r = heap->load_blocks(meta_pos-dsk.meta_block_size, read_len, data, true, entries_loaded);
-            {
-                obj_fn(heap, obj, block_num);
-            }, [](uint32_t, uint32_t, uint8_t*){});
             meta_pos += read_len;
         }
+        heap->iterate_objects([&](heap_entry_t* obj, uint32_t meta_block_num)
+        {
+            obj_fn(heap, obj, meta_block_num);
+        });
         delete heap;
     }
     else if (hdr->zero == 0 && hdr->magic == BLOCKSTORE_META_MAGIC_V1)
@@ -286,7 +288,10 @@ int disk_tool_t::dump_meta()
             if (dump_as_old)
                 dump_heap_entry_as_old(heap, obj);
             else
-                dump_heap_entry(heap, obj);
+            {
+                for (auto wr = obj; wr; wr = heap->prev(wr))
+                    dump_heap_entry(heap, obj);
+            }
         },
         [this](uint64_t block_num, clean_disk_entry *entry, uint8_t *bitmap) { dump_meta_entry(block_num, entry, bitmap); },
         true, true
@@ -340,7 +345,8 @@ void disk_tool_t::dump_heap_entry_as_old(blockstore_heap_t *heap, heap_entry_t *
     for (wr = obj; wr && !wr->is_overwrite(); wr = heap->prev(wr))
     {
     }
-    if (!wr || wr->entry_type != (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE))
+    if (!wr || wr->entry_type != (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) &&
+        wr->entry_type != (BS_HEAP_BIG_INTENT|BS_HEAP_STABLE))
     {
         return;
     }
@@ -391,18 +397,23 @@ void disk_tool_t::dump_heap_entry(blockstore_heap_t *heap, heap_entry_t *wr)
         INODE_POOL(wr->inode), INODE_NO_POOL(wr->inode), wr->stripe,
         wr->lsn, wr->version,
             t == BS_HEAP_BIG_WRITE ? "big" : (
+            t == BS_HEAP_BIG_INTENT ? "big_intent" : (
             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")))))),
+            "unknown"))))))),
         (wr->entry_type & BS_HEAP_STABLE) ? "true" : "false"
     );
     if (t == BS_HEAP_BIG_WRITE)
     {
         printf(",\"location\":%ju", wr->big_location(heap));
     }
+    else if (t == BS_HEAP_BIG_INTENT)
+    {
+        printf(",\"location\":%ju,\"offset\":%u,\"len\":%u", wr->big_location(heap), wr->big_intent().offset, wr->big_intent().len);
+    }
     else if (t == BS_HEAP_INTENT_WRITE)
     {
         printf(",\"offset\":%u,\"len\":%u", wr->small().offset, wr->small().len);
@@ -646,6 +657,8 @@ int disk_tool_t::write_json_heap(json11::Json meta, json11::Json journal)
                 wr_type = BS_HEAP_INTENT_WRITE;
             else if (meta_entry["type"] == "big")
                 wr_type = BS_HEAP_BIG_WRITE;
+            else if (meta_entry["type"] == "big_intent")
+                wr_type = BS_HEAP_BIG_INTENT;
             else if (meta_entry["type"] == "delete")
                 wr_type = BS_HEAP_DELETE;
             else if (meta_entry["type"] == "commit")
@@ -664,7 +677,11 @@ close_err0:
             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()));
+                : (wr_type == BS_HEAP_BIG_WRITE
+                    ? heap.get_big_entry_size()
+                    : (wr_type == BS_HEAP_BIG_INTENT
+                        ? heap.get_big_intent_entry_size()
+                        : heap.get_simple_entry_size())));
             if (!(wr = get_wr(wr_size)))
                 goto close_err0;
             wr->inode = oid.inode;
@@ -697,6 +714,16 @@ close_err0:
                 assert((loc / dsk.data_block_size) < 0xFFFF0000);
                 wr->set_big_location(&heap, loc);
             }
+            else if (wr_type == BS_HEAP_BIG_INTENT)
+            {
+                uint64_t loc = meta_entry["location"].uint64_value();
+                assert(!(loc % dsk.data_block_size));
+                assert((loc / dsk.data_block_size) < 0xFFFF0000);
+                auto & bi = wr->big_intent();
+                bi.block_num = loc / dsk.data_block_size;
+                bi.offset = wr_offset;
+                bi.len = wr_len;
+            }
             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));
@@ -714,7 +741,6 @@ close_err0:
                 *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);
         }
     }

src/disk_tool/disk_tool_resize.cpp

@@ -41,7 +41,8 @@ int disk_tool_t::raw_resize()
         {
             for (auto wr = obj; wr; wr = heap->prev(wr))
             {
-                if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
+                if ((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE ||
+                    (wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_BIG_INTENT)
                 {
                     data_alloc->set(wr->big().block_num, true);
                 }
@@ -517,31 +518,103 @@ int disk_tool_t::resize_write_new_journal()
     return 0;
 }
 
-int disk_tool_t::resize_rebuild_meta()
+void disk_tool_t::remap_big_write(heap_entry_t *wr)
 {
-    if (new_meta_format == BLOCKSTORE_META_FORMAT_HEAP)
+    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)
     {
-        new_dsk = dsk;
+        fprintf(stderr, "BUG: remapped block %ju not in range %ju..%ju\n", block_num, free_first, total_blocks-free_last);
-        new_dsk.data_offset = new_data_offset;
+        exit(1);
-        new_dsk.data_len = new_data_len;
+    }
-        new_dsk.block_count = new_data_len / dsk.data_block_size;
+    block_num += data_idx_diff;
-        new_dsk.journal_device = new_journal_device;
+    wr->big().block_num = block_num;
-        new_dsk.journal_offset = new_journal_offset;
+}
-        new_dsk.journal_len = new_journal_len;
+
-        new_dsk.meta_device = new_meta_device;
+void disk_tool_t::remap_small_write(heap_entry_t *wr)
-        new_dsk.meta_offset = new_meta_offset;
+{
-        new_dsk.meta_area_size = new_meta_len;
+    if (new_meta_format == BLOCKSTORE_META_FORMAT_HEAP && wr->small().len > 0)
-        new_dsk.meta_format = new_meta_format;
+    {
-        new_heap = new blockstore_heap_t(&new_dsk, NULL, 0);
+        if (new_journal_ptr-new_journal_buf+wr->small().len > new_journal_len)
-        new_meta_hdr = (blockstore_meta_header_v3_t *)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size);
+        {
-        memset(new_meta_hdr, 0, dsk.meta_block_size);
+            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;
+    }
+}
+
+void disk_tool_t::fill_old_clean_entry(blockstore_heap_t *heap, heap_entry_t *big_wr)
+{
+    uint64_t block_num = big_wr->big().block_num;
+    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 = big_wr->inode, .stripe = big_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);
+    memcpy(new_entry->bitmap + 2*new_clean_entry_bitmap_size, big_wr->get_checksums(heap), new_data_csum_size);
+    uint32_t *new_entry_csum = (uint32_t*)(((uint8_t*)new_entry) + new_clean_entry_size - 4);
+    *new_entry_csum = crc32c(0, new_entry, new_clean_entry_size - 4);
+}
+
+void disk_tool_t::fill_old_journal_entry(blockstore_heap_t *heap, heap_entry_t *wr)
+{
+    assert(wr->type() == BS_HEAP_SMALL_WRITE ||
+        wr->type() == BS_HEAP_BIG_WRITE ||
+        wr->type() == BS_HEAP_BIG_INTENT);
+    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)
+    {
+        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
     {
-        new_meta_buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, new_meta_len);
+        je->big_write.location = wr->big_location(heap);
-        memset(new_meta_buf, 0, new_meta_len);
-        new_meta_hdr = (blockstore_meta_header_v3_t *)new_meta_buf;
     }
+    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;
+}
+
+int disk_tool_t::resize_rebuild_meta()
+{
+    new_meta_buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, new_meta_len);
+    memset(new_meta_buf, 0, new_meta_len);
+    new_meta_hdr = (blockstore_meta_header_v3_t *)new_meta_buf;
+    meta_pos = dsk.meta_block_size;
+    uint64_t next_lsn = 0;
     std::vector<heap_entry_t*> writes;
     int r = process_meta(
         [&](blockstore_meta_header_v3_t *hdr)
@@ -563,107 +636,71 @@ int disk_tool_t::resize_rebuild_meta()
                 build_journal_start();
             }
         },
-        [&](blockstore_heap_t *heap, heap_entry_t *wr, uint32_t meta_block_num)
+        [&](blockstore_heap_t *heap, heap_entry_t *obj, uint32_t meta_block_num)
         {
-            if (wr->type() == BS_HEAP_BIG_WRITE)
+            heap->iterate_with_stable(obj, obj->lsn, [&](heap_entry_t *wr, bool stable)
             {
-                uint64_t block_num = wr->big().block_num;
+                if (wr->type() == BS_HEAP_DELETE && stable)
-                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);
+                    // Object is deleted, skip it
+                    return false;
+                }
+                else if (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_BIG_INTENT)
+                {
+                    remap_big_write(wr);
+                }
+                else if (wr->type() == BS_HEAP_SMALL_WRITE)
+                {
+                    remap_small_write(wr);
+                }
+                else if (new_meta_format != BLOCKSTORE_META_FORMAT_HEAP)
+                {
+                    fprintf(stderr, "Object %jx:%jx can't be converted to the old format because it contains an entry of type 0x%x%s\n",
+                        wr->inode, wr->stripe, wr->entry_type,
+                        (wr->type() == BS_HEAP_INTENT_WRITE ? " (intent_write)" : ""));
                     exit(1);
                 }
-                block_num += data_idx_diff;
+                if (new_meta_format == BLOCKSTORE_META_FORMAT_HEAP)
-                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)
+                    // New -> New
+                    if ((meta_pos % dsk.meta_block_size) + wr->size > dsk.meta_block_size)
                     {
-                        fprintf(stderr, "Small write data doesn't fit into the new buffer area\n");
+                        meta_pos = (meta_pos % dsk.meta_block_size) + dsk.meta_block_size;
-                        exit(1);
+                        if (meta_pos >= new_meta_len)
+                        {
+                            fprintf(stderr, "New metadata doesn't fit into the provided area\n");
+                            exit(1);
+                        }
                     }
-                    memcpy(new_journal_ptr, buffer_area+wr->small().location, wr->small().len);
+                    memcpy(new_meta_buf + meta_pos, wr, wr->size);
-                    wr->small().location = new_journal_ptr-new_journal_buf;
+                    if (wr->type() == BS_HEAP_BIG_WRITE && stable)
-                    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);
-            }
-            else
-            {
-                // Fill journal
-                // 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)
-                {
-                    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");
+                        // Skip older writes
-                        exit(1);
+                        return false;
                     }
-                    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);
+                    // New -> Old
+                    if (wr->type() == BS_HEAP_BIG_WRITE && stable)
+                    {
+                        fill_old_clean_entry(heap, wr);
+                        return false;
+                    }
+                    else
+                    {
+                        writes.push_back(wr);
+                    }
                 }
-                memcpy((uint8_t*)je + je->size, wr->get_ext_bitmap(heap), new_clean_entry_bitmap_size);
+                return true;
-                if (dsk.data_csum_type != 0 && wr->get_checksums(heap))
+            });
+            if (writes.size())
+            {
+                for (size_t i = writes.size(); i > 0; i--)
                 {
-                    memcpy((uint8_t*)je + je->size + new_clean_entry_bitmap_size, wr->get_checksums(heap), heap->get_csum_size(wr));
+                    fill_old_journal_entry(heap, writes[i-1]);
-                }
-                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)
-                {
-                    auto big_wr = writes[writes.size()-1];
-                    uint64_t block_num = big_wr->big().block_num;
-                    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 = 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);
-                    memcpy(new_entry->bitmap + 2*new_clean_entry_bitmap_size, big_wr->get_checksums(heap), new_data_csum_size);
-                    uint32_t *new_entry_csum = (uint32_t*)(((uint8_t*)new_entry) + new_clean_entry_size - 4);
-                    *new_entry_csum = crc32c(0, new_entry, new_clean_entry_size - 4);
                 }
+                writes.clear();
             }
         },
         [&](uint64_t block_num, clean_disk_entry *entry, uint8_t *bitmap)
@@ -677,23 +714,35 @@ int disk_tool_t::resize_rebuild_meta()
                 exit(1);
             }
             block_num += data_idx_diff;
-            if (new_heap)
+            if (new_meta_format == BLOCKSTORE_META_FORMAT_HEAP)
             {
                 // Old -> New
-                heap_entry_t *wr = NULL;
+                auto big_entry_size = sizeof(heap_big_write_t) + dsk.clean_entry_bitmap_size*2 +
+                    (!dsk.data_csum_type ? 0 : dsk.data_block_size/dsk.csum_block_size * (dsk.data_csum_type & 0xFF));
+                if ((meta_pos % dsk.meta_block_size) + big_entry_size > dsk.meta_block_size)
+                {
+                    meta_pos = (meta_pos % dsk.meta_block_size) + dsk.meta_block_size;
+                    if (meta_pos >= new_meta_len)
+                    {
+                        fprintf(stderr, "New metadata doesn't fit into the provided area\n");
+                        exit(1);
+                    }
+                }
+                heap_entry_t *wr = (heap_entry_t*)(new_meta_buf + meta_pos);
+                wr->size = big_entry_size;
                 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->size = wr->get_size(new_heap);
+                wr->lsn = ++next_lsn;
                 if (bitmap)
                 {
-                    memcpy(wr->get_ext_bitmap(new_heap), bitmap, new_clean_entry_bitmap_size);
+                    memcpy(((uint8_t*)wr) + sizeof(heap_big_write_t), bitmap, new_clean_entry_bitmap_size);
-                    memcpy(wr->get_int_bitmap(new_heap), bitmap+new_clean_entry_bitmap_size, new_clean_entry_bitmap_size);
+                    memcpy(((uint8_t*)wr) + sizeof(heap_big_write_t) + new_clean_entry_bitmap_size, 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);
+                    memcpy(((uint8_t*)wr) + sizeof(heap_big_write_t) + 2*new_clean_entry_bitmap_size, bitmap+2*new_clean_entry_bitmap_size, new_data_csum_size);
                 }
-                // FIXME add
+                wr->crc32c = wr->calc_crc32c();
             }
             else
             {
@@ -725,27 +774,7 @@ int disk_tool_t::resize_write_new_meta()
         return 1;
     }
     lseek64(new_meta_fd, new_meta_offset, 0);
-    if (new_meta_buf)
+    write_blocking(new_meta_fd, new_meta_buf, new_meta_len);
-    {
-        write_blocking(new_meta_fd, new_meta_buf, new_meta_len);
-    }
-    else
-    {
-        assert(new_heap);
-        uint32_t new_meta_blocks = new_meta_len / dsk.meta_block_size - 1;
-        const uint32_t nb = 1024;
-        uint8_t *data = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size*nb);
-        write_blocking(new_meta_fd, new_meta_hdr, dsk.meta_block_size);
-        for (uint32_t i = 0; i < new_meta_blocks; )
-        {
-            uint32_t j = 0;
-            for (j = 0; j < nb && i < new_meta_blocks; j++, i++)
-            {
-                new_heap->get_meta_block(i, data + j*dsk.meta_block_size);
-            }
-            write_blocking(new_meta_fd, data, j*dsk.meta_block_size);
-        }
-    }
     fsync(new_meta_fd);
     close(new_meta_fd);
     new_meta_fd = -1;
@@ -754,11 +783,6 @@ int disk_tool_t::resize_write_new_meta()
 
 void disk_tool_t::free_new_meta()
 {
-    if (new_heap)
-    {
-        delete new_heap;
-        new_heap = NULL;
-    }
     if ((uint8_t*)new_meta_hdr != new_meta_buf)
     {
         free(new_meta_hdr);

src/test/test_heap.cpp

@@ -550,7 +550,7 @@ void test_recheck(bool async, bool csum, bool intent)
 
         blockstore_heap_t heap(&dsk, async ? NULL : buffer_area.data(), 10);
         uint64_t entries_loaded;
-        heap.load_blocks(0, dsk.meta_block_size, tmp.data(), entries_loaded);
+        heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded);
 
         int calls = 0;
         bool done = heap.recheck_small_writes([&](bool is_data, uint64_t offset, uint64_t len, uint8_t *buf, std::function<void()> cb)
@@ -642,7 +642,7 @@ void test_corruption()
         blockstore_heap_t heap(&dsk, buffer_area.data());
         tmp.data()[10]++; // corrupt the first object
         uint64_t entries_loaded;
-        assert(heap.load_blocks(0, dsk.meta_block_size, tmp.data(), entries_loaded) == EDOM);
+        assert(heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded) == EDOM);
     }
 
     // reload heap with bad entry size
@@ -652,7 +652,7 @@ void test_corruption()
         entry->size++;
         entry->crc32c = entry->calc_crc32c();
         uint64_t entries_loaded;
-        assert(heap.load_blocks(0, dsk.meta_block_size, tmp.data(), entries_loaded) == EDOM);
+        assert(heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded) == EDOM);
     }
 
     printf("OK test_corruption\n");