Fix false-positive -EDOM checksum errors on overlapping unaligned reads

When csum_block_size > bitmap_granularity and perfect_csum_update = false
(the default behavior), an unaligned INTENT_WRITE modifies the physical
block directly in-place without recalculating and updating the large
block's checksum.

During a subsequent read that requires padding to the checksum block
boundary (e.g., reading adjacent unaligned data), the read logic must
skip checksum verification for the underlying BIG_WRITE because the
physical data has been safely tampered with by the INTENT_WRITE.

Previously, the need_skip logic in blockstore_read.cpp correctly
identified and skipped the verification for overlapping SMALL_WRITEs
but mistakenly omitted INTENT_WRITEs. This caused the padded read to
verify the stale BIG_WRITE checksum against the newly modified physical
data, leading to a fatal false-positive -EDOM (checksum mismatch) IO
error on reads.

This commit fixes the bug by ensuring BS_HEAP_INTENT_WRITE is evaluated
alongside SMALL_WRITE to apply the COPY_BUF_SKIP_CSUM bypass during
overlapping padded reads.

A regression test case (test_intent_write_corruption) is also added to
reproduce this specific data-layout scenario and verify the fix.

.gitea/workflows/test.yml

@@ -324,6 +324,24 @@ jobs:
           echo ""
         done
 
+  test_no_uring:
+    runs-on: ubuntu-latest
+    needs: build
+    container: ${{env.TEST_IMAGE}}:${{github.sha}}
+    steps:
+    - name: Run test
+      id: test
+      timeout-minutes: 3
+      run: /root/vitastor/tests/test_no_uring.sh
+    - name: Print logs
+      if: always() && steps.test.outcome == 'failure'
+      run: |
+        for i in /root/vitastor/testdata/*.log /root/vitastor/testdata/*.txt; do
+          echo "-------- $i --------"
+          cat $i
+          echo ""
+        done
+
   test_interrupted_rebalance:
     runs-on: ubuntu-latest
     needs: build

csi/src/controllerserver.go

@@ -27,7 +27,6 @@ const (
     MB int64 = 1024 * KB
     GB int64 = 1024 * MB
     TB int64 = 1024 * GB
-    ETCD_TIMEOUT time.Duration = 15*time.Second
 )
 
 type InodeIndex struct
@@ -687,7 +686,7 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 
     return &csi.ControllerExpandVolumeResponse{
         CapacityBytes: int64(inodeCfg[0].Size),
-        NodeExpansionRequired: false,
+        NodeExpansionRequired: true,
     }, nil
 }
 

csi/src/nodeserver.go

@@ -1030,10 +1030,131 @@ func (ns *NodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVo
     return nil, status.Error(codes.Unimplemented, "")
 }
 
-// NodeExpandVolume expanding the file system on the node
+// NodeExpandVolume expands the block device and, for filesystem volumes, the filesystem on the node
 func (ns *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received node expand volume request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Error(codes.InvalidArgument, "request cannot be empty")
+    }
+    if (req.VolumeId == "")
+    {
+        return nil, status.Error(codes.InvalidArgument, "VolumeId is required")
+    }
+
+    ctxVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.VolumeId), &ctxVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "volume ID not in JSON format")
+    }
+    volName := ctxVars["name"]
+
+    if (ctxVars["vitastorfs"] != "")
+    {
+        // VitastorFS CSI volumes are NFS-backed and currently don't have per-volume quotas to resize here.
+        return &csi.NodeExpandVolumeResponse{CapacityBytes: req.GetCapacityRange().GetRequiredBytes()}, nil
+    }
+
+    ns.lockVolume(ctxVars["configPath"]+":block:"+volName)
+    defer ns.unlockVolume(ctxVars["configPath"]+":block:"+volName)
+
+    targetPath := req.GetStagingTargetPath()
+    if (targetPath == "")
+    {
+        targetPath = req.GetVolumePath()
+    }
+    if (targetPath == "")
+    {
+        return nil, status.Error(codes.InvalidArgument, "VolumePath or StagingTargetPath is required")
+    }
+
+    devicePath, err := GetDeviceNameFromMount(targetPath)
+    if (err != nil)
+    {
+        if (os.IsNotExist(err))
+        {
+            return nil, status.Error(codes.NotFound, "Target path not found")
+        }
+        return nil, err
+    }
+    if (devicePath == "")
+    {
+        return nil, status.Errorf(codes.NotFound, "%s is not a mountpoint", targetPath)
+    }
+    if (devicePath[0] != '/')
+    {
+        return nil, status.Errorf(codes.Internal, "failed to resolve device for %s: %s", targetPath, devicePath)
+    }
+    if (len(devicePath) < 5 || devicePath[0:5] != "/dev/")
+    {
+        // NodeExpandVolume may be called with the published target path. For block volumes it is
+        // usually a bind mount of the staging target, so resolve one more mount layer if needed.
+        var realDevicePath string
+        realDevicePath, err = GetDeviceNameFromMount(devicePath)
+        if (err == nil && realDevicePath != "")
+        {
+            devicePath = realDevicePath
+        }
+    }
+
+    // All newest Vitastor driver versions (nbd, ublk and vduse) support online resize.
+    // So all we need to do is just wait a bit until the device is resized by the driver.
+    requiredBytes := req.GetCapacityRange().GetRequiredBytes()
+    if (requiredBytes > 0)
+    {
+        timeout := 30
+        var actualSize uint64
+        var err error
+        for i := 0; i < timeout && actualSize < requiredBytes; i++
+        {
+            actualSize, err = GetBlockDeviceSize(devicePath)
+            if (err != nil)
+            {
+                return nil, err
+            }
+            time.Sleep(1 * time.Second)
+        }
+        if (actualSize < requiredBytes)
+        {
+            return nil, status.Error(codes.NotFound, "Failed to wait for online resize of %s to %v bytes: actual size is still %v after %v seconds",
+                devicePath, requiredBytes, actualSize, timeout)
+        }
+    }
+
+    isBlock := req.GetVolumeCapability().GetBlock() != nil
+    if (isBlock)
+    {
+        return &csi.NodeExpandVolumeResponse{CapacityBytes: requiredBytes}, nil
+    }
+
+    diskMounter := &mount.SafeFormatAndMount{Interface: ns.mounter, Exec: utilexec.New()}
+    fsType := req.GetVolumeCapability().GetMount().GetFsType()
+    if (fsType == "")
+    {
+        fsType, err = diskMounter.GetDiskFormat(devicePath)
+        if (err != nil)
+        {
+            return nil, err
+        }
+    }
+
+    switch (fsType)
+    {
+        case "ext4":
+            _, err = systemCombined("resize2fs", devicePath)
+        case "xfs":
+            _, err = systemCombined("xfs_growfs", targetPath)
+        default:
+            return nil, status.Errorf(codes.InvalidArgument, "unsupported filesystem for resize: %s", fsType)
+    }
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    return &csi.NodeExpandVolumeResponse{CapacityBytes: requiredBytes}, nil
 }
 
 // NodeGetCapabilities returns the supported capabilities of the node server
@@ -1048,6 +1169,13 @@ func (ns *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetC
                     },
                 },
             },
+            &csi.NodeServiceCapability{
+                Type: &csi.NodeServiceCapability_Rpc{
+                    Rpc: &csi.NodeServiceCapability_RPC{
+                        Type: csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
+                    },
+                },
+            },
         },
     }, nil
 }

csi/src/utils.go

@@ -20,6 +20,7 @@ import (
 
     "google.golang.org/grpc/codes"
     "google.golang.org/grpc/status"
+    "golang.org/x/sys/unix"
 )
 
 type MountMethod int
@@ -519,3 +520,22 @@ func GetDeviceNameFromMount(mountPath string) (string, error)
 
     return device, nil
 }
+
+func GetBlockDeviceSize(devicePath string) (uint64, error)
+{
+    file, err := os.Open(devicePath)
+    if (err != nil)
+    {
+        return 0, err
+    }
+    defer file.Close()
+
+    // BLKGETSIZE64 returns the size in bytes as a uint64/int64
+    size, err := unix.IoctlGetInt(int(file.Fd()), unix.BLKGETSIZE64)
+    if (err != nil)
+    {
+        return 0, fmt.Errorf("ioctl BLKGETSIZE64 failed: %w", err)
+    }
+
+    return uint64(size), nil
+}

mon/etcd_schema.js

@@ -65,6 +65,8 @@ const etcd_tree = {
             placement_levels: { datacenter: 1, rack: 2, host: 3, osd: 4, ... },
             use_old_pg_combinator: false,
             osd_backfillfull_ratio: 0.99,
+            mon_metrics_path: '/prometheus',
+            mon_healthcheck_path: '/healthcheck',
             // client and osd
             tcp_header_buffer_size: 65536,
             use_sync_send_recv: false,

mon/mon.js

@@ -87,7 +87,7 @@ class Mon
             this.http = await create_http_server(this.config, (req, res) =>
             {
                 const u = new URL(req.url, 'http://'+(req.headers.host || 'localhost'));
-                if (u.pathname.replace(/\/+$/, '') == (this.config.prometheus_path||'/metrics'))
+                if (u.pathname.replace(/\/+$/, '') == (this.config.mon_metrics_path||'/metrics'))
                 {
                     if (!this.watcher_active)
                     {
@@ -100,6 +100,19 @@ class Mon
                         res.write(export_prometheus_metrics(this.state));
                     }
                 }
+                else if (u.pathname.replace(/\/+$/, '') == (this.config.mon_healthcheck_path||'/healthcheck'))
+                {
+                    if (!this.watcher_active)
+                    {
+                        res.writeHead(503);
+                        res.write('Monitor is in standby mode. Please retrieve metrics from master monitor instance\n');
+                    }
+                    else
+                    {
+                        res.setHeader('Content-Type', 'text/plain; version=0.0.4; charset=utf-8');
+                        res.write('Monitor is active');
+                    }
+                }
                 else
                 {
                     res.writeHead(404);

src/CMakeLists.txt

@@ -11,6 +11,7 @@ set(WITH_QEMU false CACHE BOOL "Build QEMU driver inside Vitastor source tree")
 set(WITH_FIO true CACHE BOOL "Build FIO driver")
 set(QEMU_PLUGINDIR qemu CACHE STRING "QEMU plugin directory suffix (qemu-kvm on RHEL)")
 set(WITH_ASAN false CACHE BOOL "Build with AddressSanitizer")
+set(WITH_UBSAN false CACHE BOOL "Build with UBSanitizer")
 set(WITH_SYSTEM_LIBURING false CACHE BOOL "Use system liburing")
 if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 	if(EXISTS "/etc/debian_version")
@@ -27,6 +28,10 @@ if (${WITH_ASAN})
 	add_definitions(-fsanitize=address)
 	add_link_options(-fsanitize=address -fno-omit-frame-pointer)
 endif (${WITH_ASAN})
+if (${WITH_UBSAN})
+	add_definitions(-fsanitize=undefined -fno-sanitize=alignment)
+	add_link_options(-fsanitize=undefined -fno-sanitize=alignment)
+endif (${WITH_UBSAN})
 set(CMAKE_CXX_FLAGS_RELEASE "${CMAKE_CXX_FLAGS_RELEASE} -fvisibility-inlines-hidden")
 set(CMAKE_CXX_FLAGS_MINSIZEREL "${CMAKE_CXX_FLAGS_MINSIZEREL} -fvisibility-inlines-hidden")
 set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "${CMAKE_CXX_FLAGS_RELWITHDEBINFO} -fvisibility-inlines-hidden")

src/blockstore/blockstore.h

@@ -167,7 +167,6 @@ struct __attribute__ ((visibility("default"))) blockstore_op_t
     uint8_t *buf = NULL;
     uint8_t *bitmap = NULL;
     int64_t retval = 0;
-    bool canceled = false;
 
     uint8_t private_data[BS_OP_PRIVATE_DATA_SIZE];
 };

src/blockstore/blockstore_disk.cpp

@@ -210,6 +210,32 @@ void blockstore_disk_t::parse_config(std::map<std::string, std::string> & config
     }
 }
 
+// A metadata entry must fit into one metadata block; a large block_size with a small
+// csum_block_size otherwise inflates the entry past it and makes the callers divide by zero.
+static uint64_t entries_per_meta_block(uint64_t entry_size, uint64_t meta_block_size,
+    uint64_t data_block_size, uint32_t bitmap_granularity, uint32_t csum_block_size)
+{
+    if (entry_size && entry_size <= meta_block_size)
+    {
+        return meta_block_size / entry_size;
+    }
+    std::string msg = "Metadata entry size ("+std::to_string(entry_size)+" bytes) exceeds meta_block_size ("+
+        std::to_string(meta_block_size)+" bytes) with block_size="+std::to_string(data_block_size)+
+        ", bitmap_granularity="+std::to_string(bitmap_granularity);
+    if (csum_block_size)
+    {
+        msg += ", csum_block_size="+std::to_string(csum_block_size)+" ("+
+            std::to_string(data_block_size/csum_block_size)+" checksums per block)";
+    }
+    msg += ". Decrease block_size";
+    if (csum_block_size)
+    {
+        msg += ", increase csum_block_size";
+    }
+    msg += " or increase meta_block_size";
+    throw std::runtime_error(msg);
+}
+
 void blockstore_disk_t::calc_lengths(bool skip_meta_check)
 {
     // data
@@ -259,17 +285,20 @@ void blockstore_disk_t::calc_lengths(bool skip_meta_check)
     clean_entry_bitmap_size = (data_block_size / bitmap_granularity + 7) / 8;
     clean_dyn_size = clean_entry_bitmap_size*2 + (csum_block_size
         ? data_block_size/csum_block_size*(data_csum_type & 0xFF) : 0);
+    uint64_t entries_per_block = 0;
 recalc:
     if (meta_format == BLOCKSTORE_META_FORMAT_HEAP)
     {
-        uint32_t entries_per_block = meta_block_size / (sizeof(heap_big_write_t) + clean_dyn_size);
+        entries_per_block = entries_per_meta_block(sizeof(heap_big_write_t) + clean_dyn_size,
+            meta_block_size, data_block_size, bitmap_granularity, csum_block_size);
         min_meta_len = (block_count+entries_per_block-1) / entries_per_block * meta_block_size;
     }
     else if (meta_format == BLOCKSTORE_META_FORMAT_V1)
     {
         clean_entry_size = 24 /*sizeof(clean_disk_entry)*/ + 2*clean_entry_bitmap_size;
-        min_meta_len = (1 + (block_count - 1 + meta_block_size / clean_entry_size)
-            / (meta_block_size / clean_entry_size)) * meta_block_size;
+        entries_per_block = entries_per_meta_block(clean_entry_size, meta_block_size,
+            data_block_size, bitmap_granularity, 0);
+        min_meta_len = (1 + (block_count - 1 + entries_per_block) / entries_per_block) * meta_block_size;
         if (!skip_meta_check && meta_area_size < min_meta_len)
         {
 too_small:
@@ -281,7 +310,9 @@ too_small:
     {
         meta_format = BLOCKSTORE_META_FORMAT_V2;
         clean_entry_size = 24 /*sizeof(clean_disk_entry)*/ + clean_dyn_size + 4 /*entry_csum*/;
-        min_meta_len = (1 + (block_count - 1 + meta_block_size / clean_entry_size) / (meta_block_size / clean_entry_size)) * meta_block_size;
+        entries_per_block = entries_per_meta_block(clean_entry_size, meta_block_size,
+            data_block_size, bitmap_granularity, csum_block_size);
+        min_meta_len = (1 + (block_count - 1 + entries_per_block) / entries_per_block) * meta_block_size;
         if (!skip_meta_check && meta_area_size < min_meta_len)
         {
             if (!data_csum_type)

src/blockstore/blockstore_read.cpp

@@ -75,10 +75,11 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
                 return false;
             }
         }
-        if (need_skip && wr->type() == BS_HEAP_SMALL_WRITE &&
+        if (need_skip && (wr->type() == BS_HEAP_SMALL_WRITE || wr->type() == BS_HEAP_INTENT_WRITE) &&
             wr->small().offset < blk_end && wr->small().offset+wr->small().len > blk_start)
         {
             // Small write may mutate big write checksums during flush
+            // Intent write directly mutates physical data in-place
             skip_csum = COPY_BUF_SKIP_CSUM;
         }
         return true;
@@ -134,7 +135,8 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *op)
         }
         else if ((vec.copy_flags & COPY_BUF_JOURNAL) && dsk.inmemory_journal)
         {
-            memcpy(op->buf + vec.offset - op->offset, buffer_area + vec.disk_loc + vec.disk_offset, vec.len);
+            // disk_loc may be negative - make ubsan happy
+            memcpy(op->buf + vec.offset - op->offset, buffer_area + (vec.disk_loc + vec.disk_offset), vec.len);
         }
         else
         {

src/blockstore/blockstore_write.cpp

@@ -34,6 +34,7 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
         assert(it != modified_blocks.end());
         free(it->second.buf);
         modified_blocks.erase(it);
+        unsynced_meta_write_count++;
         heap->complete_block_write(modified_block);
         ringloop->wakeup();
     };
@@ -41,7 +42,6 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
     io_uring_prep_writev(
         sqe, dsk.meta_fd, &data->iov, 1, dsk.meta_offset + ((uint64_t)modified_block+1)*dsk.meta_block_size
     );
-    unsynced_meta_write_count++;
     pending_modified_blocks.push_back(modified_block);
     modified_blocks[modified_block] = { .sent = false, .buf = buf };
 }

src/blockstore/v1/flush.cpp

@@ -35,7 +35,6 @@ journal_flusher_t::journal_flusher_t(blockstore_impl_t *bs)
 
 journal_flusher_co::journal_flusher_co()
 {
-    wait_state = 0;
     simple_callback_r = [this](ring_data_t* data)
     {
         bs->live = true;

src/blockstore/v1/flush.h

@@ -3,34 +3,34 @@
 
 struct copy_buffer_t
 {
-    int copy_flags;
-    uint64_t offset, len, disk_offset;
-    uint64_t journal_sector; // only for reads: sector+1 if used and !journal.inmemory, otherwise 0
-    void *buf;
-    uint8_t *csum_buf;
-    int *dyn_data;
+    int copy_flags = 0;
+    uint64_t offset = 0, len = 0, disk_offset = 0;
+    uint64_t journal_sector = 0; // only for reads: sector+1 if used and !journal.inmemory, otherwise 0
+    void *buf = NULL;
+    uint8_t *csum_buf = NULL;
+    int *dyn_data = NULL;
 };
 
 struct meta_sector_t
 {
-    uint64_t offset, len;
-    int state;
-    void *buf;
-    int usage_count;
+    uint64_t offset = 0, len = 0;
+    int state = 0;
+    void *buf = NULL;
+    int usage_count = 0;
 };
 
 struct flusher_sync_t
 {
-    bool fsync_meta;
-    int ready_count;
-    int state;
+    bool fsync_meta = false;
+    int ready_count = 0;
+    int state = 0;
 };
 
 struct flusher_meta_write_t
 {
-    uint64_t sector, pos;
-    bool submitted;
-    void *buf;
+    uint64_t sector = 0, pos = 0;
+    bool submitted = 0;
+    void *buf = NULL;
     std::map<uint64_t, meta_sector_t>::iterator it;
 };
 
@@ -39,36 +39,36 @@ class journal_flusher_t;
 // Journal flusher coroutine
 class journal_flusher_co
 {
-    blockstore_impl_t *bs;
-    journal_flusher_t *flusher;
-    int wait_state, wait_count, wait_journal_count;
-    struct io_uring_sqe *sqe;
-    struct ring_data_t *data;
+    blockstore_impl_t *bs = NULL;
+    journal_flusher_t *flusher = NULL;
+    int wait_state = 0, wait_count = 0, wait_journal_count = 0;
+    struct io_uring_sqe *sqe = NULL;
+    struct ring_data_t *data = NULL;
 
     std::list<flusher_sync_t>::iterator cur_sync;
 
-    obj_ver_id cur;
+    obj_ver_id cur = {};
     std::map<obj_ver_id, dirty_entry>::iterator dirty_it, dirty_start, dirty_end;
     std::map<object_id, uint64_t>::iterator repeat_it;
     std::function<void(ring_data_t*)> simple_callback_r, simple_callback_rj, simple_callback_w;
 
     bool try_trim = false;
-    bool skip_copy, has_delete, has_writes;
+    bool skip_copy = false, has_delete = false, has_writes = false;
     std::vector<copy_buffer_t> v;
     std::vector<copy_buffer_t>::iterator it;
-    int i;
-    bool fill_incomplete, cleared_incomplete;
-    int read_to_fill_incomplete;
-    int copy_count;
-    uint64_t clean_loc, clean_ver, old_clean_loc, old_clean_ver;
+    int i = 0;
+    bool fill_incomplete = false, cleared_incomplete = false;
+    int read_to_fill_incomplete = 0;
+    int copy_count = 0;
+    uint64_t clean_loc = 0, clean_ver = 0, old_clean_loc = 0, old_clean_ver = 0;
     flusher_meta_write_t meta_old, meta_new;
-    bool clean_init_bitmap;
-    uint64_t clean_bitmap_offset, clean_bitmap_len;
-    uint8_t *clean_init_dyn_ptr;
-    uint8_t *new_clean_bitmap;
+    bool clean_init_bitmap = false;
+    uint64_t clean_bitmap_offset = 0, clean_bitmap_len = 0;
+    uint8_t *clean_init_dyn_ptr = NULL;
+    uint8_t *new_clean_bitmap = NULL;
     std::unordered_set<uint32_t> mangle_csum_blocks;
 
-    uint64_t new_trim_pos;
+    uint64_t new_trim_pos = 0;
 
     friend class journal_flusher_t;
     void scan_dirty();
@@ -94,19 +94,19 @@ public:
 class journal_flusher_t
 {
     int trim_wanted = 0;
-    bool dequeuing;
-    int min_flusher_count, max_flusher_count, cur_flusher_count, target_flusher_count;
-    int flusher_start_threshold;
-    journal_flusher_co *co;
-    blockstore_impl_t *bs;
+    bool dequeuing = false;
+    int min_flusher_count = 0, max_flusher_count = 0, cur_flusher_count = 0, target_flusher_count = 0;
+    int flusher_start_threshold = 0;
+    journal_flusher_co *co = NULL;
+    blockstore_impl_t *bs = NULL;
     friend class journal_flusher_co;
 
-    int journal_trim_counter;
-    bool trimming;
-    void* journal_superblock;
+    int journal_trim_counter = 0;
+    bool trimming = false;
+    void* journal_superblock = NULL;
 
-    int active_flushers;
-    int syncing_flushers;
+    int active_flushers = 0;
+    int syncing_flushers = 0;
     std::list<flusher_sync_t> syncs;
     std::map<object_id, uint64_t> sync_to_repeat;
 

src/blockstore/v1/impl.cpp

@@ -24,6 +24,7 @@ blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i *
         calc_lengths();
         alloc_dyn_data = dsk.clean_dyn_size > sizeof(void*) || dsk.csum_block_size > 0;
         zero_object = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.data_block_size);
+        memset(zero_object, 0, dsk.data_block_size);
         data_alloc = new allocator_t(dsk.block_count);
     }
     catch (std::exception & e)

src/blockstore/v1/impl.h

@@ -238,7 +238,7 @@ class blockstore_impl_t: public blockstore_i
         uint64_t offset, uint64_t submit_len, uint64_t & blk_begin, uint64_t & blk_end, uint8_t* & blk_buf);
     bool read_range_fulfilled(std::vector<copy_buffer_t> & rv, uint64_t & fulfilled, uint8_t *read_buf,
         uint8_t *clean_entry_bitmap, uint32_t item_start, uint32_t item_end);
-    bool read_checksum_block(blockstore_op_t *op, int rv_pos, uint64_t &fulfilled, uint64_t clean_loc);
+    bool read_checksum_block(blockstore_op_t *op, int rv_pos, uint64_t &fulfilled, uint64_t clean_loc, bool padded);
     uint8_t* read_clean_meta_block(blockstore_op_t *read_op, uint64_t clean_loc, int rv_pos);
     bool verify_padded_checksums(uint8_t *clean_entry_bitmap, uint8_t *csum_buf, uint32_t offset,
         iovec *iov, int n_iov, std::function<void(uint32_t, uint32_t, uint32_t)> bad_block_cb);

src/blockstore/v1/init.cpp

@@ -1171,6 +1171,12 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                     clean_it->second.version < je->del.version);
                 if (!clean_exists && dirty_exists)
                 {
+                    // Stabilize previous entries to fix inode_space_stats and used_blocks (FIXME - remove it)
+                    obj_ver_id ov = {
+                        .oid = je->del.oid,
+                        .version = dirty_it->first.version,
+                    };
+                    bs->mark_stable(ov, true);
                     // Clean entry doesn't exist. This means that the delete is already flushed.
                     // So we must not flush this object anymore.
                     erase_dirty_object(dirty_it);

src/blockstore/v1/read.cpp

@@ -110,7 +110,7 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *read_op,
                     (*dyn_data)++;
                 }
                 // Submit the journal checksum block read
-                if (!read_checksum_block(read_op, 1, fulfilled, item_location - item_start))
+                if (!read_checksum_block(read_op, 1, fulfilled, item_location - item_start, false))
                 {
                     r = 0;
                 }
@@ -228,10 +228,12 @@ int blockstore_impl_t::fill_partial_checksum_blocks(std::vector<copy_buffer_t> &
             }
             end_block++;
             // OK, mark this range as required
+            uint32_t rv_start = start_block*dsk.csum_block_size < item_start ? item_start : start_block*dsk.csum_block_size;
+            uint32_t rv_end = end_block*dsk.csum_block_size > item_end ? item_end : end_block*dsk.csum_block_size;
             rv.push_back((copy_buffer_t){
                 .copy_flags = COPY_BUF_CSUM_FILL | (from_journal ? COPY_BUF_JOURNALED_BIG : 0),
-                .offset = start_block*dsk.csum_block_size,
-                .len = (end_block-start_block)*dsk.csum_block_size,
+                .offset = rv_start,
+                .len = rv_end - rv_start,
                 // save checksum reference if we're reading clean data from the journal
                 .csum_buf = from_journal
                     ? clean_entry_bitmap + dsk.clean_entry_bitmap_size + (start_block-item_start_block)*(dsk.data_csum_type & 0xFF)
@@ -300,7 +302,7 @@ bool blockstore_impl_t::read_range_fulfilled(std::vector<copy_buffer_t> & rv, ui
     return all_done;
 }
 
-bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uint64_t &fulfilled, uint64_t clean_loc)
+bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uint64_t &fulfilled, uint64_t clean_loc, bool padded)
 {
     auto & rv = PRIV(op)->read_vec;
     auto *vi = &rv[rv.size()-rv_pos];
@@ -335,10 +337,22 @@ bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uin
         }
         return 0;
     });
+    if (padded && (item_start % dsk.csum_block_size))
+    {
+        n_iov++;
+    }
+    if (padded && (item_end % dsk.csum_block_size))
+    {
+        n_iov++;
+    }
     void *buf = memalign_or_die(MEM_ALIGNMENT, fill_size + n_iov*sizeof(struct iovec));
     iovec *iov = (struct iovec*)((uint8_t*)buf+fill_size);
     n_iov = 0;
     fill_size = 0;
+    if (padded && (item_start % dsk.csum_block_size))
+    {
+        iov[n_iov++] = (struct iovec){ zero_object, item_start % dsk.csum_block_size };
+    }
     find_holes(rv, item_start, item_end, [&](int pos, bool alloc, uint32_t cur_start, uint32_t cur_end)
     {
         int res = 0;
@@ -376,18 +390,31 @@ bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uin
         }
         return res;
     });
+    if (padded && (item_end % dsk.csum_block_size))
+    {
+        iov[n_iov++] = (struct iovec){ zero_object, dsk.csum_block_size - (item_end % dsk.csum_block_size) };
+    }
     vi = &rv[rv.size()-rv_pos];
     // Save buf into read_vec too but in a creepy way
     // FIXME: Shit, something else should be invented %)
     *vi = (copy_buffer_t){
         .copy_flags = vi->copy_flags,
-        .offset = vi->offset,
+        .offset = (padded ? vi->offset - vi->offset % dsk.csum_block_size : vi->offset),
         .len = ((uint64_t)n_iov << 32) | fill_size,
-        .disk_offset = clean_loc + item_start,
+        .disk_offset = clean_loc + (padded ? vi->offset - vi->offset % dsk.csum_block_size : vi->offset),
         .buf = (uint8_t*)buf,
         .csum_buf = vi->csum_buf,
         .dyn_data = vi->dyn_data,
     };
+    if (padded && (item_start % dsk.csum_block_size))
+    {
+        iov++;
+        n_iov--;
+    }
+    if (padded && (item_end % dsk.csum_block_size))
+    {
+        n_iov--;
+    }
     int submit_fd = (vi->copy_flags & COPY_BUF_JOURNAL ? dsk.journal_fd : dsk.data_fd);
     uint64_t submit_offset = (vi->copy_flags & COPY_BUF_JOURNAL ? journal.offset : dsk.data_offset);
     uint32_t d_pos = 0;
@@ -420,12 +447,6 @@ bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uin
 
 int blockstore_impl_t::dequeue_read(blockstore_op_t *read_op)
 {
-    if (read_op->canceled)
-    {
-        read_op->retval = -EPIPE;
-        FINISH_OP(read_op);
-        return 2;
-    }
     auto & clean_db = clean_db_shard(read_op->oid);
     auto clean_it = clean_db.find(read_op->oid);
     auto dirty_it = dirty_db.upper_bound((obj_ver_id){
@@ -681,7 +702,7 @@ bool blockstore_impl_t::fulfill_clean_read(blockstore_op_t *read_op, uint64_t &
         }
         for (int i = req; i > 0; i--)
         {
-            if (!read_checksum_block(read_op, i, fulfilled, clean_loc))
+            if (!read_checksum_block(read_op, i, fulfilled, clean_loc, true))
             {
                 return false;
             }
@@ -1096,6 +1117,14 @@ int blockstore_impl_t::read_bitmap(object_id oid, uint64_t target_version, void
             // Condition has to be the same as in dequeue_read()
             if (!IS_IN_FLIGHT(dirty_it->second.state) && target_version >= dirty_it->first.version)
             {
+                if (IS_DELETE(dirty_it->second.state))
+                {
+                    if (result_version)
+                        *result_version = 0;
+                    if (bitmap)
+                        memset(bitmap, 0, dsk.clean_entry_bitmap_size);
+                    return -ENOENT;
+                }
                 if (result_version)
                     *result_version = dirty_it->first.version;
                 if (bitmap)

src/blockstore/v1/rollback.cpp

@@ -12,12 +12,6 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
     {
         return continue_rollback(op);
     }
-    if (op->canceled)
-    {
-        op->retval = -EPIPE;
-        FINISH_OP(op);
-        return 2;
-    }
     int r = split_stab_op(op, [this](obj_ver_id ov)
     {
         // Check that there are some versions greater than v->version (which may be zero),

src/blockstore/v1/stable.cpp

@@ -291,12 +291,6 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
     {
         return continue_stable(op);
     }
-    if (op->canceled)
-    {
-        op->retval = -EPIPE;
-        FINISH_OP(op);
-        return 2;
-    }
     int r = split_stab_op(op, [this](obj_ver_id ov)
     {
         auto dirty_it = dirty_db.find(ov);
@@ -451,9 +445,9 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
                 {
                     break;
                 }
-                if (!IS_STABLE(back_it->second.state))
+                if (!IS_SYNCED(back_it->second.state))
                 {
-                    // There are preceding unstable versions, can't flush <v>
+                    // There are preceding unsynced versions, can't flush <v>
                     return;
                 }
             }
@@ -469,6 +463,7 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
                 v.version = dirty_it->first.version;
             }
         }
+        auto erase_end = dirty_db.end();
         while (1)
         {
             bool was_stable = IS_STABLE(dirty_it->second.state);
@@ -476,6 +471,7 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
             {
                 dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_STABLE;
                 // Allocations and deletions are counted when they're stabilized
+                // FIXME: Count-on-stabilize logic is actually really strange, reconsider it
                 if (IS_BIG_WRITE(dirty_it->second.state))
                 {
                     int exists = -1;
@@ -529,25 +525,10 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
                 exit(1);
             }
             if (forget_dirty && (IS_BIG_WRITE(dirty_it->second.state) ||
-                IS_DELETE(dirty_it->second.state)))
+                IS_DELETE(dirty_it->second.state)) && erase_end == dirty_db.end())
             {
                 // Big write overrides all previous dirty entries
-                auto erase_end = dirty_it;
-                while (dirty_it != dirty_db.begin())
-                {
-                    dirty_it--;
-                    if (dirty_it->first.oid != v.oid)
-                    {
-                        dirty_it++;
-                        break;
-                    }
-                }
-                auto & clean_db = clean_db_shard(v.oid);
-                auto clean_it = clean_db.find(v.oid);
-                uint64_t clean_loc = clean_it != clean_db.end()
-                    ? clean_it->second.location : UINT64_MAX;
-                erase_dirty(dirty_it, erase_end, clean_loc);
-                break;
+                erase_end = dirty_it;
             }
             if (was_stable || dirty_it == dirty_db.begin())
             {
@@ -556,10 +537,20 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
             dirty_it--;
             if (dirty_it->first.oid != v.oid)
             {
+                dirty_it++;
                 break;
             }
         }
         flusher->enqueue_flush(v);
+        if (erase_end != dirty_db.end())
+        {
+            // Erase only after marking previous entries stable to not skip used_blocks++
+            auto & clean_db = clean_db_shard(v.oid);
+            auto clean_it = clean_db.find(v.oid);
+            uint64_t clean_loc = clean_it != clean_db.end()
+                ? clean_it->second.location : UINT64_MAX;
+            erase_dirty(dirty_it, erase_end, clean_loc);
+        }
     }
     auto unstab_it = unstable_writes.find(v.oid);
     if (unstab_it != unstable_writes.end() &&

src/blockstore/v1/sync.cpp

@@ -176,18 +176,15 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
 void blockstore_impl_t::ack_sync(blockstore_op_t *op)
 {
     // Handle states
-    for (auto it = PRIV(op)->sync_big_writes.begin(); it != PRIV(op)->sync_big_writes.end(); it++)
+    auto mark_sync = [&](obj_ver_id & ov)
     {
-#ifdef BLOCKSTORE_DEBUG
-        printf("Ack sync big %jx:%jx v%ju\n", it->oid.inode, it->oid.stripe, it->version);
-#endif
-        auto & unstab = unstable_writes[it->oid];
-        unstab = unstab < it->version ? it->version : unstab;
-        auto dirty_it = dirty_db.find(*it);
+        auto dirty_it = dirty_db.find(ov);
+        bool is_big = ((dirty_it->second.state & BS_ST_TYPE_MASK) == BS_ST_BIG_WRITE);
         dirty_it->second.state = ((dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_SYNCED);
-        if (dirty_it->second.state & BS_ST_INSTANT)
+        uint64_t stable_ver = 0;
+        if (IS_INSTANT(dirty_it->second.state))
         {
-            mark_stable(dirty_it->first);
+            stable_ver = ov.version;
         }
         else
         {
@@ -195,41 +192,37 @@ void blockstore_impl_t::ack_sync(blockstore_op_t *op)
             assert(unstable_unsynced >= 0);
         }
         dirty_it++;
-        while (dirty_it != dirty_db.end() && dirty_it->first.oid == it->oid)
+        while (dirty_it != dirty_db.end() && dirty_it->first.oid == ov.oid)
         {
-            if ((dirty_it->second.state & BS_ST_WORKFLOW_MASK) == BS_ST_WAIT_BIG)
+            if (is_big && (dirty_it->second.state & BS_ST_WORKFLOW_MASK) == BS_ST_WAIT_BIG)
             {
                 dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_IN_FLIGHT;
             }
+            else if ((dirty_it->second.state & BS_ST_WORKFLOW_MASK) == BS_ST_SYNCED &&
+                IS_INSTANT(dirty_it->second.state))
+            {
+                stable_ver = dirty_it->first.version;
+            }
             dirty_it++;
         }
-    }
-    for (auto it = PRIV(op)->sync_small_writes.begin(); it != PRIV(op)->sync_small_writes.end(); it++)
+        if (stable_ver != 0)
+        {
+            mark_stable((obj_ver_id){ .oid = ov.oid, .version = stable_ver });
+        }
+    };
+    for (auto & ov: PRIV(op)->sync_big_writes)
     {
 #ifdef BLOCKSTORE_DEBUG
-        printf("Ack sync small %jx:%jx v%ju\n", it->oid.inode, it->oid.stripe, it->version);
+        printf("Ack sync big %jx:%jx v%ju\n", ov.oid.inode, ov.oid.stripe, ov.version);
 #endif
-        auto & unstab = unstable_writes[it->oid];
-        unstab = unstab < it->version ? it->version : unstab;
-        if (dirty_db[*it].state == (BS_ST_DELETE | BS_ST_WRITTEN))
-        {
-            dirty_db[*it].state = (BS_ST_DELETE | BS_ST_SYNCED);
-            // Deletions are treated as immediately stable
-            mark_stable(*it);
-        }
-        else /* (BS_ST_INSTANT?) | BS_ST_SMALL_WRITE | BS_ST_WRITTEN */
-        {
-            dirty_db[*it].state = (dirty_db[*it].state & ~BS_ST_WORKFLOW_MASK) | BS_ST_SYNCED;
-            if (dirty_db[*it].state & BS_ST_INSTANT)
-            {
-                mark_stable(*it);
-            }
-            else
-            {
-                unstable_unsynced--;
-                assert(unstable_unsynced >= 0);
-            }
-        }
+        mark_sync(ov);
+    }
+    for (auto & ov: PRIV(op)->sync_small_writes)
+    {
+#ifdef BLOCKSTORE_DEBUG
+        printf("Ack sync small %jx:%jx v%ju\n", ov.oid.inode, ov.oid.stripe, ov.version);
+#endif
+        mark_sync(ov);
     }
     op->retval = 0;
     FINISH_OP(op);

src/blockstore/v1/write.cpp

@@ -12,19 +12,23 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
     bool found = false, deleted = false, unsynced = false, is_del = (op->opcode == BS_OP_DELETE);
     bool wait_big = false, wait_del = false;
     void *dyn = NULL;
+    uint8_t *dyn_ptr = NULL;
     if (is_del)
     {
         op->len = 0;
     }
-    size_t dyn_size = dsk.dirty_dyn_size(op->offset, op->len);
-    if (!is_del && alloc_dyn_data)
+    else
     {
-        // FIXME: Working with `dyn_data` has to be refactored somehow but I first have to decide how :)
-        // +sizeof(int) = refcount
-        dyn = calloc_or_die(1, dyn_size+sizeof(int));
-        *((int*)dyn) = 1;
+        if (alloc_dyn_data)
+        {
+            // FIXME: Working with `dyn_data` has to be refactored somehow but I first have to decide how :)
+            // +sizeof(int) = refcount
+            size_t dyn_size = dsk.dirty_dyn_size(op->offset, op->len);
+            dyn = calloc_or_die(1, dyn_size+sizeof(int));
+            *((int*)dyn) = 1;
+        }
+        dyn_ptr = (alloc_dyn_data ? (uint8_t*)dyn+sizeof(int) : (uint8_t*)&dyn);
     }
-    uint8_t *dyn_ptr = (alloc_dyn_data ? (uint8_t*)dyn+sizeof(int) : (uint8_t*)&dyn);
     uint64_t version = 1;
     if (dirty_db.size() > 0)
     {
@@ -214,11 +218,6 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
     return true;
 }
 
-// Why cancel all writes:
-// 1) enqueue_write forms dirty_db entries based on one another.
-//    So if we remove an entry, newer entries become possibly incorrect.
-// 2) small writes may wait for the previous big write. if we cancel it,
-//    we also have to unblock them, but it's easier to cancel them too.
 void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, blockstore_dirty_db_t::iterator dirty_it, int retval)
 {
     while (dirty_it != dirty_db.end() && dirty_it->first.oid == op->oid)
@@ -241,7 +240,9 @@ void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, blockstore_dirty_
         else if (found && other_op->oid == op->oid &&
             (other_op->opcode == BS_OP_WRITE || other_op->opcode == BS_OP_WRITE_STABLE))
         {
-            other_op->canceled = true;
+            // Mark operations to cancel them
+            PRIV(other_op)->real_version = UINT64_MAX;
+            other_op->retval = retval;
         }
     }
     op->retval = retval;
@@ -260,11 +261,6 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         .version = op->version,
     });
     assert(dirty_it != dirty_db.end());
-    if (op->canceled)
-    {
-        cancel_all_writes(op, dirty_it, -EPIPE);
-        return 2;
-    }
     if ((dirty_it->second.state & BS_ST_WORKFLOW_MASK) < BS_ST_IN_FLIGHT)
     {
         // Don't dequeue
@@ -272,6 +268,12 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     }
     if (PRIV(op)->real_version != 0)
     {
+        if (PRIV(op)->real_version == UINT64_MAX)
+        {
+            // This is the flag value used to cancel operations
+            FINISH_OP(op);
+            return 2;
+        }
         // Restore original low version number for unblocked operations
 #ifdef BLOCKSTORE_DEBUG
         printf("Restoring %jx:%jx version: v%ju -> v%ju\n", op->oid.inode, op->oid.stripe, op->version, PRIV(op)->real_version);
@@ -552,16 +554,6 @@ resume_2:
             .version = op->version,
         });
         assert(dirty_it != dirty_db.end());
-        if (op->canceled)
-        {
-            auto uo_it = used_clean_objects.find(dirty_it->second.location);
-            if (uo_it != used_clean_objects.end())
-                uo_it->second.was_freed = true;
-            else
-                data_alloc->set(dirty_it->second.location / dsk.data_block_size, false);
-            cancel_all_writes(op, dirty_it, -EPIPE);
-            return 2;
-        }
         uint64_t dyn_size = dsk.dirty_dyn_size(op->offset, op->len);
         blockstore_journal_check_t space_check(this);
         if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size,
@@ -779,11 +771,6 @@ int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
         .version = op->version,
     });
     assert(dirty_it != dirty_db.end());
-    if (op->canceled)
-    {
-        cancel_all_writes(op, dirty_it, -EPIPE);
-        return 2;
-    }
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, 1, sizeof(journal_entry_del), (unstable_writes.size()+unstable_unsynced)*journal.block_size))
     {

src/client/http_client.cpp

@@ -746,6 +746,10 @@ void http_co_t::handle_events()
             {
                 handle_connect_result();
             }
+            else
+            {
+                break;
+            }
         }
         else
         {
@@ -1301,8 +1305,8 @@ static bool ws_parse_frame(std::string & buf, uint8_t & type, std::string & res)
     {
         return false;
     }
-    type = buf[0] & ~0x80;
-    bool mask = !!(buf[1] & 0x80);
+    type = (uint8_t)buf[0] & ~0x80;
+    bool mask = !!((uint8_t)buf[1] & 0x80);
     hdr += mask ? 4 : 0;
     uint64_t len = ((uint8_t)buf[1] & ~0x80);
     if (len == 126)

src/client/http_client.h

@@ -39,7 +39,7 @@ struct http_message_t
     std::map<std::string, std::string> headers;
     std::string tls_cn;
     int tls_ca_idx = -1;
-    uint8_t ws_msg_type = -1;
+    uint8_t ws_msg_type = 0;
     std::string body;
 
     void parse_json_response(std::string & error, json11::Json & r) const;

src/client/messenger.cpp

@@ -504,18 +504,21 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd)
     int one = 1;
     setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
     cl->peer_state = PEER_CONNECTED;
-    tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
+    // EPOLLOUT is only required without io_uring, to continue partially sent messages
+    tfd->set_fd_handler(peer_fd, use_sync_send_recv, [this](int peer_fd, int epoll_events)
     {
         handle_peer_epoll(peer_fd, epoll_events);
     });
     // Check OSD number
-    init_tls_client(cl);
+    if (!init_tls_client(cl))
+    {
+        return;
+    }
     check_peer_config(cl);
 }
 
 void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
 {
-    // Mark client as ready (i.e. some data is available)
     auto cl = clients_by_fd.at(peer_fd);
     if (epoll_events & EPOLLRDHUP)
     {
@@ -525,8 +528,9 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
             fprintf(stderr, "[OSD %ju] client %ju disconnected\n", this->osd_num, cl->client_id);
         }
         stop_client(cl->client_id);
+        return;
     }
-    else if (epoll_events & EPOLLIN)
+    if (epoll_events & EPOLLIN)
     {
         // Mark client as ready (i.e. some data is available)
         cl->read_ready++;
@@ -539,6 +543,23 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
                 read_requests();
         }
     }
+    if (epoll_events & EPOLLOUT)
+    {
+        // The socket became writable again - continue the interrupted use_sync_send_recv
+        // read_requests() above may have stopped the client, so look it up again
+        auto cl_it = clients.find(cl->client_id);
+        if (cl_it == clients.end())
+        {
+            return;
+        }
+        cl = cl_it->second;
+        cl->write_ready = true;
+        // send_list is only left non-empty when we're waiting for EPOLLOUT
+        if (cl->send_list.size() && cl->peer_state != PEER_RDMA)
+        {
+            try_send(cl);
+        }
+    }
 }
 
 void osd_messenger_t::on_connect_peer(osd_num_t peer_osd, int errcode, uint64_t client_id)
@@ -634,8 +655,16 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
         op->iov.push_back(op->buf, payload_str.size());
         memcpy(op->buf, payload_str.c_str(), payload_str.size());
     }
-    op->callback = [this, cl](osd_op_t *op)
+    op->callback = [this, client_id = cl->client_id, peer_osd = cl->osd_num](osd_op_t *op)
     {
+        auto cl_it = clients.find(client_id);
+        if (cl_it == clients.end())
+        {
+            on_connect_peer(peer_osd, -EPIPE, 0);
+            delete op;
+            return;
+        }
+        auto cl = cl_it->second;
         std::string json_err;
         json11::Json config;
         bool err = false;
@@ -693,7 +722,6 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
         }
         if (err)
         {
-            osd_num_t peer_osd = cl->osd_num;
             stop_client(op->client_id);
             on_connect_peer(peer_osd, -EINVAL, 0);
             delete op;
@@ -763,9 +791,12 @@ void osd_messenger_t::accept_connections(int listen_fd)
         cl->peer_fd = peer_fd;
         cl->peer_state = PEER_CONNECTED;
         cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
-        init_tls_client(cl);
-        // Add FD to epoll
-        tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
+        if (!init_tls_client(cl))
+        {
+            continue;
+        }
+        // Add FD to epoll. EPOLLOUT is only required without io_uring
+        tfd->set_fd_handler(peer_fd, use_sync_send_recv, [this](int peer_fd, int epoll_events)
         {
             handle_peer_epoll(peer_fd, epoll_events);
         });

src/client/messenger.h

@@ -147,6 +147,8 @@ struct osd_client_t
     size_t write_op_pos = 0;
     msghdr write_msg = { 0 };
     int write_state = 0;
+    // Socket is writable. Only tracked with use_sync_send_recv, io_uring doesn't need it
+    bool write_ready = true;
     std::vector<iovec> send_list;
     size_t send_list_size = 0;
     std::deque<osd_op_t*> send_free_ops;
@@ -248,7 +250,7 @@ protected:
 
     void init_tls();
     void destroy_tls();
-    void init_tls_client(osd_client_t *cl);
+    bool init_tls_client(osd_client_t *cl);
     bool do_tls_handshake(osd_client_t *cl, bool from_recv = false);
     bool derive_aes_keys(osd_client_t *cl, bool update_my, bool update_peer);
 
@@ -301,7 +303,6 @@ public:
     void stop_client(uint64_t client_id, bool force_delete = false);
     void destroy_client(osd_client_t *cl);
     void outbox_push(osd_op_t *cur_op);
-    bool cancel_out(osd_op_t *cur_op);
     std::function<void(osd_op_t*)> exec_op;
     std::function<void(osd_num_t)> repeer_pgs;
     std::function<void(osd_num_t)> break_pg_locks;
@@ -342,10 +343,10 @@ protected:
     void check_peer_config(osd_client_t *cl);
     void cancel_osd_ops(osd_client_t *cl);
     void cancel_op(osd_op_t *op);
-    osd_op_t *copy_out_op_to_ignore(osd_op_t *cur_op);
 
     bool try_send(osd_client_t *cl);
-    void handle_send(int result, bool prev, bool more, osd_client_t *cl);
+    bool wakeup_send(osd_client_t *cl);
+    bool handle_send(int result, bool prev, bool more, osd_client_t *cl);
     static void post_send_free(osd_op_t *op);
     bool op_write_to(osd_client_t *cl, msgr_op_writer_t & wr);
     void next_write_op(osd_client_t *cl);
@@ -372,7 +373,7 @@ protected:
     void op_decrypt_free(osd_client_t* cl);
 
 #ifdef WITH_RDMA
-    void try_send_rdma(osd_client_t *cl);
+    bool try_send_rdma(osd_client_t *cl);
     bool init_recv_rdma(osd_client_t *cl);
     void handle_rdma_events(msgr_rdma_context_t *rdma_context);
     msgr_rdma_context_t* choose_rdma_context(osd_client_t *cl);

src/client/msgr_encrypt.cpp

@@ -616,7 +616,7 @@ void osd_messenger_t::init_tls()
     }
 }
 
-void osd_messenger_t::init_tls_client(osd_client_t *cl)
+bool osd_messenger_t::init_tls_client(osd_client_t *cl)
 {
     if (gcm_enabled)
     {
@@ -625,13 +625,13 @@ void osd_messenger_t::init_tls_client(osd_client_t *cl)
         cl->hs->init(cl->is_incoming);
         if (cl->hs->out_size())
         {
-            if (cl->write_state == 0)
+            if (!wakeup_send(cl))
             {
-                cl->write_state = CL_WRITE_READY;
-                write_ready_clients.push_back(cl->client_id);
+                return false;
             }
         }
     }
+    return true;
 }
 
 void osd_messenger_t::destroy_tls()

src/client/msgr_op.h

@@ -188,7 +188,6 @@ struct __attribute__((visibility("default"))) osd_op_t
     std::shared_ptr<osd_op_enc_t> enc;
     uint64_t csum = 0; // network layer checksum
     osd_primary_op_data_t* op_data = NULL;
-    bool canceled = false;
     std::function<void(osd_op_t*)> callback;
 
     osd_op_buf_list_t iov;

src/client/msgr_rdma.cpp

@@ -571,7 +571,7 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
     cl->rdma_conn->cur_send++;
 }
 
-void osd_messenger_t::try_send_rdma(osd_client_t *cl)
+bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
 {
     auto rc = cl->rdma_conn;
     if (!rc->send_out_size)
@@ -602,7 +602,7 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
         if (cl->io_error)
         {
             stop_client(cl->client_id);
-            return;
+            return false;
         }
         if (copied > 0)
         {
@@ -622,6 +622,7 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
             cl->send_free_ops.push_back(NULL); // end marker
         }
     }
+    return true;
 }
 
 static void try_recv_rdma_wr(osd_client_t *cl, void *buf)
@@ -750,6 +751,7 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context)
             }
         }
     } while (event_count > 0);
+    handle_immediate_ops();
 }
 
 void osd_messenger_t::destroy_rdma_conn(msgr_rdma_connection_t *rdma_conn)

src/client/msgr_rdmacm.cpp

@@ -60,7 +60,10 @@ rdma_cm_id *osd_messenger_t::rdmacm_listen(const std::string & bind_address, int
     }
     return listener;
 fail:
-    rdma_destroy_id(listener);
+    if (listener)
+    {
+        rdma_destroy_id(listener);
+    }
     return NULL;
 }
 
@@ -500,7 +503,6 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev)
     cl->osd_num = peer_osd;
     cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
     cl->rdma_conn = rc;
-    init_tls_client(cl);
     clients[conn->client_id] = cl;
     if (conn->timeout_id >= 0)
     {
@@ -514,6 +516,10 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev)
         fprintf(stderr, "Successfully connected with OSD %ju using RDMA-CM\n", peer_osd);
     }
     // Add initial receive request(s)
+    if (!init_tls_client(cl))
+    {
+        return;
+    }
     init_recv_rdma(cl);
     if (peer_osd)
     {

src/client/msgr_receive.cpp

@@ -458,8 +458,7 @@ void osd_messenger_t::read_requests()
             {
                 result = -errno;
             }
-            // like set_immediate
-            tfd->set_timer_us(0, false, [this, result, cl](int){ handle_read(result, cl); });
+            handle_read(result, cl);
         }
     }
     read_ready_clients.clear();
@@ -490,8 +489,10 @@ void osd_messenger_t::handle_read(int result, osd_client_t *cl)
         }
         stop_client(cl->client_id);
 out_wakeup:
-        if (set_immediate_ops.size())
+        if (ringloop && set_immediate_ops.size())
+        {
             ringloop->wakeup();
+        }
         return;
     }
     bool full_read = false;
@@ -501,7 +502,9 @@ out_wakeup:
         {
             full_read = result >= cl->read_iov.iov_len;
             if (!handle_read_buffer(cl, cl->in_buf, result))
+            {
                 goto out_wakeup;
+            }
         }
         else
         {
@@ -567,15 +570,11 @@ void osd_messenger_t::handle_immediate_ops()
             else
                 delete op;
         }
-        else if (op->callback)
+        else
         {
             // Copy lambda to be unaffected by `delete op`
             std::function<void(osd_op_t*)>(op->callback)(op);
         }
-        else
-        {
-            delete op;
-        }
     }
 }
 
@@ -601,13 +600,16 @@ bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size
             bufsize -= done;
             if (cl->hs->out_size())
             {
-                if (cl->write_state == 0)
+                if (!wakeup_send(cl))
                 {
-                    cl->write_state = CL_WRITE_READY;
-                    write_ready_clients.push_back(cl->client_id);
+                    return false;
                 }
             }
-            if (cl->hs->done() && !cl->hs->out_size())
+            if (!cl->hs)
+            {
+                // Deleted by wakeup_send with use_sync_send_recv
+            }
+            else if (cl->hs->done() && !cl->hs->out_size())
             {
                 // Delete hs when done and nothing to send
                 delete cl->hs;

src/client/msgr_send.cpp

@@ -17,6 +17,10 @@
 
 #define GCM_TMP_BUF_SIZE 4096
 
+// How many times to retry sendmsg() in place before falling back to EPOLLOUT.
+// Only used with use_sync_send_recv - io_uring waits for the socket by itself
+#define MSGR_SEND_RETRIES 4
+
 class msgr_op_writer_t
 {
 public:
@@ -432,7 +436,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     auto cl_it = clients.find(cur_op->client_id);
     if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
     {
-        delete cur_op;
+        cur_op->cancel();
         return;
     }
     osd_client_t *cl = cl_it->second;
@@ -467,137 +471,60 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
         return;
     }
 #endif
-    if (!ringloop)
-    {
-        // FIXME: It's worse because it doesn't allow batching
-        while (cl->write_op || cl->write_ops.size())
-        {
-            try_send(cl);
-        }
-    }
-    else
-    {
-        if (!try_send(cl) && cl->write_state == 0)
-        {
-            cl->write_state = CL_WRITE_READY;
-            write_ready_clients.push_back(cl->client_id);
-        }
-        ringloop->wakeup();
-    }
+    try_send(cl);
 }
 
-// Try to cancel an outbound operation
-// Returns true when successfully canceled, returns false if the reply is already received
-bool osd_messenger_t::cancel_out(osd_op_t *cur_op)
+bool osd_messenger_t::wakeup_send(osd_client_t *cl)
 {
-    assert(cur_op->client_id);
-    assert(cur_op->op_type == OSD_OP_OUT);
-    auto cl_it = clients.find(cur_op->client_id);
-    if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
+    if (cl->peer_state == PEER_RDMA)
     {
-        // Client is stopped => messenger will call the callback anyway
-        return false;
+        return try_send_rdma(cl);
     }
-    osd_client_t *cl = cl_it->second;
-    auto sent_it = cl->sent_ops.find(cur_op->req.hdr.id);
-    if (sent_it == cl->sent_ops.end())
+    if (use_sync_send_recv)
     {
-        // Reply header is already received => callback is pending
-        return false;
+        // FIXME: Batch send/recv without io_uring too.
+        return try_send(cl);
     }
-    for (auto op: set_immediate_ops)
+    if (cl->write_state == 0)
     {
-        if (op == cur_op)
-        {
-            // Reply is already received, callback is pending
-            return false;
-        }
+        cl->write_state = CL_WRITE_READY;
+        write_ready_clients.push_back(cl->client_id);
     }
-    for (auto wr_it = cl->write_ops.begin(); wr_it != cl->write_ops.end(); wr_it++)
-    {
-        if (*wr_it == cur_op)
-        {
-            // Operation is just queued and not sent and can be canceled easily
-            cl->sent_ops.erase(sent_it);
-            cl->write_ops.erase(wr_it);
-            cur_op->callback = NULL;
-            return true;
-        }
-    }
-    // Not in queue: sent or in progress of sending.
-    bool in_progress = (cl->write_op == cur_op);
-    if (!in_progress)
-    {
-        for (osd_op_t *op: cl->send_free_ops)
-        {
-            if (op == cur_op)
-            {
-                in_progress = true;
-                break;
-            }
-        }
-    }
-    if (!in_progress)
-    {
-        for (osd_op_t *op: cl->zc_free_list)
-        {
-            if (op == cur_op)
-            {
-                in_progress = true;
-                break;
-            }
-        }
-    }
-    sent_it->second = copy_out_op_to_ignore(cur_op);
-    cur_op->callback = NULL;
-    if (in_progress)
-    {
-        // We CAN cancel it, but only when the reference from send_free_ops/zc_free_list is cleared
-        cur_op->canceled = true;
-        return false;
-    }
-    // Make a copy to have some space to place the reply and forget the original object.
+    ringloop->wakeup();
     return true;
 }
 
-osd_op_t *osd_messenger_t::copy_out_op_to_ignore(osd_op_t *cur_op)
-{
-    osd_op_t *op_copy = new osd_op_t;
-    op_copy->tv_begin = cur_op->tv_begin;
-    op_copy->client_id = cur_op->client_id;
-    op_copy->osd_num = cur_op->osd_num;
-    op_copy->bitmap_len = cur_op->bitmap_len;
-    memcpy(op_copy->req.buf, cur_op->req.buf, OSD_PACKET_SIZE);
-    if (op_copy->req.hdr.opcode == OSD_OP_SEC_READ)
-    {
-        // will be freed
-        op_copy->buf = (uint8_t*)malloc_or_die(op_copy->bitmap_len + op_copy->req.sec_rw.len);
-        // won't be freed
-        op_copy->iov.push_back(op_copy->buf, op_copy->req.sec_rw.len);
-        op_copy->bitmap = op_copy->buf + op_copy->req.sec_rw.len;
-    }
-    else if (op_copy->req.hdr.opcode == OSD_OP_READ)
-    {
-        // will be freed
-        op_copy->buf = (uint8_t*)malloc_or_die(op_copy->bitmap_len + op_copy->req.rw.len);
-        // won't be freed
-        op_copy->iov.push_back(op_copy->buf, op_copy->req.rw.len);
-        op_copy->bitmap = op_copy->buf + op_copy->req.rw.len;
-    }
-    return op_copy;
-}
-
 bool osd_messenger_t::try_send(osd_client_t *cl)
 {
+    assert(cl->peer_state != PEER_RDMA);
+restart:
     if (cl->peer_state == PEER_STOPPED || cl->peer_fd < 0)
     {
         return true;
     }
-    if (cl->write_msg.msg_iovlen > 0 || !ringloop->space_left() && !use_sync_send_recv)
+    if (cl->write_msg.msg_iovlen > 0)
     {
-        return false;
+        return true;
+    }
+    if (!use_sync_send_recv)
+    {
+        if (!ringloop->space_left())
+        {
+            if (cl->write_state == 0)
+            {
+                cl->write_state = CL_WRITE_READY;
+                write_ready_clients.push_back(cl->client_id);
+            }
+            ringloop->wakeup();
+            return true;
+        }
+    }
+    else if (!cl->write_ready)
+    {
+        // Socket buffer is full, the unsent tail is still in send_list.
+        // Do not copy more data into it, just wait for EPOLLOUT
+        return true;
     }
-    assert(cl->peer_state != PEER_RDMA);
     if (cl->hs)
     {
         // Send handshake message
@@ -624,13 +551,19 @@ copy_ops:
     if (cl->io_error)
     {
         stop_client(cl->client_id);
-        return true;
+        return false;
     }
     if (!cl->send_list.size())
     {
-        cl->write_state = 0;
         return true;
     }
+    // get_op_writer_t already guarantees that it's under IOV_MAX
+    assert(cl->send_list.size() <= IOV_MAX);
+    cl->send_list_size = 0;
+    for (auto & iov: cl->send_list)
+    {
+        cl->send_list_size += iov.iov_len;
+    }
     if (!use_sync_send_recv)
     {
         auto iothread = iothreads.size() ? iothreads[cl->peer_fd % iothreads.size()] : NULL;
@@ -643,13 +576,8 @@ copy_ops:
             data_local = {};
         }
         assert(sqe);
-        cl->send_list_size = 0;
-        for (auto & iov: cl->send_list)
-        {
-            cl->send_list_size += iov.iov_len;
-        }
         cl->write_msg.msg_iov = cl->send_list.data();
-        cl->write_msg.msg_iovlen = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
+        cl->write_msg.msg_iovlen = cl->send_list.size();
         cl->refs++;
         ring_data_t* data = ((ring_data_t*)sqe->user_data);
         data->callback = [this, cl](ring_data_t *data) { handle_send(data->res, data->prev, data->more, cl); };
@@ -674,16 +602,29 @@ copy_ops:
     }
     else
     {
-        cl->write_msg.msg_iov = cl->send_list.data();
-        cl->write_msg.msg_iovlen = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
-        cl->refs++;
-        int result = sendmsg(cl->peer_fd, &cl->write_msg, MSG_NOSIGNAL);
-        if (result < 0)
+        // Try to first send with a busy loop, then postpone to the EPOLLOUT handler
+        int send_retries = MSGR_SEND_RETRIES;
+        while (cl->send_list.size())
         {
-            result = -errno;
+            cl->write_msg.msg_iov = cl->send_list.data();
+            cl->write_msg.msg_iovlen = cl->send_list.size();
+            cl->refs++;
+            int result = sendmsg(cl->peer_fd, &cl->write_msg, MSG_NOSIGNAL);
+            if (result < 0)
+            {
+                result = -errno;
+            }
+            if (!handle_send(result, false, false, cl))
+            {
+                return false;
+            }
+            if (cl->send_list.size() && --send_retries <= 0)
+            {
+                cl->write_ready = false;
+                return true;
+            }
         }
-        // like set_immediate
-        tfd->set_timer_us(0, false, [this, result, cl](int){ handle_send(result, false, false, cl); });
+        goto restart;
     }
     return true;
 }
@@ -733,9 +674,9 @@ size_t osd_messenger_t::copy_ops_to_with(osd_client_t *cl, uint8_t *dst, size_t
                 return 0;
             break;
         }
-        if (!cl->write_op)
+        if (!cl->write_op && op->op_type == OSD_OP_IN)
         {
-            // remember the operation for the current send-list
+            // this is a reply, free the op after sending it
             cl->send_free_ops.push_back(op);
         }
     }
@@ -763,21 +704,42 @@ void osd_messenger_t::send_replies()
     {
         uint64_t client_id = write_ready_clients[i];
         auto cl_it = clients.find(client_id);
-        if (cl_it != clients.end() && cl_it->second->peer_state != PEER_RDMA && !try_send(cl_it->second))
+        if (cl_it != clients.end() && cl_it->second->peer_state != PEER_RDMA)
         {
-            write_ready_clients.erase(write_ready_clients.begin(), write_ready_clients.begin() + i);
-            return;
+            if (!ringloop->space_left())
+            {
+                write_ready_clients.erase(write_ready_clients.begin(), write_ready_clients.begin() + i);
+                return;
+            }
+            cl_it->second->write_state = 0;
+            try_send(cl_it->second);
         }
     }
     write_ready_clients.clear();
 }
 
-void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t *cl)
+static void eat_send_list(osd_client_t *cl, size_t n)
+{
+    int i = 0;
+    while (i < cl->send_list.size() && n >= cl->send_list[i].iov_len)
+    {
+        n -= cl->send_list[i].iov_len;
+        i++;
+    }
+    if (i > 0)
+        cl->send_list.erase(cl->send_list.begin(), cl->send_list.begin()+i);
+    if (cl->send_list.size())
+    {
+        cl->send_list[0].iov_base = (uint8_t*)cl->send_list[0].iov_base + n;
+        cl->send_list[0].iov_len -= n;
+    }
+}
+
+bool osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t *cl)
 {
     if (!prev)
     {
         cl->write_msg.msg_iovlen = 0;
-        cl->send_list.clear();
     }
     if (!more)
     {
@@ -789,14 +751,15 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         {
             destroy_client(cl);
         }
-        return;
+        // FIXME: Return error codes instead of true/false (?)
+        return false;
     }
     if (result < 0 && result != -EAGAIN && result != -EINTR)
     {
         // this is a client socket, so don't panic. just disconnect it
         fprintf(stderr, "Client %ju socket write error: %d (%s). Disconnecting client\n", cl->client_id, -result, strerror(-result));
         stop_client(cl->client_id);
-        return;
+        return false;
     }
     if (result >= 0)
     {
@@ -808,14 +771,26 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
                 post_send_free(cl->zc_free_list[i]);
             if (i > 0)
                 cl->zc_free_list.erase(cl->zc_free_list.begin(), cl->zc_free_list.begin()+i+1);
-            return;
+            return true;
         }
-        if (cl->send_list_size > result)
+        if (cl->send_list_size <= result)
+        {
+            cl->send_list.clear();
+            cl->send_list_size = 0;
+        }
+        else if (!use_sync_send_recv)
         {
             fprintf(stderr, "Client %ju socket write error: expected to send "
                 "%zu bytes with MSG_WAITALL but sent %u. Disconnecting client\n", cl->client_id, cl->send_list_size, result);
             stop_client(cl->client_id);
-            return;
+            return false;
+        }
+        else
+        {
+            // Partially sent
+            eat_send_list(cl, result);
+            cl->send_list_size -= result;
+            return true;
         }
         for (auto op: cl->send_free_ops)
         {
@@ -827,9 +802,6 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         if (more)
             cl->zc_free_list.push_back(NULL); // end marker
         cl->send_free_ops.clear();
-        cl->write_state = 0;
-        if (cl->write_op || cl->write_ops.size())
-            cl->write_state = CL_WRITE_READY;
         if ((cl->proto_csum_status & MSGR_CSUM_NEG) && !cl->write_op && !cl->write_ops.size())
         {
             // Checksums negotiated, enable
@@ -849,10 +821,16 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         }
 #endif
     }
-    if (cl->write_state != 0)
+    if ((cl->write_op || cl->write_ops.size() || cl->hs && cl->hs->out_size()) && !use_sync_send_recv)
     {
-        write_ready_clients.push_back(cl->client_id);
+        if (cl->write_state == 0)
+        {
+            cl->write_state = CL_WRITE_READY;
+            write_ready_clients.push_back(cl->client_id);
+        }
+        ringloop->wakeup();
     }
+    return true;
 }
 
 static inline bool op_has_data_for_ssl(osd_op_t *op)

src/client/msgr_stop.cpp

@@ -48,15 +48,7 @@ void osd_messenger_t::post_send_free(osd_op_t *op)
     if (!op)
         return;
     else if (!((size_t)op & 7))
-    {
-        if (op->op_type == OSD_OP_IN)
-        {
-            // this is a reply, free the op after sending it
-            delete op;
-        }
-        else if (op->canceled)
-            op->cancel();
-    }
+        delete op;
     else
         free((void*)((size_t)op & ~(size_t)7));
 }
@@ -166,22 +158,6 @@ void osd_messenger_t::destroy_client(osd_client_t *cl)
     if (cl->peer_fd >= 0)
     {
         tfd->set_fd_handler(cl->peer_fd, false, NULL);
-        for (auto rit = read_ready_clients.begin(); rit != read_ready_clients.end(); rit++)
-        {
-            if (*rit == cl->client_id)
-            {
-                read_ready_clients.erase(rit);
-                break;
-            }
-        }
-        for (auto wit = write_ready_clients.begin(); wit != write_ready_clients.end(); wit++)
-        {
-            if (*wit == cl->client_id)
-            {
-                write_ready_clients.erase(wit);
-                break;
-            }
-        }
         clients_by_fd.erase(cl->peer_fd);
     }
 #ifdef WITH_RDMA

src/client/nbd_proxy.cpp

@@ -361,6 +361,26 @@ protected:
 public:
     ~nbd_proxy()
     {
+        if (change_thread)
+        {
+            change_thread->join();
+            change_thread.reset();
+        }
+        if (cli)
+        {
+            delete cli;
+            cli = NULL;
+        }
+        if (epmgr)
+        {
+            delete epmgr;
+            epmgr = NULL;
+        }
+        if (ringloop)
+        {
+            delete ringloop;
+            ringloop = NULL;
+        }
         if (recv_buf)
         {
             free(recv_buf);
@@ -403,6 +423,11 @@ public:
         return cfg;
     }
 
+    void request_stop()
+    {
+        stop = true;
+    }
+
     void exec(json11::Json cfg)
     {
         if (cfg["help"].bool_value())
@@ -525,10 +550,10 @@ help:
         if (!inode)
         {
             // Load image metadata
-            while (!cli->is_ready())
+            while (!cli->is_ready() && !stop)
             {
                 ringloop->loop();
-                if (cli->is_ready())
+                if (cli->is_ready() || stop)
                     break;
                 ringloop->wait();
             }
@@ -542,6 +567,10 @@ help:
             }
             watch->callback = [this](inode_watch_t*) { on_change_watch(); };
         }
+        if (stop)
+        {
+            return;
+        }
 
         // cli->config contains merged config
         if (cli->config.find("nbd_max_devices") != cli->config.end())
@@ -708,17 +737,6 @@ help:
             ringloop->wait();
         }
         cli->flush();
-        if (change_thread)
-        {
-            change_thread->join();
-            change_thread.reset();
-        }
-        delete cli;
-        delete epmgr;
-        delete ringloop;
-        cli = NULL;
-        epmgr = NULL;
-        ringloop = NULL;
     }
 
     void on_change_watch()
@@ -1271,13 +1289,25 @@ protected:
     }
 };
 
+nbd_proxy *srv = NULL;
+
+static void handle_sigint(int sig)
+{
+    if (srv)
+    {
+        srv->request_stop();
+    }
+}
+
 int main(int narg, const char *args[])
 {
     setvbuf(stdout, NULL, _IONBF, 0);
     setvbuf(stderr, NULL, _IONBF, 0);
     exe_name = args[0];
-    nbd_proxy *p = new nbd_proxy();
-    p->exec(nbd_proxy::parse_args(narg, args));
-    delete p;
+    srv = new nbd_proxy();
+    signal(SIGINT, handle_sigint);
+    signal(SIGTERM, handle_sigint);
+    srv->exec(nbd_proxy::parse_args(narg, args));
+    delete srv;
     return 0;
 }

src/client/ublk_server.cpp

@@ -99,6 +99,21 @@ public:
 
     ~ublk_server()
     {
+        if (cli)
+        {
+            delete cli;
+            cli = NULL;
+        }
+        if (epmgr)
+        {
+            delete epmgr;
+            epmgr = NULL;
+        }
+        if (ringloop)
+        {
+            delete ringloop;
+            ringloop = NULL;
+        }
         if (ctrl_fd >= 0)
         {
             close(ctrl_fd);
@@ -114,11 +129,6 @@ public:
             free(buf);
         }
         buffers.clear();
-        if (ringloop)
-        {
-            delete ringloop;
-            ringloop = NULL;
-        }
     }
 
     static json11::Json::object parse_args(int narg, const char *args[])
@@ -159,6 +169,11 @@ public:
         return cfg;
     }
 
+    void request_stop()
+    {
+        stop = true;
+    }
+
     void exec(json11::Json cfg)
     {
         if (cfg["help"].bool_value())
@@ -272,13 +287,17 @@ help:
         }
 
         // Load image metadata
-        while (!cli->is_ready())
+        while (!cli->is_ready() && !stop)
         {
             ringloop->loop();
-            if (cli->is_ready())
+            if (cli->is_ready() || stop)
                 break;
             ringloop->wait();
         }
+        if (stop)
+        {
+            return;
+        }
         if (!inode)
         {
             watch = cli->st_cli->watch_inode(image_name);
@@ -367,10 +386,6 @@ help:
         }
         cli->flush();
         ringloop->unregister_consumer(&consumer);
-        delete cli;
-        delete epmgr;
-        cli = NULL;
-        epmgr = NULL;
     }
 
     void load_module()
@@ -980,13 +995,25 @@ protected:
     }
 };
 
+ublk_server *srv = NULL;
+
+static void handle_sigint(int sig)
+{
+    if (srv)
+    {
+        srv->request_stop();
+    }
+}
+
 int main(int narg, const char *args[])
 {
     setvbuf(stdout, NULL, _IONBF, 0);
     setvbuf(stderr, NULL, _IONBF, 0);
     exe_name = args[0];
-    ublk_server *p = new ublk_server();
-    p->exec(ublk_server::parse_args(narg, args));
-    delete p;
+    srv = new ublk_server();
+    signal(SIGINT, handle_sigint);
+    signal(SIGTERM, handle_sigint);
+    srv->exec(ublk_server::parse_args(narg, args));
+    delete srv;
     return 0;
 }

src/disk_tool/disk_simple_offsets.cpp

@@ -130,6 +130,19 @@ void disk_tool_simple_offsets(json11::Json cfg, bool json_output)
     uint64_t clean_entry_bitmap_size = data_block_size/bitmap_granularity/8;
     uint64_t object_count = ((device_size-meta_offset)/data_block_size);
     uint64_t meta_size;
+    // A metadata entry must fit into a single metadata block, otherwise entries_per_block below is 0
+    auto check_entry_size = [&](uint64_t entry_size)
+    {
+        if (entry_size && entry_size <= device_block_size)
+            return;
+        fprintf(
+            stderr, "Metadata entry size (%ju bytes) exceeds meta_block_size (%ju bytes) with block_size=%ju,"
+            " bitmap_granularity=%ju, csum_block_size=%u. Decrease block_size, increase csum_block_size"
+            " or increase device_block_size\n",
+            entry_size, device_block_size, data_block_size, bitmap_granularity, csum_block_size
+        );
+        exit(1);
+    };
     if (meta_format == BLOCKSTORE_META_FORMAT_HEAP)
     {
         uint32_t min_object_size = sizeof(heap_big_intent_t) + (data_csum_size ? data_csum_size : 4) + 2*clean_entry_bitmap_size;
@@ -138,18 +151,21 @@ void disk_tool_simple_offsets(json11::Json cfg, bool json_output)
             meta_reserve = 1.5;
         else if (meta_reserve < 1)
             meta_reserve = 1;
+        check_entry_size(min_object_size);
         uint32_t entries_per_block = device_block_size / min_object_size;
         meta_size = device_block_size * (uint64_t)((object_count+entries_per_block-1) / entries_per_block * meta_reserve);
     }
     else if (meta_format == BLOCKSTORE_META_FORMAT_V2)
     {
         uint64_t clean_entry_size = 24 /*sizeof(clean_disk_entry)*/ + 2*clean_entry_bitmap_size + data_csum_size + 4 /*entry_csum*/;
+        check_entry_size(clean_entry_size);
         uint64_t entries_per_block = device_block_size / clean_entry_size;
         meta_size = (1 + (object_count+entries_per_block-1)/entries_per_block) * device_block_size;
     }
     else if (meta_format == BLOCKSTORE_META_FORMAT_V1)
     {
         uint64_t clean_entry_size = 24 /*sizeof(clean_disk_entry)*/ + 2*clean_entry_bitmap_size;
+        check_entry_size(clean_entry_size);
         uint64_t entries_per_block = device_block_size / clean_entry_size;
         meta_size = (1 + (object_count+entries_per_block-1)/entries_per_block) * device_block_size;
     }

src/nfs/nfs_proxy.cpp

@@ -39,23 +39,43 @@ nfs_proxy_t::~nfs_proxy_t()
 #ifdef WITH_RDMACM
     destroy_rdma();
 #endif
+    // Destroy the client
     if (kvfs)
+    {
         delete kvfs;
+        kvfs = NULL;
+    }
     if (blockfs)
+    {
         delete blockfs;
+        blockfs = NULL;
+    }
     if (db)
+    {
         delete db;
+        db = NULL;
+    }
     if (cmd)
+    {
         delete cmd;
+        cmd = NULL;
+    }
     if (cli)
     {
         cli->flush();
         delete cli;
+        cli = NULL;
     }
     if (epmgr)
+    {
         delete epmgr;
+        epmgr = NULL;
+    }
     if (ringloop)
+    {
         delete ringloop;
+        ringloop = NULL;
+    }
 }
 
 static const char* help_text =
@@ -340,13 +360,17 @@ void nfs_proxy_t::run(json11::Json cfg)
         blockfs->init(this, cfg);
     }
     // Load image metadata
-    while (!cli->is_ready())
+    while (!cli->is_ready() && !finished)
     {
         ringloop->loop();
-        if (cli->is_ready())
+        if (cli->is_ready() || finished)
             break;
         ringloop->wait();
     }
+    if (finished)
+    {
+        return;
+    }
     // Check default pool
     check_default_pool();
     // Daemonize before initializing messenger and RDMA because otherwise RDMA doesn't survive fork()
@@ -392,29 +416,11 @@ void nfs_proxy_t::run(json11::Json cfg)
         ringloop->loop();
         ringloop->wait();
     }
-    // Destroy the client
-    cli->flush();
-    if (kvfs)
-    {
-        delete kvfs;
-        kvfs = NULL;
-    }
-    if (blockfs)
-    {
-        delete blockfs;
-        blockfs = NULL;
-    }
-    if (db)
-    {
-        delete db;
-        db = NULL;
-    }
-    delete cli;
-    delete epmgr;
-    delete ringloop;
-    cli = NULL;
-    epmgr = NULL;
-    ringloop = NULL;
+}
+
+void nfs_proxy_t::request_stop()
+{
+    finished = true;
 }
 
 void nfs_proxy_t::run_server(json11::Json cfg)
@@ -1551,14 +1557,26 @@ void nfs_proxy_t::check_exit()
     finished = true;
 }
 
+nfs_proxy_t *srv = NULL;
+
+static void handle_sigint(int sig)
+{
+    if (srv)
+    {
+        srv->request_stop();
+    }
+}
+
 int main(int narg, const char *args[])
 {
     setvbuf(stdout, NULL, _IONBF, 0);
     setvbuf(stderr, NULL, _IONBF, 0);
     exe_name = strrchr(args[0], '/');
     exe_name = exe_name ? exe_name+1 : args[0];
-    nfs_proxy_t *p = new nfs_proxy_t();
-    p->run(nfs_proxy_t::parse_args(narg, args));
-    delete p;
+    srv = new nfs_proxy_t();
+    signal(SIGINT, handle_sigint);
+    signal(SIGTERM, handle_sigint);
+    srv->run(nfs_proxy_t::parse_args(narg, args));
+    delete srv;
     return 0;
 }

src/nfs/nfs_proxy.h

@@ -79,6 +79,7 @@ public:
 
     static json11::Json::object parse_args(int narg, const char *args[]);
     void run(json11::Json cfg);
+    void request_stop();
     void run_server(json11::Json cfg);
     void watch_stats();
     void parse_stats(etcd_kv_t & kv);

src/osd/osd_cluster.cpp

@@ -623,7 +623,7 @@ void osd_t::create_osd_state()
             {
                 printf("  listening at: %s:%jd\n", addr.string_value().c_str(), port);
             }
-            force_stop(0);
+            force_stop(1);
             return;
         }
         if (run_primary)

src/osd/osd_primary_subops.cpp

@@ -101,15 +101,7 @@ void osd_t::finish_op(osd_op_t *cur_op, int retval)
     else
     {
         // FIXME add separate magic number for primary ops
-        auto cl_it = msgr.clients.find(cur_op->client_id);
-        if (cl_it != msgr.clients.end())
-        {
-            msgr.outbox_push(cur_op);
-        }
-        else
-        {
-            delete cur_op;
-        }
+        msgr.outbox_push(cur_op);
     }
 }
 

src/osd/osd_primary_write.cpp

@@ -7,24 +7,27 @@
 bool osd_t::check_write_queue(osd_op_t *cur_op, pg_t & pg)
 {
     osd_primary_op_data_t *op_data = cur_op->op_data;
+    op_data->st = 1;
     // First check if PG is not active anymore
     if (!(pg.state & PG_ACTIVE))
     {
         pg_cancel_write_queue(pg, cur_op, op_data->oid, -EPIPE);
         return false;
     }
-    // Check if actions are pending for this object
-    auto act_it = pg.flush_actions.lower_bound((obj_piece_id_t){
-        .oid = op_data->oid,
-        .osd_num = 0,
-    });
-    op_data->st = 1;
-    if (act_it != pg.flush_actions.end() &&
-        act_it->first.oid.inode == op_data->oid.inode &&
-        (act_it->first.oid.stripe & ~STRIPE_MASK) == op_data->oid.stripe)
+    if (pg.state & PG_HAS_UNCLEAN)
     {
-        pg.write_queue.emplace(op_data->oid, cur_op);
-        return false;
+        // Check if actions are pending for this object
+        auto act_it = pg.flush_actions.lower_bound((obj_piece_id_t){
+            .oid = op_data->oid,
+            .osd_num = 0,
+        });
+        if (act_it != pg.flush_actions.end() &&
+            act_it->first.oid.inode == op_data->oid.inode &&
+            (act_it->first.oid.stripe & ~STRIPE_MASK) == op_data->oid.stripe)
+        {
+            pg.write_queue.emplace(op_data->oid, cur_op);
+            return false;
+        }
     }
     // Check if there are other write requests to the same object
     auto vo_it = pg.write_queue.find(op_data->oid);
@@ -166,11 +169,14 @@ resume_3:
         if (pg.cur_set.data() != op_data->prev_set && (op_data->stripes[0].write_start != 0 ||
             op_data->stripes[0].write_end != bs_block_size))
         {
-            memcpy(
-                (uint8_t*)op_data->stripes[0].read_buf + op_data->stripes[0].req_start,
-                op_data->stripes[0].write_buf,
-                op_data->stripes[0].req_end - op_data->stripes[0].req_start
-            );
+            if (op_data->stripes[0].req_end > op_data->stripes[0].req_start)
+            {
+                memcpy(
+                    (uint8_t*)op_data->stripes[0].read_buf + op_data->stripes[0].req_start,
+                    op_data->stripes[0].write_buf,
+                    op_data->stripes[0].req_end - op_data->stripes[0].req_start
+                );
+            }
             op_data->stripes[0].write_buf = op_data->stripes[0].read_buf;
             op_data->stripes[0].write_start = 0;
             op_data->stripes[0].write_end = bs_block_size;

src/osd/osd_rmw.cpp

@@ -886,11 +886,14 @@ static void calc_rmw_parity_copy_mod(osd_rmw_stripe_t *stripes, int pg_size, int
                 (stripes[role].req_start != 0 || stripes[role].req_end != chunk_size))
             {
                 // Copy modified chunk into the read buffer to write it back
-                memcpy(
-                    (uint8_t*)stripes[role].read_buf + stripes[role].req_start,
-                    stripes[role].write_buf,
-                    stripes[role].req_end - stripes[role].req_start
-                );
+                if (stripes[role].req_end > stripes[role].req_start)
+                {
+                    memcpy(
+                        (uint8_t*)stripes[role].read_buf + stripes[role].req_start,
+                        stripes[role].write_buf,
+                        stripes[role].req_end - stripes[role].req_start
+                    );
+                }
                 stripes[role].write_buf = stripes[role].read_buf;
                 stripes[role].write_start = 0;
                 stripes[role].write_end = chunk_size;

src/test/mock/messenger.cpp

@@ -21,7 +21,13 @@ osd_messenger_t::~osd_messenger_t()
 
 void osd_messenger_t::outbox_push(osd_op_t *cur_op)
 {
-    auto cl = clients.at(cur_op->client_id);
+    auto cl_it = clients.find(cur_op->client_id);
+    if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
+    {
+        cur_op->cancel();
+        return;
+    }
+    auto cl = cl_it->second;
     if (cur_op->op_type == OSD_OP_OUT)
         cur_op->req.hdr.id = ++cl->send_op_id;
     cl->sent_ops[cur_op->req.hdr.id] = cur_op;
@@ -67,6 +73,11 @@ void osd_messenger_t::accept_connections(int listen_fd)
 {
 }
 
+bool osd_messenger_t::wakeup_send(osd_client_t *cl)
+{
+    return true;
+}
+
 #ifdef WITH_RDMA
 json11::Json osd_messenger_t::connect_rdma(uint64_t client_id, std::string rdma_address, uint64_t client_max_msg)
 {

src/test/osd_test.cpp

@@ -55,8 +55,11 @@ static osd_op_t *make_write_op(inode_t inode, uint64_t offset, uint64_t len, uin
     op->req.rw.offset = offset;
     op->req.rw.len = len;
     op->req.rw.version = 0;
-    op->buf = malloc(len);
-    memset(op->buf, fill, len);
+    if (len > 0)
+    {
+        op->buf = malloc(len);
+        memset(op->buf, fill, len);
+    }
     return op;
 }
 

src/test/test_blockstore.cpp

@@ -362,6 +362,121 @@ static void test_fsync(bool separate_meta)
     free(op2.buf);
 }
 
+static void test_fsync_meta_before_complete()
+{
+    printf("\n-- test_fsync_meta_before_complete\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["disable_data_fsync"] = "0";
+    test.config["immediate_commit"] = "none";
+    test.config["meta_device"] = "./test_meta.bin";
+    test.config["disable_meta_fsync"] = "0";
+    test.config["meta_device_size"] = "33554432";
+    test.config["meta_device_sect"] = "4096";
+    test.config["data_offset"] = "0";
+    test.init();
+    test.meta_disk->trace = 1;
+
+    // Write
+    printf("writing 16K+4K v1\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 16384;
+    op.len = 4096;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
+    memset(op.buf, 0xaa, 4096);
+
+    // Block metadata write execution
+    uint64_t mod_off = 0;
+    void *mod_buf = NULL;
+    ring_data_t *mod_data = NULL;
+    test.sqe_handler = [&](io_uring_sqe *sqe)
+    {
+        if (sqe->fd == MOCK_META_FD && sqe->opcode == IORING_OP_WRITEV)
+        {
+            assert(sqe->len == 1);
+            mod_off = sqe->off;
+            mod_buf = ((iovec*)sqe->addr)[0].iov_base;
+            mod_data = (ring_data_t*)sqe->user_data;
+            assert(((iovec*)sqe->addr)[0].iov_len == test.bs->dsk.meta_block_size);
+            return true;
+        }
+        return false;
+    };
+
+    bool wr_done = false;
+    op.callback = [&](blockstore_op_t *op)
+    {
+        printf("op opcode=%ju completed retval=%jd\n", op->opcode, op->retval);
+        wr_done = true;
+    };
+    test.bs->enqueue_op(&op);
+
+    while (!mod_buf)
+        test.ringloop->loop();
+
+    // Sync
+    printf("syncing\n");
+    blockstore_op_t sync_op;
+    sync_op.opcode = BS_OP_SYNC;
+    test.exec_op(&sync_op);
+    assert(sync_op.retval == 0);
+
+    // Now send the metadata write
+    {
+        ring_data_t data;
+        io_uring_sqe sqe;
+        iovec iov;
+        sqe.opcode = IORING_OP_WRITEV;
+        sqe.off = test.bs->dsk.meta_offset + mod_off;
+        sqe.addr = (uint64_t)&iov;
+        sqe.len = 1;
+        sqe.rw_flags = 0;
+        iov.iov_base = mod_buf;
+        iov.iov_len = test.bs->dsk.meta_block_size;
+        sqe.user_data = (uint64_t)mod_data;
+        bool ok = test.meta_disk->submit(&sqe);
+        assert(ok);
+        test.ringloop->mark_completed(mod_data);
+    }
+
+    while (!wr_done)
+        test.ringloop->loop();
+    assert(op.retval == op.len);
+
+    // Sync again
+    printf("syncing again\n");
+    test.exec_op(&sync_op);
+    assert(sync_op.retval == 0);
+
+    // Discard and restart
+    printf("destroying again\n");
+    test.destroy_bs();
+    test.meta_disk->discard_buffers(true, 0);
+    test.init();
+
+    // Check that the write isn't lost
+    printf("checking for OK\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, 16*1024));
+    assert(memcmp(op2.buf+16*1024, op.buf, 4*1024) == 0);
+    assert(is_zero(op2.buf+20*1024, 108*1024));
+
+    free(op.buf);
+    free(op2.buf);
+}
+
 static void test_intent_over_unstable()
 {
     printf("\n-- test_intent_over_unstable\n");
@@ -976,6 +1091,63 @@ static void test_list_limit()
     free(op.buf);
 }
 
+static void test_intent_write_corruption()
+{
+    printf("\n-- test_intent_write_corruption\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["csum_block_size"] = "16384";
+    test.config["bitmap_granularity"] = "4096";
+    test.config["perfect_csum_update"] = "false";
+    test.init();
+
+    // 1. Write BIG_WRITE (0-48K)
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 49152; // 48K
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op.len);
+    memset(op.buf, 0xaa, op.len);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+
+    // 2. Write unaligned INTENT_WRITE B (8K-12K)
+    free(op.buf);
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 2;
+    op.offset = 8192; // 8K
+    op.len = 4096; // 4K
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op.len);
+    memset(op.buf, 0xbb, op.len);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+    
+    // 3. Read 12-40K
+    // This will hit the hole logic where B provided 8-12K, but we don't even request it.
+    // The request is for 12-40K.
+    // 12K is not aligned to 16K, so A must pad by reading 0-16K.
+    // However, 8-12K of the physical block has been overwritten by B.
+    // A will try to verify the 0-16K block with its old checksum and fail.
+    free(op.buf);
+    op.opcode = BS_OP_READ;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = UINT64_MAX; // latest version
+    op.offset = 12288; // 12K
+    op.len = 28672; // 28K (so end = 40K)
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op.len);
+    test.exec_op(&op);
+
+    // Assert that the read succeeds without checksum corruption errors
+    assert(op.retval == op.len);
+
+    free(op.buf);
+}
+
 // FIXME Add a simple intent_write / big_intent test
 
 int main(int narg, char *args[])
@@ -983,7 +1155,9 @@ int main(int narg, char *args[])
     test_simple();
     test_fsync(false);
     test_fsync(true);
+    test_fsync_meta_before_complete();
     test_intent_over_unstable();
+    test_intent_write_corruption();
     test_padded_csum_intent(false);
     test_padded_csum_intent(true);
     test_padded_csum_sparse_leading_hole();

src/test/test_blockstore_v1.cpp

@@ -5,6 +5,7 @@
 #include "str_util.h"
 #include "ringloop_mock.h"
 #include "blockstore/v1/impl.h"
+#include "blockstore/v1/internal.h"
 
 struct bs_test_t
 {
@@ -45,11 +46,22 @@ struct bs_test_t
         return bs->used_blocks;
     }
 
+    uint64_t free_block_count()
+    {
+        return bs->get_free_block_count();
+    }
+
     bool is_data_loc_used(uint64_t loc)
     {
         return bs->data_alloc->get(loc / bs->dsk.data_block_size);
     }
 
+    v1::dirty_entry* find_dirty_entry(object_id oid, uint64_t version)
+    {
+        auto it = bs->dirty_db.find((obj_ver_id){ .oid = oid, .version = version });
+        return it == bs->dirty_db.end() ? NULL : &it->second;
+    }
+
     void force_compaction()
     {
         printf("triggering compaction\n");
@@ -315,6 +327,25 @@ static void test_validate_padded_journal()
     test.init();
     printf("blockstore initialized\n");
 
+    // initialize the beginning of data area with non-zero data to check
+    // explicit padding instead of the on-disk data verification
+    {
+        std::vector<uint8_t> nonzero_data(1048576, 0xF8);
+        ring_data_t data;
+        io_uring_sqe sqe;
+        iovec iov;
+        sqe.opcode = IORING_OP_WRITEV;
+        sqe.off = test.dsk().data_offset;
+        sqe.addr = (uint64_t)&iov;
+        sqe.len = 1;
+        sqe.rw_flags = 0;
+        iov.iov_base = nonzero_data.data();
+        iov.iov_len = nonzero_data.size();
+        sqe.user_data = (uint64_t)&data;
+        bool ok = test.data_disk->submit(&sqe);
+        assert(ok);
+    }
+
     printf("write v1 4+32k\n");
     blockstore_op_t op;
     op.opcode = BS_OP_WRITE_STABLE;
@@ -852,6 +883,402 @@ static void test_meta_tail_reload()
     printf("blockstore re-initialized with an overflowing entry\n");
 }
 
+// Check null dereference in read_bitmap over DELETE
+static void test_read_bitmap_of_deleted_object()
+{
+    printf("\n-- test_read_bitmap_of_deleted_object\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    // 128k/16k*4 = 32 bytes of checksums --> dyn_data allocated dynamically
+    test.config["data_csum_type"] = "crc32c";
+    test.config["csum_block_size"] = "16384";
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint8_t bmp[16];
+    memset(bmp, 0xCC, sizeof(bmp));
+
+    printf("write v1 0+128k\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op.len);
+    op.bitmap = bmp;
+    memset(op.buf, 0xAA, op.len);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+    {
+        // Check that big_write is here and with dyn_data
+        v1::dirty_entry *de = test.find_dirty_entry(op.oid, 1);
+        assert(de != NULL);
+        assert(de->dyn_data != NULL);
+    }
+
+    {
+        uint64_t version = 0;
+        assert(test.dsk().clean_entry_bitmap_size <= sizeof(bmp));
+        memset(bmp, 0, sizeof(bmp));
+        assert(test.bs->read_bitmap(op.oid, UINT64_MAX, bmp, &version) == 0);
+        assert(version == 1);
+        assert(memcheck(bmp, 0xCC, test.dsk().clean_entry_bitmap_size));
+    }
+
+    {
+        printf("delete v2\n");
+        blockstore_op_t del_op;
+        del_op.opcode = BS_OP_DELETE;
+        del_op.oid = { .inode = 1, .stripe = 0 };
+        del_op.version = 2;
+        test.exec_op(&del_op);
+        assert(del_op.retval == 0);
+    }
+
+    {
+        // Check that the delete is still in dirty_db
+        v1::dirty_entry *de = test.find_dirty_entry(op.oid, 2);
+        assert(de != NULL);
+        assert((de->state & BS_ST_TYPE_MASK) == BS_ST_DELETE);
+        assert(de->dyn_data == NULL);
+    }
+
+    auto check_deleted_bmp = [&]()
+    {
+        uint64_t version = UINT64_MAX;
+        memset(bmp, 0xCC, sizeof(bmp));
+        assert(test.bs->read_bitmap(op.oid, UINT64_MAX, bmp, &version) == -ENOENT);
+        assert(version == 0);
+        assert(memcheck(bmp, 0, test.dsk().clean_entry_bitmap_size));
+    };
+    printf("read_bitmap of the deleted object - should be -ENOENT\n");
+    check_deleted_bmp();
+
+    {
+        // Reading the data must agree with that
+        printf("read of the deleted object - should be -ENOENT\n");
+        blockstore_op_t read_op;
+        read_op.opcode = BS_OP_READ;
+        read_op.oid = { .inode = 1, .stripe = 0 };
+        read_op.version = UINT64_MAX;
+        read_op.offset = 0;
+        read_op.len = 128*1024;
+        read_op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, read_op.len);
+        test.exec_op(&read_op);
+        assert(read_op.retval == -ENOENT);
+        free(read_op.buf);
+    }
+
+    // And once the delete is flushed too, via the clean_db path
+    test.force_compaction();
+    printf("read_bitmap after the delete is flushed - should be -ENOENT\n");
+    check_deleted_bmp();
+
+    // Write over delete to also check that it doesn't read NULL dyn_data
+    printf("write v3 0+128k over the delete\n");
+    op.version = 3;
+    memset(op.buf, 0xAB, op.len);
+    memset(bmp, 0xDE, sizeof(bmp));
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+    {
+        uint64_t version = 0;
+        memset(bmp, 0, sizeof(bmp));
+        assert(test.bs->read_bitmap(op.oid, UINT64_MAX, bmp, &version) == 0);
+        assert(memcheck(bmp, 0xDE, test.dsk().clean_entry_bitmap_size));
+        assert(version == 3);
+    }
+
+    free(op.buf);
+}
+
+// used_blocks and inode_space_stats count one data block per object and only when a big_write
+// or a delete is marked stable. counters should persist correctly over a restart.
+static void test_used_blocks_replay_stable_over_small_write()
+{
+    printf("\n-- test_used_blocks_replay_stable_over_small_write\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint64_t block_count = test.dsk().block_count;
+    assert(!test.used_blocks());
+
+    blockstore_op_t op;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    memset(op.buf, 0xAA, 128*1024);
+
+    // Unstable big write. BS_OP_WRITE (not WRITE_STABLE) so that it needs an explicit
+    // BS_OP_STABLE and stays in dirty_db until then.
+    printf("write v1 0+128k (big)\n");
+    op.opcode = BS_OP_WRITE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    // Small write - the entry that used to be misread as "the block is already counted"
+    printf("write v2 0+4k (small)\n");
+    op.version = 2;
+    op.len = 4096;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("write v3 0+128k (big)\n");
+    op.version = 3;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("stabilize v3\n");
+    op.opcode = BS_OP_STABLE;
+    op.len = 1;
+    *((obj_ver_id*)op.buf) = (obj_ver_id){ .oid = { .inode = 1, .stripe = 0 }, .version = 3 };
+    test.exec_op(&op);
+    assert(op.retval == 0);
+    // The live path walks the whole range, so it has always got this one right
+    printf("used_blocks after stabilize = %ju (expected 1)\n", test.used_blocks());
+    assert(test.used_blocks() == 1);
+    // Nothing must have been flushed - the journal alone has to reproduce the counters below
+    assert(test.flusher()->get_queue_size());
+
+    printf("restarting the blockstore (journal replay)\n");
+    test.destroy_bs();
+    test.init();
+    uint64_t after_replay = test.used_blocks();
+    printf("used_blocks after replay = %ju (expected 1)\n", after_replay);
+
+    // Deleting the object must bring the counter back to zero, not below it
+    printf("delete v4\n");
+    op.opcode = BS_OP_DELETE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 4;
+    op.offset = 0;
+    op.len = 0;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+    uint64_t after_delete = test.used_blocks();
+    uint64_t free_after_delete = test.free_block_count();
+    printf("used_blocks after delete = %ju (expected 0), free_block_count = %ju (expected %ju)\n",
+        after_delete, free_after_delete, block_count);
+
+    assert(after_replay == 1);
+    assert(after_delete == 0);
+    assert(free_after_delete == block_count);
+
+    free(op.buf);
+}
+
+// The same accounting rule for deletes. used_space counter must not underflow after a restart.
+static void test_used_blocks_replay_delete_over_unstable_big_write()
+{
+    printf("\n-- test_used_blocks_replay_delete_over_unstable_big_write\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint64_t block_count = test.dsk().block_count;
+    assert(!test.used_blocks());
+
+    blockstore_op_t op;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    memset(op.buf, 0xAA, 128*1024);
+
+    printf("write v1 0+128k (big, left unstable)\n");
+    op.opcode = BS_OP_WRITE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+    // An unstable big write owns a block but is not counted yet - that is what mark_stable() is for
+    assert(!test.used_blocks());
+
+    printf("delete v2\n");
+    op.opcode = BS_OP_DELETE;
+    op.version = 2;
+    op.offset = 0;
+    op.len = 0;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+    assert(!test.used_blocks());
+
+    printf("restarting the blockstore (journal replay)\n");
+    test.destroy_bs();
+    test.init();
+    uint64_t after_replay = test.used_blocks();
+    uint64_t free_after_replay = test.free_block_count();
+    printf("used_blocks after replay = %ju (expected 0), free_block_count = %ju (expected %ju)\n",
+        after_replay, free_after_replay, block_count);
+    assert(after_replay == 0);
+    assert(free_after_replay == block_count);
+
+    free(op.buf);
+}
+
+static void test_delete_over_ec_write()
+{
+    printf("\n-- test_delete_over_ec_write\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.init();
+    printf("blockstore initialized\n");
+
+    blockstore_op_t op;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    memset(op.buf, 0xAA, 128*1024);
+
+    printf("write v1 0+128k (big, stable)\n");
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+    assert(test.used_blocks() > 0);
+
+    printf("write v2 0+4k (big, unstable)\n");
+    op.opcode = BS_OP_WRITE;
+    op.version = 2;
+    op.len = 4*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("delete v3 - should mark all previous versions stable\n");
+    op.opcode = BS_OP_DELETE;
+    op.version = 3;
+    op.offset = 0;
+    op.len = 0;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 2);
+        assert(dirty);
+        assert(IS_STABLE(dirty->state));
+    }
+
+    printf("restarting the blockstore (journal replay)\n");
+    test.destroy_bs();
+    test.init();
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 2);
+        if (!dirty)
+        {
+            dirty = test.find_dirty_entry(op.oid, 3);
+            assert(!dirty);
+        }
+        else
+        {
+            assert(IS_STABLE(dirty->state));
+            dirty = test.find_dirty_entry(op.oid, 3);
+            assert(IS_DELETE(dirty->state));
+        }
+    }
+
+    free(op.buf);
+}
+
+static void test_delete_over_ec_write_unstable()
+{
+    printf("\n-- test_delete_over_ec_write_unstable\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["disable_data_fsync"] = "0";
+    test.config["immediate_commit"] = "none";
+    test.init();
+    printf("blockstore initialized\n");
+
+    blockstore_op_t op;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    memset(op.buf, 0xAA, 128*1024);
+
+    printf("write v1 0+128k (big, stable)\n");
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("write v2 0+4k (big, unstable)\n");
+    op.opcode = BS_OP_WRITE;
+    op.version = 2;
+    op.len = 4*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("delete v3 - should mark all previous versions stable, but only after sync\n");
+    bool del_done = false;
+    op.opcode = BS_OP_DELETE;
+    op.version = 3;
+    op.offset = 0;
+    op.len = 0;
+    op.callback = [&](blockstore_op_t *op)
+    {
+        del_done = true;
+    };
+    test.exec_op(&op);
+    assert(op.retval == 0);
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 3);
+        assert(dirty);
+        assert(!IS_SYNCED(dirty->state));
+        assert(!IS_STABLE(dirty->state));
+    }
+
+    blockstore_op_t sync_op;
+    sync_op.opcode = BS_OP_SYNC;
+    test.exec_op(&sync_op);
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 2);
+        assert(dirty);
+        assert(IS_STABLE(dirty->state));
+        dirty = test.find_dirty_entry(op.oid, 3);
+        assert(dirty);
+        assert(IS_STABLE(dirty->state));
+    }
+
+    printf("restarting the blockstore (journal replay)\n");
+    test.destroy_bs();
+    test.init();
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 2);
+        if (!dirty)
+        {
+            dirty = test.find_dirty_entry(op.oid, 3);
+            assert(!dirty);
+        }
+        else
+        {
+            assert(IS_STABLE(dirty->state));
+            dirty = test.find_dirty_entry(op.oid, 3);
+            assert(IS_DELETE(dirty->state));
+        }
+    }
+
+    free(op.buf);
+}
+
 int main(int narg, char *args[])
 {
     test_preserve_corruption();
@@ -863,5 +1290,10 @@ int main(int narg, char *args[])
     test_read_retry_on_ring_full_1M_csum4k_clean();
     test_read_free_clean_loc_used();
     test_meta_tail_reload();
+    test_read_bitmap_of_deleted_object();
+    test_used_blocks_replay_stable_over_small_write();
+    test_used_blocks_replay_delete_over_unstable_big_write();
+    test_delete_over_ec_write();
+    test_delete_over_ec_write_unstable();
     return 0;
 }

tests/run_tests.sh

@@ -27,6 +27,8 @@ ETCD_SCHEME=https ./test_snapshot.sh
 
 ./test_img_online_resize.sh
 
+./test_no_uring.sh
+
 ./test_interrupted_rebalance.sh
 IMMEDIATE_COMMIT=1 ./test_interrupted_rebalance.sh
 SCHEME=ec ./test_interrupted_rebalance.sh

tests/test_no_uring.sh

@@ -0,0 +1,8 @@
+#!/bin/bash -ex
+
+. `dirname $0`/run_3osds.sh
+
+$VITASTOR_FIO -bs=128k -direct=1 -numjobs=1 -iodepth=4 \
+    -rw=write -pool=1 -inode=1 -size=128M -runtime=10 -no_io_uring=1
+
+format_green OK