WIP refactor dequeue_write/continue_write into a single "coroutine"

Test scenarios:
- 128KB aligned I/O (may work despite bug)
- 4KB random I/O (triggers assertion failure)
- 4KB mixed read/write (triggers crash)
- 4KB sequential write (triggers crash)

I accept Vitastor CLA agreement: https://git.yourcmc.ru/vitalif/vitastor/src/branch/master/CLA-en.md

src/blockstore/blockstore_disk.cpp

@@ -171,6 +171,12 @@ void blockstore_disk_t::parse_config(std::map<std::string, std::string> & config
     {
         throw std::runtime_error("Data block size must be a multiple of sparse write tracking granularity");
     }
+    if (data_block_size / bitmap_granularity < 8)
+    {
+        fprintf(stderr, "Warning: block_size (%u) / bitmap_granularity (%u) = %u bits. "
+            "Consider using larger block_size or bitmap_granularity for better performance.\n",
+            data_block_size, bitmap_granularity, data_block_size / bitmap_granularity);
+    }
     if (!data_csum_type)
     {
         csum_block_size = 0;
@@ -259,7 +265,7 @@ void blockstore_disk_t::calc_lengths(bool skip_meta_check)
     }
     // required metadata size
     block_count = data_len / data_block_size;
-    clean_entry_bitmap_size = data_block_size / bitmap_granularity / 8;
+    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);
 recalc:

src/blockstore/blockstore_flush.h

@@ -58,7 +58,6 @@ class journal_flusher_co
     int i, res;
     bool read_to_fill_incomplete;
     int copy_count;
-    bool do_repeat = false;
 
     friend class journal_flusher_t;
 

src/blockstore/blockstore_heap.cpp

@@ -1347,6 +1347,7 @@ int blockstore_heap_t::add_small_write(object_id oid, heap_entry_t **obj_ptr, ui
     // Small writes are written in parallel with buffered data so they require explicit_complete
     return add_entry(wr_size, modified_block, false, true, [&](heap_entry_t *wr)
     {
+        printf("add_small_write t%u %lx:%lx l%lu v%lu %u +%u loc:%lx\n", type, oid.inode, oid.stripe, wr->lsn, version, offset, len, location);
         wr->entry_type = type;
         wr->inode = oid.inode;
         wr->stripe = oid.stripe;
@@ -1376,6 +1377,7 @@ int blockstore_heap_t::add_big_write(object_id oid, heap_entry_t *old_head, bool
     // Big writes are written after writing data so they don't require explicit_complete
     return add_entry(wr_size, modified_block, false, false, [&](heap_entry_t *wr)
     {
+        printf("add_big_write %lx:%lx l%lu v%lu loc:%lx\n", oid.inode, oid.stripe, wr->lsn, version, location);
         wr->entry_type = BS_HEAP_BIG_WRITE | (stable ? BS_HEAP_STABLE : 0);
         wr->inode = oid.inode;
         wr->stripe = oid.stripe;
@@ -1402,6 +1404,7 @@ int blockstore_heap_t::add_redirect_intent(object_id oid, heap_entry_t **obj_ptr
     // Big-redirect intents, just like regular big writes, are written after writing data so they don't require explicit_complete
     return add_entry(wr_size, modified_block, false, false, [&](heap_entry_t *wr)
     {
+        printf("add_redir_intent %lx:%lx l%lu v%lu %u +%u loc:%lx\n", oid.inode, oid.stripe, wr->lsn, version, offset, len, location);
         wr->entry_type = BS_HEAP_BIG_INTENT|BS_HEAP_STABLE;
         wr->inode = oid.inode;
         wr->stripe = oid.stripe;
@@ -1438,6 +1441,7 @@ int blockstore_heap_t::add_big_intent(object_id oid, heap_entry_t **obj_ptr, uin
     // Big intents are written before writing data so they require explicit_complete
     return add_entry(wr_size, modified_block, false, true, [&](heap_entry_t *wr)
     {
+        printf("add_big_intent %lx:%lx l%lu v%lu %u +%u loc:%lx\n", oid.inode, oid.stripe, wr->lsn, version, offset, len, obj->big_location(this));
         wr->entry_type = BS_HEAP_BIG_INTENT | BS_HEAP_STABLE;
         wr->inode = oid.inode;
         wr->stripe = oid.stripe;
@@ -1493,6 +1497,7 @@ int blockstore_heap_t::add_compact(heap_entry_t *obj, uint64_t compact_version,
     // Compaction entry is added after copying data so it doesn't require explicit_complete
     return add_entry(wr_size, modified_block, true, false, [&](heap_entry_t *new_wr)
     {
+        printf("add_compact %lx:%lx l%lu v%lu loc:%lx\n", obj->inode, obj->stripe, compact_lsn, compact_version, compact_location);
         new_wr->entry_type = BS_HEAP_BIG_WRITE|BS_HEAP_STABLE;
         new_wr->inode = obj->inode;
         new_wr->stripe = obj->stripe;

src/blockstore/blockstore_heap.h

@@ -163,7 +163,7 @@ using heap_mvcc_map_t = robin_hood::unordered_flat_map<object_id, heap_object_mv
 
 class blockstore_heap_t
 {
-    friend class heap_entry_t;
+    friend struct heap_entry_t;
 
     blockstore_disk_t *dsk = NULL;
     uint8_t* buffer_area = NULL;

src/blockstore/blockstore_init.cpp

@@ -269,7 +269,7 @@ resume_6:
         }
         GET_SQE();
         data->iov = (iovec){ buf, len };
-        data->callback = [this, offset, cb](ring_data_t *data)
+        data->callback = [offset, cb](ring_data_t *data)
         {
             if (data->res < 0)
             {

src/blockstore/blockstore_write.cpp

@@ -13,10 +13,13 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
     return true;
 }
 
-void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
+bool blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
 {
-    if (modified_blocks.find(modified_block) != modified_blocks.end())
-        return;
+    auto mod_it = modified_blocks.find(modified_block);
+    if (mod_it != modified_blocks.end())
+    {
+        return !mod_it->second.sent;
+    }
     io_uring_sqe *sqe = get_sqe();
     assert(sqe != NULL);
     ring_data_t *data = ((ring_data_t*)sqe->user_data);
@@ -41,6 +44,7 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
     unsynced_meta_write_count++;
     pending_modified_blocks.push_back(modified_block);
     modified_blocks[modified_block] = { .sent = false, .buf = buf };
+    return true;
 }
 
 bool blockstore_impl_t::meta_block_is_pending(uint32_t modified_block)
@@ -121,6 +125,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     heap_entry_t *obj = heap->read_entry(op->oid);
     if (op->opcode == BS_OP_DELETE)
     {
+        return continue_delete(op, 0);
         // Delete
         if (!obj || obj->type() == BS_HEAP_DELETE)
         {
@@ -133,17 +138,38 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         BS_SUBMIT_CHECK_SQES(1);
         int res = heap->add_delete(obj, &PRIV(op)->modified_block);
         if (res == ENOSPC)
+        {
             goto enospc;
+        }
         assert(res == 0);
-        prepare_meta_block_write(PRIV(op)->modified_block);
-        PRIV(op)->pending_ops++;
-        PRIV(op)->op_state = 5;
         write_iodepth++;
+resume_1:
+        while (!prepare_meta_block_write(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 1;
+            return 1;
+        }
+rseume_2:
+        while (meta_block_is_pending(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 2;
+            return 1;
+        }
+resume_3:
+resume_4:
+        if (!throttle_write(op, 3))
+        {
+            return 1;
+        }
+        write_iodepth--;
+        ack_write(op);
+        return 2;
     }
     // FIXME: Allow to do initial writes as buffered, not redirected
     // FIXME: Allow to do direct writes over holes
     else if (!obj || obj->type() == BS_HEAP_DELETE || op->offset == 0 && op->len == dsk.data_block_size)
     {
+        return continue_big_write(op, 10);
         // Big (redirect) write
         PRIV(op)->write_type = dsk.disable_data_fsync || op->opcode != BS_OP_WRITE_STABLE ? BS_HEAP_BIG_WRITE : _REDIRECT_INTENT;
         BS_SUBMIT_CHECK_SQES(1);
@@ -163,6 +189,7 @@ enospc:
             flusher->request_trim();
             return 0;
         }
+        write_iodepth++;
         uint64_t loc = PRIV(op)->location;
 #ifdef BLOCKSTORE_DEBUG
         printf(
@@ -176,18 +203,72 @@ enospc:
         data->iov = (struct iovec){ op->buf, op->len };
         data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
         io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + loc + op->offset);
+        if (PRIV(op)->write_type == BS_HEAP_BIG_WRITE)
+            inflight_big++;
         PRIV(op)->pending_ops++;
-        write_iodepth++;
+resume_10:
+        if (PRIV(op)->pending_ops > 0)
+        {
+            PRIV(op)->op_state = 10;
+            return 1;
+        }
         if (PRIV(op)->write_type == BS_HEAP_BIG_WRITE)
         {
-            PRIV(op)->op_state = 1;
-            inflight_big++;
+            inflight_big--;
+resume_11:
+resume_12:
+resume_13:
+            if (!fsync_big_write(op, 11))
+                return 1;
+        }
+        heap_entry_t *obj = heap->read_entry(op->oid);
+        int res = 0;
+        if (PRIV(op)->write_type == _REDIRECT_INTENT)
+        {
+            res = heap->add_redirect_intent(op->oid, &obj, op->version, op->offset, op->len,
+                PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
         }
         else
-            PRIV(op)->op_state = 3;
+        {
+            res = heap->add_big_write(op->oid, obj, op->opcode == BS_OP_WRITE_STABLE,
+                op->version, op->offset, op->len, PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
+        }
+        if (res == ENOSPC)
+        {
+            if (!heap->get_to_compact_count())
+            {
+                // no space
+                heap->free_data(op->oid.inode, PRIV(op)->location);
+                write_iodepth--;
+                op->retval = -ENOSPC;
+                FINISH_OP(op);
+                return 2;
+            }
+            PRIV(op)->wait_for = WAIT_COMPACTION;
+            PRIV(op)->wait_detail = heap->get_compacted_count();
+            flusher->request_trim();
+            return 0;
+        }
+        assert(res == 0);
+resume_14:
+        while (!prepare_meta_block_write(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 14;
+            return 1;
+        }
+resume_15:
+        while (meta_block_is_pending(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 15;
+            return 1;
+        }
+        write_iodepth--;
+        ack_write(op);
+        return 2;
     }
     else if (intent_write_allowed(op, obj))
     {
+        return continue_intent_write(op, 20);
         // Direct intent-write
         BS_SUBMIT_CHECK_SQES(1);
         int res = 0;
@@ -223,13 +304,41 @@ enospc:
             assert(res == 0);
             PRIV(op)->lsn = obj->lsn;
         }
-        prepare_meta_block_write(PRIV(op)->modified_block);
-        PRIV(op)->pending_ops++;
-        PRIV(op)->op_state = 9;
         write_iodepth++;
+resume_20:
+        while (!prepare_meta_block_write(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 20;
+            return 1;
+        }
+resume_21:
+        while (meta_block_is_pending(PRIV(op)->modified_block))
+        {
+            PRIV(op)->op_state = 21;
+            return 1;
+        }
+        // Direct intent-write
+        // LSN is not marked as completed so big_write won't be freed
+        BS_SUBMIT_GET_SQE(sqe, data);
+        data->iov = (struct iovec){ op->buf, op->len };
+        data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
+        io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + PRIV(op)->location + op->offset);
+        if (dsk.use_atomic_flag)
+            sqe->rw_flags = RWF_ATOMIC;
+        PRIV(op)->pending_ops++;
+resume_22:
+        if (PRIV(op)->pending_ops > 0)
+        {
+            PRIV(op)->op_state = 22;
+            return 1;
+        }
+        write_iodepth--;
+        ack_write(op);
+        return 2;
     }
     else
     {
+        return continue_small_write(op, 30);
         // Small (buffered) overwrite
         // First check if there is free buffer space
         PRIV(op)->write_type = BS_HEAP_SMALL_WRITE;
@@ -242,7 +351,7 @@ enospc:
             return 0;
         }
         // There is sufficient space. Check SQE(s)
-        BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0));
+        BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0)); ---> refactor too
         int res = heap->add_small_write(op->oid, &obj, (BS_HEAP_SMALL_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)),
             op->version, op->offset, op->len, loc, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
         if (res == ENOSPC)
@@ -311,27 +420,46 @@ again:
         goto again;
     }
 resume_2:
+
+
+resume_4:
+resume_6:
+
+resume_8:
+    ack
+    return 2;
+resume_10:
+    return 1;
+}
+
+bool blockstore_impl_t::fsync_big_write(blockstore_op_t *op, int base_state)
+{
+    if (PRIV(op)->state == base_state)
+        goto resume_0;
+    else if (PRIV(op)->state == base_state+1)
+        goto resume_1;
+    else if (PRIV(op)->state == base_state+2)
+        goto resume_2;
     // We must fsync all big writes to avoid complex write workflows
     // It's OK for all HDDs and for server SSDs, but slightly worse for desktop SSDs
-    inflight_big--;
     if (!dsk.disable_data_fsync)
     {
         // fsync data in a batch
-resume_11:
+resume_0:
         if (inflight_big > 0)
         {
-            PRIV(op)->op_state = 11;
-            return 1;
+            PRIV(op)->op_state = base_state;
+            return false;
         }
         if (fsyncing_data)
         {
-resume_12:
+resume_1:
             if (fsyncing_data)
             {
-                PRIV(op)->op_state = 12;
-                return 1;
+                PRIV(op)->op_state = base_state+1;
+                return false;
             }
-            goto resume_4;
+            return true;
         }
         fsyncing_data = true;
         BS_SUBMIT_GET_SQE(sqe, data);
@@ -343,47 +471,23 @@ resume_12:
             handle_write_event(data, op);
         };
         PRIV(op)->pending_ops++;
-        PRIV(op)->op_state = 3;
-        return 1;
+resume_2:
+        if (PRIV(op)->pending_ops > 0)
+        {
+            PRIV(op)->op_state = base_state+2;
+            return false;
+        }
     }
-resume_4:
-    {
-        auto obj = heap->read_entry(op->oid);
-        int res = 0;
-        if (PRIV(op)->write_type == _REDIRECT_INTENT)
-        {
-            res = heap->add_redirect_intent(op->oid, &obj, op->version, op->offset, op->len,
-                PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
-        }
-        else
-        {
-            res = heap->add_big_write(op->oid, obj, op->opcode == BS_OP_WRITE_STABLE,
-                op->version, op->offset, op->len, PRIV(op)->location, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
-        }
-        if (res == ENOSPC)
-        {
-            if (!heap->get_to_compact_count())
-            {
-                // no space
-                heap->free_data(op->oid.inode, PRIV(op)->location);
-                write_iodepth--;
-                op->retval = -ENOSPC;
-                FINISH_OP(op);
-                return 2;
-            }
-            PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = heap->get_compacted_count();
-            flusher->request_trim();
-            return 0;
-        }
-        assert(res == 0);
-        prepare_meta_block_write(PRIV(op)->modified_block);
-        PRIV(op)->pending_ops++;
-        PRIV(op)->op_state = 5;
-        return 1;
-    }
-resume_6:
+    return true;
+}
+
+bool blockstore_impl_t::throttle_write(blockstore_op_t *op, int base_state)
+{
     // Apply throttling to not fill the journal too quickly for the SSD+HDD case
+    if (PRIV(op)->op_state >= base_state+1)
+    {
+        return true;
+    }
     if (PRIV(op)->write_type == BS_HEAP_SMALL_WRITE && throttle_small_writes)
     {
         // Apply throttling
@@ -404,17 +508,21 @@ resume_6:
         if (ref_us > exec_us + throttle_threshold_us)
         {
             // Pause reply
-            PRIV(op)->op_state = 7;
+            PRIV(op)->op_state = base_state;
             // Remember that the timer can in theory be called right here
             tfd->set_timer_us(ref_us-exec_us, false, [this, op](int timer_id)
             {
-                PRIV(op)->op_state = 8;
+                PRIV(op)->op_state++;
                 ringloop->wakeup();
             });
-            return 1;
+            return false;
         }
     }
-resume_8:
+    return true;
+}
+
+void blockstore_impl_t::ack_write(blockstore_op_t *op)
+{
     // Acknowledge write
 #ifdef BLOCKSTORE_DEBUG
     printf("Ack write %jx:%jx v%ju\n", op->oid.inode, op->oid.stripe, op->version);
@@ -441,21 +549,7 @@ resume_8:
         unsynced_data_write_count++;
         intent_write_counter++;
     }
-    write_iodepth--;
     FINISH_OP(op);
-    return 2;
-resume_10:
-    // Direct intent-write
-    // LSN is not marked as completed so big_write won't be freed
-    BS_SUBMIT_GET_SQE(sqe, data);
-    data->iov = (struct iovec){ op->buf, op->len };
-    data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
-    io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + PRIV(op)->location + op->offset);
-    if (dsk.use_atomic_flag)
-        sqe->rw_flags = RWF_ATOMIC;
-    PRIV(op)->pending_ops++;
-    PRIV(op)->op_state = 7;
-    return 1;
 }
 
 void blockstore_impl_t::handle_write_event(ring_data_t *data, blockstore_op_t *op)

src/client/cluster_client.cpp

@@ -898,7 +898,7 @@ void cluster_client_t::execute_cas(cluster_op_t *op)
                         .opcode = OSD_OP_SYNC,
                     },
                 },
-                .callback = [this, op](osd_op_t *part)
+                .callback = [op](osd_op_t *part)
                 {
                     if (part->reply.hdr.retval != 0)
                     {
@@ -1257,7 +1257,7 @@ void cluster_client_t::slice_rw(cluster_op_t *op)
         {
             op->bitmap_buf = realloc_or_die(op->bitmap_buf, bitmap_mem);
             op->part_bitmaps = (uint8_t*)op->bitmap_buf + object_bitmap_size;
-            memset(op->bitmap_buf+op->bitmap_buf_size, 0, bitmap_mem-op->bitmap_buf_size);
+            memset((uint8_t*)op->bitmap_buf+op->bitmap_buf_size, 0, bitmap_mem-op->bitmap_buf_size);
             op->bitmap_buf_size = bitmap_mem;
         }
     }

src/client/messenger.cpp

@@ -583,13 +583,7 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
 
 void osd_messenger_t::on_connect_peer(osd_num_t peer_osd, int peer_fd)
 {
-    auto wp_it = wanted_peers.find(peer_osd);
-    if (wp_it == wanted_peers.end())
-    {
-        fprintf(stderr, "on_connect_peer: no wanted peer entry for OSD %ju\n", peer_osd);
-        return;
-    }
-    auto & wp = wp_it->second;
+    auto & wp = wanted_peers.at(peer_osd);
     wp.connecting = false;
     if (peer_fd < 0)
     {

src/cmd/cli_dd.cpp

@@ -890,12 +890,12 @@ resume_2:
         clock_gettime(CLOCK_REALTIME, &tv_begin);
         tv_progress = tv_begin;
 resume_3:
-        while ((ignore_errors || !copy_error) && (!in_eof || read_buffers.size() || in_waiting > 0 || out_waiting > 0))
+        while ((ignore_errors || !copy_error) && (!in_eof || read_buffers.size() || in_waiting > 0 || out_waiting > 0 || short_writes.size()))
         {
             print_progress(false);
             while ((ignore_errors || !copy_error) &&
                 (!in_eof && in_waiting < in_iodepth && read_buffers.size() < out_iodepth ||
-                read_buffers.size() && out_waiting < out_iodepth))
+                (read_buffers.size() || short_writes.size()) && out_waiting < out_iodepth))
             {
                 if (!in_eof && in_waiting < in_iodepth && read_buffers.size() < out_iodepth)
                 {
@@ -904,7 +904,7 @@ resume_3:
                         break;
                     }
                 }
-                if (read_buffers.size() && out_waiting < out_iodepth)
+                if ((read_buffers.size() || short_writes.size()) && out_waiting < out_iodepth)
                 {
                     if (!add_write_op())
                     {

src/disk_tool/disk_tool_discard.cpp

@@ -54,7 +54,7 @@ int disk_tool_t::trim_data(std::string device)
     fprintf(stderr, "Reading metadata\n");
     data_alloc = new allocator_t(dsk.block_count);
     r = process_meta(
-        [this](blockstore_meta_header_v3_t *hdr) {},
+        [](blockstore_meta_header_v3_t *hdr) {},
         [this](blockstore_heap_t *heap, heap_entry_t *obj, uint32_t meta_block_num)
         {
             for (auto wr = obj; wr; wr = heap->prev(wr))

src/osd/osd.h

@@ -328,8 +328,8 @@ class osd_t
     void exec_show_config(osd_op_t *cur_op);
     void exec_secondary(osd_op_t *cur_op);
     void exec_secondary_real(osd_op_t *cur_op);
-    void exec_sec_read_bmp(osd_op_t *cur_op, osd_client_t *cl);
-    void exec_sec_lock(osd_op_t *cur_op, osd_client_t *cl);
+    void exec_sec_read_bmp(osd_op_t *cur_op);
+    void exec_sec_lock(osd_op_t *cur_op);
     void secondary_op_callback(osd_op_t *cur_op);
 
     // primary ops

src/osd/osd_peering.cpp

@@ -584,7 +584,7 @@ void osd_t::submit_list_subop(osd_num_t role_osd, pg_peering_state_t *ps)
                     .opcode = OSD_OP_SEC_LIST,
                 },
                 .list_pg = ps->pg_num,
-                .pg_count = pool_cfg.applied_pg_count,
+                .pg_count = (uint32_t)pool_cfg.applied_pg_count,
                 .pg_stripe_size = pool_cfg.applied_pg_stripe_size,
                 .min_inode = ((uint64_t)(ps->pool_id) << (64 - POOL_ID_BITS)),
                 .max_inode = ((uint64_t)(ps->pool_id+1) << (64 - POOL_ID_BITS)) - 1,

src/osd/osd_secondary.cpp

@@ -112,13 +112,6 @@ bool osd_t::sec_check_pg_lock(osd_num_t primary_osd, const object_id &oid, uint3
 
 void osd_t::exec_secondary_real(osd_op_t *cur_op)
 {
-    auto cl_it = msgr.clients.find(cur_op->peer_fd);
-    if (cl_it == msgr.clients.end())
-    {
-        finish_op(cur_op, -EPIPE);
-        return;
-    }
-    auto cl = cl_it->second;
     if (cur_op->req.hdr.opcode == OSD_OP_SEC_LIST &&
         (cur_op->req.sec_list.flags & OSD_LIST_PRIMARY))
     {
@@ -127,14 +120,15 @@ void osd_t::exec_secondary_real(osd_op_t *cur_op)
     }
     if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
     {
-        exec_sec_read_bmp(cur_op, cl);
+        exec_sec_read_bmp(cur_op);
         return;
     }
     else if (cur_op->req.hdr.opcode == OSD_OP_SEC_LOCK)
     {
-        exec_sec_lock(cur_op, cl);
+        exec_sec_lock(cur_op);
         return;
     }
+    auto cl = msgr.clients.at(cur_op->peer_fd);
     cur_op->bs_op = new blockstore_op_t();
     cur_op->bs_op->callback = [this, cur_op](blockstore_op_t* bs_op) { secondary_op_callback(cur_op); };
     cur_op->bs_op->opcode = (cur_op->req.hdr.opcode == OSD_OP_SEC_READ ? BS_OP_READ
@@ -253,8 +247,9 @@ void osd_t::exec_secondary_real(osd_op_t *cur_op)
 #endif
 }
 
-void osd_t::exec_sec_read_bmp(osd_op_t *cur_op, osd_client_t *cl)
+void osd_t::exec_sec_read_bmp(osd_op_t *cur_op)
 {
+    auto cl = msgr.clients.at(cur_op->peer_fd);
     int n = cur_op->req.sec_read_bmp.len / sizeof(obj_ver_id);
     if (n > 0)
     {
@@ -280,9 +275,10 @@ void osd_t::exec_sec_read_bmp(osd_op_t *cur_op, osd_client_t *cl)
 }
 
 // Lock/Unlock PG
-void osd_t::exec_sec_lock(osd_op_t *cur_op, osd_client_t *cl)
+void osd_t::exec_sec_lock(osd_op_t *cur_op)
 {
     cur_op->reply.sec_lock.cur_primary = 0;
+    auto cl = msgr.clients.at(cur_op->peer_fd);
     if (!cl->in_osd_num ||
         cur_op->req.sec_lock.flags != OSD_SEC_LOCK_PG &&
         cur_op->req.sec_lock.flags != OSD_SEC_UNLOCK_PG ||
@@ -344,13 +340,7 @@ void osd_t::exec_show_config(osd_op_t *cur_op)
         ? json11::Json::parse(std::string((char *)cur_op->buf), json_err)
         : json11::Json();
     auto peer_osd_num = req_json["osd_num"].uint64_value();
-    auto cl_it = msgr.clients.find(cur_op->peer_fd);
-    if (cl_it == msgr.clients.end())
-    {
-        finish_op(cur_op, -EPIPE);
-        return;
-    }
-    auto cl = cl_it->second;
+    auto cl = msgr.clients.at(cur_op->peer_fd);
     cl->in_osd_num = peer_osd_num;
     if (req_json["features"]["check_sequencing"].bool_value())
     {

src/test/ringloop_mock.cpp

@@ -199,7 +199,7 @@ void disk_mock_t::erase_buffers(uint64_t begin, uint64_t end)
         {
             // Cut beginning & end & stop
             uint8_t *ce = (uint8_t*)malloc_or_die(be-end);
-            memcpy(ce, it->second.iov_base + (end-bs), be-end);
+            memcpy(ce, (uint8_t*)it->second.iov_base + (end-bs), be-end);
             uint8_t *cs = (uint8_t*)realloc(it->second.iov_base, begin-bs);
             if (!cs)
                 throw std::bad_alloc();
@@ -221,7 +221,7 @@ void disk_mock_t::erase_buffers(uint64_t begin, uint64_t end)
             // Cut end & stop
             assert(be > end);
             uint8_t *ce = (uint8_t*)malloc_or_die(be-end);
-            memcpy(ce, it->second.iov_base + (end-bs), be-end);
+            memcpy(ce, (uint8_t*)it->second.iov_base + (end-bs), be-end);
             buffers[be] = (iovec){ .iov_base = ce, .iov_len = be-end };
             buffers.erase(it);
             break;
@@ -308,7 +308,7 @@ void disk_mock_t::read_item(uint8_t *to, uint64_t offset, uint64_t len)
             last = offset;
         }
         uint64_t cur_end = be < offset+len ? be : offset+len;
-        memcpy(to+last-offset, it->second.iov_base+last-bs, cur_end-last);
+        memcpy(to+last-offset, (uint8_t*)it->second.iov_base+last-bs, cur_end-last);
         last = be;
     }
     if (last < offset+len)

src/test/test_heap.cpp

@@ -436,12 +436,13 @@ void test_compact(bool csum, bool stable)
 
     bitmap_set(ref_int_bitmap, 8192, 4096, 4096);
     {
-        uint32_t csums[dsk.data_block_size/(dsk.csum_block_size ? dsk.csum_block_size : 4096)] = {};
+        size_t csum_count = dsk.data_block_size/(dsk.csum_block_size ? dsk.csum_block_size : 4096);
+        std::vector<uint32_t> csums(csum_count);
         csums[0] = crc32c(0, buffer_area.data(), 4096);
         csums[2] = crc32c(0, buffer_area.data()+8192, 4096);
         res = heap.add_compact(obj, compact_info.compact_version, compact_info.compact_lsn,
             compact_info.clean_wr->big_location(&heap), compact_info.do_delete,
-            &mblock, ref_int_bitmap, ref_int_bitmap, (uint8_t*)csums);
+            &mblock, ref_int_bitmap, ref_int_bitmap, (uint8_t*)csums.data());
         assert(res == 0);
     }
     assert(mblock == 0);
@@ -461,10 +462,11 @@ void test_compact(bool csum, bool stable)
     if (csum)
     {
         assert(heap.calc_checksums(obj, buffer_area.data(), false));
-        uint32_t csums[dsk.data_block_size/(dsk.csum_block_size ? dsk.csum_block_size : 4096)] = {};
+        size_t csum_count = dsk.data_block_size/(dsk.csum_block_size ? dsk.csum_block_size : 4096);
+        std::vector<uint32_t> csums(csum_count);
         csums[0] = crc32c(0, buffer_area.data(), 4096);
         csums[2] = crc32c(0, buffer_area.data()+8192, 4096);
-        assert(!memcmp(obj->get_checksums(&heap), csums, dsk.data_block_size/dsk.csum_block_size*4));
+        assert(!memcmp(obj->get_checksums(&heap), csums.data(), dsk.data_block_size/dsk.csum_block_size*4));
     }
 
     obj = heap.read_entry({ .inode = INODE_WITH_POOL(1, 2), .stripe = 0 });

tests/test_small_block_size.sh

@@ -0,0 +1,47 @@
+#!/bin/bash -ex
+
+# Test for Issue #112: Integer division bug causes crash with block_size < 32KB
+# This test verifies that small block sizes (4KB, 8KB, 16KB) work correctly
+# with 4KB bitmap_granularity, both for aligned and unaligned I/O
+
+# Arrange: Set up test environment with small block sizes
+export SCHEME=replicated
+export OSD_COUNT=3
+export PG_COUNT=1
+export PG_SIZE=2
+export OSD_SIZE=256
+
+# Test with 16KB block_size and 4KB bitmap_granularity
+# This should trigger the bug: 16384 / 4096 / 8 = 4 / 8 = 0 bytes
+export OFFSET_ARGS="--data_block_size 16384 --bitmap_granularity 4096"
+
+. `dirname $0`/run_3osds.sh
+
+# Act: Run I/O tests that exercise the bitmap code paths
+
+echo "Test 1: 128KB aligned I/O (may work even with bug)"
+LD_PRELOAD="build/src/client/libfio_vitastor.so" \
+    fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so \
+        -bs=128k -direct=1 -iodepth=4 -rw=randwrite \
+        -etcd=$ETCD_URL -pool=1 -inode=1 -size=64M -runtime=5
+
+echo "Test 2: 4KB random I/O (will crash with bug)"
+LD_PRELOAD="build/src/client/libfio_vitastor.so" \
+    fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so \
+        -bs=4k -direct=1 -iodepth=16 -rw=randwrite \
+        -etcd=$ETCD_URL -pool=1 -inode=1 -size=64M -runtime=5
+
+echo "Test 3: Mixed read/write with 4KB I/O"
+LD_PRELOAD="build/src/client/libfio_vitastor.so" \
+    fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so \
+        -bs=4k -direct=1 -iodepth=16 -rw=randrw -rwmixread=50 \
+        -etcd=$ETCD_URL -pool=1 -inode=1 -size=64M -runtime=5
+
+echo "Test 4: Sequential 4KB writes"
+LD_PRELOAD="build/src/client/libfio_vitastor.so" \
+    fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so \
+        -bs=4k -direct=1 -iodepth=1 -rw=write \
+        -etcd=$ETCD_URL -pool=1 -inode=1 -size=64M
+
+# Assert: If we reach here without crash, test passed
+format_green "OK: 16KB block_size with 4KB bitmap_granularity works correctly"