Support canceling blockstore v1 operations (v2 doesn't need it)

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

.gitea/workflows/buildenv.Dockerfile

@@ -23,7 +23,7 @@ RUN apt-get update
 RUN apt-get -y install etcd qemu-system-x86 qemu-block-extra qemu-utils fio libasan8 \
     libgoogle-perftools-dev devscripts libjerasure-dev cmake libibverbs-dev libisal-dev
 RUN apt-get -y build-dep fio qemu=`dpkg -s qemu-system-x86|grep ^Version:|awk '{print $2}'`
-RUN apt-get update && apt-get -y install jq lp-solve sudo nfs-common fdisk parted libc-ares-dev udev
+RUN apt-get update && apt-get -y install jq lp-solve sudo nfs-common fdisk parted libc-ares-dev udev iproute2
 RUN apt-get --download-only source fio qemu=`dpkg -s qemu-system-x86|grep ^Version:|awk '{print $2}'`
 
 RUN set -ex; \

.gitea/workflows/test.yml

@@ -162,6 +162,24 @@ jobs:
           echo ""
         done
 
+  test_change_pg_minsize:
+    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_change_pg_minsize.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_change_pg_size:
     runs-on: ubuntu-latest
     needs: build
@@ -288,6 +306,24 @@ jobs:
           echo ""
         done
 
+  test_img_online_resize:
+    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_img_online_resize.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/nodeserver.go

@@ -754,7 +754,7 @@ func (ns *NodeServer) mountNFS(ctxVars map[string]string) (string, error)
     // Mount NFS
     _, _, err = system(
         "mount", "-t", "nfs", "127.0.0.1:/", state.Path,
-        "-o", fmt.Sprintf("port=%d,mountport=%d,nfsvers=3,soft,nolock,tcp", port, port),
+        "-o", fmt.Sprintf("port=%d,mountport=%d,nfsvers=3,soft,sync,nolock,tcp", port, port),
     )
     if (err != nil)
     {

docs/usage/nfs.en.md

@@ -54,7 +54,7 @@ vitastor-nfs start --block --etcd_address 192.168.5.10:2379 --portmap 0 --port 2
 To mount the FS exported by this server, run:
 
 ```
-mount server:/ /mnt/ -o port=2050,mountport=2050,nfsvers=3,soft,nolock,tcp
+mount server:/ /mnt/ -o port=2050,mountport=2050,nfsvers=3,soft,sync,nolock,tcp
 ```
 
 ## VitastorFS
@@ -155,11 +155,14 @@ By default, Linux uses rather small readahead value of 128 KB for NFS mounts.
 For simple reading/copuing of files it's way too low because it stops VitastorFS from
 using parallel reads.
 
-However, you can increase readahead - either with `vitastor-nfs mount --readahead <SIZE>`
-option (for example, `--readahead 2M`), or with the following command (here, `2048` is the
-desired readahead in KB, and `/mnt/nfs` is your FS mountpoint):
+However, you can increase readahead. You have 3 options:
 
-`echo 2048 > /sys/class/bdi/$(findmnt -no MAJ:MIN /mnt/nfs 2>/dev/null | tr -d ' ')/read_ahead_kb`
+1. Use the built-in `vitastor-nfs mount --readahead <SIZE>` option (for example, `--readahead 2M`).
+2. Edit `[nfsrahead]` section in `/etc/nfs.conf` - add `nfs=2048` in it (global setting for all NFS 3.0 mounts).
+3. Manually with the following command: \
+   `echo 2048 > /sys/class/bdi/$(findmnt -no MAJ:MIN /mnt/nfs 2>/dev/null | tr -d ' ')/read_ahead_kb`
+
+Here, `2048` is the desired readahead in KB, and `/mnt/nfs` is your FS mountpoint.
 
 You can easily see 2x-3x linear read improvement after you increase the parameter, so we
 really urge you to try increasing it!

docs/usage/nfs.ru.md

@@ -54,7 +54,7 @@ vitastor-nfs start --block --etcd_address 192.168.5.10:2379 --portmap 0 --port 2
 Примонтировать ФС, запущенную с такими опциями, можно следующей командой:
 
 ```
-mount server:/ /mnt/ -o port=2050,mountport=2050,nfsvers=3,soft,nolock,tcp
+mount server:/ /mnt/ -o port=2050,mountport=2050,nfsvers=3,soft,sync,nolock,tcp
 ```
 
 ## VitastorFS
@@ -159,13 +159,17 @@ NFS в Linux имеет некоторые важные для Vitastor особ
 Для простого чтения/копирования файлов его не достаточно, потому что оно мешает VitastorFS
 нормально утилизировать параллелизм чтения.
 
-Однако, вы можете увеличить его - либо с помощью опции `vitastor-nfs mount --readahead <SIZE>`
-(например, `--readahead 2M`), или следующей командой (здесь `2048` - размер readahead, а `/mnt/nfs` -
-путь к вашей ФС):
+Однако, вы можете увеличить эту настройку целыми 3 способами:
 
-`echo 2048 > /sys/class/bdi/$(findmnt -no MAJ:MIN /mnt/nfs 2>/dev/null | tr -d ' ')/read_ahead_kb`
+1. Встроенной опцией `vitastor-nfs mount --readahead <SIZE>` (например, `--readahead 2M`).
+2. Добавив в секцию `[nfsrahead]` файла конфигурации `/etc/nfs.conf` строку `nfs=2048` (глобальная
+   настройка для всех NFS 3.0 точек монтирования).
+3. Вручную следующей командой: \
+   `echo 2048 > /sys/class/bdi/$(findmnt -no MAJ:MIN /mnt/nfs 2>/dev/null | tr -d ' ')/read_ahead_kb`
 
-После увеличения параметра вы легко можете заметить улучшение скорости чтения в 2-3 раза, так что
+Здесь `2048` - размер readahead, а `/mnt/nfs` - путь к вашей ФС.
+
+После увеличения параметра скорость линейного чтения легко может увеличиться в 2-3 раза, так что
 советуем обязательно попробовать!
 
 ### Размер записи Linux NFS

src/blockstore/CMakeLists.txt

@@ -12,6 +12,7 @@ add_library(vitastor_blk STATIC
 target_compile_options(vitastor_blk PUBLIC -fPIC)
 target_link_libraries(vitastor_blk
 	util
+	crc32c
 )
 set_target_properties(vitastor_blk PROPERTIES VERSION ${VITASTOR_VERSION} SOVERSION 0)
 

src/blockstore/blockstore.h

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

src/blockstore/blockstore_heap.cpp

@@ -633,8 +633,9 @@ int blockstore_heap_t::mark_used_blocks()
                     li = NULL;
                 }
                 bool overwritten = false;
-                for (; li; li = li->prev, wr = &li->entry)
+                for (; li; li = li->prev)
                 {
+                    auto wr = &li->entry;
                     if (overwritten)
                     {
                         wr->set_garbage();

src/blockstore/fio_engine.cpp

@@ -262,7 +262,7 @@ static enum fio_q_status _bs_queue(struct thread_data *td, struct io_u *io, bool
             bsd->inflight--;
             bsd->completed.push_back(io);
             if (bsd->trace)
-                printf("--- OP_READ %zx n=%d retval=%d\n", (size_t)op, n, op->retval);
+                printf("--- OP_READ %zx n=%d retval=%jd\n", (size_t)op, n, op->retval);
             delete op;
         };
         break;
@@ -283,7 +283,7 @@ static enum fio_q_status _bs_queue(struct thread_data *td, struct io_u *io, bool
             {
                 bs_data *bsd = ((bs_data*)((thread_data*)io->engine_data)->io_ops_data);
                 if (bsd->trace)
-                    printf("--- OP_WRITE %zx n=%d retval=%d\n", (size_t)op, n, op->retval);
+                    printf("--- OP_WRITE %zx n=%d retval=%jd\n", (size_t)op, n, op->retval);
                 if (op->retval < 0)
                 {
                     io->error = op->retval < 0 ? -op->retval : 0;
@@ -305,7 +305,7 @@ static enum fio_q_status _bs_queue(struct thread_data *td, struct io_u *io, bool
                     {
                         bs_data *bsd = ((bs_data*)((thread_data*)io->engine_data)->io_ops_data);
                         if (bsd->trace)
-                            printf("--- OP_STABLE %zx n=%d retval=%d\n", (size_t)op, n, op->retval);
+                            printf("--- OP_STABLE %zx n=%d retval=%jd\n", (size_t)op, n, op->retval);
                         io->error = op->retval < 0 ? -op->retval : 0;
                         bsd->inflight--;
                         bsd->completed.push_back(io);
@@ -323,7 +323,7 @@ static enum fio_q_status _bs_queue(struct thread_data *td, struct io_u *io, bool
             {
                 bs_data *bsd = ((bs_data*)((thread_data*)io->engine_data)->io_ops_data);
                 if (bsd->trace)
-                    printf("--- OP_WRITE_STABLE %zx n=%d retval=%d\n", (size_t)op, n, op->retval);
+                    printf("--- OP_WRITE_STABLE %zx n=%d retval=%jd\n", (size_t)op, n, op->retval);
                 io->error = op->retval < 0 ? -op->retval : 0;
                 bsd->inflight--;
                 bsd->completed.push_back(io);
@@ -342,7 +342,7 @@ static enum fio_q_status _bs_queue(struct thread_data *td, struct io_u *io, bool
             bsd->completed.push_back(io);
             bsd->inflight--;
             if (bsd->trace)
-                printf("--- OP_SYNC %zx n=%d retval=%d\n", (size_t)op, n, op->retval);
+                printf("--- OP_SYNC %zx n=%d retval=%jd\n", (size_t)op, n, op->retval);
             delete op;
         };
         bsd->last_sync = true;

src/blockstore/v1/read.cpp

@@ -173,7 +173,8 @@ int blockstore_impl_t::fill_partial_checksum_blocks(std::vector<copy_buffer_t> &
     if (read_end == read_offset)
         return 0;
     int required = 0;
-    read_buf -= read_offset;
+    if (read_buf)
+        read_buf -= read_offset;
     uint32_t last_block = (read_end-1)/dsk.csum_block_size;
     uint32_t start_block = read_offset/dsk.csum_block_size;
     uint32_t item_start_block = item_start/dsk.csum_block_size;
@@ -189,20 +190,20 @@ int blockstore_impl_t::fill_partial_checksum_blocks(std::vector<copy_buffer_t> &
         };
         rv.insert(rv.begin() + pos, el);
         if (read_buf)
-            memset(read_buf + el.offset - read_offset, 0, el.len);
+            memset(read_buf + el.offset, 0, el.len);
         fulfilled += el.len;
         return 1;
     };
     if (read_offset < item_start)
     {
         // Zero-fill the beginning
-        find_holes(rv, read_offset, item_start, zero_range);
+        find_holes(rv, read_offset, item_start < read_end ? item_start : read_end, zero_range);
         read_offset = item_start;
     }
     if (read_end > item_end)
     {
         // Zero-fill the end
-        find_holes(rv, item_end, read_end, zero_range);
+        find_holes(rv, read_offset < item_end ? item_end : read_offset, read_end, zero_range);
         read_end = item_end;
     }
     while (start_block <= last_block)
@@ -278,7 +279,7 @@ bool blockstore_impl_t::read_range_fulfilled(std::vector<copy_buffer_t> & rv, ui
                     .offset = bmp_start*dsk.bitmap_granularity,
                     .len = (bmp_pos-bmp_start)*dsk.bitmap_granularity,
                 };
-                rv.insert(rv.begin() + pos, el);
+                rv.insert(rv.begin() + pos + diff, el);
                 if (read_buf)
                     memset(read_buf + el.offset, 0, el.len);
                 fulfilled += el.len;
@@ -419,6 +420,12 @@ 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){

src/blockstore/v1/rollback.cpp

@@ -12,6 +12,12 @@ 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,6 +291,12 @@ 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);

src/blockstore/v1/write.cpp

@@ -214,6 +214,11 @@ 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)
@@ -236,9 +241,7 @@ 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))
         {
-            // Mark operations to cancel them
-            PRIV(other_op)->real_version = UINT64_MAX;
-            other_op->retval = retval;
+            other_op->canceled = true;
         }
     }
     op->retval = retval;
@@ -257,6 +260,11 @@ 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
@@ -264,12 +272,6 @@ 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);
@@ -550,6 +552,16 @@ 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,
@@ -767,6 +779,11 @@ 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/cluster_client.cpp

@@ -883,7 +883,13 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
         execute_internal(sync_op);
         return;
     }
-    slice_rw(op);
+    if (!slice_rw(op))
+    {
+        op->retval = -EINVAL;
+        auto cb = std::move(op->callback);
+        cb(op);
+        return;
+    }
     op->needs_reslice = false;
     int res = try_send(op, 0, [this, op](osd_op_t *part)
     {
@@ -1086,7 +1092,12 @@ int cluster_client_t::continue_rw(cluster_op_t *op)
         goto resume_2;
 resume_0:
     // Slice the operation into parts
-    slice_rw(op);
+    if (!slice_rw(op))
+    {
+        op->retval = -EINVAL;
+        erase_op(op);
+        return 1;
+    }
     op->needs_reslice = false;
 resume_1:
     // Send unsent parts, if they're not subject to change
@@ -1220,8 +1231,16 @@ resume_2:
         op->retval = op->len;
         if (op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP)
         {
-            auto & pool_cfg = st_cli->pool_config.at(INODE_POOL(op->inode));
-            op->retval = op->len / pool_cfg.bitmap_granularity;
+            auto pool_it = st_cli->pool_config.find(INODE_POOL(op->inode));
+            if (pool_it == st_cli->pool_config.end())
+            {
+                op->retval = -EINVAL;
+            }
+            else
+            {
+                auto & pool_cfg = pool_it->second;
+                op->retval = op->len / pool_cfg.bitmap_granularity;
+            }
         }
         if (op->flush_id)
         {
@@ -1296,11 +1315,17 @@ static void add_iov(int size, int skip, cluster_op_t *op, int &iov_idx, size_t &
     }
 }
 
-void cluster_client_t::slice_rw(cluster_op_t *op)
+bool cluster_client_t::slice_rw(cluster_op_t *op)
 {
     // Slice the request into individual object stripe requests
     // Primary OSDs still operate individual stripes, but their size is multiplied by PG minsize in case of EC
-    auto & pool_cfg = st_cli->pool_config.at(INODE_POOL(op->cur_inode));
+    auto pool_it = st_cli->pool_config.find(INODE_POOL(op->cur_inode));
+    if (pool_it == st_cli->pool_config.end() ||
+        !pool_it->second.real_pg_count)
+    {
+        return false;
+    }
+    auto & pool_cfg = pool_it->second;
     uint32_t pg_data_size = (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pool_cfg.pg_size-pool_cfg.parity_chunks);
     uint64_t pg_block_size = pool_cfg.data_block_size * pg_data_size;
     uint64_t first_stripe = (op->offset / pg_block_size) * pg_block_size;
@@ -1395,6 +1420,7 @@ void cluster_client_t::slice_rw(cluster_op_t *op)
         op->parts[i].osd_num = 0;
         i++;
     }
+    return true;
 }
 
 bool cluster_client_t::affects_pg(uint64_t inode, uint64_t offset, uint64_t len, pool_id_t pool_id, pg_num_t pg_num)
@@ -1403,7 +1429,12 @@ bool cluster_client_t::affects_pg(uint64_t inode, uint64_t offset, uint64_t len,
     {
         return false;
     }
-    auto & pool_cfg = st_cli->pool_config.at(INODE_POOL(inode));
+    auto pool_it = st_cli->pool_config.find(INODE_POOL(inode));
+    if (pool_it == st_cli->pool_config.end())
+    {
+        return false;
+    }
+    auto & pool_cfg = pool_it->second;
     uint32_t pg_data_size = (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pool_cfg.pg_size-pool_cfg.parity_chunks);
     uint64_t pg_block_size = pool_cfg.data_block_size * pg_data_size;
     uint64_t first_stripe = (offset / pg_block_size) * pg_block_size;
@@ -1422,7 +1453,12 @@ bool cluster_client_t::affects_pg(uint64_t inode, uint64_t offset, uint64_t len,
 
 bool cluster_client_t::affects_osd(uint64_t inode, uint64_t offset, uint64_t len, osd_num_t osd)
 {
-    auto & pool_cfg = st_cli->pool_config.at(INODE_POOL(inode));
+    auto pool_it = st_cli->pool_config.find(INODE_POOL(inode));
+    if (pool_it == st_cli->pool_config.end())
+    {
+        return false;
+    }
+    auto & pool_cfg = pool_it->second;
     uint32_t pg_data_size = (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pool_cfg.pg_size-pool_cfg.parity_chunks);
     uint64_t pg_block_size = pool_cfg.data_block_size * pg_data_size;
     uint64_t first_stripe = (offset / pg_block_size) * pg_block_size;
@@ -1446,7 +1482,12 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
         init_msgr();
     }
     auto part = &op->parts[i];
-    auto & pool_cfg = st_cli->pool_config.at(INODE_POOL(op->cur_inode));
+    auto pool_it = st_cli->pool_config.find(INODE_POOL(op->cur_inode));
+    if (pool_it == st_cli->pool_config.end())
+    {
+        return TRY_SEND_OFFLINE;
+    }
+    auto & pool_cfg = pool_it->second;
     auto pg_it = pool_cfg.pg_config.find(part->pg_num);
     if (pg_it != pool_cfg.pg_config.end() &&
         !pg_it->second.pause && pg_it->second.cur_primary &&
@@ -1708,7 +1749,12 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
 void cluster_client_t::copy_part_bitmap(cluster_op_t *op, cluster_op_part_t *part)
 {
     // Copy (OR) bitmap
-    auto & pool_cfg = st_cli->pool_config.at(INODE_POOL(op->cur_inode));
+    auto pool_it = st_cli->pool_config.find(INODE_POOL(op->cur_inode));
+    if (pool_it == st_cli->pool_config.end())
+    {
+        return;
+    }
+    auto & pool_cfg = pool_it->second;
     uint32_t pg_block_size = pool_cfg.data_block_size * (
         pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pool_cfg.pg_size-pool_cfg.parity_chunks
     );

src/client/cluster_client.h

@@ -226,7 +226,7 @@ protected:
     void unshift_op(cluster_op_t *op);
     int continue_rw(cluster_op_t *op);
     bool check_rw(cluster_op_t *op);
-    void slice_rw(cluster_op_t *op);
+    bool slice_rw(cluster_op_t *op);
     void reset_retry_timer(int new_duration);
     int try_send(cluster_op_t *op, int i, std::function<void(osd_op_t *op_part)> cb = nullptr);
     int continue_sync(cluster_op_t *op);

src/client/cluster_client_list.cpp

@@ -63,14 +63,15 @@ void cluster_client_t::list_inode(inode_t inode, uint64_t min_offset, uint64_t m
 {
     init_msgr();
     pool_id_t pool_id = INODE_POOL(inode);
-    if (!pool_id || st_cli->pool_config.find(pool_id) == st_cli->pool_config.end())
+    auto pool_it = st_cli->pool_config.find(pool_id);
+    if (!pool_id || pool_it == st_cli->pool_config.end())
     {
         if (log_level > 0)
             fprintf(stderr, "Pool %u does not exist\n", pool_id);
         pg_callback(-EINVAL, 0, 0, std::set<object_id>());
         return;
     }
-    auto pg_stripe_size = st_cli->pool_config.at(pool_id).pg_stripe_size;
+    auto pg_stripe_size = pool_it->second.pg_stripe_size;
     if (min_offset)
         min_offset = (min_offset/pg_stripe_size) * pg_stripe_size;
     inode_list_t *lst = new inode_list_t();
@@ -248,7 +249,12 @@ void cluster_client_t::set_list_retry_timeout(int ms, timespec new_time)
 
 int cluster_client_t::start_pg_listing(inode_list_pg_t *pg)
 {
-    auto & pool_cfg = st_cli->pool_config.at(pg->lst->pool_id);
+    auto pool_it = st_cli->pool_config.find(pg->lst->pool_id);
+    if (pool_it == st_cli->pool_config.end())
+    {
+        return LIST_PG_WAIT_ACTIVE;
+    }
+    auto & pool_cfg = pool_it->second;
     auto pg_it = pool_cfg.pg_config.find(pg->pg_num);
     assert(pg->lst->real_pg_count == pool_cfg.real_pg_count);
     if (pg_it == pool_cfg.pg_config.end() ||

src/client/etcd_state_client.cpp

@@ -959,6 +959,8 @@ void etcd_state_client_t::insert_inode_config(const inode_config_t & cfg)
             if (w->name == cfg.name)
             {
                 w->cfg = cfg;
+                if (w->callback)
+                    w->callback(w);
             }
         }
     }

src/client/etcd_state_client.h

@@ -101,6 +101,7 @@ struct inode_watch_t
 {
     std::string name;
     inode_config_t cfg = {};
+    std::function<void(inode_watch_t*)> callback;
 };
 
 struct http_url_t

src/client/messenger.h

@@ -206,6 +206,7 @@ protected:
     friend class copy_op_writer_t;
     friend class gcm_op_writer_t;
     friend class get_op_writer_t;
+    friend class osd_client_t;
 
     int keepalive_timer_id = -1;
 
@@ -300,6 +301,7 @@ 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;
@@ -340,9 +342,11 @@ 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);
+    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);
     bool op_write_buf(osd_client_t *cl, uint8_t *src, size_t src_len, uint8_t *dst, size_t dst_len, bool skip_csum, size_t & from, size_t & done);

src/client/msgr_op.h

@@ -188,6 +188,7 @@ 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

@@ -737,10 +737,7 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context)
                 assert(rc->send_done_pos < rc->send_out_size);
                 while (osd_op_t *op = cl->send_free_ops.front())
                 {
-                    if (!((size_t)op & 7))
-                        delete op;
-                    else
-                        free((void*)((size_t)op & ~(size_t)7));
+                    post_send_free(op);
                     cl->send_free_ops.pop_front();
                 }
                 cl->send_free_ops.pop_front();

src/client/msgr_receive.cpp

@@ -522,6 +522,8 @@ out_wakeup:
             }
             if (i < cl->recv_list.size())
             {
+                if (result > 0 && cl->read_csum_state && !(cl->recv_flags[i] & RDR_NO_CSUM))
+                    XXH3_64bits_update(cl->read_csum_state, cl->recv_list[i].iov_base, result);
                 cl->recv_list[i].iov_base += result;
                 cl->recv_list[i].iov_len -= result;
             }
@@ -531,7 +533,7 @@ out_wakeup:
             }
             cl->recv_list.erase(cl->recv_list.begin(), cl->recv_list.begin()+i);
             cl->recv_flags.erase(cl->recv_flags.begin(), cl->recv_flags.begin()+i);
-            if (!handle_finished_op(cl))
+            if (!cl->recv_list.size() && !handle_finished_op(cl))
             {
                 goto out_wakeup;
             }
@@ -565,11 +567,15 @@ void osd_messenger_t::handle_immediate_ops()
             else
                 delete op;
         }
-        else
+        else if (op->callback)
         {
             // Copy lambda to be unaffected by `delete op`
             std::function<void(osd_op_t*)>(op->callback)(op);
         }
+        else
+        {
+            delete op;
+        }
     }
 }
 

src/client/msgr_send.cpp

@@ -486,6 +486,107 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     }
 }
 
+// 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)
+{
+    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)
+    {
+        // Client is stopped => messenger will call the callback anyway
+        return false;
+    }
+    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())
+    {
+        // Reply header is already received => callback is pending
+        return false;
+    }
+    for (auto op: set_immediate_ops)
+    {
+        if (op == cur_op)
+        {
+            // Reply is already received, callback is pending
+            return false;
+        }
+    }
+    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.
+    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)
 {
     if (cl->peer_state == PEER_STOPPED || cl->peer_fd < 0)
@@ -632,9 +733,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 && op->op_type == OSD_OP_IN)
+        if (!cl->write_op)
         {
-            // this is a reply, free the op after sending it
+            // remember the operation for the current send-list
             cl->send_free_ops.push_back(op);
         }
     }
@@ -704,12 +805,7 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
             // Second notification - only free a batch of postponed ops
             int i = 0;
             for (; i < cl->zc_free_list.size() && cl->zc_free_list[i]; i++)
-            {
-                if (!((size_t)cl->zc_free_list[i] & 7))
-                    delete cl->zc_free_list[i];
-                else
-                    free((void*)((size_t)cl->zc_free_list[i] & ~(size_t)7));
-            }
+                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;
@@ -725,10 +821,8 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         {
             if (more)
                 cl->zc_free_list.push_back(op);
-            else if (!((size_t)op & 7))
-                delete op;
             else
-                free((void*)((size_t)op & ~(size_t)7));
+                post_send_free(op);
         }
         if (more)
             cl->zc_free_list.push_back(NULL); // end marker

src/client/msgr_stop.cpp

@@ -43,6 +43,24 @@ void osd_op_t::cancel()
     }
 }
 
+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();
+    }
+    else
+        free((void*)((size_t)op & ~(size_t)7));
+}
+
 // force_delete means stop the client anyway, even if there are refs to it in the event loop.
 // the flag should be used in the destructor.
 // why? - because yes, we could close the FD first and let it fail all requests in the event loop,
@@ -200,23 +218,11 @@ osd_client_t::~osd_client_t()
     cancel_ops();
     for (osd_op_t *op: send_free_ops)
     {
-        if (op)
-        {
-            if (!((size_t)op & 7))
-                delete op;
-            else
-                free((void*)((size_t)op & ~(size_t)7));
-        }
+        osd_messenger_t::post_send_free(op);
     }
     for (osd_op_t *op: zc_free_list)
     {
-        if (op)
-        {
-            if (!((size_t)op & 7))
-                delete op;
-            else
-                free((void*)((size_t)op & ~(size_t)7));
-        }
+        osd_messenger_t::post_send_free(op);
     }
     if (read_csum_state)
     {

src/client/nbd_proxy.cpp

@@ -2,9 +2,12 @@
 // License: VNPL-1.1 (see README.md for details)
 // Similar to qemu-nbd, but sets timeout and uses io_uring
 
-#include <cerrno>
-#include <cstdint>
-#include <cstdio>
+#include <thread>
+#include <mutex>
+
+#include <errno.h>
+#include <stdint.h>
+#include <stdio.h>
 #include <linux/genetlink.h>
 #include <linux/nbd.h>
 #include <linux/netlink.h>
@@ -328,6 +331,9 @@ protected:
     int nbd_max_devices = 64;
     int nbd_max_part = 3;
     inode_watch_t *watch = NULL;
+    std::mutex new_size_mu;
+    bool changing_size = false;
+    std::unique_ptr<std::thread> change_thread;
 
     ring_loop_t *ringloop = NULL;
     epoll_manager_t *epmgr = NULL;
@@ -336,6 +342,7 @@ protected:
 
     std::vector<iovec> send_list, next_send_list;
     std::vector<void*> to_free;
+    int dev_num = -1;
     int nbd_fd = -1;
     void *recv_buf = NULL;
     int receive_buffer_size = 9000;
@@ -448,7 +455,7 @@ help:
     {
         char path[64] = { 0 };
         // Check if mapped
-        sprintf(path, "/sys/block/nbd%d/pid", dev_num);
+        snprintf(path, sizeof(path), "/sys/block/nbd%d/pid", dev_num);
         if (access(path, F_OK) != 0)
         {
             fprintf(stderr, "/dev/nbd%d is not mapped: /sys/block/nbd%d/pid does not exist\n", dev_num, dev_num);
@@ -456,7 +463,7 @@ help:
                 exit(1);
         }
         // Run unmap
-        sprintf(path, "/dev/nbd%d", dev_num);
+        snprintf(path, sizeof(path), "/dev/nbd%d", dev_num);
         int r, nbd = open(path, O_RDWR);
         if (nbd < 0)
         {
@@ -533,6 +540,7 @@ help:
                 fprintf(stderr, "Image %s does not exist\n", image_name.c_str());
                 exit(1);
             }
+            watch->callback = [this](inode_watch_t*) { on_change_watch(); };
         }
 
         // cli->config contains merged config
@@ -574,10 +582,9 @@ help:
         if (netlink)
         {
 #ifdef HAVE_NBD_NETLINK_H
-            int devnum = -1;
             if (!cfg["dev_num"].is_null())
             {
-                devnum = (int)cfg["dev_num"].uint64_value();
+                dev_num = (int)cfg["dev_num"].uint64_value();
             }
             uint64_t flags = NBD_FLAG_SEND_FLUSH;
             uint64_t cflags = 0;
@@ -595,15 +602,16 @@ help:
             {
                 daemonize_fork();
             }
-            int err = netlink_configure(sockfd + 1, 1, devnum, device_size, 4096, flags, cflags, nbd_timeout, nbd_conn_timeout, NULL, revive);
+            int err = netlink_configure(sockfd + 1, 1, dev_num, device_size, 4096, flags, cflags, nbd_timeout, nbd_conn_timeout, NULL, revive);
             if (err < 0)
             {
                 errno = (err == -NLE_BUSY ? EBUSY : EIO);
                 fprintf(stderr, "netlink_configure failed: %s (code %d)\n", nl_geterror(err), err);
                 exit(1);
             }
+            dev_num = err;
             close(sockfd[1]);
-            printf("/dev/nbd%d\n", err);
+            printf("/dev/nbd%d\n", dev_num);
             if (bg)
             {
                 daemonize_reopen_stdio();
@@ -621,6 +629,7 @@ help:
                 uint64_t flags = NBD_FLAG_SEND_FLUSH;
                 if (!cfg["readonly"].is_null())
                     flags |= NBD_FLAG_READ_ONLY;
+                dev_num = cfg["dev_num"].int64_value();
                 if ((r = run_nbd(sockfd, cfg["dev_num"].int64_value(), device_size, flags, nbd_timeout, bg)) != 0)
                 {
                     fprintf(stderr, "run_nbd: %s\n", strerror(-r));
@@ -643,6 +652,7 @@ help:
                     if (r == 0)
                     {
                         printf("/dev/nbd%d\n", i);
+                        dev_num = i;
                         break;
                     }
                     else if (r == -ENOENT)
@@ -698,6 +708,11 @@ help:
             ringloop->wait();
         }
         cli->flush();
+        if (change_thread)
+        {
+            change_thread->join();
+            change_thread.reset();
+        }
         delete cli;
         delete epmgr;
         delete ringloop;
@@ -706,6 +721,62 @@ help:
         ringloop = NULL;
     }
 
+    void on_change_watch()
+    {
+        new_size_mu.lock();
+        if (device_size != watch->cfg.size)
+        {
+            bool chg = changing_size;
+            device_size = watch->cfg.size;
+            changing_size = true;
+            new_size_mu.unlock();
+            if (!chg)
+            {
+                // NBD quiesces the device queue before changing device size, so we have
+                // to do ioctl in a separate thread to avoid blocking the main I/O handler
+                if (change_thread)
+                    change_thread->join();
+                change_thread = std::make_unique<std::thread>(&nbd_proxy::change_size, this);
+            }
+        }
+        else
+            new_size_mu.unlock();
+    }
+
+    void change_size()
+    {
+        uint64_t new_size = 0;
+        char path[64] = { 0 };
+        snprintf(path, sizeof(path), "/dev/nbd%d", dev_num);
+        int r, nbd = open(path, O_RDWR);
+        if (nbd < 0)
+        {
+            fprintf(stderr, "Failed to update device size: %s (code %u)\n", strerror(errno), errno);
+            return;
+        }
+        new_size_mu.lock();
+        new_size = device_size;
+        new_size_mu.unlock();
+        bool done = false;
+        while (!done)
+        {
+            r = ioctl(nbd, NBD_SET_SIZE, new_size);
+            if (r < 0)
+            {
+                fprintf(stderr, "Failed to update device size: %s (code %u)\n", strerror(errno), errno);
+            }
+            new_size_mu.lock();
+            if (new_size == device_size)
+            {
+                done = true;
+                changing_size = false;
+            }
+            new_size = device_size;
+            new_size_mu.unlock();
+        }
+        close(nbd);
+    }
+
     void load_module()
     {
         if (access("/sys/module/nbd", F_OK) == 0)
@@ -769,10 +840,10 @@ help:
         while (true)
         {
             dev_num++;
-            sprintf(path, "/sys/block/nbd%d", dev_num);
+            snprintf(path, sizeof(path), "/sys/block/nbd%d", dev_num);
             if (access(path, F_OK) != 0)
                 break;
-            sprintf(path, "/sys/block/nbd%d/pid", dev_num);
+            snprintf(path, sizeof(path), "/sys/block/nbd%d/pid", dev_num);
             std::string pid_str = read_file(path);
             if (pid_str == "")
                 continue;
@@ -781,7 +852,7 @@ help:
                 printf("Failed to read pid from /sys/block/nbd%d/pid\n", dev_num);
                 continue;
             }
-            sprintf(path, "/proc/%d/cmdline", pid);
+            snprintf(path, sizeof(path), "/proc/%d/cmdline", pid);
             std::string cmdline = read_file(path);
             std::vector<const char*> argv;
             int last = 0;
@@ -896,7 +967,7 @@ protected:
                 daemonize_fork();
             }
             close(notifyfd[0]);
-            sprintf(path, "/dev/nbd%d", dev_num);
+            snprintf(path, sizeof(path), "/dev/nbd%d", dev_num);
             int r, nbd = open(path, O_RDWR), qd_fd;
             if (nbd < 0)
             {
@@ -928,7 +999,7 @@ protected:
                 }
             }
             // Configure request size
-            sprintf(path, "/sys/block/nbd%d/queue/max_sectors_kb", dev_num);
+            snprintf(path, sizeof(path), "/sys/block/nbd%d/queue/max_sectors_kb", dev_num);
             qd_fd = open(path, O_WRONLY);
             if (qd_fd < 0)
             {

src/client/qemu_driver.c

@@ -67,6 +67,11 @@ void __attribute__((visibility("default"))) DSO_STAMP_FUN(void)
 #endif
 
 typedef struct VitastorFdData VitastorFdData;
+typedef struct VitastorRPC VitastorRPC;
+
+#if QEMU_VERSION_MAJOR >= 4 && defined VITASTOR_C_API_VERSION && VITASTOR_C_API_VERSION >= 5
+#define VITASTOR_ONLINE_RESIZE
+#endif
 
 typedef struct VitastorClient
 {
@@ -75,6 +80,9 @@ typedef struct VitastorClient
     int auto_loop;
 
     void *watch;
+    VitastorRPC *open_task;
+    int bh_resize_scheduled;
+    int closing;
     char *config_path;
     char *etcd_host;
     char *etcd_prefix;
@@ -305,6 +313,88 @@ static void vitastor_schedule_uring_handler(VitastorClient *client)
 }
 #endif
 
+#ifdef VITASTOR_ONLINE_RESIZE
+// Notify the parents (i.e. the guest OS or, in case of qemu-storage-daemon with VDUSE,
+// the Linux kernel) that the size of the image has changed.
+// Runs in a coroutine because bdrv_co_parent_cb_resize() requires the graph read lock.
+static void coroutine_fn vitastor_co_resize(void *opaque)
+{
+    BlockDriverState *bs = (BlockDriverState*)opaque;
+    VitastorClient *client = bs->opaque;
+    bs->total_sectors = client->size / BDRV_SECTOR_SIZE;
+#if QEMU_VERSION_MAJOR > 10 || QEMU_VERSION_MAJOR == 10 && QEMU_VERSION_MINOR >= 2
+    {
+        GRAPH_RDLOCK_GUARD();
+        bdrv_co_parent_cb_resize(bs);
+    }
+#else
+    BdrvChild *c;
+#if QEMU_VERSION_MAJOR >= 9
+    assert_bdrv_graph_readable();
+#endif
+    QLIST_FOREACH(c, &bs->parents, next_parent) {
+#if QEMU_VERSION_MAJOR >= 6
+        if (c->klass->resize) {
+            c->klass->resize(c);
+        }
+#else
+        if (c->role->resize) {
+            c->role->resize(c);
+        }
+#endif
+    }
+#endif
+    bdrv_dec_in_flight(bs);
+}
+
+// B/H handler - runs in the AioContext of the BlockDriverState, without client->mutex held
+static void vitastor_size_changed_bh(void *opaque)
+{
+    BlockDriverState *bs = (BlockDriverState*)opaque;
+    VitastorClient *client = bs->opaque;
+    uint64_t new_size = 0;
+    qemu_mutex_lock(&client->mutex);
+    client->bh_resize_scheduled = 0;
+    if (client->watch)
+        new_size = vitastor_c_inode_get_size(client->watch);
+    qemu_mutex_unlock(&client->mutex);
+    if (!new_size || new_size == client->size)
+        return;
+    client->size = new_size;
+    bdrv_inc_in_flight(bs);
+    aio_co_enter(bdrv_get_aio_context(bs), qemu_coroutine_create(vitastor_co_resize, bs));
+}
+
+// Called by the Vitastor client with client->mutex held - both as the reply to
+// vitastor_c_watch_image() itself and every time the image is changed in etcd
+static void vitastor_watch_cb(void *opaque, long retval)
+{
+    BlockDriverState *bs = (BlockDriverState*)opaque;
+    VitastorClient *client = bs->opaque;
+    if (client->closing)
+        return;
+    client->watch = (void*)retval;
+    if (client->open_task)
+    {
+        // The first call is the reply to vitastor_c_watch_image() itself
+        VitastorRPC *task = client->open_task;
+        client->open_task = NULL;
+        vitastor_co_generic_cb(task, retval);
+    }
+    else if (!client->bh_resize_scheduled && client->size != vitastor_c_inode_get_size(client->watch))
+    {
+        // The image is resized - notify the guest, but do it in a B/H because
+        // we can't call into the block layer with client->mutex held
+        client->bh_resize_scheduled = 1;
+#if QEMU_VERSION_MAJOR > 4 || QEMU_VERSION_MAJOR == 4 && QEMU_VERSION_MINOR >= 2
+        replay_bh_schedule_oneshot_event(client->ctx, vitastor_size_changed_bh, bs);
+#else
+        aio_bh_schedule_oneshot(client->ctx, vitastor_size_changed_bh, bs);
+#endif
+    }
+}
+#endif
+
 static void coroutine_fn vitastor_co_get_metadata(VitastorRPC *task)
 {
     BlockDriverState *bs = task->bs;
@@ -312,7 +402,18 @@ static void coroutine_fn vitastor_co_get_metadata(VitastorRPC *task)
     task->co = qemu_coroutine_self();
 
     qemu_mutex_lock(&client->mutex);
-    vitastor_c_watch_inode(client->proxy, client->image, vitastor_co_generic_cb, task);
+    if (client->image)
+    {
+        // Watch the image to be able to follow online size changes
+        client->open_task = task;
+#ifdef VITASTOR_ONLINE_RESIZE
+        vitastor_c_watch_image(client->proxy, client->image, vitastor_watch_cb, bs);
+#else
+        vitastor_c_watch_inode(client->proxy, client->image, vitastor_co_generic_cb, task);
+#endif
+    }
+    else
+        vitastor_c_on_ready(client->proxy, vitastor_co_generic_cb, task);
     if (!client->auto_loop)
         vitastor_schedule_uring_handler(client);
     qemu_mutex_unlock(&client->mutex);
@@ -590,8 +691,6 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
     image = client->image = g_strdup(qdict_get_try_str(options, "image"));
     client->readonly = (flags & BDRV_O_RDWR) ? 1 : 0;
     // Get image metadata (size and readonly flag) or just wait until the client is ready
-    if (!image)
-        client->image = (char*)"x";
     task.complete = 0;
     task.bs = bs;
     if (qemu_in_coroutine())
@@ -669,6 +768,17 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
 static void vitastor_close(BlockDriverState *bs)
 {
     VitastorClient *client = bs->opaque;
+#ifdef VITASTOR_ONLINE_RESIZE
+    // Prevent scheduling of new size change B/H and wait for the scheduled one
+    qemu_mutex_lock(&client->mutex);
+    client->closing = 1;
+    qemu_mutex_unlock(&client->mutex);
+    if (client->bh_resize_scheduled)
+    {
+        BDRV_POLL_WHILE(bs, client->bh_resize_scheduled);
+    }
+    client->watch = NULL;
+#endif
     if (client->uring_eventfd >= 0)
     {
         // clear the eventfd handler

src/client/ublk_server.cpp

@@ -86,6 +86,7 @@ protected:
     epoll_manager_t *epmgr = NULL;
     cluster_client_t *cli = NULL;
     inode_watch_t *watch = NULL;
+    bool size_changed = false;
 
     std::string logfile = "/dev/null";
     std::string pidfile;
@@ -288,6 +289,7 @@ help:
                 fprintf(stderr, "Image %s does not exist\n", image_name.c_str());
                 exit(1);
             }
+            watch->callback = [this](inode_watch_t* watch) { on_change_watch(); };
         }
         const bool writeback = !cli->get_immediate_commit(inode ? inode : watch->cfg.num);
         auto pool_it = cli->st_cli->pool_config.find(INODE_POOL(inode ? inode : watch->cfg.num));
@@ -340,6 +342,8 @@ help:
         }
         consumer.loop = [this]()
         {
+            if (size_changed)
+                try_resize();
             submit_postponed();
         };
         ringloop->register_consumer(&consumer);
@@ -664,6 +668,34 @@ protected:
         }
     }
 
+    void on_change_watch()
+    {
+        if (device_size != watch->cfg.size)
+        {
+            device_size = watch->cfg.size;
+            size_changed = true;
+            ringloop->wakeup();
+        }
+    }
+
+    void try_resize()
+    {
+        io_uring_sqe *sqe = ublk_cmd(UBLK_U_CMD_UPDATE_SIZE, NULL, 0, 0, device_size / 512);
+        if (!sqe)
+        {
+            return;
+        }
+        size_changed = true;
+        ring_data_t* data = ((ring_data_t*)sqe->user_data);
+        data->callback = [this, size = device_size](ring_data_t *data)
+        {
+            if (data->res != 0)
+            {
+                fprintf(stderr, "Failed to update size of /dev/ublkb%d to %lu bytes: %s (code %d)\n", ublk_dev.dev_id, size, strerror(-data->res), data->res);
+            }
+        };
+    }
+
     void map_ublk_queue()
     {
         const unsigned page_sz = getpagesize();
@@ -745,8 +777,6 @@ protected:
             fprintf(stderr, "Failed to start ublk device: %s (code %d)\n", strerror(-res), res);
             exit(1);
         }
-        close(ctrl_fd);
-        ctrl_fd = -1;
     }
 
     bool submit_request(uint64_t ublk_cmd, int i, int res)
@@ -904,23 +934,12 @@ protected:
 
     int sync_ublk_cmd(uint32_t cmd_op, void *addr, uint32_t len, uint16_t dev_path_len = 0, uint64_t data0 = 0)
     {
-        io_uring_sqe *sqe = ringloop->get_sqe();
+        io_uring_sqe *sqe = ublk_cmd(cmd_op, addr, len, dev_path_len, data0);
         if (!sqe)
         {
             fprintf(stderr, "Error: io_uring is full when trying to execute a control command\n");
             exit(1);
         }
-        sqe->fd = ctrl_fd;
-        sqe->opcode = IORING_OP_URING_CMD;
-        sqe->ioprio = 0;
-        sqe->off = cmd_op;
-        ublksrv_ctrl_cmd *cmd = (ublksrv_ctrl_cmd *)&sqe->addr3; // sqe128 command buffer address
-        cmd->dev_id = ublk_dev.dev_id;
-        cmd->queue_id = -1;
-        cmd->addr = (uint64_t)addr;
-        cmd->len = len;
-        cmd->data[0] = data0;
-        cmd->dev_path_len = dev_path_len;
         ring_data_t* data = ((ring_data_t*)sqe->user_data);
         bool done = false;
         int res = 0;
@@ -938,6 +957,27 @@ protected:
         }
         return res;
     }
+
+    io_uring_sqe *ublk_cmd(uint32_t cmd_op, void *addr, uint32_t len, uint16_t dev_path_len = 0, uint64_t data0 = 0)
+    {
+        io_uring_sqe *sqe = ringloop->get_sqe();
+        if (!sqe)
+        {
+            return NULL;
+        }
+        sqe->fd = ctrl_fd;
+        sqe->opcode = IORING_OP_URING_CMD;
+        sqe->ioprio = 0;
+        sqe->off = cmd_op;
+        ublksrv_ctrl_cmd *cmd = (ublksrv_ctrl_cmd *)&sqe->addr3; // sqe128 command buffer address
+        cmd->dev_id = ublk_dev.dev_id;
+        cmd->queue_id = -1;
+        cmd->addr = (uint64_t)addr;
+        cmd->len = len;
+        cmd->data[0] = data0;
+        cmd->dev_path_len = dev_path_len;
+        return sqe;
+    }
 };
 
 int main(int narg, const char *args[])

src/client/vitastor_c.cpp

@@ -405,6 +405,23 @@ void vitastor_c_watch_inode(vitastor_c *client, char *image, VitastorIOHandler c
     }
 }
 
+void vitastor_c_watch_image(vitastor_c *client, char *image, VitastorIOHandler cb, void *opaque)
+{
+    client->cli->on_ready([=]()
+    {
+        inode_watch_t *watch = client->cli->st_cli->watch_inode(std::string(image));
+        cb(opaque, (long)watch);
+        watch->callback = [cb, opaque](inode_watch_t *watch)
+        {
+            cb(opaque, (long)watch);
+        };
+    });
+    if (client->ringloop)
+    {
+        client->ringloop->loop();
+    }
+}
+
 void vitastor_c_close_watch(vitastor_c *client, void *handle)
 {
     client->cli->st_cli->close_watch((inode_watch_t*)handle);

src/client/vitastor_c.h

@@ -7,7 +7,7 @@
 #define VITASTOR_QEMU_PROXY_H
 
 // C API wrapper version
-#define VITASTOR_C_API_VERSION 4
+#define VITASTOR_C_API_VERSION 5
 
 #ifndef POOL_ID_BITS
 #define POOL_ID_BITS 16
@@ -71,6 +71,7 @@ void vitastor_c_read_bitmap(vitastor_c *client, uint64_t inode, uint64_t offset,
     int with_parents, VitastorReadBitmapHandler cb, void *opaque);
 void vitastor_c_sync(vitastor_c *client, VitastorIOHandler cb, void *opaque);
 void vitastor_c_watch_inode(vitastor_c *client, char *image, VitastorIOHandler cb, void *opaque);
+void vitastor_c_watch_image(vitastor_c *client, char *image, VitastorIOHandler cb, void *opaque);
 void vitastor_c_close_watch(vitastor_c *client, void *handle);
 uint64_t vitastor_c_inode_get_size(void *handle);
 uint64_t vitastor_c_inode_get_num(void *handle);

src/cmd/cli_dd.cpp

@@ -49,7 +49,6 @@ struct dd_in_info_t
 
     void open_input(cli_tool_t *parent)
     {
-        in_seekable = true;
         if (iimg != "")
         {
             iwatch = parent->cli->st_cli->watch_inode(iimg);
@@ -73,8 +72,10 @@ struct dd_in_info_t
             {
                 in_size = iwatch->cfg.size;
             }
+            in_seekable = true;
+            return;
         }
-        else if (ifile != "")
+        if (ifile != "")
         {
             ifd = open(ifile.c_str(), (in_direct ? O_DIRECT : 0) | O_RDONLY);
             if (ifd < 0)
@@ -82,48 +83,45 @@ struct dd_in_info_t
                 result = (cli_result_t){ .err = errno, .text = "Failed to open "+ifile+": "+std::string(strerror(errno)) };
                 return;
             }
-            if (detect_size)
-            {
-                struct stat st;
-                if (fstat(ifd, &st) < 0)
-                {
-                    result = (cli_result_t){ .err = errno, .text = "Failed to stat "+ifile+": "+std::string(strerror(errno)) };
-                    close(ifd);
-                    ifd = -1;
-                    return;
-                }
-                if (S_ISREG(st.st_mode))
-                {
-                    in_size = st.st_size;
-                }
-                else if (S_ISBLK(st.st_mode))
-                {
-                    if (ioctl(ifd, BLKGETSIZE64, &in_size) < 0)
-                    {
-                        result = (cli_result_t){ .err = errno, .text = "Failed to get "+ifile+" size: "+std::string(strerror(errno)) };
-                        close(ifd);
-                        ifd = -1;
-                        return;
-                    }
-                }
-            }
             if (in_direct)
             {
                 in_granularity = 512;
             }
-            if (lseek(ifd, 1, SEEK_SET) != (off_t)1)
-            {
-                in_seekable = false;
-            }
-            else
-            {
-                lseek(ifd, 0, SEEK_SET);
-            }
         }
         else
         {
             ifd = 0;
-            in_seekable = false;
+        }
+        if (detect_size)
+        {
+            struct stat st;
+            if (fstat(ifd, &st) < 0)
+            {
+                result = (cli_result_t){ .err = errno, .text = "Failed to stat "+ifile+": "+std::string(strerror(errno)) };
+                close(ifd);
+                ifd = -1;
+                return;
+            }
+            if (S_ISREG(st.st_mode))
+            {
+                in_size = st.st_size;
+            }
+            else if (S_ISBLK(st.st_mode))
+            {
+                if (ioctl(ifd, BLKGETSIZE64, &in_size) < 0)
+                {
+                    result = (cli_result_t){ .err = errno, .text = "Failed to get "+ifile+" size: "+std::string(strerror(errno)) };
+                    close(ifd);
+                    ifd = -1;
+                    return;
+                }
+            }
+        }
+        in_seekable = false;
+        if (lseek(ifd, 1, SEEK_SET) == (off_t)1)
+        {
+            in_seekable = true;
+            lseek(ifd, 0, SEEK_SET);
         }
     }
 
@@ -286,8 +284,9 @@ resume_2:
                 state = base_state+1;
                 return false;
             }
+            return true;
         }
-        else if (ofile != "")
+        if (ofile != "")
         {
             ofd = open(ofile.c_str(), (out_direct ? O_DIRECT : 0) | (out_append ? O_APPEND : O_RDWR) | (out_create ? O_CREAT : 0), 0666);
             if (ofd < 0)
@@ -304,12 +303,16 @@ resume_2:
             {
                 out_granularity = 512;
             }
-            out_seekable = !out_append;
         }
         else
         {
             ofd = 1;
-            out_seekable = false;
+        }
+        out_seekable = false;
+        if (!out_append && lseek(ofd, 1, SEEK_SET) == (off_t)1)
+        {
+            out_seekable = true;
+            lseek(ofd, 0, SEEK_SET);
         }
         return true;
     }

src/disk_tool/CMakeLists.txt

@@ -22,7 +22,5 @@ add_executable(vitastor-disk
 )
 target_link_libraries(vitastor-disk
 	util
-	${LIBURING_LIBRARIES}
-	# for crc32_iscsi
-	${ISAL_LIBRARIES}
+	crc32c
 )

src/nfs/nfs_proxy.cpp

@@ -145,7 +145,7 @@ static const char* help_text =
     "\n"
     "Example start and mount commands for a custom NFS port:\n"
     "  vitastor-nfs start --block --etcd_address 192.168.5.10:2379 --portmap 0 --port 2050 --pool testpool\n"
-    "  mount localhost:/ /mnt/ -o port=2050,mountport=2050,nfsvers=3,soft,nolock,tcp\n"
+    "  mount localhost:/ /mnt/ -o port=2050,mountport=2050,nfsvers=3,soft,sync,nolock,tcp\n"
     "Or just:\n"
     "  vitastor-nfs mount --block --pool testpool /mnt/\n"
 ;

src/osd/osd_cluster.cpp

@@ -465,6 +465,14 @@ void osd_t::apply_pg_locks_localize_only()
             printf("[PG %u/%u] Repeer to enable PG locks\n", pg.pool_id, pg.pg_num);
             repeer_pg(pg);
         }
+        else if (pg.pg_minsize != pool_cfg.pg_minsize &&
+            (pg.state == PG_INCOMPLETE) != (pg.pg_cursize < pool_cfg.pg_minsize))
+        {
+            // Repeer PG with the new pg_minsize
+            printf("[PG %u/%u] Repeer to apply pg_minsize\n", pg.pool_id, pg.pg_num);
+            pg.pg_minsize = pool_cfg.pg_minsize;
+            repeer_pg(pg);
+        }
     }
 }
 

src/osd/osd_peering.cpp

@@ -540,13 +540,13 @@ void osd_t::submit_list_subop(osd_num_t role_osd, pg_peering_state_t *ps)
         {
             if (op->bs_op->retval < 0)
             {
-                printf("Local OP_LIST failed: retval=%d\n", op->bs_op->retval);
+                printf("Local OP_LIST failed: retval=%jd\n", op->bs_op->retval);
                 force_stop(1);
                 return;
             }
             add_bs_subop_stats(op);
             printf(
-                "[PG %u/%u] Got object list from OSD %ju (local): %d object versions (%ju of them stable)\n",
+                "[PG %u/%u] Got object list from OSD %ju (local): %jd object versions (%ju of them stable)\n",
                 ps->pool_id, ps->pg_num, role_osd, bs_op->retval, bs_op->version
             );
             ps->list_results[role_osd] = {

src/osd/osd_primary_subops.cpp

@@ -317,7 +317,7 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop)
 {
     osd_op_t *cur_op = (osd_op_t*)subop->op_type;
     blockstore_op_t *bs_op = subop->bs_op;
-    int expected = bs_op->opcode == BS_OP_READ || bs_op->opcode == BS_OP_WRITE
+    int64_t expected = bs_op->opcode == BS_OP_READ || bs_op->opcode == BS_OP_WRITE
         || bs_op->opcode == BS_OP_WRITE_STABLE ? bs_op->len : 0;
     if (bs_op->retval != expected && bs_op->opcode != BS_OP_READ &&
         (bs_op->opcode != BS_OP_WRITE && bs_op->opcode != BS_OP_WRITE_STABLE ||
@@ -327,7 +327,7 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop)
         if (bs_op->opcode == BS_OP_WRITE || bs_op->opcode == BS_OP_WRITE_STABLE)
         {
             printf(
-                "%s subop to %jx:%jx v%ju failed locally: retval = %d (expected %d)\n",
+                "%s subop to %jx:%jx v%ju failed locally: retval = %jd (expected %jd)\n",
                 osd_op_names[bs_op_to_osd_op[bs_op->opcode]],
                 bs_op->oid.inode, bs_op->oid.stripe, bs_op->version, bs_op->retval, expected
             );
@@ -335,7 +335,7 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop)
         else
         {
             printf(
-                "%s subop failed locally: retval = %d (expected %d)\n",
+                "%s subop failed locally: retval = %jd (expected %jd)\n",
                 osd_op_names[bs_op_to_osd_op[bs_op->opcode]], bs_op->retval, expected
             );
         }

src/osd/osd_scrub.cpp

@@ -36,7 +36,7 @@ void osd_t::scrub_list(pool_pg_num_t pg_id, osd_num_t role_osd, object_id min_oi
             scrub_list_op = NULL;
             if (op->bs_op->retval < 0)
             {
-                printf("Local OP_LIST failed: retval=%d\n", op->bs_op->retval);
+                printf("Local OP_LIST failed: retval=%jd\n", op->bs_op->retval);
                 force_stop(1);
                 return;
             }

src/test/CMakeLists.txt

@@ -42,8 +42,8 @@ add_executable(test_heap
 	../blockstore/blockstore_disk.cpp
 )
 target_link_libraries(test_heap
-	${ISAL_LIBRARIES}
 	util
+	crc32c
 )
 add_dependencies(build_tests test_heap)
 add_test(NAME test_heap COMMAND test_heap)
@@ -66,13 +66,13 @@ add_executable(test_crc32
 # test_blockstore
 add_executable(test_blockstore EXCLUDE_FROM_ALL test_blockstore.cpp ringloop_mock.cpp)
 add_dependencies(build_tests test_blockstore)
-target_link_libraries(test_blockstore vitastor_blk vitastor_common ${ISAL_LIBRARIES})
+target_link_libraries(test_blockstore vitastor_blk vitastor_common)
 add_test(NAME test_blockstore COMMAND test_blockstore)
 
 # test_blockstore_v1
 add_executable(test_blockstore_v1 EXCLUDE_FROM_ALL test_blockstore_v1.cpp ringloop_mock.cpp)
 add_dependencies(build_tests test_blockstore_v1)
-target_link_libraries(test_blockstore_v1 vitastor_blk vitastor_common ${ISAL_LIBRARIES})
+target_link_libraries(test_blockstore_v1 vitastor_blk vitastor_common)
 add_test(NAME test_blockstore_v1 COMMAND test_blockstore_v1)
 
 ## test_shit

src/test/osd_test.cpp

@@ -1970,7 +1970,6 @@ void test_ec42_inconsistent_read()
     use_ec(6, 4, false);
     memcpy(chunks.data() + 4*chunk_size, (uint8_t*)rmw_buf + 0*chunk_size, chunk_size);
     memcpy(chunks.data() + 5*chunk_size, (uint8_t*)rmw_buf + 1*chunk_size, chunk_size);
-    printf("XXXXX %02x %02x\n", chunks[4*chunk_size], chunks[5*chunk_size]);
     free(rmw_buf);
 
     // Scrub sees one missing data role and one bad survivor. The other four

src/test/test_atomic.cpp

@@ -54,7 +54,7 @@ int main(int narg, char *args[])
                 {
                     if (data->res != write_size)
                     {
-                        fprintf(stderr, "Error writing at offset %lu: %s (code %d)\n", offset, strerror(-data->res), -data->res);
+                        fprintf(stderr, "Error writing at offset %ju: %s (code %d)\n", offset, strerror(-data->res), -data->res);
                         exit(1);
                     }
                     inflight--;

src/test/test_blockstore.cpp

@@ -142,7 +142,7 @@ struct bs_test_t
         bool done = false;
         op->callback = [&](blockstore_op_t *op)
         {
-            printf("op opcode=%lu completed retval=%d\n", op->opcode, op->retval);
+            printf("op opcode=%ju completed retval=%jd\n", op->opcode, op->retval);
             done = true;
         };
         bs->enqueue_op(op);

src/test/test_blockstore_v1.cpp

@@ -391,6 +391,263 @@ static void test_validate_padded_journal()
     free(read_op.buf);
 }
 
+// Check that the flusher can pad a small write which covers only a part of a checksum
+// block when the rest of that checksum block is a hole in the clean object.
+// The flusher asks for the list of required reads without a buffer (read_buf = NULL),
+// so fill_partial_checksum_blocks() must not rebase NULL into a bogus pointer.
+static void test_flush_partial_csum_block_over_hole()
+{
+    printf("\n-- test_flush_partial_csum_block_over_hole\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["data_csum_type"] = "crc32c";
+    test.config["csum_block_size"] = "8192";
+    test.init();
+    printf("blockstore initialized\n");
+
+    // Clean version covering 0..8k only, so 8k..16k stays a hole in the clean bitmap
+    // while the checksum block 8k..16k gets partially overwritten below
+    printf("write v1 0+8k\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 = 8*1024;
+    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);
+
+    test.force_compaction();
+
+    // Small write aligned at the beginning of checksum block 8k..16k, but not at its end,
+    // so the flusher has to pad it with old data - and the padding is a hole
+    printf("small_write v2 8+4k\n");
+    op.version = 2;
+    op.offset = 8*1024;
+    op.len = 4*1024;
+    memset(op.buf, 0xAB, op.len);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+    assert(test.flusher()->get_queue_size());
+
+    // Used to segfault in journal_flusher_co::scan_dirty()
+    test.force_compaction();
+
+    printf("read v2 0+128k\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 == read_op.len);
+    assert(memcheck(read_op.buf, 0xAA, 8*1024));
+    assert(memcheck(read_op.buf + 8*1024, 0xAB, 4*1024));
+    assert(memcheck(read_op.buf + 12*1024, 0, (128-12)*1024));
+
+    free(op.buf);
+    free(read_op.buf);
+}
+
+// Check that a read which starts before the object's data zero-fills the correct
+// part of the read buffer. read_buf is already rebased by read_offset when the
+// zero-filling callback runs, so it must not be subtracted a second time.
+static void test_read_padded_big_write_at_offset()
+{
+    printf("\n-- test_read_padded_big_write_at_offset\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["data_csum_type"] = "crc32c";
+    test.config["csum_block_size"] = "8192";
+    test.init();
+    printf("blockstore initialized\n");
+
+    // The first write to a non-existent object is a big_write, so the object's data
+    // starts at 16k and 0..16k is a hole
+    printf("write v1 16+16k\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 16*1024;
+    op.len = 16*1024;
+    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);
+
+    // Read 8k..24k. 8k..16k has to be zero-filled, 16k..24k has to be read from the disk.
+    // Keep 16k of slack in front of the buffer so that a misplaced zero-fill lands in the
+    // slack and fails the assertions below instead of corrupting the heap.
+    printf("read v1 8+16k\n");
+    uint8_t *read_base = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 32*1024);
+    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 = 8*1024;
+    read_op.len = 16*1024;
+    read_op.buf = read_base + 16*1024;
+    memset(read_base, 0xCC, 32*1024);
+    test.exec_op(&read_op);
+    assert(read_op.retval == read_op.len);
+    assert(memcheck((uint8_t*)read_op.buf, 0, 8*1024));
+    assert(memcheck((uint8_t*)read_op.buf + 8*1024, 0xAA, 8*1024));
+    // The slack in front of the buffer must be untouched
+    assert(memcheck(read_base, 0xCC, 16*1024));
+
+    free(op.buf);
+    free(read_base);
+}
+
+// Check that a clean object with several holes inside one checksum block is read correctly.
+// Such a read used to abort with assert(fulfilled == read_op->len) in dequeue_read():
+// read_range_fulfilled() inserted all zero-fill ranges of a single hole at the same position,
+// so read_vec ended up unsorted and the next find_holes() reported an already zero-filled
+// range as a hole again, adding its length to `fulfilled` for the second time
+static void test_read_sparse_clean_bitmap_multi_zero_runs()
+{
+    printf("\n-- test_read_sparse_clean_bitmap_multi_zero_runs\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["block_size"] = "131072";
+    test.config["data_csum_type"] = "crc32c";
+    // 8 bitmap granules per checksum block
+    test.config["csum_block_size"] = "32768";
+    test.init();
+    printf("blockstore initialized\n");
+
+    // The first write to a non-existent object is a big_write, so the bitmap built for the
+    // clean entry during compaction has granule 0 and granules 6..7 of checksum block 0
+    // clear and granules 1..5 set - that is, two separate zero runs inside one checksum block
+    printf("write v1 4+20k\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 4*1024;
+    op.len = 20*1024;
+    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);
+
+    // read_range_fulfilled() is only reached for clean objects - the journal path passes
+    // clean_entry_bitmap = NULL because journal writes don't have holes - so compact first
+    test.force_compaction();
+
+    // Read the whole checksum block: 0..4k and 24..32k are zero-filled from the bitmap,
+    // 4..24k is read from the disk.
+    // Keep 32k of slack in front of the buffer so that a misplaced zero-fill lands in the
+    // slack and fails the assertions below instead of corrupting the heap.
+    printf("read v1 0+32k\n");
+    uint8_t *read_base = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 64*1024);
+    memset(read_base, 0xCC, 64*1024);
+    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 = 32*1024;
+    read_op.buf = read_base + 32*1024;
+    test.exec_op(&read_op);
+    assert(read_op.retval == read_op.len);
+    assert(memcheck(read_op.buf, 0, 4*1024));
+    assert(memcheck(read_op.buf + 4*1024, 0xAA, 20*1024));
+    assert(memcheck(read_op.buf + 24*1024, 0, 8*1024));
+    // The slack in front of the buffer must be untouched
+    assert(memcheck(read_base, 0xCC, 32*1024));
+
+    // The same over the whole object - checksum blocks 1..3 are completely unallocated
+    printf("read v1 0+128k\n");
+    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 == read_op.len);
+    assert(memcheck(read_op.buf, 0, 4*1024));
+    assert(memcheck(read_op.buf + 4*1024, 0xAA, 20*1024));
+    assert(memcheck(read_op.buf + 24*1024, 0, (128-24)*1024));
+
+    free(read_op.buf);
+    free(read_base);
+    free(op.buf);
+}
+
+// Check that a read which doesn't intersect the object's data at all is zero-filled
+// inside the read buffer. The first write to a non-existent object is a big_write with
+// the offset and length of that write, so the dirty entry may cover any part of the
+// block - and a read of any other part of the same block gets an item range which lies
+// completely before or completely after it.
+static void test_read_padded_big_write_outside_read()
+{
+    printf("\n-- test_read_padded_big_write_outside_read\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["data_csum_type"] = "crc32c";
+    test.config["csum_block_size"] = "8192";
+    test.init();
+    printf("blockstore initialized\n");
+
+    // The first write to a non-existent object is a big_write, so the object's data
+    // is 16k..32k and the rest of the block is a hole
+    printf("write v1 16+16k\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 16*1024;
+    op.len = 16*1024;
+    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);
+
+    // Keep 128k of slack on both sides of the read buffer so that a misplaced zero-fill
+    // lands in the slack and fails the assertions below instead of corrupting the heap
+    uint8_t *read_base = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 3*128*1024);
+    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.buf = read_base + 128*1024;
+
+    // Read 0..8k - the whole read is in front of the object's data
+    printf("read v1 0+8k\n");
+    read_op.offset = 0;
+    read_op.len = 8*1024;
+    memset(read_base, 0xCC, 3*128*1024);
+    test.exec_op(&read_op);
+    assert(read_op.retval == read_op.len);
+    assert(memcheck((uint8_t*)read_op.buf, 0, 8*1024));
+    // The slack around the buffer must be untouched
+    assert(memcheck(read_base, 0xCC, 128*1024));
+    assert(memcheck(read_base + 128*1024 + 8*1024, 0xCC, 2*128*1024 - 8*1024));
+
+    // Read 40k..56k - the whole read is behind the object's data
+    printf("read v1 40+16k\n");
+    read_op.offset = 40*1024;
+    read_op.len = 16*1024;
+    memset(read_base, 0xCC, 3*128*1024);
+    test.exec_op(&read_op);
+    assert(read_op.retval == read_op.len);
+    assert(memcheck((uint8_t*)read_op.buf, 0, 16*1024));
+    // The slack around the buffer must be untouched
+    assert(memcheck(read_base, 0xCC, 128*1024));
+    assert(memcheck(read_base + 128*1024 + 16*1024, 0xCC, 2*128*1024 - 16*1024));
+
+    free(op.buf);
+    free(read_base);
+}
+
 // Check that read is retried and temporary read buffers are freed correctly (LSAN)
 // TODO: Check retries in other configuration
 static void test_read_retry_on_ring_full_1M_csum4k_clean()
@@ -537,7 +794,7 @@ static void test_read_free_clean_loc_used()
     test.force_compaction();
 
     // Check that the old location is still not freed
-    printf("checking that data location 0x%lx is still used\n", read_loc);
+    printf("checking that data location 0x%jx is still used\n", read_loc);
     assert(test.is_data_loc_used(read_loc));
     printf("completing read\n");
     test.ringloop->mark_completed(data_read);
@@ -545,7 +802,7 @@ static void test_read_free_clean_loc_used()
     while (!done_read)
         test.ringloop->loop();
     read_op.callback = NULL;
-    printf("checking that data location 0x%lx is freed\n", read_loc);
+    printf("checking that data location 0x%jx is freed\n", read_loc);
     assert(!test.is_data_loc_used(read_loc));
 
     assert(read_op.retval == read_op.len);
@@ -599,6 +856,10 @@ int main(int narg, char *args[])
 {
     test_preserve_corruption();
     test_validate_padded_journal();
+    test_flush_partial_csum_block_over_hole();
+    test_read_padded_big_write_at_offset();
+    test_read_sparse_clean_bitmap_multi_zero_runs();
+    test_read_padded_big_write_outside_read();
     test_read_retry_on_ring_full_1M_csum4k_clean();
     test_read_free_clean_loc_used();
     test_meta_tail_reload();

src/test/test_cluster_client.cpp

@@ -87,6 +87,34 @@ int *test_write(cluster_client_t *cli, uint64_t offset, uint64_t len, uint8_t c,
     return r;
 }
 
+int *test_read(cluster_client_t *cli, uint64_t offset, uint64_t len, std::function<void()> cb = NULL)
+{
+    printf("Post read %jx+%jx\n", offset, len);
+    int *r = new int;
+    *r = -1000;
+    cluster_op_t *op = new cluster_op_t();
+    op->opcode = OSD_OP_READ;
+    op->inode = 0x1000000000001;
+    op->offset = offset;
+    op->len = len;
+    op->iov.push_back(malloc_or_die(len), len);
+    op->callback = [r, cb](cluster_op_t *op)
+    {
+        if (*r == -1000)
+            printf("Error: Not allowed to complete yet\n");
+        assert(*r != -1000);
+        assert(op->retval == op->len || op->retval < 0);
+        *r = op->retval == op->len ? 1 : op->retval;
+        free(op->iov.buf[0].iov_base);
+        printf("Done read %jx+%jx r=%d\n", op->offset, op->len, op->retval);
+        delete op;
+        if (cb != NULL)
+            cb();
+    };
+    cli->execute(op);
+    return r;
+}
+
 int *test_sync(cluster_client_t *cli)
 {
     printf("Post sync\n");
@@ -1039,6 +1067,59 @@ void cluster_client_test_t::test_vault()
     printf("[ok] basic vault key fetch test\n");
 }
 
+void test_pg_config_disappears()
+{
+    json11::Json config;
+    timerfd_manager_t *tfd = new timerfd_manager_t([](int fd, bool wr, std::function<void(int, int)> callback){});
+    etcd_state_client_mock_t *mock = new etcd_state_client_mock_t();
+    mock->pause();
+    cluster_client_t *cli = new cluster_client_t(NULL, tfd, config, std::unique_ptr<etcd_state_client_t>(mock));
+
+    configure_single_pg_pool(mock);
+    mock->resume();
+    int *r1 = test_write(cli, 0, 4096, 0x55);
+
+    // A bad guy removes /pg/config. Client shouldn't crash but should return EINVAL.
+    can_complete(r1);
+    mock->set("/vitastor/pg/config", json11::Json::object {
+        { "items", json11::Json::object { } }
+    });
+    assert(*r1 == -EINVAL);
+    delete r1;
+
+    delete cli;
+    delete tfd;
+    printf("[ok] test_pg_config_disappears\n");
+}
+
+void test_snapshot_over_nonexistent_pool()
+{
+    json11::Json config;
+    timerfd_manager_t *tfd = new timerfd_manager_t([](int fd, bool wr, std::function<void(int, int)> callback){});
+    etcd_state_client_mock_t *mock = new etcd_state_client_mock_t();
+    mock->pause();
+    cluster_client_t *cli = new cluster_client_t(NULL, tfd, config, std::unique_ptr<etcd_state_client_t>(mock));
+
+    configure_single_pg_pool(mock);
+    mock->set("/vitastor/config/inode/1/1", json11::Json::object {
+        { "name", "parent" },
+        { "parent_id", 1 },
+        { "parent_pool", 2 },
+    });
+    mock->resume();
+    pretend_connected(cli, 1);
+    int *r1 = test_read(cli, 0, 4096);
+    check_op_count(cli, 1, 1);
+    can_complete(r1);
+    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_READ, 0, 4096), 0);
+    assert(*r1 == -EINVAL);
+    delete r1;
+
+    delete cli;
+    delete tfd;
+    printf("[ok] test_snapshot_over_nonexistent_pool\n");
+}
+
 int main(int narg, char *args[])
 {
     test1();
@@ -1050,5 +1131,7 @@ int main(int narg, char *args[])
     test_msgr_encrypt();
     test_msgr_decrypt_chain();
     cluster_client_test_t::test_vault();
+    test_pg_config_disappears();
+    test_snapshot_over_nonexistent_pool();
     return 0;
 }

src/util/CMakeLists.txt

@@ -32,6 +32,16 @@ endif (RDMACM_LIBRARIES)
 target_compile_options(util PUBLIC -fPIC)
 target_link_libraries(util
 	${LIBURING_LIBRARIES}
+	${OPENSSL_LIBRARIES}
+)
+
+# libcrc32c.a to isolate the isa-l dependency
+add_library(crc32c STATIC
+	crc32c.c
+)
+target_compile_options(crc32c PUBLIC -fPIC)
+target_link_libraries(crc32c
+	${ISAL_LIBRARIES}
 )
 
 # test_rdma_alloc

tests/common.sh

@@ -114,11 +114,11 @@ wait_condition()
     i=0
     while [[ $i -lt $sec ]]; do
         eval "$check" && break
+        i=$((i+1))
         if [ $i -eq $sec ]; then
             format_error "$proc couldn't finish in $sec seconds"
         fi
         sleep 1
-        i=$((i+1))
     done
 }
 

tests/run_tests.sh

@@ -12,6 +12,8 @@ SCHEME=ec ./test_change_pg_count.sh
 
 ./test_change_pg_count_online.sh
 
+./test_change_pg_minsize.sh
+
 ./test_change_pg_size.sh
 
 ./test_create_nomaxid.sh
@@ -23,6 +25,8 @@ ETCD_SCHEME=https ./test_etcd_fail.sh
 ETCD_SCHEME=https ANTIETCD=1 ./test_etcd_fail.sh
 ETCD_SCHEME=https ./test_snapshot.sh
 
+./test_img_online_resize.sh
+
 ./test_interrupted_rebalance.sh
 IMMEDIATE_COMMIT=1 ./test_interrupted_rebalance.sh
 SCHEME=ec ./test_interrupted_rebalance.sh

tests/test_change_pg_minsize.sh

@@ -0,0 +1,23 @@
+#!/bin/bash -ex
+
+OSD_TLS=0
+PG_COUNT=8
+OSD_COUNT=3
+PG_SIZE=3
+PG_MINSIZE=3
+SCHEME=replicated
+
+. `dirname $0`/run_3osds.sh
+
+kill $OSD1_PID
+
+wait_condition 10 "$ETCDCTL get /vitastor/pg/state/1/ --prefix --print-value-only |\
+    jq -s -e '[ .[] | select(.state | contains(["'"'"incomplete"'"'"])) ] | length == '$PG_COUNT" "wait for incomplete"
+
+PG_MINSIZE=2
+$ETCDCTL put /vitastor/config/pools '{"1":{'$POOLCFG',"pg_size":'$PG_SIZE',"pg_minsize":'$PG_MINSIZE',"pg_count":'$PG_COUNT'}}'
+
+wait_condition 10 "$ETCDCTL get /vitastor/pg/state/1/ --prefix --print-value-only |\
+    jq -s -e '[ .[] | select(.state | contains(["'"'"active"'"'"])) ] | length == '$PG_COUNT" "wait for active"
+
+format_green OK

tests/test_dd.sh

@@ -16,5 +16,12 @@ $VITASTOR_CLI snap-create testimg@snap1
 $VITASTOR_CLI dd iodepth=4 if=./testdata/over oimg=testimg bs=1M seek=17
 $VITASTOR_CLI dd iodepth=4 iimg=testimg of=./testdata/testfile1
 diff ./testdata/testfile ./testdata/testfile1
+rm ./testdata/testfile1
+
+# copy image using dd
+$VITASTOR_CLI dd iimg=testimg iodepth=4 bs=1M count=128 | $VITASTOR_CLI dd oimg=testimg2 iodepth=4 bs=1M count=128
+$VITASTOR_CLI dd iimg=testimg2 iodepth=4 bs=1M count=128 > ./testdata/testfile1
+diff ./testdata/testfile ./testdata/testfile1
+rm ./testdata/testfile1
 
 format_green OK

tests/test_img_online_resize.sh

@@ -0,0 +1,62 @@
+#!/bin/bash -ex
+
+PG_COUNT=16
+. `dirname $0`/run_3osds.sh
+
+$VITASTOR_CLI create -s 100M testimg
+
+if ! (mount | grep '/dev type devtmpfs'); then
+    sudo bash -c 'mount udev /dev/ -t devtmpfs && mount devpts /dev/pts -t devpts'
+fi
+
+NBD_DEV=$(sudo build/src/client/vitastor-nbd --config_path $VITASTOR_CFG map --image testimg)
+trap "sudo build/src/client/vitastor-nbd unmap $NBD_DEV"' || true; kill -9 $(jobs -p)' EXIT
+UBLK_DEV=$(sudo build/src/client/vitastor-ublk --config_path $VITASTOR_CFG map --image testimg)
+trap "sudo build/src/client/vitastor-ublk unmap $UBLK_DEV || true; sudo build/src/client/vitastor-nbd unmap $NBD_DEV || true; "'kill -9 $(jobs -p)' EXIT
+
+VDUSE_DEV=
+if sudo vdpa dev list; then
+    # VDUSE doesn't work in Docker without host network at all
+    check_qemu
+    [[ -e /sys/module/vduse ]] || sudo modprobe vduse
+    [[ -e /sys/module/virtio_vdpa ]] || sudo modprobe virtio-vdpa
+    VDPA_ID=vpda$$
+    sudo -E qemu-storage-daemon --pidfile ./testdata/qsd.pid --daemonize --blockdev '{"node-name":"'$VDPA_ID'","driver":"vitastor",
+        "config-path":"'$VITASTOR_CFG'","image":"testimg","cache":{"direct":true,"no-flush":false},"discard":"unmap"}' \
+        --export vduse-blk,id=$VDPA_ID,node-name=$VDPA_ID,name=$VDPA_ID,num-queues=16,queue-size=128,writable=true
+    QSD_PID=$(sudo cat ./testdata/qsd.pid)
+    trap "sudo kill -9 $QSD_PID; sudo build/src/client/vitastor-ublk unmap $UBLK_DEV || true; sudo build/src/client/vitastor-nbd unmap $NBD_DEV || true; "'kill -9 $(jobs -p)' EXIT
+    sudo vdpa dev add name $VDPA_ID mgmtdev vduse
+    trap "sudo vdpa dev del $VDPA_ID; sudo kill -9 $QSD_PID; sudo build/src/client/vitastor-ublk unmap $UBLK_DEV || true; sudo build/src/client/vitastor-nbd unmap $NBD_DEV || true; "'kill -9 $(jobs -p)' EXIT
+    VDUSE_DEV=/dev/$(ls /sys/bus/vdpa/devices/$VDPA_ID/virtio*/block/)
+fi
+
+check_bdev_size() {
+    if [[ $(sudo blockdev --getsize64 $1) -ne $2 ]]; then
+        format_error "$1 device size is not $2"
+    fi
+}
+
+check_bdev_size $NBD_DEV 104857600
+check_bdev_size $UBLK_DEV 104857600
+if [[ -n "$VDUSE_DEV" ]]; then
+    check_bdev_size $VDUSE_DEV 104857600
+fi
+
+$VITASTOR_CLI modify --resize 200M testimg
+
+for i in {1..10}; do
+    if [[ "$(sudo blockdev --getsize64 $NBD_DEV)" -ne 209715200 || "$(sudo blockdev --getsize64 $UBLK_DEV)" -ne 209715200 || -n "$VDUSE_DEV" && "$(sudo blockdev --getsize64 $VDUSE_DEV)" -ne 209715200 ]]; then
+        sleep 1
+    else
+        break
+    fi
+done
+
+check_bdev_size $NBD_DEV 209715200
+check_bdev_size $UBLK_DEV 209715200
+if [[ -n "$VDUSE_DEV" ]]; then
+    check_bdev_size $VDUSE_DEV 209715200
+fi
+
+format_green OK