Reject all-zero blocks, mark empty space with a special entry type during init

src/blockstore/blockstore_heap.cpp

@@ -16,9 +16,10 @@
 
 #define BS_HEAP_FREE_MVCC 1
 #define BS_HEAP_FREE_MAIN 2
-#define FREE_SPACE_BIT 0x8000
 #define META_ALLOC_LEVELS 8
 
+#define BS_HEAP_FREE_SPACE 0xAB8F
+
 #define HEAP_INFLIGHT_DONE 1
 #define HEAP_INFLIGHT_COMPACTABLE 2
 #define HEAP_INFLIGHT_COMPACTED 4
@@ -289,20 +290,11 @@ int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uin
         {
             uint8_t *data = buf + buf_offset + block_offset;
             heap_entry_t *wr = (heap_entry_t*)data;
-            if (!wr->size)
+            if (wr->size > dsk->meta_block_size-block_offset)
-            {
-                // Block or the rest of block is apparently empty
-                // FIXME: Prevent all-zero blocks
-                if (block_offset > 0)
-                {
-                    wr->size = FREE_SPACE_BIT | (dsk->meta_block_size - block_offset);
-                }
-                break;
-            }
-            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). ",
-                    block_num, block_offset, (wr->size & ~FREE_SPACE_BIT), dsk->meta_block_size-block_offset);
+                    block_num, block_offset, wr->size, dsk->meta_block_size-block_offset);
+corrupted_block:
                 if (allow_corrupted)
                 {
                     fprintf(stderr, "Metadata block is corrupted, skipping\n");
@@ -314,11 +306,21 @@ int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uin
                     return EDOM;
                 }
             }
-            if (wr->size & FREE_SPACE_BIT)
+            if (dsk->meta_block_size-block_offset < sizeof(heap_entry_t) ||
+                wr->size >= 4 && wr->entry_type == BS_HEAP_FREE_SPACE)
             {
-                // Free space
+                // Empty end of the block - required to be filled with heap_empty_pattern
-                block_offset += (wr->size & ~FREE_SPACE_BIT);
+                if (wr->size != dsk->meta_block_size-block_offset || wr->size >= 4 && wr->entry_type != BS_HEAP_FREE_SPACE)
-                continue;
+                {
+                    goto corrupted_block;
+                }
+                break;
+            }
+            if (wr->size < sizeof(heap_entry_t))
+            {
+                fprintf(stderr, "Error: entry is too small in metadata block %u at %u (%u < min %zu bytes). ",
+                    block_num, block_offset, wr->size, sizeof(heap_entry_t));
+                goto corrupted_block;
             }
             wr->entry_type &= ~BS_HEAP_GARBAGE;
             if ((wr->entry_type & BS_HEAP_TYPE) < BS_HEAP_BIG_WRITE ||
@@ -343,18 +345,22 @@ corrupted_object:
                     return EDOM;
                 }
             }
+            if (((wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_SMALL_WRITE ||
+                (wr->entry_type & BS_HEAP_TYPE) == BS_HEAP_INTENT_WRITE) &&
+                wr->size < sizeof(heap_small_write_t))
+            {
+                // Small writes require accessing offset & len to calculate correct length,
+                // so require at least sizeof(heap_small_write_t) for them
+                fprintf(stderr, "Error: entry %jx:%jx v%ju has invalid size in metadata block %u at %u (%u < min %zu bytes). Metadata is corrupted, aborting\n",
+                    wr->inode, wr->stripe, wr->version, block_num, block_offset, wr->size, sizeof(heap_small_write_t));
+                goto corrupted_object;
+            }
             if (wr->entry_type == BS_HEAP_COMMIT && !wr->version)
             {
                 fprintf(stderr, "Error: commit entry has zero version in metadata block %u at %u. ",
                     block_num, block_offset);
                 goto corrupted_object;
             }
-            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));
-                goto corrupted_object;
-            }
             // Verify crc
             uint32_t expected_crc32c = wr->calc_crc32c();
             if (wr->crc32c != expected_crc32c)
@@ -1989,7 +1995,39 @@ void blockstore_heap_t::get_meta_block(uint32_t block_num, uint8_t *buffer)
         pos += li->entry.size;
     }
     assert(pos <= dsk->meta_block_size);
-    memset(buffer+pos, 0, dsk->meta_block_size-pos);
+    if (pos <= dsk->meta_block_size-2)
+    {
+        *((uint16_t*)(buffer+pos)) = dsk->meta_block_size-pos;
+        pos += 2;
+    }
+    if (pos <= dsk->meta_block_size-2)
+    {
+        *((uint16_t*)(buffer+pos)) = BS_HEAP_FREE_SPACE;
+        pos += 2;
+    }
+    if (pos < dsk->meta_block_size)
+    {
+        memset(buffer+pos, 0, dsk->meta_block_size-pos);
+    }
+}
+
+void blockstore_heap_t::fill_block_empty_space(uint8_t *buffer, uint32_t pos)
+{
+    if (pos > dsk->meta_block_size)
+    {
+        buffer += (pos / dsk->meta_block_size) * dsk->meta_block_size;
+        pos = pos % dsk->meta_block_size;
+    }
+    if (pos <= dsk->meta_block_size-2)
+    {
+        *((uint16_t*)(buffer+pos)) = dsk->meta_block_size-pos;
+        pos += 2;
+    }
+    if (pos <= dsk->meta_block_size-2)
+    {
+        *((uint16_t*)(buffer+pos)) = BS_HEAP_FREE_SPACE;
+        pos += 2;
+    }
 }
 
 uint32_t blockstore_heap_t::get_meta_block_used_space(uint32_t block_num)

src/blockstore/blockstore_heap.h

@@ -333,6 +333,7 @@ public:
 
     // get metadata block data buffer and used space
     void get_meta_block(uint32_t block_num, uint8_t *buffer);
+    void fill_block_empty_space(uint8_t *buffer, uint32_t pos);
     uint32_t get_meta_block_used_space(uint32_t block_num);
 
     // get space usage statistics

src/blockstore/blockstore_init.cpp

@@ -211,8 +211,10 @@ resume_4:
                     io_uring_prep_readv(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset + bufs[i].offset);
                 else
                 {
-                    // Fill metadata with zeroes
+                    // Fill metadata with empty block pattern
-                    memset(data->iov.iov_base, 0, data->iov.iov_len);
+                    memset(bufs[i].buf, 0, bufs[i].size);
+                    for (uint64_t o = 0; o < bufs[i].size; o += bs->dsk.meta_block_size)
+                        bs->heap->fill_block_empty_space(bufs[i].buf + o, 0);
                     io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset + bufs[i].offset);
                 }
                 bs->ringloop->submit();

src/blockstore/blockstore_open.cpp

@@ -96,6 +96,10 @@ void blockstore_impl_t::parse_config(blockstore_config_t & config, bool init)
     {
         metadata_buf_size = 4*1024*1024;
     }
+    if (metadata_buf_size % dsk.meta_block_size)
+    {
+        throw std::runtime_error("metadata_buf_size should be a multiple of meta_block_size");
+    }
     if (!meta_write_recheck_parallelism)
     {
         meta_write_recheck_parallelism = 16;

src/disk_tool/disk_tool_meta.cpp

@@ -116,6 +116,7 @@ close_error:
         {
             obj_fn(heap, obj, meta_block_num);
         });
+        obj_fn(heap, NULL, 0);
         delete heap;
     }
     else if (hdr->zero == 0 && hdr->magic == BLOCKSTORE_META_MAGIC_V1)
@@ -288,6 +289,8 @@ int disk_tool_t::dump_meta()
         },
         [this](blockstore_heap_t *heap, heap_entry_t *obj, uint32_t meta_block_num)
         {
+            if (!obj)
+                return;
             if (dump_as_old)
                 dump_heap_entry_as_old(heap, obj);
             else

src/disk_tool/disk_tool_resize.cpp

@@ -641,6 +641,21 @@ int disk_tool_t::resize_rebuild_meta()
         },
         [&](blockstore_heap_t *heap, heap_entry_t *obj, uint32_t meta_block_num)
         {
+            if (!obj)
+            {
+                // Finish
+                if (new_meta_format == BLOCKSTORE_META_FORMAT_HEAP)
+                {
+                    heap->fill_block_empty_space(new_meta_buf, new_meta_pos);
+                    new_meta_pos = (new_meta_pos/dsk.meta_block_size + 1) * dsk.meta_block_size;
+                    while (new_meta_pos < new_meta_len)
+                    {
+                        heap->fill_block_empty_space(new_meta_buf, new_meta_pos);
+                        new_meta_pos += dsk.meta_block_size;
+                    }
+                }
+                return;
+            }
             auto handle_write = [&](heap_entry_t *wr, bool stable)
             {
                 if (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_BIG_INTENT)
@@ -663,6 +678,7 @@ int disk_tool_t::resize_rebuild_meta()
                     // New -> New
                     if ((new_meta_pos % dsk.meta_block_size) + wr->size > dsk.meta_block_size)
                     {
+                        heap->fill_block_empty_space(new_meta_buf, new_meta_pos);
                         new_meta_pos = (new_meta_pos/dsk.meta_block_size + 1) * dsk.meta_block_size;
                         if (new_meta_pos >= new_meta_len)
                         {