Release 1.10.1

New features:

- Add "deleted" image flag which is set when vitastor-cli rm starts to delete an image,
  but can't delete it fully due to inactive PGs or stopped OSDs
- Support JSON output in vitastor-disk prepare and purge
- Show backfillfull pools in vitastor-cli status
- Make object listings consistent (used in vitastor-cli rm/rm-data/merge/etc).
  This means that there is now a guarantee that if a data block is present when you invoke rm,
  rm will attempt to delete it, even if rm is invoked when the PG switches state. Previously in
  such cases rm could skip and leave some objects behind as garbage, and merge probably could
  incorrectly move data between snapshots.
- Make deletions (rm/rm-data) consistent. This means that rm/rm-data will either complete
  successfully and delete all requested image data or complete with an error if some objects
  could not be deleted or if there is a possibility that some data is left on stopped OSDs.
  Previously, when some PGs or OSDs were inactive at the moment of deletion, rm-data was
  behaving incorrectly: it wasn't retrying deletions failed due to dropped OSD connections,
  it could hang waiting for PGs to activate, and it could return with a successful error
  code while some garbage was still possibly left on some OSDs. Deletions are not fully atomic
  cluster-wide yet, which means that you still have to repeat the deletion request after you
  return stopped OSDs back, but now you always know for sure if you have to repeat it.

Bug fixes:

- Fix vitastor-cli rm --exact / --matching command not working
- Finally fix "Unexpected status" in the Proxmox plugin
- Fix vitastor-cli create-snap incorrectly linking multiple snapshots in a different pool
- Fix incomplete image parent_id loop check in OSD
- Fix reads from snapshots in a different pool not working if there are more than 2 snapshots
- Fix append of VITASTOR_CONF to cmdline in the opennebula prebackup script
- Fix OSDs crashing again when the cluster is full with EC (was meant to work since 1.6.0 but didn't)
- Improve logging of subop failures

.gitea/workflows/test.yml

@@ -414,6 +414,24 @@ jobs:
           echo ""
         done
 
+  test_rm_degraded:
+    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_rm_degraded.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_snapshot_chain:
     runs-on: ubuntu-latest
     needs: build

CMakeLists.txt

@@ -2,6 +2,6 @@ cmake_minimum_required(VERSION 2.8.12)
 
 project(vitastor)
 
-set(VITASTOR_VERSION "1.10.0")
+set(VITASTOR_VERSION "1.10.1")
 
 add_subdirectory(src)

csi/Makefile

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v1.10.0
+VITASTOR_VERSION ?= v1.10.1
 
 all: build push
 

csi/deploy/004-csi-nodeplugin.yaml

@@ -49,7 +49,7 @@ spec:
             capabilities:
               add: ["SYS_ADMIN"]
             allowPrivilegeEscalation: true
-          image: vitalif/vitastor-csi:v1.10.0
+          image: vitalif/vitastor-csi:v1.10.1
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/deploy/007-csi-provisioner.yaml

@@ -121,7 +121,7 @@ spec:
             privileged: true
             capabilities:
               add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v1.10.0
+          image: vitalif/vitastor-csi:v1.10.1
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/src/config.go

@@ -5,7 +5,7 @@ package vitastor
 
 const (
     vitastorCSIDriverName    = "csi.vitastor.io"
-    vitastorCSIDriverVersion = "1.10.0"
+    vitastorCSIDriverVersion = "1.10.1"
 )
 
 // Config struct fills the parameters of request or user input

debian/changelog

@@ -1,4 +1,4 @@
-vitastor (1.10.0-1) unstable; urgency=medium
+vitastor (1.10.1-1) unstable; urgency=medium
 
   * Bugfixes
 

docs/config/src/client.yml

@@ -61,6 +61,24 @@
   info_ru: |
     Повторять запросы записи, завершившиеся с ошибками нехватки места, т.е.
     ожидать, пока на OSD не освободится место.
+- name: client_wait_up_timeout
+  type: sec
+  default: 16
+  online: true
+  info: |
+    Wait for this number of seconds until PGs are up when doing operations
+    which require all PGs to be up. Currently only used by object listings
+    in delete and merge-based commands ([vitastor-cli rm](../usage/cli.en.md#rm), merge and so on).
+
+    The default value is calculated as `1 + OSD lease timeout`, which is
+    `1 + etcd_report_interval + max_etcd_attempts*2*etcd_quick_timeout`.
+  info_ru: |
+    Время ожидания поднятия PG при операциях, требующих активности всех PG.
+    В данный момент используется листингами объектов в командах, использующих
+    удаление и слияние ([vitastor-cli rm](../usage/cli.ru.md#rm), merge и подобные).
+
+    Значение по умолчанию вычисляется как `1 + время lease OSD`, равное
+    `1 + etcd_report_interval + max_etcd_attempts*2*etcd_quick_timeout`.
 - name: client_max_dirty_bytes
   type: int
   default: 33554432

docs/usage/cli.en.md

@@ -146,6 +146,7 @@ Rename, resize image or change its readonly status. Images with children can't b
 If the new size is smaller than the old size, extra data will be purged.
 You should resize file system in the image, if present, before shrinking it.
 
+* `--deleted 1|0` - Set/clear 'deleted image' flag (set automatically during unfinished deletes).
 * `-f|--force` - Proceed with shrinking or setting readwrite flag even if the image has children.
 * `--down-ok` - Proceed with shrinking even if some data will be left on unavailable OSDs.
 
@@ -221,6 +222,7 @@ Remove inode data without changing metadata.
               Requires more memory, but allows to show correct removal progress.
 --min-offset  Purge only data starting with specified offset.
 --max-offset  Purge only data before specified offset.
+--client_wait_up_timeout 16  Timeout for waiting until PGs are up in seconds.
 ```
 
 ## merge-data

docs/usage/cli.ru.md

@@ -149,6 +149,7 @@ vitastor-cli snap-create [-p|--pool <id|name>] <image>@<snapshot>
 Если новый размер меньше старого, "лишние" данные будут удалены, поэтому перед уменьшением
 образа сначала уменьшите файловую систему в нём.
 
+* `--deleted 1|0` - Установить/снять флаг "образ удалён" (устанавливается при незавершённом удалении).
 * `-f|--force` - Разрешить уменьшение или перевод в чтение-запись образа, у которого есть клоны.
 * `--down-ok` - Разрешить уменьшение, даже если часть данных останется неудалённой на недоступных OSD.
 
@@ -226,6 +227,7 @@ vitastor-cli dd [iimg=<image> | if=<file>] [oimg=<image> | of=<file>] [bs=1M] \
               Требует больше памяти, но позволяет правильно печатать прогресс удаления.
 --min-offset  Удалять только данные, начиная с заданного смещения.
 --max-offset  Удалять только данные до (исключительно) заданного смещения.
+--client_wait_up_timeout 16  Время ожидания поднятия PG в секундах.
 ```
 
 ## merge-data

mon/antietcd_adapter.js

@@ -23,6 +23,9 @@ class AntiEtcdAdapter
             }, {}));
             const cfg_port = config.antietcd_port;
             const is_local = local_ips(true).reduce((a, c) => { a[c] = true; return a; }, {});
+            is_local['0.0.0.0'] = true;
+            is_local['::'] = true;
+            is_local[''] = true;
             const selected = cluster.map(s => s.split(':', 2)).filter(ip => is_local[ip[0]] && (!cfg_port || ip[1] == cfg_port));
             if (selected.length > 1)
             {

mon/etcd_schema.js

@@ -216,6 +216,7 @@ const etcd_tree = {
                     parent_pool?: <pool_id>,
                     parent_id?: <inode_t>,
                     readonly?: boolean,
+                    deleted?: boolean,
                 }
             }
         }, */

mon/package.json

@@ -1,6 +1,6 @@
 {
   "name": "vitastor-mon",
-  "version": "1.10.0",
+  "version": "1.10.1",
   "description": "Vitastor SDS monitor service",
   "main": "mon-main.js",
   "scripts": {

opennebula/remotes/tm/vitastor/prebackup

@@ -98,8 +98,8 @@ vm.elements.each 'TEMPLATE/DISK' do |d|
     cmd = 'vitastor-cli'
     qemu_arg = ''
     if d.elements['VITASTOR_CONF']
-        cmd = cmd + ' --config_path ' + d.elements['VITASTOR_CONF']
-        qemu_arg += 'config_path='+d.elements['VITASTOR_CONF']+':'
+        cmd = cmd + ' --config_path ' + d.elements['VITASTOR_CONF'].text
+        qemu_arg += 'config_path=' + d.elements['VITASTOR_CONF'].text + ':'
     end
 
     draw = "#{bck_dir}/disk.#{did}.raw"

patches/VitastorPlugin.pm

@@ -37,16 +37,6 @@ sub run_cli
     $json = 1 if !defined $json;
     my $binary = delete $args{binary};
     $binary = '/usr/bin/vitastor-cli' if !defined $binary;
-    if (!exists($args{errfunc}))
-    {
-        $args{errfunc} = sub
-        {
-            my $line = shift;
-            print STDERR $line;
-            *STDERR->flush();
-            $stderr .= $line;
-        };
-    }
     if (!exists($args{outfunc}))
     {
         $retval = '';

patches/cinder-vitastor.py

@@ -50,7 +50,7 @@ from cinder.volume import configuration
 from cinder.volume import driver
 from cinder.volume import volume_utils
 
-VITASTOR_VERSION = '1.10.0'
+VITASTOR_VERSION = '1.10.1'
 
 LOG = logging.getLogger(__name__)
 

rpm/vitastor-el7.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        1.10.0
+Version:        1.10.1
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-1.10.0.el7.tar.gz
+Source0:        vitastor-1.10.1.el7.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

rpm/vitastor-el8.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        1.10.0
+Version:        1.10.1
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-1.10.0.el8.tar.gz
+Source0:        vitastor-1.10.1.el8.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

rpm/vitastor-el9.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        1.10.0
+Version:        1.10.1
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-1.10.0.el9.tar.gz
+Source0:        vitastor-1.10.1.el9.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

src/CMakeLists.txt

@@ -19,7 +19,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 	set(CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/${CMAKE_INSTALL_LIBDIR}")
 endif()
 
-add_definitions(-DVITASTOR_VERSION="1.10.0")
+add_definitions(-DVITASTOR_VERSION="1.10.1")
 add_definitions(-D_LARGEFILE64_SOURCE -D_FILE_OFFSET_BITS=64 -Wall -Wno-sign-compare -Wno-comment -Wno-parentheses -Wno-pointer-arith -fdiagnostics-color=always -fno-omit-frame-pointer -I ${CMAKE_SOURCE_DIR}/src)
 add_link_options(-fno-omit-frame-pointer)
 if (${WITH_ASAN})

src/client/cluster_client.cpp

@@ -6,6 +6,10 @@
 #include "cluster_client_impl.h"
 #include "json_util.h"
 
+#define TRY_SEND_OFFLINE 0
+#define TRY_SEND_CONNECTING 1
+#define TRY_SEND_OK 2
+
 cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json config)
 {
     wb = new writeback_cache_t();
@@ -67,6 +71,18 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
 
 cluster_client_t::~cluster_client_t()
 {
+    if (retry_timeout_id >= 0)
+    {
+        tfd->clear_timer(retry_timeout_id);
+        retry_timeout_duration = 0;
+        retry_timeout_id = -1;
+    }
+    if (list_retry_timeout_id >= 0)
+    {
+        tfd->clear_timer(list_retry_timeout_id);
+        list_retry_timeout_id = -1;
+        list_retry_time = {};
+    }
     msgr.repeer_pgs = [](osd_num_t){};
     if (ringloop)
     {
@@ -87,6 +103,46 @@ cluster_op_t::~cluster_op_t()
     }
 }
 
+bool cluster_op_t::support_left_on_dead()
+{
+    if (!parts.size())
+    {
+        return false;
+    }
+    for (auto & part: parts)
+    {
+        if (!(part.flags & PART_DONE) ||
+            part.op.reply.hdr.opcode != OSD_OP_DELETE ||
+            part.op.reply.hdr.retval != 0 ||
+            !(part.op.reply.del.flags & OSD_DEL_SUPPORT_LEFT_ON_DEAD))
+        {
+            return false;
+        }
+    }
+    return true;
+}
+
+std::vector<osd_num_t> cluster_op_t::get_left_on_dead()
+{
+    std::set<osd_num_t> osds;
+    for (auto & part: parts)
+    {
+        if ((part.flags & PART_DONE) ||
+            part.op.reply.hdr.opcode == OSD_OP_DELETE &&
+            part.op.reply.hdr.retval == 0 &&
+            (part.op.reply.del.flags & OSD_DEL_LEFT_ON_DEAD) != 0)
+        {
+            int del_count = (OSD_PACKET_SIZE-sizeof(part.op.reply.del)) / sizeof(uint32_t);
+            if (del_count > part.op.reply.del.left_on_dead_count)
+                del_count = part.op.reply.del.left_on_dead_count;
+            uint32_t *left_on_dead = (uint32_t*)((&part.op.reply.del) + 1);
+            for (int i = 0; i < del_count; i++)
+                osds.insert(left_on_dead[i]);
+        }
+    }
+    return std::vector<osd_num_t>(osds.begin(), osds.end());
+}
+
 void cluster_client_t::continue_raw_ops(osd_num_t peer_osd)
 {
     auto it = raw_ops.find(peer_osd);
@@ -134,12 +190,12 @@ void cluster_client_t::unshift_op(cluster_op_t *op)
 void cluster_client_t::calc_wait(cluster_op_t *op)
 {
     op->prev_wait = 0;
-    if (op->opcode == OSD_OP_WRITE)
+    if (op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE)
     {
         for (auto prev = op->prev; prev; prev = prev->prev)
         {
             if (prev->opcode == OSD_OP_SYNC ||
-                prev->opcode == OSD_OP_WRITE && !(op->flags & OP_FLUSH_BUFFER) && (prev->flags & OP_FLUSH_BUFFER))
+                (prev->opcode == OSD_OP_WRITE || prev->opcode == OSD_OP_DELETE) && !(op->flags & OP_FLUSH_BUFFER) && (prev->flags & OP_FLUSH_BUFFER))
             {
                 op->prev_wait++;
             }
@@ -151,7 +207,8 @@ void cluster_client_t::calc_wait(cluster_op_t *op)
     {
         for (auto prev = op->prev; prev; prev = prev->prev)
         {
-            if (prev->opcode == OSD_OP_SYNC || prev->opcode == OSD_OP_WRITE && (!(prev->flags & OP_IMMEDIATE_COMMIT) || enable_writeback))
+            if (prev->opcode == OSD_OP_SYNC || (prev->opcode == OSD_OP_WRITE || prev->opcode == OSD_OP_DELETE) &&
+                (!(prev->flags & OP_IMMEDIATE_COMMIT) || enable_writeback))
             {
                 op->prev_wait++;
             }
@@ -167,7 +224,7 @@ void cluster_client_t::calc_wait(cluster_op_t *op)
 
 void cluster_client_t::inc_wait(uint64_t opcode, uint64_t flags, cluster_op_t *next, int inc)
 {
-    if (opcode != OSD_OP_WRITE && opcode != OSD_OP_SYNC)
+    if (opcode != OSD_OP_WRITE && opcode != OSD_OP_DELETE && opcode != OSD_OP_SYNC)
     {
         return;
     }
@@ -176,10 +233,10 @@ void cluster_client_t::inc_wait(uint64_t opcode, uint64_t flags, cluster_op_t *n
     while (next)
     {
         auto n2 = next->next;
-        if (opcode == OSD_OP_WRITE
+        if ((opcode == OSD_OP_WRITE || opcode == OSD_OP_DELETE)
             ? (next->opcode == OSD_OP_SYNC && (!(flags & OP_IMMEDIATE_COMMIT) || enable_writeback) ||
-                next->opcode == OSD_OP_WRITE && (flags & OP_FLUSH_BUFFER) && !(next->flags & OP_FLUSH_BUFFER))
-            : (next->opcode == OSD_OP_SYNC || next->opcode == OSD_OP_WRITE))
+                (next->opcode == OSD_OP_WRITE || next->opcode == OSD_OP_DELETE) && (flags & OP_FLUSH_BUFFER) && !(next->flags & OP_FLUSH_BUFFER))
+            : (next->opcode == OSD_OP_SYNC || next->opcode == OSD_OP_WRITE || next->opcode == OSD_OP_DELETE))
         {
             next->prev_wait += inc;
             assert(next->prev_wait >= 0);
@@ -314,7 +371,7 @@ void cluster_client_t::reset_retry_timer(int new_duration)
     {
         return;
     }
-    if (retry_timeout_id)
+    if (retry_timeout_id >= 0)
     {
         tfd->clear_timer(retry_timeout_id);
     }
@@ -322,7 +379,7 @@ void cluster_client_t::reset_retry_timer(int new_duration)
     retry_timeout_id = tfd->set_timer(retry_timeout_duration, false, [this](int)
     {
         int time_passed = retry_timeout_duration;
-        retry_timeout_id = 0;
+        retry_timeout_id = -1;
         retry_timeout_duration = 0;
         continue_ops(time_passed);
     });
@@ -397,6 +454,16 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_co
     }
     // client_retry_enospc
     client_retry_enospc = config["client_retry_enospc"].is_null() ? true : config["client_retry_enospc"].bool_value();
+    // client_wait_up_timeout
+    if (!config["client_wait_up_timeout"].is_null())
+        client_wait_up_timeout = config["client_wait_up_timeout"].uint64_value();
+    else
+    {
+        auto etcd_report_interval = config["etcd_report_interval"].uint64_value();
+        if (!etcd_report_interval)
+            etcd_report_interval = 5;
+        client_wait_up_timeout = 1+etcd_report_interval+(st_cli.max_etcd_attempts*(2*st_cli.etcd_quick_timeout)+999)/1000;
+    }
     // log_level
     log_level = config["log_level"].uint64_value();
     msgr.parse_config(config);
@@ -434,7 +501,7 @@ void cluster_client_t::on_change_pool_config_hook()
             // And now they have to be resliced!
             for (auto op = op_queue_head; op; op = op->next)
             {
-                if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_READ ||
+                if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE || op->opcode == OSD_OP_READ ||
                     op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP) &&
                     INODE_POOL(op->cur_inode) == pool_item.first)
                 {
@@ -457,6 +524,7 @@ void cluster_client_t::on_change_pg_state_hook(pool_id_t pool_id, pg_num_t pg_nu
     }
     // Always continue to resume operations hung because of lack of the primary OSD
     continue_ops();
+    continue_lists();
 }
 
 bool cluster_client_t::get_immediate_commit(uint64_t inode)
@@ -477,6 +545,7 @@ void cluster_client_t::on_change_osd_state_hook(uint64_t peer_osd)
     if (msgr.wanted_peers.find(peer_osd) != msgr.wanted_peers.end())
     {
         msgr.connect_peer(peer_osd, st_cli.peer_states[peer_osd]);
+        continue_lists();
     }
 }
 
@@ -559,7 +628,8 @@ bool cluster_client_t::flush()
 void cluster_client_t::execute(cluster_op_t *op)
 {
     if (op->opcode != OSD_OP_SYNC && op->opcode != OSD_OP_READ &&
-        op->opcode != OSD_OP_READ_BITMAP && op->opcode != OSD_OP_READ_CHAIN_BITMAP && op->opcode != OSD_OP_WRITE)
+        op->opcode != OSD_OP_READ_BITMAP && op->opcode != OSD_OP_READ_CHAIN_BITMAP &&
+        op->opcode != OSD_OP_WRITE && op->opcode != OSD_OP_DELETE)
     {
         op->retval = -EINVAL;
         auto cb = std::move(op->callback);
@@ -571,7 +641,7 @@ void cluster_client_t::execute(cluster_op_t *op)
         offline_ops.push_back(op);
         return;
     }
-    op->flags = op->flags & OSD_OP_IGNORE_READONLY; // the only allowed flag
+    op->flags = op->flags & (OSD_OP_IGNORE_READONLY | OSD_OP_WAIT_UP_TIMEOUT); // allowed client flags
     execute_internal(op);
 }
 
@@ -592,7 +662,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
     {
         return;
     }
-    if (op->opcode == OSD_OP_WRITE && enable_writeback && !(op->flags & OP_FLUSH_BUFFER) &&
+    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && enable_writeback && !(op->flags & OP_FLUSH_BUFFER) &&
         !op->version /* no CAS writeback */)
     {
         if (wb->writebacks_active >= client_max_writeback_iodepth)
@@ -613,7 +683,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
         cb(op);
         return;
     }
-    if (op->opcode == OSD_OP_WRITE && !(op->flags & OP_IMMEDIATE_COMMIT))
+    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && !(op->flags & OP_IMMEDIATE_COMMIT))
     {
         if (!(op->flags & OP_FLUSH_BUFFER) && !op->version /* no CAS write-repeat */)
         {
@@ -633,7 +703,10 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
             };
             execute_internal(sync_op);
         }
-        dirty_bytes += op->len;
+        if (op->opcode != OSD_OP_DELETE)
+        {
+            dirty_bytes += op->len;
+        }
         dirty_ops++;
     }
     else if (op->opcode == OSD_OP_SYNC)
@@ -777,9 +850,48 @@ resume_1:
     {
         if (!(op->parts[i].flags & PART_SENT))
         {
-            if (!try_send(op, i))
+            int is_ok = try_send(op, i);
+            if (is_ok != TRY_SEND_OK)
             {
                 // We'll need to retry again
+                if (op->flags & OSD_OP_WAIT_UP_TIMEOUT)
+                {
+                    if (is_ok != TRY_SEND_OFFLINE)
+                    {
+                        // Reset "wait_up" timer
+                        op->wait_up_until = {};
+                    }
+                    else if (!op->wait_up_until.tv_sec && !client_wait_up_timeout)
+                    {
+                        // Don't wait for the PG to come up at all and fail
+                        op->parts[i].flags |= PART_ERROR;
+                        if (!op->retval)
+                            op->retval = -ETIMEDOUT;
+                        break;
+                    }
+                    else if (!op->wait_up_until.tv_sec)
+                    {
+                        // Set "wait_up" timer
+                        clock_gettime(CLOCK_REALTIME, &op->wait_up_until);
+                        op->wait_up_until.tv_sec += client_wait_up_timeout;
+                    }
+                    else
+                    {
+                        // Check if the timeout expired
+                        timespec tv;
+                        clock_gettime(CLOCK_REALTIME, &tv);
+                        if (tv.tv_sec > op->wait_up_until.tv_sec ||
+                            tv.tv_sec == op->wait_up_until.tv_sec &&
+                            tv.tv_nsec > op->wait_up_until.tv_nsec)
+                        {
+                            // Fail
+                            op->parts[i].flags |= PART_ERROR;
+                            if (!op->retval)
+                                op->retval = -ETIMEDOUT;
+                            break;
+                        }
+                    }
+                }
                 if (op->parts[i].flags & PART_RETRY)
                 {
                     op->retry_after = client_retry_interval;
@@ -1053,7 +1165,7 @@ bool cluster_client_t::affects_osd(uint64_t inode, uint64_t offset, uint64_t len
     return false;
 }
 
-bool cluster_client_t::try_send(cluster_op_t *op, int i)
+int cluster_client_t::try_send(cluster_op_t *op, int i)
 {
     if (!msgr_initialized)
     {
@@ -1079,7 +1191,7 @@ bool cluster_client_t::try_send(cluster_op_t *op, int i)
             uint64_t meta_rev = 0;
             if (op->opcode != OSD_OP_READ_BITMAP && op->opcode != OSD_OP_DELETE)
             {
-                auto ino_it = st_cli.inode_config.find(op->inode);
+                auto ino_it = st_cli.inode_config.find(op->cur_inode);
                 if (ino_it != st_cli.inode_config.end())
                     meta_rev = ino_it->second.mod_revision;
             }
@@ -1109,14 +1221,15 @@ bool cluster_client_t::try_send(cluster_op_t *op, int i)
             };
             part->op.iov = part->iov;
             msgr.outbox_push(&part->op);
-            return true;
+            return TRY_SEND_OK;
         }
         else if (msgr.wanted_peers.find(primary_osd) == msgr.wanted_peers.end())
         {
             msgr.connect_peer(primary_osd, st_cli.peer_states[primary_osd]);
+            return TRY_SEND_CONNECTING;
         }
     }
-    return false;
+    return TRY_SEND_OFFLINE;
 }
 
 int cluster_client_t::continue_sync(cluster_op_t *op)
@@ -1188,13 +1301,12 @@ resume_1:
 
 void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part)
 {
-    auto peer_it = msgr.osd_peer_fds.find(part->osd_num);
-    assert(peer_it != msgr.osd_peer_fds.end());
+    auto peer_fd = msgr.osd_peer_fds.at(part->osd_num);
     part->flags |= PART_SENT;
     op->inflight_count++;
     part->op = (osd_op_t){
         .op_type = OSD_OP_OUT,
-        .peer_fd = peer_it->second,
+        .peer_fd = peer_fd,
         .req = {
             .hdr = {
                 .magic = SECONDARY_OSD_OP_MAGIC,
@@ -1228,9 +1340,11 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
     {
         // Operation failed, retry
         part->flags |= PART_ERROR;
-        if (!op->retval || op->retval == -EPIPE || part->op.reply.hdr.retval == -EIO)
+        if (!op->retval || op->retval == -EPIPE ||
+            part->op.reply.hdr.retval == -ENOSPC && op->retval == -ETIMEDOUT ||
+            part->op.reply.hdr.retval == -EIO)
         {
-            // Error priority: EIO > ENOSPC > EPIPE
+            // Error priority: EIO > ENOSPC > ETIMEDOUT > EPIPE
             op->retval = part->op.reply.hdr.retval;
         }
         int stop_fd = -1;
@@ -1293,7 +1407,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
                 op->version = op->parts.size() == 1 ? part->op.reply.rw.version : 0;
             }
         }
-        else if (op->opcode == OSD_OP_WRITE)
+        else if (op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE)
         {
             op->version = op->parts.size() == 1 ? part->op.reply.rw.version : 0;
         }

src/client/cluster_client.h

@@ -11,12 +11,11 @@
 #define DEFAULT_CLIENT_MAX_BUFFERED_BYTES 32*1024*1024
 #define DEFAULT_CLIENT_MAX_BUFFERED_OPS 1024
 #define DEFAULT_CLIENT_MAX_WRITEBACK_IODEPTH 256
-#define INODE_LIST_DONE 1
-#define INODE_LIST_HAS_UNSTABLE 2
 #define OSD_OP_READ_BITMAP OSD_OP_SEC_READ_BMP
 #define OSD_OP_READ_CHAIN_BITMAP 0x102
 
 #define OSD_OP_IGNORE_READONLY 0x08
+#define OSD_OP_WAIT_UP_TIMEOUT 0x10
 
 struct cluster_op_t;
 
@@ -41,7 +40,8 @@ struct cluster_op_t
     // for reads and writes within a single object (stripe),
     // reads can return current version and writes can use "CAS" semantics
     uint64_t version = 0;
-    // now only OSD_OP_IGNORE_READONLY is supported
+    // flags: OSD_OP_IGNORE_READONLY - ignore inode readonly flag
+    // OSD_OP_WAIT_UP_TIMEOUT - do not retry the operation infinitely if PG is inactive, only for for <wait_up_timeout>
     uint64_t flags = 0;
     // negative retval is an error number
     // write and read return len on success
@@ -53,12 +53,17 @@ struct cluster_op_t
     void *bitmap_buf = NULL;
     std::function<void(cluster_op_t*)> callback;
     ~cluster_op_t();
+
+    // for deletions, remove after 'atomic delete':
+    bool support_left_on_dead();
+    std::vector<osd_num_t> get_left_on_dead();
 protected:
     int state = 0;
     uint64_t cur_inode; // for snapshot reads
     bool needs_reslice = false;
     int retry_after = 0;
     int inflight_count = 0, done_count = 0;
+    timespec wait_up_until = {};
     std::vector<cluster_op_part_t> parts;
     void *part_bitmaps = NULL;
     unsigned bitmap_buf_size = 0;
@@ -71,6 +76,7 @@ protected:
 
 struct inode_list_t;
 struct inode_list_osd_t;
+struct inode_list_pg_t;
 class writeback_cache_t;
 
 // FIXME: Split into public and private interfaces
@@ -95,8 +101,9 @@ class cluster_client_t
     int client_retry_interval = 50; // ms
     int client_eio_retry_interval = 1000; // ms
     bool client_retry_enospc = true;
+    int client_wait_up_timeout = 16; // sec (for listings)
 
-    int retry_timeout_id = 0;
+    int retry_timeout_id = -1;
     int retry_timeout_duration = 0;
     std::vector<cluster_op_t*> offline_ops;
     cluster_op_t *op_queue_head = NULL, *op_queue_tail = NULL;
@@ -110,6 +117,8 @@ class cluster_client_t
     bool pgs_loaded = false;
     ring_consumer_t consumer;
     std::vector<std::function<void(void)>> on_ready_hooks;
+    int list_retry_timeout_id = -1;
+    timespec list_retry_time;
     std::vector<inode_list_t*> lists;
     std::multimap<osd_num_t, osd_op_t*> raw_ops;
     int continuing_ops = 0;
@@ -135,11 +144,10 @@ public:
     bool get_immediate_commit(uint64_t inode);
 
     void continue_ops(int time_passed = 0);
-    inode_list_t *list_inode_start(inode_t inode,
-        std::function<void(inode_list_t* lst, std::set<object_id>&& objects, pg_num_t pg_num, osd_num_t primary_osd, int status)> callback);
-    int list_pg_count(inode_list_t *lst);
-    const std::vector<osd_num_t> & list_inode_get_inactive_osds(inode_list_t *lst);
-    void list_inode_next(inode_list_t *lst, int next_pgs);
+
+    void list_inode(inode_t inode, uint64_t min_offset, uint64_t max_offset, int max_parallel_pgs, std::function<void(
+        int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects)> pg_callback);
+
     //inline uint32_t get_bs_bitmap_granularity() { return st_cli.global_bitmap_granularity; }
     //inline uint64_t get_bs_block_size() { return st_cli.global_block_size; }
     uint64_t next_op_id();
@@ -158,7 +166,7 @@ protected:
     bool check_rw(cluster_op_t *op);
     void slice_rw(cluster_op_t *op);
     void reset_retry_timer(int new_duration);
-    bool try_send(cluster_op_t *op, int i);
+    int try_send(cluster_op_t *op, int i);
     int continue_sync(cluster_op_t *op);
     void send_sync(cluster_op_t *op, cluster_op_part_t *part);
     void handle_op_part(cluster_op_part_t *part);
@@ -167,8 +175,14 @@ protected:
     void calc_wait(cluster_op_t *op);
     void inc_wait(uint64_t opcode, uint64_t flags, cluster_op_t *next, int inc);
     void continue_lists();
-    void continue_listing(inode_list_t *lst);
+    bool continue_listing(inode_list_t *lst);
+    bool restart_listing(inode_list_t* lst);
+    void retry_start_pg_listing(inode_list_pg_t *pg);
+    int start_pg_listing(inode_list_pg_t *pg);
     void send_list(inode_list_osd_t *cur_list);
+    void set_list_retry_timeout(int ms, timespec new_time);
+    void finish_list_pg(inode_list_pg_t *pg, bool retry_epipe);
+    bool check_finish_listing(inode_list_t *lst);
     void continue_raw_ops(osd_num_t peer_osd);
 
     friend class writeback_cache_t;

src/client/cluster_client_list.cpp

@@ -2,9 +2,17 @@
 // License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
 
 #include <algorithm>
+#include "assert.h"
 #include "pg_states.h"
 #include "cluster_client.h"
 
+#define LIST_PG_INIT 0
+#define LIST_PG_WAIT_ACTIVE 1
+#define LIST_PG_WAIT_CONNECT 2
+#define LIST_PG_WAIT_RETRY 3
+#define LIST_PG_SENT 4
+#define LIST_PG_DONE 5
+
 struct inode_list_t;
 
 struct inode_list_pg_t;
@@ -13,20 +21,22 @@ struct inode_list_osd_t
 {
     inode_list_pg_t *pg = NULL;
     osd_num_t osd_num = 0;
-    bool sent = false;
 };
 
 struct inode_list_pg_t
 {
     inode_list_t *lst = NULL;
-    int pos = 0;
-    pg_num_t pg_num;
-    osd_num_t cur_primary;
-    bool has_unstable = false;
-    int sent = 0;
-    int done = 0;
+    int errcode = 0;
+    pg_num_t pg_num = 0;
+    osd_num_t cur_primary = 0;
+    int state = 0;
+    int inflight_ops = 0;
+    timespec wait_until;
     std::vector<inode_list_osd_t> list_osds;
+
+    bool has_unstable = false;
     std::set<object_id> objects;
+    std::vector<osd_num_t> inactive_osds;
 };
 
 struct inode_list_t
@@ -34,175 +44,295 @@ struct inode_list_t
     cluster_client_t *cli = NULL;
     pool_id_t pool_id = 0;
     inode_t inode = 0;
+    uint64_t min_offset = 0;
+    uint64_t max_offset = 0;
+    int max_parallel_pgs = 16;
+
+    bool fallback = false;
+    int inflight_pgs = 0;
+    std::map<osd_num_t, int> inflight_per_osd;
     int done_pgs = 0;
-    int want = 0;
-    std::vector<osd_num_t> inactive_osds;
+    int onstack = 0;
     std::vector<inode_list_pg_t*> pgs;
-    std::function<void(inode_list_t* lst, std::set<object_id>&& objects, pg_num_t pg_num, osd_num_t primary_osd, int status)> callback;
+    pg_num_t real_pg_count = 0;
+    std::function<void(int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects)> callback;
 };
 
-inode_list_t* cluster_client_t::list_inode_start(inode_t inode,
-    std::function<void(inode_list_t* lst, std::set<object_id>&& objects, pg_num_t pg_num, osd_num_t primary_osd, int status)> callback)
+void cluster_client_t::list_inode(inode_t inode, uint64_t min_offset, uint64_t max_offset, int max_parallel_pgs, std::function<void(
+    int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects)> pg_callback)
 {
     init_msgr();
-    int skipped_pgs = 0;
     pool_id_t pool_id = INODE_POOL(inode);
     if (!pool_id || st_cli.pool_config.find(pool_id) == st_cli.pool_config.end())
     {
         if (log_level > 0)
-        {
             fprintf(stderr, "Pool %u does not exist\n", pool_id);
-        }
-        return NULL;
+        pg_callback(-EINVAL, 0, 0, std::set<object_id>());
+        return;
     }
+    auto pg_stripe_size = st_cli.pool_config.at(pool_id).pg_stripe_size;
+    if (min_offset)
+        min_offset = (min_offset/pg_stripe_size) * pg_stripe_size;
     inode_list_t *lst = new inode_list_t();
     lst->cli = this;
     lst->pool_id = pool_id;
     lst->inode = inode;
-    lst->callback = callback;
-    auto pool_cfg = st_cli.pool_config[pool_id];
-    std::set<osd_num_t> inactive_osd_set;
-    for (auto & pg_item: pool_cfg.pg_config)
-    {
-        auto & pg = pg_item.second;
-        if (pg.pause || !pg.cur_primary || !(pg.cur_state & PG_ACTIVE))
-        {
-            skipped_pgs++;
-            if (log_level > 0)
-            {
-                fprintf(stderr, "PG %u is inactive, skipping\n", pg_item.first);
-            }
-            continue;
-        }
-        inode_list_pg_t *r = new inode_list_pg_t();
-        r->lst = lst;
-        r->pg_num = pg_item.first;
-        r->cur_primary = pg.cur_primary;
-        if (pg.cur_state != PG_ACTIVE)
-        {
-            // Not clean
-            std::set<osd_num_t> all_peers;
-            for (osd_num_t pg_osd: pg.target_set)
-            {
-                if (pg_osd != 0)
-                {
-                    all_peers.insert(pg_osd);
-                }
-            }
-            for (osd_num_t pg_osd: pg.all_peers)
-            {
-                if (pg_osd != 0)
-                {
-                    all_peers.insert(pg_osd);
-                }
-            }
-            for (auto & hist_item: pg.target_history)
-            {
-                for (auto pg_osd: hist_item)
-                {
-                    if (pg_osd != 0)
-                    {
-                        all_peers.insert(pg_osd);
-                    }
-                }
-            }
-            for (osd_num_t peer_osd: all_peers)
-            {
-                if (st_cli.peer_states.find(peer_osd) != st_cli.peer_states.end())
-                {
-                    r->list_osds.push_back((inode_list_osd_t){
-                        .pg = r,
-                        .osd_num = peer_osd,
-                        .sent = false,
-                    });
-                }
-                else
-                {
-                    inactive_osd_set.insert(peer_osd);
-                }
-            }
-        }
-        else
-        {
-            // Clean
-            r->list_osds.push_back((inode_list_osd_t){
-                .pg = r,
-                .osd_num = pg.cur_primary,
-                .sent = false,
-            });
-        }
-        lst->pgs.push_back(r);
-    }
-    std::sort(lst->pgs.begin(), lst->pgs.end(), [](inode_list_pg_t *a, inode_list_pg_t *b)
-    {
-        return a->cur_primary < b->cur_primary ? true : false;
-    });
-    for (int i = 0; i < lst->pgs.size(); i++)
-    {
-        lst->pgs[i]->pos = i;
-    }
-    lst->inactive_osds.insert(lst->inactive_osds.end(), inactive_osd_set.begin(), inactive_osd_set.end());
+    lst->min_offset = min_offset;
+    lst->max_offset = max_offset;
+    lst->callback = pg_callback;
+    lst->max_parallel_pgs = max_parallel_pgs <= 0 ? 16 : max_parallel_pgs;
     lists.push_back(lst);
-    return lst;
-}
-
-int cluster_client_t::list_pg_count(inode_list_t *lst)
-{
-    return lst->pgs.size();
-}
-
-const std::vector<osd_num_t> & cluster_client_t::list_inode_get_inactive_osds(inode_list_t *lst)
-{
-    return lst->inactive_osds;
-}
-
-void cluster_client_t::list_inode_next(inode_list_t *lst, int next_pgs)
-{
-    if (next_pgs >= 0)
-    {
-        lst->want += next_pgs;
-    }
     continue_listing(lst);
 }
 
-void cluster_client_t::continue_listing(inode_list_t *lst)
+bool cluster_client_t::continue_listing(inode_list_t *lst)
 {
-    if (lst->done_pgs >= lst->pgs.size())
+    if (lst->onstack > 0)
     {
-        return;
+        return true;
     }
-    if (lst->want <= 0)
+    lst->onstack++;
+    if (restart_listing(lst))
     {
-        return;
-    }
-    for (int i = 0; i < lst->pgs.size(); i++)
-    {
-        if (lst->pgs[i] && lst->pgs[i]->sent < lst->pgs[i]->list_osds.size())
+        for (int i = 0; i < lst->pgs.size() && lst->inflight_pgs < lst->max_parallel_pgs; i++)
         {
-            for (int j = 0; j < lst->pgs[i]->list_osds.size(); j++)
+            retry_start_pg_listing(lst->pgs[i]);
+        }
+    }
+    if (check_finish_listing(lst))
+    {
+        // Do not change lst->onstack because it's already freed
+        return false;
+    }
+    lst->onstack--;
+    return true;
+}
+
+bool cluster_client_t::restart_listing(inode_list_t* lst)
+{
+    auto pool_it = st_cli.pool_config.find(lst->pool_id);
+    // We want listing to be consistent. To achieve it we should:
+    // 1) retry listing of each PG if its state changes
+    // 2) abort listing if PG count changes during listing
+    // 3) ideally, only talk to the primary OSD - this will be done separately
+    // So first we add all PGs without checking their state
+    if (pool_it == st_cli.pool_config.end() ||
+        lst->real_pg_count != pool_it->second.real_pg_count)
+    {
+        for (auto pg: lst->pgs)
+        {
+            if (pg->inflight_ops > 0)
             {
-                send_list(&lst->pgs[i]->list_osds[j]);
-                if (lst->want <= 0)
-                {
-                    return;
-                }
+                // Wait until all in-progress listings complete or fail
+                return false;
+            }
+        }
+        for (auto pg: lst->pgs)
+        {
+            delete pg;
+        }
+        if (log_level > 0 && lst->real_pg_count)
+        {
+            fprintf(stderr, "PG count in pool %u changed during listing\n", lst->pool_id);
+        }
+        lst->pgs.clear();
+        if (pool_it == st_cli.pool_config.end())
+        {
+            // Unknown pool
+            lst->callback(-EINVAL, 0, 0, std::set<object_id>());
+            return false;
+        }
+        else if (lst->done_pgs)
+        {
+            // PG count changed during listing, it should fail
+            lst->callback(-EAGAIN, 0, 0, std::set<object_id>());
+            return false;
+        }
+        else
+        {
+            lst->real_pg_count = pool_it->second.real_pg_count;
+            for (pg_num_t pg_num = 1; pg_num <= lst->real_pg_count; pg_num++)
+            {
+                inode_list_pg_t *pg = new inode_list_pg_t();
+                pg->lst = lst;
+                pg->pg_num = pg_num;
+                lst->pgs.push_back(pg);
             }
         }
     }
+    return true;
+}
+
+void cluster_client_t::retry_start_pg_listing(inode_list_pg_t *pg)
+{
+    if (pg->state == LIST_PG_SENT || pg->state == LIST_PG_DONE)
+    {
+        return;
+    }
+    if (pg->state == LIST_PG_WAIT_RETRY)
+    {
+        // Check if the timeout expired
+        timespec tv;
+        clock_gettime(CLOCK_REALTIME, &tv);
+        if (tv.tv_sec < pg->wait_until.tv_sec ||
+            tv.tv_sec == pg->wait_until.tv_sec && tv.tv_nsec < pg->wait_until.tv_nsec)
+        {
+            return;
+        }
+    }
+    int new_st = start_pg_listing(pg);
+    if (new_st == LIST_PG_SENT || new_st == LIST_PG_WAIT_CONNECT)
+    {
+        // sent => wait for completion
+        // not connected, but OSD state exists => wait for PG or OSD state change infinitely
+        pg->state = new_st;
+        return;
+    }
+    if (new_st == LIST_PG_WAIT_ACTIVE && pg->state != LIST_PG_WAIT_ACTIVE)
+    {
+        if (!client_wait_up_timeout)
+        {
+            fprintf(stderr, "PG %u/%u is inactive, skipping listing\n", pg->lst->pool_id, pg->pg_num);
+            pg->errcode = -EPIPE;
+            pg->list_osds.clear();
+            pg->objects.clear();
+            finish_list_pg(pg, false);
+            return;
+        }
+        pg->state = new_st;
+        clock_gettime(CLOCK_REALTIME, &pg->wait_until);
+        pg->wait_until.tv_sec += client_wait_up_timeout;
+        if (log_level > 1)
+        {
+            fprintf(stderr, "Waiting for PG %u/%u to become active for %d seconds\n", pg->lst->pool_id, pg->pg_num, client_wait_up_timeout);
+        }
+        set_list_retry_timeout(client_wait_up_timeout*1000, pg->wait_until);
+        return;
+    }
+    assert(pg->state == LIST_PG_WAIT_ACTIVE);
+    // Check if the timeout expired
+    timespec tv;
+    clock_gettime(CLOCK_REALTIME, &tv);
+    if (tv.tv_sec > pg->wait_until.tv_sec ||
+        tv.tv_sec == pg->wait_until.tv_sec && tv.tv_nsec >= pg->wait_until.tv_nsec)
+    {
+        fprintf(stderr, "Failed to wait for PG %u/%u to become active, skipping listing\n", pg->lst->pool_id, pg->pg_num);
+        pg->errcode = -EPIPE;
+        pg->list_osds.clear();
+        pg->objects.clear();
+        finish_list_pg(pg, false);
+    }
+}
+
+void cluster_client_t::set_list_retry_timeout(int ms, timespec new_time)
+{
+    if (!list_retry_time.tv_sec || list_retry_time.tv_sec > new_time.tv_sec ||
+        list_retry_time.tv_sec == new_time.tv_sec && list_retry_time.tv_nsec > new_time.tv_nsec)
+    {
+        list_retry_time = new_time;
+        if (list_retry_timeout_id >= 0)
+        {
+            tfd->clear_timer(list_retry_timeout_id);
+        }
+        list_retry_timeout_id = tfd->set_timer(ms, false, [this](int timer_id)
+        {
+            list_retry_timeout_id = -1;
+            list_retry_time = {};
+            continue_lists();
+        });
+    }
+}
+
+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 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() ||
+        pg_it->second.pause ||
+        !pg_it->second.cur_primary ||
+        !(pg_it->second.cur_state & PG_ACTIVE))
+    {
+        // PG is (temporarily?) unavailable
+        return LIST_PG_WAIT_ACTIVE;
+    }
+    pg->inactive_osds.clear();
+    std::set<osd_num_t> all_peers;
+    if (pg_it->second.cur_state != PG_ACTIVE && pg->lst->fallback)
+    {
+        // Not clean and OSDs don't support listing from primary
+        for (osd_num_t pg_osd: pg_it->second.target_set)
+            all_peers.insert(pg_osd);
+        for (osd_num_t pg_osd: pg_it->second.all_peers)
+            all_peers.insert(pg_osd);
+        for (auto & hist_item: pg_it->second.target_history)
+            for (auto pg_osd: hist_item)
+                all_peers.insert(pg_osd);
+        // Remove zero OSD number
+        all_peers.erase(0);
+        // Remove unconnectable peers except cur_primary
+        for (auto peer_it = all_peers.begin(); peer_it != all_peers.end(); )
+        {
+            if (*peer_it != pg_it->second.cur_primary &&
+                st_cli.peer_states[*peer_it].is_null())
+            {
+                pg->inactive_osds.push_back(*peer_it);
+                all_peers.erase(peer_it++);
+            }
+            else
+                peer_it++;
+        }
+    }
+    else
+    {
+        // Clean
+        all_peers.insert(pg_it->second.cur_primary);
+    }
+    // Check that we're connected to all PG OSDs
+    bool conn = true;
+    for (osd_num_t peer_osd: all_peers)
+    {
+        if (msgr.osd_peer_fds.find(peer_osd) == msgr.osd_peer_fds.end())
+        {
+            // Initiate connection
+            if (st_cli.peer_states[peer_osd].is_null())
+            {
+                return LIST_PG_WAIT_ACTIVE;
+            }
+            msgr.connect_peer(peer_osd, st_cli.peer_states[peer_osd]);
+            conn = false;
+        }
+    }
+    if (!conn)
+    {
+        return LIST_PG_WAIT_CONNECT;
+    }
+    // Send all listings at once as the simplest way to guarantee that we connect
+    // to the exact same OSDs that are listed in PG state
+    pg->errcode = 0;
+    pg->list_osds.clear();
+    pg->has_unstable = false;
+    pg->objects.clear();
+    pg->cur_primary = pg_it->second.cur_primary;
+    for (osd_num_t peer_osd: all_peers)
+    {
+        pg->list_osds.push_back((inode_list_osd_t){
+            .pg = pg,
+            .osd_num = peer_osd,
+        });
+    }
+    for (auto & list_osd: pg->list_osds)
+    {
+        send_list(&list_osd);
+    }
+    return LIST_PG_SENT;
 }
 
 void cluster_client_t::send_list(inode_list_osd_t *cur_list)
 {
-    if (cur_list->sent)
-    {
-        return;
-    }
-    if (msgr.osd_peer_fds.find(cur_list->osd_num) == msgr.osd_peer_fds.end())
-    {
-        // Initiate connection
-        msgr.connect_peer(cur_list->osd_num, st_cli.peer_states[cur_list->osd_num]);
-        return;
-    }
+    if (!cur_list->pg->inflight_ops)
+        cur_list->pg->lst->inflight_pgs++;
+    cur_list->pg->inflight_ops++;
     auto & pool_cfg = st_cli.pool_config[cur_list->pg->lst->pool_id];
     osd_op_t *op = new osd_op_t();
     op->op_type = OSD_OP_OUT;
@@ -220,6 +350,9 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
             .pg_stripe_size = pool_cfg.pg_stripe_size,
             .min_inode = cur_list->pg->lst->inode,
             .max_inode = cur_list->pg->lst->inode,
+            .min_stripe = cur_list->pg->lst->min_offset,
+            .max_stripe = cur_list->pg->lst->max_offset,
+            .flags = (uint64_t)(cur_list->pg->lst->fallback ? 0 : OSD_LIST_PRIMARY),
         },
     };
     op->callback = [this, cur_list](osd_op_t *op)
@@ -228,6 +361,29 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
         {
             fprintf(stderr, "Failed to get PG %u/%u object list from OSD %ju (retval=%jd), skipping\n",
                 cur_list->pg->lst->pool_id, cur_list->pg->pg_num, cur_list->osd_num, op->reply.hdr.retval);
+            if (!cur_list->pg->errcode ||
+                cur_list->pg->errcode == -EPIPE ||
+                op->reply.hdr.retval != -EPIPE)
+            {
+                cur_list->pg->errcode = op->reply.hdr.retval;
+            }
+        }
+        else if ((op->req.sec_list.flags & OSD_LIST_PRIMARY) &&
+            !(op->reply.sec_list.flags & OSD_LIST_PRIMARY))
+        {
+            // OSD is old and doesn't support listing from primary
+            if (log_level > 0)
+            {
+                fprintf(
+                    stderr, "[PG %u/%u] Primary OSD doesn't support consistent listings, falling back to listings from all peers\n",
+                    cur_list->pg->lst->pool_id, cur_list->pg->pg_num
+                );
+            }
+            cur_list->pg->lst->fallback = true;
+            if (!cur_list->pg->errcode)
+            {
+                cur_list->pg->errcode = -EPIPE;
+            }
         }
         else
         {
@@ -256,55 +412,66 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
             }
         }
         delete op;
-        auto lst = cur_list->pg->lst;
-        auto pg = cur_list->pg;
-        pg->done++;
-        if (pg->done >= pg->list_osds.size())
-        {
-            int status = 0;
-            lst->done_pgs++;
-            if (lst->done_pgs >= lst->pgs.size())
-            {
-                status |= INODE_LIST_DONE;
-            }
-            if (pg->has_unstable)
-            {
-                status |= INODE_LIST_HAS_UNSTABLE;
-            }
-            lst->callback(lst, std::move(pg->objects), pg->pg_num, pg->cur_primary, status);
-            lst->pgs[pg->pos] = NULL;
-            delete pg;
-            if (lst->done_pgs >= lst->pgs.size())
-            {
-                // All done
-                for (int i = 0; i < lists.size(); i++)
-                {
-                    if (lists[i] == lst)
-                    {
-                        lists.erase(lists.begin()+i, lists.begin()+i+1);
-                        break;
-                    }
-                }
-                delete lst;
-                return;
-            }
-        }
-        else
-        {
-            lst->want++;
-        }
-        continue_listing(lst);
+        cur_list->pg->inflight_ops--;
+        if (!cur_list->pg->inflight_ops)
+            cur_list->pg->lst->inflight_pgs--;
+        finish_list_pg(cur_list->pg, true);
+        continue_listing(cur_list->pg->lst);
     };
     msgr.outbox_push(op);
-    cur_list->sent = true;
-    cur_list->pg->sent++;
-    cur_list->pg->lst->want--;
+}
+
+void cluster_client_t::finish_list_pg(inode_list_pg_t *pg, bool retry_epipe)
+{
+    auto lst = pg->lst;
+    if (pg->inflight_ops == 0)
+    {
+        if (pg->errcode == -EPIPE && retry_epipe)
+        {
+            // Retry listing after <client_retry_interval> ms on EPIPE
+            pg->state = LIST_PG_WAIT_RETRY;
+            clock_gettime(CLOCK_REALTIME, &pg->wait_until);
+            pg->wait_until.tv_nsec += client_retry_interval*1000000;
+            pg->wait_until.tv_sec += (pg->wait_until.tv_nsec / 1000000000);
+            pg->wait_until.tv_nsec = (pg->wait_until.tv_nsec % 1000000000);
+            set_list_retry_timeout(client_retry_interval, pg->wait_until);
+            return;
+        }
+        lst->done_pgs++;
+        pg->state = LIST_PG_DONE;
+        lst->callback(pg->errcode, lst->pgs.size()-lst->done_pgs, pg->pg_num, std::move(pg->objects));
+        pg->objects.clear();
+        pg->inactive_osds.clear();
+    }
 }
 
 void cluster_client_t::continue_lists()
 {
-    for (auto lst: lists)
+    for (int i = lists.size()-1; i >= 0; i--)
     {
-        continue_listing(lst);
+        continue_listing(lists[i]);
     }
 }
+
+bool cluster_client_t::check_finish_listing(inode_list_t *lst)
+{
+    if (lst->done_pgs >= lst->pgs.size())
+    {
+        for (auto pg: lst->pgs)
+        {
+            delete pg;
+        }
+        lst->pgs.clear();
+        for (int i = 0; i < lists.size(); i++)
+        {
+            if (lists[i] == lst)
+            {
+                lists.erase(lists.begin()+i, lists.begin()+i+1);
+                break;
+            }
+        }
+        delete lst;
+        return true;
+    }
+    return false;
+}

src/client/cluster_client_wb.cpp

@@ -43,6 +43,7 @@ bool writeback_cache_t::is_left_merged(dirty_buf_it_t dirty_it)
         auto prev_it = dirty_it;
         prev_it--;
         if (prev_it->first.inode == dirty_it->first.inode &&
+            (prev_it->second.buf != NULL) == (dirty_it->second.buf != NULL) &&
             prev_it->first.stripe+prev_it->second.len == dirty_it->first.stripe &&
             prev_it->second.state == CACHE_DIRTY)
         {
@@ -58,6 +59,7 @@ bool writeback_cache_t::is_right_merged(dirty_buf_it_t dirty_it)
     next_it++;
     if (next_it != dirty_buffers.end() &&
         next_it->first.inode == dirty_it->first.inode &&
+        (next_it->second.buf != NULL) == (dirty_it->second.buf != NULL) &&
         next_it->first.stripe == dirty_it->first.stripe+dirty_it->second.len &&
         next_it->second.state == CACHE_DIRTY)
     {
@@ -99,7 +101,7 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
                     .inode = op->inode,
                     .stripe = new_end,
                 }, (cluster_buffer_t){
-                    .buf = dirty_it->second.buf + new_end - dirty_it->first.stripe,
+                    .buf = dirty_it->second.buf ? dirty_it->second.buf + new_end - dirty_it->first.stripe : NULL,
                     .len = old_end - new_end,
                     .state = dirty_it->second.state,
                     .flush_id = dirty_it->second.flush_id,
@@ -143,7 +145,7 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
                 .inode = op->inode,
                 .stripe = new_end,
             }, (cluster_buffer_t){
-                .buf = dirty_it->second.buf + new_end - dirty_it->first.stripe,
+                .buf = dirty_it->second.buf ? dirty_it->second.buf + new_end - dirty_it->first.stripe : NULL,
                 .len = old_end - new_end,
                 .state = dirty_it->second.state,
                 .flush_id = dirty_it->second.flush_id,
@@ -170,8 +172,9 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
         }
     }
     // Overlapping buffers are removed, just insert the new one
-    uint64_t *refcnt = (uint64_t*)malloc_or_die(sizeof(uint64_t) + op->len);
-    uint8_t *buf = (uint8_t*)refcnt + sizeof(uint64_t);
+    bool is_del = op->opcode == OSD_OP_DELETE;
+    uint64_t *refcnt = is_del ? NULL : (uint64_t*)malloc_or_die(sizeof(uint64_t) + op->len);
+    uint8_t *buf = is_del ? NULL : ((uint8_t*)refcnt + sizeof(uint64_t));
     *refcnt = 1;
     dirty_it = dirty_buffers.emplace_hint(dirty_it, (object_id){
         .inode = op->inode,
@@ -185,7 +188,7 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
     });
     if (state == CACHE_DIRTY)
     {
-        writeback_bytes += op->len;
+        writeback_bytes += is_del ? 0 : op->len;
         // Track consecutive write-back operations
         if (!is_merged(dirty_it))
         {
@@ -199,13 +202,16 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
             });
         }
     }
-    uint64_t pos = 0, len = op->len, iov_idx = 0;
-    while (len > 0 && iov_idx < op->iov.count)
+    if (!is_del)
     {
-        auto & iov = op->iov.buf[iov_idx];
-        memcpy(buf + pos, iov.iov_base, iov.iov_len);
-        pos += iov.iov_len;
-        iov_idx++;
+        uint64_t pos = 0, len = op->len, iov_idx = 0;
+        while (len > 0 && iov_idx < op->iov.count)
+        {
+            auto & iov = op->iov.buf[iov_idx];
+            memcpy(buf + pos, iov.iov_base, iov.iov_len);
+            pos += iov.iov_len;
+            iov_idx++;
+        }
     }
 }
 
@@ -250,7 +256,7 @@ void writeback_cache_t::flush_buffers(cluster_client_t *cli, dirty_buf_it_t from
     bool is_writeback = from_it->second.state == CACHE_DIRTY;
     cluster_op_t *op = new cluster_op_t;
     op->flags = OSD_OP_IGNORE_READONLY|OP_FLUSH_BUFFER;
-    op->opcode = OSD_OP_WRITE;
+    op->opcode = from_it->second.buf ? OSD_OP_WRITE : OSD_OP_DELETE;
     op->cur_inode = op->inode = from_it->first.inode;
     op->offset = from_it->first.stripe;
     op->len = prev_it->first.stripe + prev_it->second.len - from_it->first.stripe;
@@ -260,9 +266,12 @@ void writeback_cache_t::flush_buffers(cluster_client_t *cli, dirty_buf_it_t from
     {
         it->second.state = CACHE_REPEATING;
         it->second.flush_id = flush_id;
-        (*it->second.refcnt)++;
-        flushed_buffers.emplace(flush_id, it->second.refcnt);
-        op->iov.push_back(it->second.buf, it->second.len);
+        if (it->second.buf)
+        {
+            (*it->second.refcnt)++;
+            flushed_buffers.emplace(flush_id, it->second.refcnt);
+            op->iov.push_back(it->second.buf, it->second.len);
+        }
         calc_len += it->second.len;
     }
     assert(calc_len == op->len);
@@ -334,10 +343,12 @@ void writeback_cache_t::start_writebacks(cluster_client_t *cli, int count)
         }
         auto from_it = dirty_it;
         uint64_t off = dirty_it->first.stripe;
+        bool is_del = (dirty_it->second.buf == NULL);
         while (from_it != dirty_buffers.begin())
         {
             from_it--;
             if (from_it->second.state != CACHE_DIRTY ||
+                (from_it->second.buf == NULL) != is_del ||
                 from_it->first.inode != req.inode ||
                 from_it->first.stripe+from_it->second.len != off)
             {
@@ -352,6 +363,7 @@ void writeback_cache_t::start_writebacks(cluster_client_t *cli, int count)
         while (to_it != dirty_buffers.end())
         {
             if (to_it->second.state != CACHE_DIRTY ||
+                (to_it->second.buf == NULL) != is_del ||
                 to_it->first.inode != req.inode ||
                 to_it->first.stripe != off)
             {
@@ -391,15 +403,27 @@ static void copy_to_op(cluster_op_t *op, uint64_t offset, uint8_t *buf, uint64_t
             auto & v = op->iov.buf[iov_idx];
             auto begin = (cur_offset < offset ? offset : cur_offset);
             auto end = (cur_offset+v.iov_len > offset+len ? offset+len : cur_offset+v.iov_len);
-            memcpy(
-                (uint8_t*)v.iov_base + begin - cur_offset,
-                buf + (cur_offset <= offset ? 0 : cur_offset-offset),
-                end - begin
-            );
+            if (!buf)
+            {
+                memset((uint8_t*)v.iov_base + begin - cur_offset, 0, end - begin);
+            }
+            else
+            {
+                memcpy(
+                    (uint8_t*)v.iov_base + begin - cur_offset,
+                    buf + (cur_offset <= offset ? 0 : cur_offset-offset),
+                    end - begin
+                );
+            }
             cur_offset += v.iov_len;
             iov_idx++;
         }
     }
+    if (!buf)
+    {
+        // Bitmap is initially zero, don't set it
+        return;
+    }
     // Set bitmap bits
     int start_bit = (offset-op->offset)/bitmap_granularity;
     int end_bit = (offset-op->offset+len)/bitmap_granularity;
@@ -449,7 +473,8 @@ bool writeback_cache_t::read_from_cache(cluster_op_t *op, uint32_t bitmap_granul
                     {
                         // Copy data
                         dirty_copied = true;
-                        copy_to_op(op, prev, dirty_it->second.buf + prev - dirty_it->first.stripe, cur-prev, bitmap_granularity);
+                        copy_to_op(op, prev, dirty_it->second.buf ? (dirty_it->second.buf + prev - dirty_it->first.stripe) : NULL,
+                            cur-prev, bitmap_granularity);
                     }
                     skip_prev = skip;
                     prev = cur;
@@ -461,7 +486,8 @@ bool writeback_cache_t::read_from_cache(cluster_op_t *op, uint32_t bitmap_granul
             {
                 // Copy data
                 dirty_copied = true;
-                copy_to_op(op, prev, dirty_it->second.buf + prev - dirty_it->first.stripe, cur-prev, bitmap_granularity);
+                copy_to_op(op, prev, dirty_it->second.buf ? (dirty_it->second.buf + prev - dirty_it->first.stripe) : NULL,
+                    cur-prev, bitmap_granularity);
             }
             dirty_it++;
         }

src/client/etcd_state_client.cpp

@@ -1217,6 +1217,7 @@ void etcd_state_client_t::parse_state(const etcd_kv_t & kv)
                     .size = value["size"].uint64_value(),
                     .parent_id = parent_inode_num,
                     .readonly = value["readonly"].bool_value(),
+                    .deleted = value["deleted"].bool_value(),
                     .meta = value["meta"],
                     .mod_revision = kv.mod_revision,
                 });
@@ -1305,6 +1306,10 @@ json11::Json::object etcd_state_client_t::serialize_inode_cfg(inode_config_t *cf
     {
         new_cfg["readonly"] = true;
     }
+    if (cfg->deleted)
+    {
+        new_cfg["deleted"] = true;
+    }
     if (cfg->meta.is_object())
     {
         new_cfg["meta"] = cfg->meta;

src/client/etcd_state_client.h

@@ -72,6 +72,7 @@ struct inode_config_t
     uint64_t size = 0;
     inode_t parent_id = 0;
     bool readonly = false;
+    bool deleted = false;
     // Arbitrary metadata
     json11::Json meta;
     // Change revision of the metadata in etcd

src/client/osd_ops.h

@@ -51,6 +51,11 @@
 #define LOC_CORRUPTED 2
 #define LOC_INCONSISTENT 4
 
+#define OSD_LIST_PRIMARY 1
+
+#define OSD_DEL_SUPPORT_LEFT_ON_DEAD 1
+#define OSD_DEL_LEFT_ON_DEAD         2
+
 // common request and reply headers
 struct __attribute__((__packed__)) osd_op_header_t
 {
@@ -196,6 +201,9 @@ struct __attribute__((__packed__)) osd_op_sec_list_t
     uint64_t min_stripe, max_stripe;
     // max stable object count
     uint32_t stable_limit;
+    // flags - OSD_LIST_PRIMARY or 0
+    // for OSD_LIST_PRIMARY, only a single-PG listing is allowed
+    uint64_t flags;
 };
 
 struct __attribute__((__packed__)) osd_reply_sec_list_t
@@ -204,9 +212,11 @@ struct __attribute__((__packed__)) osd_reply_sec_list_t
     // stable object version count. header.retval = total object version count
     // FIXME: maybe change to the number of bytes in the reply...
     uint64_t stable_count;
+    // flags - OSD_LIST_PRIMARY or 0
+    uint64_t flags;
 };
 
-// read or write to the primary OSD (must be within individual stripe)
+// read, write or delete command for the primary OSD (must be within individual stripe)
 struct __attribute__((__packed__)) osd_op_rw_t
 {
     osd_op_header_t header;
@@ -235,6 +245,20 @@ struct __attribute__((__packed__)) osd_reply_rw_t
     uint64_t version;
 };
 
+struct __attribute__((__packed__)) osd_reply_del_t
+{
+    osd_reply_header_t header;
+    // OSD_DEL_SUPPORT_LEFT_ON_DEAD and/or OSD_DEL_LEFT_ON_DEAD or 0
+    uint32_t flags;
+    // for deletes, if flags & OSD_DEL_LEFT_ON_DEAD:
+    // count of OSDs from which the object could be not deleted
+    // these come directly after this del_left_on_dead_list_size as uint32_t[]
+    // FIXME it's kind of a hack and will be removed in the future, when Vitastor will
+    // have 'atomic deletions', i.e. when it will be able to remember deleted objects
+    // and complete deletions automatically after extra OSDs are started
+    uint32_t left_on_dead_count;
+};
+
 // sync to the primary OSD
 struct __attribute__((__packed__)) osd_op_sync_t
 {
@@ -307,6 +331,7 @@ union osd_any_reply_t
     osd_reply_sec_list_t sec_list;
     osd_reply_show_config_t show_conf;
     osd_reply_rw_t rw;
+    osd_reply_del_t del;
     osd_reply_sync_t sync;
     osd_reply_describe_t describe;
     uint8_t buf[OSD_PACKET_SIZE];

src/client/vitastor.pc.in

@@ -6,7 +6,7 @@ includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@
 
 Name: Vitastor
 Description: Vitastor client library
-Version: 1.10.0
+Version: 1.10.1
 Libs: -L${libdir} -lvitastor_client
 Cflags: -I${includedir}
 

src/cmd/cli.cpp

@@ -51,8 +51,9 @@ static const char* help_text =
     "  Rename, resize image or change its readonly status. Images with children can't be made read-write.\n"
     "  If the new size is smaller than the old size, extra data will be purged.\n"
     "  You should resize file system in the image, if present, before shrinking it.\n"
-    "  -f|--force  Proceed with shrinking or setting readwrite flag even if the image has children.\n"
-    "  --down-ok   Proceed with shrinking even if some data will be left on unavailable OSDs.\n"
+    "  --deleted 1|0  Set/clear 'deleted image' flag (set automatically during unfinished deletes).\n"
+    "  -f|--force     Proceed with shrinking or setting readwrite flag even if the image has children.\n"
+    "  --down-ok      Proceed with shrinking even if some data will be left on unavailable OSDs.\n"
     "\n"
     "vitastor-cli dd [iimg=<image> | if=<file>] [oimg=<image> | of=<file>] [bs=1M]\n"
     "    [count=N] [seek/oseek=N] [skip/iseek=M] [iodepth=N] [status=progress]\n"
@@ -101,6 +102,7 @@ static const char* help_text =
     "                Requires more memory, but allows to show correct removal progress.\n"
     "  --min-offset  Purge only data starting with specified offset.\n"
     "  --max-offset  Purge only data before specified offset.\n"
+    "  --client_wait_up_timeout 16  Timeout for waiting until PGs are up in seconds.\n"
     "\n"
     "vitastor-cli merge-data <from> <to> [--target <target>]\n"
     "  Merge layer data without changing metadata. Merge <from>..<to> to <target>.\n"
@@ -430,13 +432,22 @@ static int run(cli_tool_t *p, json11::Json::object cfg)
     else if (cmd[0] == "rm")
     {
         // Remove multiple snapshots and rebase their children
-        if (cmd.size() > 1)
+        if (cfg["exact"].bool_value() || cfg["matching"].bool_value())
         {
-            cfg["from"] = cmd[1];
-            if (cmd.size() > 2)
-                cfg["to"] = cmd[2];
+            cmd.erase(cmd.begin(), cmd.begin()+1);
+            cfg["globs"] = cmd;
+            action_cb = p->start_rm_wildcard(cfg);
+        }
+        else
+        {
+            if (cmd.size() > 1)
+            {
+                cfg["from"] = cmd[1];
+                if (cmd.size() > 2)
+                    cfg["to"] = cmd[2];
+            }
+            action_cb = p->start_rm(cfg);
         }
-        action_cb = p->start_rm(cfg);
     }
     else if (cmd[0] == "describe")
     {

src/cmd/cli.h

@@ -30,7 +30,7 @@ struct cli_result_t
 class cli_tool_t
 {
 public:
-    uint64_t iodepth = 4, parallel_osds = 32;
+    uint64_t iodepth = 32, parallel_osds = 4;
     bool progress = false;
     bool list_first = false;
     bool json_output = false;

src/cmd/cli_create.cpp

@@ -192,7 +192,7 @@ resume_3:
             }
         } while (!parent->etcd_result["succeeded"].bool_value());
         // Save into inode_config for library users to be able to take it from there immediately
-        new_cfg.mod_revision = parent->etcd_result["responses"][0]["response_put"]["header"]["revision"].uint64_value();
+        new_cfg.mod_revision = parent->etcd_result["header"]["revision"].uint64_value();
         parent->cli->st_cli.insert_inode_config(new_cfg);
         result = (cli_result_t){
             .err = 0,
@@ -269,7 +269,7 @@ resume_4:
             }
         } while (!parent->etcd_result["succeeded"].bool_value());
         // Save into inode_config for library users to be able to take it from there immediately
-        new_cfg.mod_revision = parent->etcd_result["responses"][0]["response_put"]["header"]["revision"].uint64_value();
+        new_cfg.mod_revision = parent->etcd_result["header"]["revision"].uint64_value();
         parent->cli->st_cli.insert_inode_config(new_cfg);
         result = (cli_result_t){
             .err = 0,
@@ -401,7 +401,7 @@ resume_3:
                 auto kv = parent->cli->st_cli.parse_etcd_kv(parent->etcd_result["responses"][0]["response_range"]["kvs"][0]);
                 size = kv.value["size"].uint64_value();
                 new_parent_id = kv.value["parent_id"].uint64_value();
-                uint64_t parent_pool_id = kv.value["parent_pool_id"].uint64_value();
+                uint64_t parent_pool_id = kv.value["parent_pool"].uint64_value();
                 if (new_parent_id)
                 {
                     new_parent_id = INODE_WITH_POOL(parent_pool_id ? parent_pool_id : old_pool_id, new_parent_id);
@@ -413,7 +413,7 @@ resume_3:
 
     void attempt_create()
     {
-        new_cfg = {
+        new_cfg = (inode_config_t){
             .num = INODE_WITH_POOL(new_pool_id, new_id),
             .name = image_name,
             .size = size,

src/cmd/cli_ls.cpp

@@ -68,6 +68,7 @@ struct image_lister_t
                 { "pool_name", good_pool ? pool_it->second.name : "? (ID:"+std::to_string(INODE_POOL(ic.second.num))+")" },
                 { "inode_num", INODE_NO_POOL(ic.second.num) },
                 { "inode_id", ic.second.num },
+                { "deleted", ic.second.deleted },
             };
             if (ic.second.parent_id)
             {
@@ -371,7 +372,8 @@ resume_1:
                 kv.second["delete_q"] = format_q(kv.second["delete_queue"].number_value());
             }
             kv.second["size_fmt"] = format_size(kv.second["size"].uint64_value());
-            kv.second["ro"] = kv.second["readonly"].bool_value() ? "RO" : "-";
+            kv.second["ro"] = kv.second["deleted"].bool_value() ? "DEL" :
+                (kv.second["readonly"].bool_value() ? "RO" : "-");
         }
         result.text = print_table(to_list(), cols, parent->color);
         state = 100;

src/cmd/cli_merge.cpp

@@ -51,6 +51,7 @@ struct snap_merger_t
     btree::safe_btree_set<uint64_t> merge_offsets;
     btree::safe_btree_set<uint64_t>::iterator oit;
     std::map<inode_t, std::vector<uint64_t>> layer_lists;
+    std::map<inode_t, int> list_errcode;
     std::map<inode_t, uint64_t> layer_block_size;
     std::map<inode_t, uint64_t> layer_list_pos;
     std::vector<snap_rw_op_t*> continue_rwo, continue_rwo2;
@@ -251,6 +252,7 @@ struct snap_merger_t
         // Get parents and so on
         start_merge();
         // First list lower layers
+        list_errcode.clear();
         list_layers(true);
         state = 1;
     resume_1:
@@ -259,6 +261,15 @@ struct snap_merger_t
             // Wait for lists
             return;
         }
+        if (list_errcode.size())
+        {
+            result = (cli_result_t){
+                .err = EIO,
+                .text = "Failed to list lower layer(s) in some PGs, merging would be incorrect",
+            };
+            state = 100;
+            return;
+        }
         if (merge_offsets.size() > 0)
         {
             state = 2;
@@ -295,6 +306,7 @@ struct snap_merger_t
         state = 3;
     resume_3:
         // Then list upper layers
+        list_errcode.clear();
         list_layers(false);
         state = 4;
     resume_4:
@@ -303,6 +315,15 @@ struct snap_merger_t
             // Wait for lists
             return;
         }
+        if (list_errcode.size() > 0)
+        {
+            result = (cli_result_t){
+                .err = EIO,
+                .text = "Failed to list upper layer(s) in some PGs, merging would be incorrect",
+            };
+            state = 100;
+            return;
+        }
         state = 5;
         processed = 0;
         to_process = merge_offsets.size();
@@ -368,9 +389,13 @@ struct snap_merger_t
             if (lower ? (sp.second < target_rank) : (sp.second > target_rank))
             {
                 lists_todo++;
-                inode_list_t* lst = parent->cli->list_inode_start(src, [this, src](
-                    inode_list_t *lst, std::set<object_id>&& objects, pg_num_t pg_num, osd_num_t primary_osd, int status)
+                parent->cli->list_inode(src, 0, 0, parent->parallel_osds, [this, src](
+                    int errcode, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects)
                 {
+                    if (errcode)
+                    {
+                        list_errcode[src] = errcode;
+                    }
                     uint64_t layer_block = layer_block_size.at(src);
                     for (object_id obj: objects)
                     {
@@ -391,12 +416,18 @@ struct snap_merger_t
                             layer_list[pos++] = obj.stripe;
                         }
                     }
-                    if (status & INODE_LIST_DONE)
+                    if (!pgs_left)
                     {
                         auto & name = parent->cli->st_cli.inode_config.at(src).name;
-                        if (parent->progress)
+                        if (list_errcode.find(src) != list_errcode.end())
                         {
-                            printf("Got listing of layer %s (inode %ju in pool %u)\n", name.c_str(), INODE_NO_POOL(src), INODE_POOL(src));
+                            fprintf(stderr, "Failed to get listing of layer %s (inode %ju in pool %u): %s (code %d)\n",
+                                name.c_str(), INODE_NO_POOL(src), INODE_POOL(src), strerror(-list_errcode[src]), list_errcode[src]);
+                        }
+                        else if (parent->progress)
+                        {
+                            fprintf(stderr, "Got listing of layer %s (inode %ju in pool %u)\n",
+                                name.c_str(), INODE_NO_POOL(src), INODE_POOL(src));
                         }
                         if (delete_source)
                         {
@@ -406,12 +437,7 @@ struct snap_merger_t
                         lists_todo--;
                         continue_merge_reent();
                     }
-                    else
-                    {
-                        parent->cli->list_inode_next(lst, 1);
-                    }
                 });
-                parent->cli->list_inode_next(lst, parent->parallel_osds);
             }
         }
     }
@@ -428,7 +454,7 @@ struct snap_merger_t
         {
             if (op->retval < 0)
             {
-                fprintf(stderr, "error reading target bitmap at offset %jx: %s\n", op->offset, strerror(-op->retval));
+                fprintf(stderr, "Warning: failed to read target bitmap at offset %jx: %s\n", op->offset, strerror(-op->retval));
             }
             else
             {
@@ -585,7 +611,7 @@ struct snap_merger_t
         subop->inode = inode_num;
         subop->offset = offset;
         subop->len = 0;
-        subop->flags = OSD_OP_IGNORE_READONLY;
+        subop->flags = OSD_OP_IGNORE_READONLY | OSD_OP_WAIT_UP_TIMEOUT;
         subop->callback = [](cluster_op_t *subop)
         {
             if (subop->retval != 0)

src/cmd/cli_modify.cpp

@@ -4,6 +4,7 @@
 #include "cli.h"
 #include "cluster_client.h"
 #include "str_util.h"
+#include "json_util.h"
 
 // Rename, resize image (and purge extra data on shrink) or change its readonly status
 struct image_changer_t
@@ -15,6 +16,7 @@ struct image_changer_t
     uint64_t new_size = 0;
     bool force_size = false, inc_size = false;
     bool set_readonly = false, set_readwrite = false, force = false;
+    bool set_deleted = false, new_deleted = false;
     bool down_ok = false;
     // interval between fsyncs
     int fsync_interval = 128;
@@ -82,6 +84,7 @@ struct image_changer_t
         }
         if ((!set_readwrite || !cfg.readonly) &&
             (!set_readonly || cfg.readonly) &&
+            (!set_deleted || cfg.deleted == new_deleted) &&
             (!new_size && !force_size || cfg.size == new_size || cfg.size >= new_size && inc_size) &&
             (new_name == "" || new_name == image_name))
         {
@@ -141,6 +144,10 @@ resume_1:
                 return;
             }
         }
+        if (set_deleted)
+        {
+            cfg.deleted = new_deleted;
+        }
         if (new_name != "")
         {
             cfg.name = new_name;
@@ -219,7 +226,7 @@ resume_2:
             return;
         }
         // Save into inode_config for library users to be able to take it from there immediately
-        cfg.mod_revision = parent->etcd_result["responses"][0]["response_put"]["header"]["revision"].uint64_value();
+        cfg.mod_revision = parent->etcd_result["header"]["revision"].uint64_value();
         if (new_name != "")
         {
             parent->cli->st_cli.inode_by_name.erase(image_name);
@@ -251,6 +258,8 @@ std::function<bool(cli_result_t &)> cli_tool_t::start_modify(json11::Json cfg)
     changer->force = cfg["force"].bool_value();
     changer->set_readonly = cfg["readonly"].bool_value();
     changer->set_readwrite = cfg["readwrite"].bool_value();
+    changer->set_deleted = !cfg["deleted"].is_null();
+    changer->new_deleted = json_is_true(cfg["deleted"]);
     changer->fsync_interval = cfg["fsync_interval"].uint64_value();
     if (!changer->fsync_interval)
         changer->fsync_interval = 128;

src/cmd/cli_pool_create.cpp

@@ -460,7 +460,7 @@ resume_8:
             }
             if (osd_bs && osd_bs != UINT32_MAX && osd_bs != parent->cli->st_cli.global_block_size)
             {
-                fprintf(stderr, "Auto-selecting block_size=%s because all pool OSDs use it\n", format_size(osd_bs).c_str());
+                fprintf(stderr, "Auto-selecting block_size=%s because all pool OSDs use it\n", format_size(osd_bs, false, true).c_str());
                 upd["block_size"] = osd_bs;
             }
         }
@@ -479,7 +479,7 @@ resume_8:
             }
             if (osd_bg && osd_bg != UINT32_MAX && osd_bg != parent->cli->st_cli.global_bitmap_granularity)
             {
-                fprintf(stderr, "Auto-selecting bitmap_granularity=%s because all pool OSDs use it\n", format_size(osd_bg).c_str());
+                fprintf(stderr, "Auto-selecting bitmap_granularity=%s because all pool OSDs use it\n", format_size(osd_bg, false, true).c_str());
                 upd["bitmap_granularity"] = osd_bg;
             }
         }

src/cmd/cli_pool_ls.cpp

@@ -185,6 +185,7 @@ resume_1:
                 { "space_efficiency", pool_stats[pool_cfg.id]["space_efficiency"].number_value() },
                 { "pg_real_size", pool_stats[pool_cfg.id]["pg_real_size"].uint64_value() },
                 { "osd_count", (uint64_t)pg_per_osd.size() },
+                { "backfillfull", pool_cfg.backfillfull },
             };
         }
         // Include full pool config
@@ -484,6 +485,7 @@ resume_3:
                     st["block_size_fmt"] = format_size(st["block_size"].uint64_value());
                 if (st["bitmap_granularity"].uint64_value())
                     st["bitmap_granularity_fmt"] = format_size(st["bitmap_granularity"].uint64_value());
+                st["backfillfull_fmt"] = st["backfillfull"].bool_value() ? "yes" : "";
             }
             // All pool parameters are only displayed in the "detailed" mode
             // because there's too many of them to show them in table
@@ -493,6 +495,7 @@ resume_3:
                 { "scheme_name", "Scheme" },
                 { "used_for_fs", "Used for VitastorFS" },
                 { "status", "Status" },
+                { "backfillfull_fmt", "Backfillfull" },
                 { "pg_count_fmt", "PGs" },
                 { "pg_minsize", "PG minsize" },
                 { "failure_domain", "Failure domain" },

src/cmd/cli_rm.cpp

@@ -95,6 +95,10 @@ struct snap_remover_t
             goto resume_7;
         else if (state == 8)
             goto resume_8;
+        else if (state == 9)
+            goto resume_9;
+        else if (state == 10)
+            goto resume_10;
         else if (state == 100)
             goto resume_100;
         assert(!state);
@@ -129,23 +133,9 @@ resume_1:
                 continue;
             rebased_images.push_back(parent->cli->st_cli.inode_config.at(merge_children[current_child]).name);
             start_merge_child(merge_children[current_child], merge_children[current_child]);
-            if (state == 100)
-                return;
 resume_2:
-            while (!cb(result))
-            {
-                state = 2;
+            while (!wait_result(2))
                 return;
-            }
-            cb = NULL;
-            if (result.err)
-            {
-                result.data = my_result(result.data);
-                state = 100;
-                return;
-            }
-            else if (parent->progress)
-                printf("%s\n", result.text.c_str());
             parent->change_parent(merge_children[current_child], new_parent, &result);
             state = 3;
 resume_3:
@@ -164,42 +154,19 @@ resume_3:
         if (inverse_child != 0)
         {
             start_merge_child(inverse_child, inverse_parent);
-            if (state == 100)
-                return;
 resume_4:
-            while (!cb(result))
-            {
-                state = 4;
+            while (!wait_result(4))
                 return;
-            }
-            cb = NULL;
-            if (result.err)
-            {
-                result.data = my_result(result.data);
-                state = 100;
+            // Mark child as deleted
+            start_mark_deleted(inverse_child);
+resume_9:
+            while (!wait_result(9))
                 return;
-            }
-            else if (parent->progress)
-                printf("%s\n", result.text.c_str());
             // Delete "inverse" child data
             start_delete_source(inverse_child);
-            if (state == 100)
-                return;
 resume_5:
-            while (!cb(result))
-            {
-                state = 5;
+            while (!wait_result(5))
                 return;
-            }
-            cb = NULL;
-            if (result.err)
-            {
-                result.data = my_result(result.data);
-                state = 100;
-                return;
-            }
-            else if (parent->progress)
-                printf("%s\n", result.text.c_str());
             // Delete "inverse" child metadata, rename parent over it,
             // and also change parent links of the previous "inverse" child
             rename_inverse_parent();
@@ -221,22 +188,15 @@ resume_6:
                     deleted_images.push_back(parent_it->second.name);
                 deleted_ids.push_back(chain_list[current_child]);
             }
+            // Mark child as deleted
+            start_mark_deleted(chain_list[current_child]);
+resume_10:
+            while (!wait_result(10))
+                return;
             start_delete_source(chain_list[current_child]);
 resume_7:
-            while (!cb(result))
-            {
-                state = 7;
+            while (!wait_result(7))
                 return;
-            }
-            cb = NULL;
-            if (result.err)
-            {
-                result.data = my_result(result.data);
-                state = 100;
-                return;
-            }
-            else if (parent->progress)
-                printf("%s\n", result.text.c_str());
             delete_inode_config(chain_list[current_child]);
             if (state == 100)
                 return;
@@ -256,6 +216,27 @@ resume_100:
         return;
     }
 
+    bool wait_result(int base_state)
+    {
+        if (state == 100 || !cb)
+            return true;
+        while (!cb(result))
+        {
+            state = base_state;
+            return false;
+        }
+        cb = NULL;
+        if (result.err)
+        {
+            result.data = my_result(result.data);
+            state = 100;
+            return false;
+        }
+        else if (parent->progress)
+            printf("%s\n", result.text.c_str());
+        return true;
+    }
+
     json11::Json my_result(json11::Json src)
     {
         auto obj = src.object_items();
@@ -667,6 +648,27 @@ resume_100:
         });
     }
 
+    void start_mark_deleted(inode_t inode)
+    {
+        auto ino_it = parent->cli->st_cli.inode_config.find(inode);
+        if (ino_it == parent->cli->st_cli.inode_config.end())
+        {
+            char buf[1024];
+            snprintf(buf, 1024, "Inode 0x%jx disappeared", inode);
+            result = (cli_result_t){ .err = EIO, .text = std::string(buf) };
+            state = 100;
+            return;
+        }
+        if (ino_it->second.deleted)
+        {
+            return;
+        }
+        cb = parent->start_modify(json11::Json::object {
+            { "image", ino_it->second.name },
+            { "deleted", true },
+        });
+    }
+
     void start_delete_source(inode_t inode)
     {
         auto source = parent->cli->st_cli.inode_config.find(inode);
@@ -689,10 +691,6 @@ resume_100:
 
 std::function<bool(cli_result_t &)> cli_tool_t::start_rm(json11::Json cfg)
 {
-    if (cfg["exact"].bool_value() || cfg["matching"].bool_value())
-    {
-        return start_rm_wildcard(cfg);
-    }
     auto snap_remover = new snap_remover_t();
     snap_remover->parent = this;
     snap_remover->from_name = cfg["from"].string_value();

src/cmd/cli_rm_data.cpp

@@ -11,7 +11,6 @@
 struct rm_pg_t
 {
     pg_num_t pg_num;
-    osd_num_t rm_osd_num;
     std::set<object_id> objects;
     std::set<object_id>::iterator obj_pos;
     uint64_t obj_count = 0, obj_done = 0;
@@ -31,160 +30,127 @@ struct rm_inode_t
     cli_tool_t *parent = NULL;
     inode_list_t *lister = NULL;
     std::vector<rm_pg_t*> lists;
-    std::vector<osd_num_t> inactive_osds;
+    std::set<osd_num_t> inactive_osds;
+    std::set<pg_num_t> inactive_pgs;
+    std::set<pg_num_t> fallback_pgs;
     uint64_t total_count = 0, total_done = 0, total_prev_pct = 0;
-    uint64_t pgs_to_list = 0;
     bool lists_done = false;
+    int pgs_to_list = 0;
     int state = 0;
     int error_count = 0;
+    bool in_continue = false;
 
     cli_result_t result;
 
     void start_delete()
     {
-        lister = parent->cli->list_inode_start(inode, [this](inode_list_t *lst,
-            std::set<object_id>&& objects, pg_num_t pg_num, osd_num_t primary_osd, int status)
+        auto pool_it = parent->cli->st_cli.pool_config.find(pool_id);
+        if (pool_it == parent->cli->st_cli.pool_config.end())
         {
-            rm_pg_t *rm = new rm_pg_t((rm_pg_t){
-                .pg_num = pg_num,
-                .rm_osd_num = primary_osd,
-                .objects = objects,
-                .obj_count = objects.size(),
-                .obj_done = 0,
-                .synced = parent->cli->get_immediate_commit(inode),
-            });
-            if (min_offset == 0 && max_offset == 0)
-            {
-                total_count += objects.size();
-            }
-            else
-            {
-                for (object_id oid: objects)
-                {
-                    if (oid.stripe >= min_offset && (!max_offset || oid.stripe < max_offset))
-                    {
-                        total_count++;
-                    }
-                }
-            }
-            rm->obj_pos = rm->objects.begin();
-            lists.push_back(rm);
-            if (parent->list_first && !(status & INODE_LIST_DONE))
-            {
-                // The listing object is dead when DONE => don't call next()
-                parent->cli->list_inode_next(lister, 1);
-            }
-            if (status & INODE_LIST_DONE)
-            {
-                lists_done = true;
-            }
-            pgs_to_list--;
-            continue_delete();
-        });
-        if (!lister)
-        {
-            result = (cli_result_t){
-                .err = EIO,
-                .text = "Failed to list objects of inode "+std::to_string(INODE_NO_POOL(inode))+
-                    " from pool "+std::to_string(INODE_POOL(inode)),
-            };
+            result = (cli_result_t){ .err = EINVAL, .text = "Pool does not exist" };
             state = 100;
             return;
         }
-        inactive_osds = parent->cli->list_inode_get_inactive_osds(lister);
-        if (inactive_osds.size() && !parent->json_output)
+        pgs_to_list = pool_it->second.real_pg_count;
+        parent->cli->list_inode(inode, min_offset, max_offset, parent->parallel_osds, [this](
+            int errcode, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects)
         {
-            fprintf(stderr, "Some data may remain after delete on OSDs which are currently down: ");
-            for (int i = 0; i < inactive_osds.size(); i++)
+            if (errcode)
             {
-                fprintf(stderr, i > 0 ? ", %ju" : "%ju", inactive_osds[i]);
+                inactive_pgs.insert(pg_num);
             }
-            fprintf(stderr, "\n");
-        }
-        pgs_to_list = parent->cli->list_pg_count(lister);
-        parent->cli->list_inode_next(lister, parent->parallel_osds);
+            else
+            {
+                rm_pg_t *rm = new rm_pg_t((rm_pg_t){
+                    .pg_num = pg_num,
+                    .objects = std::move(objects),
+                    .obj_done = 0,
+                    .synced = !objects.size() || parent->cli->get_immediate_commit(inode),
+                });
+                if (min_offset == 0 && max_offset == 0)
+                {
+                    total_count += objects.size();
+                }
+                else
+                {
+                    for (object_id oid: objects)
+                    {
+                        if (oid.stripe >= min_offset && (!max_offset || oid.stripe < max_offset))
+                        {
+                            total_count++;
+                        }
+                    }
+                }
+                rm->obj_pos = rm->objects.begin();
+                lists.push_back(rm);
+            }
+            pgs_to_list = pgs_left;
+            lists_done = !pgs_to_list;
+            continue_delete();
+        });
     }
 
     void send_ops(rm_pg_t *cur_list)
     {
-        parent->cli->init_msgr();
-        if (parent->cli->msgr.osd_peer_fds.find(cur_list->rm_osd_num) ==
-            parent->cli->msgr.osd_peer_fds.end())
-        {
-            // Initiate connection
-            parent->cli->msgr.connect_peer(cur_list->rm_osd_num, parent->cli->st_cli.peer_states[cur_list->rm_osd_num]);
-            return;
-        }
         while (cur_list->in_flight < parent->iodepth && cur_list->obj_pos != cur_list->objects.end())
         {
             if (cur_list->obj_pos->stripe >= min_offset && (!max_offset || cur_list->obj_pos->stripe < max_offset))
             {
-                osd_op_t *op = new osd_op_t();
-                op->op_type = OSD_OP_OUT;
-                // Already checked that it exists above, but anyway
-                op->peer_fd = parent->cli->msgr.osd_peer_fds.at(cur_list->rm_osd_num);
-                op->req = (osd_any_op_t){
-                    .rw = {
-                        .header = {
-                            .magic = SECONDARY_OSD_OP_MAGIC,
-                            .id = parent->cli->next_op_id(),
-                            .opcode = OSD_OP_DELETE,
-                        },
-                        .inode = cur_list->obj_pos->inode,
-                        .offset = cur_list->obj_pos->stripe,
-                        .len = 0,
-                    },
-                };
-                op->callback = [this, cur_list](osd_op_t *op)
+                cluster_op_t *op = new cluster_op_t;
+                op->opcode = OSD_OP_DELETE;
+                op->inode = cur_list->obj_pos->inode;
+                op->offset = cur_list->obj_pos->stripe;
+                op->len = 0;
+                op->flags = OSD_OP_IGNORE_READONLY | OSD_OP_WAIT_UP_TIMEOUT;
+                op->callback = [this, cur_list](cluster_op_t *op)
                 {
                     cur_list->in_flight--;
-                    if (op->reply.hdr.retval < 0)
+                    if (op->retval < 0)
                     {
-                        fprintf(stderr, "Failed to remove object %jx:%jx from PG %u (OSD %ju) (retval=%jd)\n",
-                            op->req.rw.inode, op->req.rw.offset,
-                            cur_list->pg_num, cur_list->rm_osd_num, op->reply.hdr.retval);
+                        fprintf(stderr, "Failed to remove object %jx:%jx from PG %u (retval=%d)\n",
+                            op->inode, op->offset, cur_list->pg_num, op->retval);
                         error_count++;
                     }
+                    else
+                    {
+                        if (!op->support_left_on_dead())
+                            fallback_pgs.insert(cur_list->pg_num);
+                        for (auto inactive_osd: op->get_left_on_dead())
+                            inactive_osds.insert(inactive_osd);
+                    }
                     delete op;
                     cur_list->obj_done++;
                     total_done++;
                     continue_delete();
                 };
                 cur_list->in_flight++;
-                parent->cli->msgr.outbox_push(op);
+                cur_list->obj_pos++;
+                parent->cli->execute(op);
+            }
+            else
+            {
+                cur_list->obj_pos++;
             }
-            cur_list->obj_pos++;
         }
         if (cur_list->in_flight == 0 && cur_list->obj_pos == cur_list->objects.end() &&
             !cur_list->synced)
         {
-            osd_op_t *op = new osd_op_t();
-            op->op_type = OSD_OP_OUT;
-            op->peer_fd = parent->cli->msgr.osd_peer_fds.at(cur_list->rm_osd_num);
-            op->req = (osd_any_op_t){
-                .sync = {
-                    .header = {
-                        .magic = SECONDARY_OSD_OP_MAGIC,
-                        .id = parent->cli->next_op_id(),
-                        .opcode = OSD_OP_SYNC,
-                    },
-                },
-            };
-            op->callback = [this, cur_list](osd_op_t *op)
+            cluster_op_t *op = new cluster_op_t;
+            op->opcode = OSD_OP_SYNC;
+            op->callback = [this, cur_list](cluster_op_t *op)
             {
                 cur_list->in_flight--;
                 cur_list->synced = true;
-                if (op->reply.hdr.retval < 0)
+                if (op->retval < 0)
                 {
-                    fprintf(stderr, "Failed to sync OSD %ju (retval=%jd)\n",
-                        cur_list->rm_osd_num, op->reply.hdr.retval);
+                    fprintf(stderr, "Failed to sync after deletion (retval=%d)\n", op->retval);
                     error_count++;
                 }
                 delete op;
                 continue_delete();
             };
             cur_list->in_flight++;
-            parent->cli->msgr.outbox_push(op);
+            parent->cli->execute(op);
         }
     }
 
@@ -194,6 +160,11 @@ struct rm_inode_t
         {
             return;
         }
+        if (in_continue)
+        {
+            return;
+        }
+        in_continue = true;
         for (int i = 0; i < lists.size(); i++)
         {
             if (!lists[i]->in_flight && lists[i]->obj_pos == lists[i]->objects.end() &&
@@ -202,10 +173,6 @@ struct rm_inode_t
                 delete lists[i];
                 lists.erase(lists.begin()+i, lists.begin()+i+1);
                 i--;
-                if (!lists_done)
-                {
-                    parent->cli->list_inode_next(lister, 1);
-                }
             }
             else
             {
@@ -215,8 +182,8 @@ struct rm_inode_t
         if (parent->progress && total_count > 0 && total_done*1000/total_count != total_prev_pct)
         {
             fprintf(stderr, parent->color
-                ? "\rRemoved %ju/%ju objects, %ju more PGs to list..."
-                : "Removed %ju/%ju objects, %ju more PGs to list...\n", total_done, total_count, pgs_to_list);
+                ? "\rRemoved %ju/%ju objects, %u more PGs to list..."
+                : "Removed %ju/%ju objects, %u more PGs to list...\n", total_done, total_count, pgs_to_list);
             total_prev_pct = total_done*1000/total_count;
         }
         if (lists_done && !lists.size())
@@ -225,28 +192,102 @@ struct rm_inode_t
             {
                 fprintf(stderr, "\n");
             }
-            bool is_error = (total_done < total_count || inactive_osds.size() > 0 || error_count > 0);
-            if (parent->progress && is_error)
+            if (fallback_pgs.size() && !parent->json_output)
             {
-                fprintf(
-                    stderr, "Warning: Pool:%u,ID:%ju inode data may not have been fully removed.\n"
-                    "Use `vitastor-cli rm-data --pool %u --inode %ju` if you encounter it in listings.\n",
-                    pool_id, INODE_NO_POOL(inode), pool_id, INODE_NO_POOL(inode)
-                );
+                fprintf(stderr, "Warning: some OSDs don't indicate left_on_dead PG OSDs"
+                    " in delete replies, falling back to simpler checks\n");
+                auto pool_it = parent->cli->st_cli.pool_config.find(pool_id);
+                if (pool_it != parent->cli->st_cli.pool_config.end())
+                {
+                    std::set<osd_num_t> all_peers;
+                    for (auto pg_num: fallback_pgs)
+                    {
+                        auto pg_it = pool_it->second.pg_config.find(pg_num);
+                        if (pg_it != pool_it->second.pg_config.end())
+                        {
+                            for (osd_num_t pg_osd: pg_it->second.target_set)
+                                all_peers.insert(pg_osd);
+                            for (osd_num_t pg_osd: pg_it->second.all_peers)
+                                all_peers.insert(pg_osd);
+                            for (auto & hist_item: pg_it->second.target_history)
+                                for (auto pg_osd: hist_item)
+                                    all_peers.insert(pg_osd);
+                        }
+                    }
+                    all_peers.erase(0);
+                    for (auto peer_osd: all_peers)
+                    {
+                        if (parent->cli->st_cli.peer_states[peer_osd].is_null())
+                            inactive_osds.insert(peer_osd);
+                    }
+                }
             }
-            result = (cli_result_t){
-                .err = is_error && !down_ok ? EIO : 0,
-                .text = is_error ? "Some blocks were not removed" : (
-                    "Done, inode "+std::to_string(INODE_NO_POOL(inode))+" from pool "+
-                    std::to_string(pool_id)+" removed"),
-                .data = json11::Json::object {
-                    { "removed_objects", total_done },
-                    { "total_objects", total_count },
-                    { "inactive_osds", inactive_osds },
-                },
+            if (inactive_osds.size() && !parent->json_output)
+            {
+                fprintf(stderr, "Some data may remain after delete on OSDs which are currently down: ");
+                int i = 0;
+                for (auto osd_num: inactive_osds)
+                {
+                    fprintf(stderr, (i++) ? ", %ju" : "%ju", osd_num);
+                }
+                fprintf(stderr, "\n");
+            }
+            if (inactive_pgs.size() && !parent->json_output)
+            {
+                fprintf(stderr, "Failed to list some PGs, deletion is not complete: PG ");
+                int i = 0;
+                for (auto pg_num: inactive_pgs)
+                {
+                    fprintf(stderr, (i++) > 0 ? ", %u" : "%u", pg_num);
+                }
+                fprintf(stderr, "\n");
+            }
+            if (error_count > 0 && !parent->json_output)
+            {
+                fprintf(stderr, "Failed to delete %u objects from active OSD(s).\n", error_count);
+            }
+            json11::Json::array inactive_pgs_json;
+            for (auto pg_num: inactive_pgs)
+            {
+                inactive_pgs_json.push_back((uint64_t)pg_num);
+            }
+            json11::Json data = json11::Json::object {
+                { "removed_objects", total_done },
+                { "total_objects", total_count },
+                { "error_count", error_count },
+                { "inactive_osds", json11::Json::array(inactive_osds.begin(), inactive_osds.end()) },
+                { "inactive_pgs", inactive_pgs_json },
             };
             state = 100;
+            if (total_done < total_count || inactive_pgs.size() > 0 || error_count > 0 ||
+                inactive_osds.size() > 0 && !down_ok)
+            {
+                // Error
+                result = (cli_result_t){
+                    .err = EIO,
+                    .text = "Failed: some blocks were not removed",
+                    .data = data,
+                };
+            }
+            else
+            {
+                if (parent->progress && inactive_osds.size() > 0 && down_ok)
+                {
+                    fprintf(
+                        stderr, "Warning: --down-ok is set and some OSDs are down.\n"
+                        "Pool:%u,ID:%ju inode data may not have been fully removed.\n"
+                        "Use `vitastor-cli rm-data --pool %u --inode %ju` if you encounter it in listings.\n",
+                        pool_id, INODE_NO_POOL(inode), pool_id, INODE_NO_POOL(inode)
+                    );
+                }
+                result = (cli_result_t){
+                    .text = "Done, inode "+std::to_string(INODE_NO_POOL(inode))+" from pool "+
+                        std::to_string(pool_id)+" removed",
+                    .data = data,
+                };
+            }
         }
+        in_continue = false;
     }
 
     bool is_done()

src/cmd/cli_rm_wildcard.cpp

@@ -73,8 +73,11 @@ struct wildcard_remover_t
             {
                 auto & inode_cfg = parent->cli->st_cli.inode_config.at(child_id);
                 ver_chain.push_back((inode_rev_t){ .inode_num = child_id, .meta_rev = inode_cfg.mod_revision });
-                child_id = inode_cfg.parent_id;
-            } while (child_id && child_id != parent_id);
+                if (child_id == parent_id)
+                    break;
+                else
+                    child_id = inode_cfg.parent_id;
+            } while (child_id);
             versioned_chains.push_back(std::move(ver_chain));
         }
         // Sort chains based on parent inode rank to first delete child-most layers
@@ -146,6 +149,7 @@ struct wildcard_remover_t
                 sub_cfg = cfg.object_items();
                 sub_cfg.erase("globs");
                 sub_cfg.erase("exact");
+                sub_cfg.erase("matching");
                 sub_cfg["from"] = from_cfg.name;
                 sub_cfg["to"] = to_cfg.name;
                 sub_cb = parent->start_rm(sub_cfg);

src/cmd/cli_status.cpp

@@ -173,6 +173,8 @@ resume_2:
             }
         });
         int pool_count = 0, pools_active = 0;
+        json11::Json::array backfillfull_pool_ids;
+        std::string backfillfull_pool_names;
         std::map<std::string, int> pgs_by_state;
         std::string pgs_by_state_str;
         for (auto & pool_pair: parent->cli->st_cli.pool_config)
@@ -204,6 +206,13 @@ resume_2:
             {
                 pools_active++;
             }
+            if (pool_cfg.backfillfull)
+            {
+                backfillfull_pool_ids.push_back((uint64_t)pool_pair.first);
+                if (backfillfull_pool_names != "")
+                    backfillfull_pool_names += ", ";
+                backfillfull_pool_names += pool_cfg.name;
+            }
         }
         for (auto & kv: pgs_by_state)
         {
@@ -242,6 +251,7 @@ resume_2:
                 { "no_scrub", no_scrub },
                 { "pool_count", pool_count },
                 { "active_pool_count", pools_active },
+                { "backfillfull_pools", backfillfull_pool_ids },
                 { "pg_states", pgs_by_state },
                 { "op_stats", agg_stats["op_stats"] },
                 { "recovery_stats", agg_stats["recovery_stats"] },
@@ -338,6 +348,13 @@ resume_2:
             }
             warning_str += "\n";
         }
+        if (backfillfull_pool_names != "")
+        {
+            if (backfillfull_pool_ids.size() > 1)
+                warning_str += "    pools "+backfillfull_pool_names+" are backfillfull\n";
+            else
+                warning_str += "    pool "+backfillfull_pool_names+" is backfillfull\n";
+        }
         if (warning_str != "")
         {
             warning_str = "\n  warning:\n"+warning_str;

src/disk_tool/disk_tool.cpp

@@ -68,6 +68,7 @@ static const char *help_text =
     "    --data_device_block 4k     Override data device block size\n"
     "    --meta_device_block 4k     Override metadata device block size\n"
     "    --journal_device_block 4k  Override journal device block size\n"
+    "    --json                     Enable JSON output\n"
     "  \n"
     "  immediate_commit setting is automatically derived from \"disable fsync\" options.\n"
     "  It's set to \"all\" when fsync is disabled on all devices, and to \"small\" if fsync\n"
@@ -102,6 +103,7 @@ static const char *help_text =
     "    --move-meta TARGET       move metadata to TARGET\n"
     "    --journal-size NEW_SIZE  resize journal to NEW_SIZE\n"
     "    --data-size NEW_SIZE     resize data device to NEW_SIZE\n"
+    "    --json                   enable JSON output\n"
     "    --dry-run                only show new layout, do not apply it\n"
     "  \n"
     "  NEW_SIZE may include k/m/g/t suffixes.\n"
@@ -131,10 +133,11 @@ static const char *help_text =
     "  Commands are passed to systemctl with vitastor-osd@<num> units as arguments.\n"
     "  When --now is added to enable/disable, OSDs are also immediately started/stopped.\n"
     "\n"
-    "vitastor-disk purge [--force] [--allow-data-loss] <device> [device2 device3 ...]\n"
+    "vitastor-disk purge [--json] [--force] [--allow-data-loss] <device> [device2 device3 ...]\n"
     "  Purge Vitastor OSD(s) on specified device(s). Uses vitastor-cli rm-osd to check\n"
     "  if deletion is possible without data loss and to actually remove metadata from etcd.\n"
     "  --force and --allow-data-loss options may be used to ignore safety check results.\n"
+    "  --json enables JSON output.\n"
     "  \n"
     "  Requires `vitastor-cli`, `sfdisk` and `partprobe` (from parted) utilities.\n"
     "\n"

src/disk_tool/disk_tool.h

@@ -42,7 +42,7 @@ struct disk_tool_t
 
     std::map<std::string, std::string> options;
     bool test_mode = false;
-    bool all, json, now;
+    bool all = false, json = false, now = false;
     bool dump_with_blocks, dump_with_data;
     blockstore_disk_t dsk;
 
@@ -127,7 +127,7 @@ struct disk_tool_t
     json11::Json read_osd_superblock(std::string device, bool expect_exist = true, bool ignore_nonref = false);
     uint32_t write_osd_superblock(std::string device, json11::Json params);
 
-    int prepare_one(std::map<std::string, std::string> options, int is_hdd = -1);
+    int prepare_one(std::map<std::string, std::string> options, int is_hdd, json11::Json::object & result);
     int check_existing_partition(std::string & dev_by_uuid);
     int fix_partition_type(std::string & dev_by_uuid);
     int prepare(std::vector<std::string> devices);

src/disk_tool/disk_tool_prepare.cpp

@@ -6,7 +6,7 @@
 #include "json_util.h"
 #include "osd_id.h"
 
-int disk_tool_t::prepare_one(std::map<std::string, std::string> options, int is_hdd)
+int disk_tool_t::prepare_one(std::map<std::string, std::string> options, int is_hdd, json11::Json::object & result)
 {
     static const char *allow_additional_params[] = {
         "autosync_writes",
@@ -203,12 +203,14 @@ int disk_tool_t::prepare_one(std::map<std::string, std::string> options, int is_
     fprintf(stderr, "Initialized OSD %ju on %s\n", osd_num, desc.c_str());
     if (!test_mode || options.find("no_init") == options.end())
     {
-        if (shell_exec({ "systemctl", "enable", "--now", "vitastor-osd@"+std::to_string(osd_num) }, "", NULL, NULL) != 0)
+        std::string out;
+        if (shell_exec({ "systemctl", "enable", "--now", "vitastor-osd@"+std::to_string(osd_num) }, "", json ? &out : NULL, NULL) != 0)
         {
             fprintf(stderr, "Failed to enable systemd unit vitastor-osd@%ju\n", osd_num);
             return 1;
         }
     }
+    result = sb;
     return 0;
 }
 
@@ -578,7 +580,13 @@ int disk_tool_t::prepare(std::vector<std::string> devices)
             fprintf(stderr, "Device list (positional arguments), --osd_per_disk, --hybrid and --fast-devices are incompatible with --data_device\n");
             return 1;
         }
-        return prepare_one(options, options.find("hdd") != options.end() ? 1 : 0);
+        json11::Json::object result;
+        int r = prepare_one(options, options.find("hdd") != options.end() ? 1 : 0, result);
+        if (r)
+            return r;
+        if (json)
+            printf("%s\n", json11::Json(result).dump().c_str());
+        return 0;
     }
     if (!devices.size())
     {
@@ -669,6 +677,7 @@ int disk_tool_t::prepare(std::vector<std::string> devices)
         options.erase("disable_meta_fsync");
         options.erase("disable_journal_fsync");
     }
+    json11::Json::array all_results, errors;
     auto journal_size = options["journal_size"];
     for (auto & dev: devinfo)
     {
@@ -688,7 +697,15 @@ int disk_tool_t::prepare(std::vector<std::string> devices)
                 options.erase("journal_size");
             }
             // Treat all disks as SSDs if not in the hybrid mode
-            prepare_one(options, dev.is_hdd ? 1 : 0);
+            json11::Json::object result;
+            int r = prepare_one(options, dev.is_hdd ? 1 : 0, result);
+            if (json)
+            {
+                if (!r)
+                    all_results.push_back(std::move(result));
+                else
+                    errors.push_back(options);
+            }
             if (hybrid)
             {
                 options["journal_size"] = journal_size;
@@ -697,5 +714,9 @@ int disk_tool_t::prepare(std::vector<std::string> devices)
             }
         }
     }
+    if (json)
+    {
+        printf("%s\n", json11::Json(json11::Json::object{ { "osds", all_results }, { "errors", errors } }).dump().c_str());
+    }
     return 0;
 }

src/disk_tool/disk_tool_udev.cpp

@@ -427,8 +427,12 @@ int disk_tool_t::purge_devices(const std::vector<std::string> & devices)
             }
         }
     }
+    json11::Json::object result;
+    result["removed_osds"] = std::vector<uint64_t>(osd_numbers.begin(), osd_numbers.end());
     if (!osd_numbers.size())
     {
+        if (json)
+            printf("%s\n", json11::Json(result).dump().c_str());
         return 0;
     }
     std::vector<std::string> rm_osd_cli = { "vitastor-cli", "rm-osd" };
@@ -457,17 +461,18 @@ int disk_tool_t::purge_devices(const std::vector<std::string> & devices)
     {
         systemctl_cli.push_back("vitastor-osd@"+std::to_string(osd_num));
     }
-    if (shell_exec(systemctl_cli, "", NULL, NULL) != 0)
+    if (shell_exec(systemctl_cli, "", json ? &dry_run_ignore_stdout : NULL, NULL) != 0)
     {
         return 1;
     }
     // Remove OSD metadata
     rm_osd_cli.pop_back();
-    if (shell_exec(rm_osd_cli, "", NULL, NULL) != 0)
+    if (shell_exec(rm_osd_cli, "", json ? &dry_run_ignore_stdout : NULL, NULL) != 0)
     {
         return 1;
     }
     // Destroy OSD superblocks
+    json11::Json::array removed_devices;
     for (auto & sb: superblocks)
     {
         for (auto dev_type: std::vector<std::string>{ "data", "meta", "journal" })
@@ -521,9 +526,15 @@ int disk_tool_t::purge_devices(const std::vector<std::string> & devices)
                             break;
                         }
                     }
+                    removed_devices.push_back(dev);
                 }
             }
         }
     }
+    if (json)
+    {
+        result["removed_devices"] = removed_devices;
+        printf("%s\n", json11::Json(result).dump().c_str());
+    }
     return 0;
 }

src/osd/osd.h

@@ -301,6 +301,7 @@ class osd_t
     void continue_primary_read(osd_op_t *cur_op);
     void continue_primary_scrub(osd_op_t *cur_op);
     void continue_primary_describe(osd_op_t *cur_op);
+    void continue_primary_list(osd_op_t *cur_op);
     void continue_primary_write(osd_op_t *cur_op);
     void cancel_primary_write(osd_op_t *cur_op);
     void continue_primary_sync(osd_op_t *cur_op);

src/osd/osd_peering.cpp

@@ -133,6 +133,7 @@ void osd_t::repeer_pgs(osd_num_t peer_osd)
 void osd_t::reset_pg(pg_t & pg)
 {
     pg.cur_peers.clear();
+    pg.dead_peers.clear();
     pg.state_dict.clear();
     copies_to_delete_after_sync_count -= pg.copies_to_delete_after_sync.size();
     pg.copies_to_delete_after_sync.clear();
@@ -235,13 +236,16 @@ void osd_t::start_pg_peering(pg_t & pg)
         return;
     }
     std::set<osd_num_t> cur_peers;
+    std::set<osd_num_t> dead_peers;
     for (auto pg_osd: pg.all_peers)
     {
         if (pg_osd == this->osd_num || msgr.osd_peer_fds.find(pg_osd) != msgr.osd_peer_fds.end())
-        {
             cur_peers.insert(pg_osd);
-        }
+        else
+            dead_peers.insert(pg_osd);
     }
+    pg.cur_peers.insert(pg.cur_peers.begin(), cur_peers.begin(), cur_peers.end());
+    pg.dead_peers.insert(pg.dead_peers.begin(), dead_peers.begin(), dead_peers.end());
     if (pg.target_history.size())
     {
         // Refuse to start PG if no peers are available from any of the historical OSD sets
@@ -269,7 +273,6 @@ void osd_t::start_pg_peering(pg_t & pg)
             }
         }
     }
-    pg.cur_peers.insert(pg.cur_peers.begin(), cur_peers.begin(), cur_peers.end());
     if (pg.peering_state)
     {
         // Adjust the peering operation that's still in progress - discard unneeded results

src/osd/osd_peering_pg.h

@@ -97,6 +97,8 @@ struct pg_t
     bool history_changed = false;
     // peer list from the last peering event
     std::vector<osd_num_t> cur_peers;
+    // dead_peers = all_peers - cur_peers
+    std::vector<osd_num_t> dead_peers;
     // target_set is the "correct" peer OSD set for this PG
     std::vector<osd_num_t> target_set;
     // cur_set is the current set of connected peer OSDs for this PG

src/osd/osd_primary.cpp

@@ -68,11 +68,20 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
             return false;
         }
         // Find parents from the same pool. Optimized reads only work within pools
-        while (inode_it != st_cli.inode_config.end() && inode_it->second.parent_id &&
-            INODE_POOL(inode_it->second.parent_id) == pg_it->second.pool_id &&
-            // Check for loops
-            inode_it->second.parent_id != cur_op->req.rw.inode)
+        while (inode_it != st_cli.inode_config.end() &&
+            inode_it->second.parent_id &&
+            INODE_POOL(inode_it->second.parent_id) == pg_it->second.pool_id)
         {
+            // Check for loops
+            if (inode_it->second.parent_id == cur_op->req.rw.inode ||
+                inode_it->second.parent_id == inode_it->second.num ||
+                chain_size > st_cli.inode_config.size())
+            {
+                printf("Inode %ju from pool %u has a parent_id loop, returning EINVAL in response to read\n",
+                    INODE_NO_POOL(cur_op->req.rw.inode), INODE_POOL(cur_op->req.rw.inode));
+                finish_op(cur_op, -EINVAL);
+                return false;
+            }
             chain_size++;
             inode_it = st_cli.inode_config.find(inode_it->second.parent_id);
         }
@@ -774,6 +783,20 @@ resume_5:
     }
     pg.total_count--;
     cur_op->reply.hdr.retval = 0;
+    // indicate possibly unfinished (left_on_dead) deletions
+    cur_op->reply.del.flags = OSD_DEL_SUPPORT_LEFT_ON_DEAD;
+    if (pg.dead_peers.size() > 0)
+    {
+        int max_del = (OSD_PACKET_SIZE-sizeof(cur_op->reply.del)) / sizeof(uint32_t);
+        cur_op->reply.del.flags |= OSD_DEL_LEFT_ON_DEAD;
+        cur_op->reply.del.left_on_dead_count = pg.dead_peers.size() < max_del
+            ? pg.dead_peers.size() : max_del;
+        uint32_t *left_on_dead = (uint32_t*)((&cur_op->reply.del) + 1);
+        for (int i = 0; i < cur_op->reply.del.left_on_dead_count; i++)
+        {
+            left_on_dead[i] = pg.dead_peers[i];
+        }
+    }
 continue_others:
     osd_op_t *next_op = NULL;
     auto next_it = pg.write_queue.find(op_data->oid);

src/osd/osd_primary_describe.cpp

@@ -138,3 +138,108 @@ void osd_t::continue_primary_describe(osd_op_t *cur_op)
         cur_op->iov.push_back(res.items, res.size * sizeof(osd_reply_describe_item_t));
     finish_op(cur_op, res.size);
 }
+
+static void add_primary_list(btree::btree_map<object_id, pg_osd_set_state_t*> & list, osd_op_sec_list_t & req, std::set<object_id> & oids)
+{
+    auto begin_it = list.begin();
+    auto end_it = list.end();
+    if (req.min_inode)
+        begin_it = list.lower_bound((object_id){ .inode = req.min_inode, .stripe = req.min_stripe });
+    if (req.max_inode)
+        end_it = list.upper_bound((object_id){ .inode = req.max_inode, .stripe = (req.max_stripe ? req.max_stripe : UINT64_MAX) });
+    for (auto list_it = begin_it; list_it != end_it; list_it++)
+        oids.insert(list_it->first);
+}
+
+void osd_t::continue_primary_list(osd_op_t *cur_op)
+{
+    auto pool_cfg_it = st_cli.pool_config.find(INODE_POOL(cur_op->req.sec_list.min_inode));
+    // Validate the request
+    if (!cur_op->req.sec_list.list_pg ||
+        !INODE_POOL(cur_op->req.sec_list.min_inode) ||
+        INODE_NO_POOL(cur_op->req.sec_list.min_inode) != INODE_NO_POOL(cur_op->req.sec_list.max_inode) ||
+        INODE_POOL(cur_op->req.sec_list.max_inode) != INODE_POOL(cur_op->req.sec_list.min_inode) ||
+        cur_op->req.sec_list.stable_limit ||
+        pool_cfg_it == st_cli.pool_config.end() ||
+        (cur_op->req.sec_list.pg_stripe_size != 0 && cur_op->req.sec_list.pg_stripe_size != pool_cfg_it->second.pg_stripe_size) ||
+        (cur_op->req.sec_list.pg_count != 0 && cur_op->req.sec_list.pg_count != pool_cfg_it->second.real_pg_count))
+    {
+        finish_op(cur_op, -EINVAL);
+        return;
+    }
+    auto pg_it = pgs.find({ .pool_id = INODE_POOL(cur_op->req.sec_list.min_inode), .pg_num = cur_op->req.sec_list.list_pg });
+    if (pg_it == pgs.end())
+    {
+        // Not primary
+        finish_op(cur_op, -EPIPE);
+        return;
+    }
+    cur_op->bs_op = new blockstore_op_t();
+    cur_op->bs_op->opcode = BS_OP_LIST;
+    cur_op->bs_op->pg_alignment = pool_cfg_it->second.pg_stripe_size;
+    cur_op->bs_op->pg_count = pool_cfg_it->second.real_pg_count;
+    cur_op->bs_op->pg_number = cur_op->req.sec_list.list_pg - 1;
+    cur_op->bs_op->min_oid.inode = cur_op->req.sec_list.min_inode;
+    cur_op->bs_op->min_oid.stripe = cur_op->req.sec_list.min_stripe;
+    cur_op->bs_op->max_oid.inode = cur_op->req.sec_list.max_inode;
+    if (cur_op->req.sec_list.max_inode && cur_op->req.sec_list.max_stripe != UINT64_MAX)
+    {
+        cur_op->bs_op->max_oid.stripe = cur_op->req.sec_list.max_stripe
+            ? cur_op->req.sec_list.max_stripe : UINT64_MAX;
+    }
+    cur_op->bs_op->callback = [this, cur_op](blockstore_op_t* bs_op)
+    {
+        if (bs_op->retval < 0)
+        {
+            auto retval = bs_op->retval;
+            delete bs_op;
+            cur_op->bs_op = NULL;
+            finish_op(cur_op, retval);
+            return;
+        }
+        // Move into a set
+        std::set<object_id> oids;
+        obj_ver_id *rbuf = (obj_ver_id*)bs_op->buf;
+        uint64_t total_count = bs_op->retval;
+        for (uint64_t i = 0; i < total_count; i++)
+        {
+            oids.insert(rbuf[i].oid);
+        }
+        if (bs_op->buf)
+        {
+            free(bs_op->buf);
+            bs_op->buf = NULL;
+        }
+        delete bs_op;
+        cur_op->bs_op = NULL;
+        // Check if still primary
+        auto pg_it = pgs.find({ .pool_id = INODE_POOL(cur_op->req.sec_list.min_inode), .pg_num = cur_op->req.sec_list.list_pg });
+        if (pg_it == pgs.end())
+        {
+            finish_op(cur_op, -EPIPE);
+            return;
+        }
+        // Add unclean objects which may be not present on the primary OSD
+        auto & pg = pg_it->second;
+        add_primary_list(pg.inconsistent_objects, cur_op->req.sec_list, oids);
+        add_primary_list(pg.incomplete_objects, cur_op->req.sec_list, oids);
+        add_primary_list(pg.degraded_objects, cur_op->req.sec_list, oids);
+        add_primary_list(pg.misplaced_objects, cur_op->req.sec_list, oids);
+        // Generate the result
+        if (oids.size())
+        {
+            rbuf = (obj_ver_id*)malloc_or_die(sizeof(obj_ver_id) * oids.size());
+            uint64_t i = 0;
+            for (auto oid: oids)
+            {
+                rbuf[i++] = (obj_ver_id){ .oid = oid };
+            }
+            cur_op->buf = rbuf;
+            cur_op->iov.push_back(rbuf, sizeof(obj_ver_id) * oids.size());
+        }
+        cur_op->reply.sec_list.stable_count = oids.size();
+        cur_op->reply.sec_list.flags = OSD_LIST_PRIMARY;
+        finish_op(cur_op, oids.size());
+    };
+    bs->enqueue_op(cur_op->bs_op);
+}

src/osd/osd_primary_subops.cpp

@@ -304,11 +304,23 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop)
         (bs_op->opcode != BS_OP_WRITE && bs_op->opcode != BS_OP_WRITE_STABLE ||
         bs_op->retval != -ENOSPC))
     {
-        // die on any error except ENOSPC
-        throw std::runtime_error(
-            "local blockstore modification failed (opcode = "+std::to_string(bs_op->opcode)+
-            " retval = "+std::to_string(bs_op->retval)+")"
-        );
+        // die on any error except ENOSPC during write
+        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",
+                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
+            );
+        }
+        else
+        {
+            printf(
+                "%s subop failed locally: retval = %d (expected %d)\n",
+                osd_op_names[bs_op_to_osd_op[bs_op->opcode]], bs_op->retval, expected
+            );
+        }
+        throw std::runtime_error("local blockstore modification failed");
     }
     bool recovery_related = cur_op->peer_fd == SELF_FD && cur_op->req.hdr.opcode != OSD_OP_SCRUB;
     add_bs_subop_stats(subop, recovery_related);
@@ -383,9 +395,11 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
     {
         uint64_t version = subop->reply.sec_rw.version;
 #ifdef OSD_DEBUG
-        uint64_t peer_osd = msgr.clients.find(subop->peer_fd) != msgr.clients.end()
-            ? msgr.clients[subop->peer_fd]->osd_num : osd_num;
-        printf("subop %s %jx:%jx from osd %ju: version = %ju\n", osd_op_names[opcode], subop->req.sec_rw.oid.inode, subop->req.sec_rw.oid.stripe, peer_osd, version);
+        int64_t peer_osd = subop->peer_fd == SELF_FD ? osd_num :
+            (msgr.clients.find(subop->peer_fd) != msgr.clients.end()
+                ? msgr.clients[subop->peer_fd]->osd_num : -subop->peer_fd);
+        printf("subop %s %jx:%jx from osd %jd: version = %ju\n", osd_op_names[opcode],
+            subop->req.sec_rw.oid.inode, subop->req.sec_rw.oid.stripe, peer_osd, version);
 #endif
         if (op_data->fact_ver != UINT64_MAX)
         {
@@ -403,21 +417,23 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
     }
     if (retval != expected)
     {
+        int64_t peer_osd = (msgr.clients.find(subop->peer_fd) != msgr.clients.end()
+            ? msgr.clients[subop->peer_fd]->osd_num : -subop->peer_fd);
         if (opcode == OSD_OP_SEC_READ || opcode == OSD_OP_SEC_WRITE || opcode == OSD_OP_SEC_WRITE_STABLE)
         {
             printf(
                 subop->peer_fd >= 0
-                    ? "%1$s subop to %2$jx:%3$jx v%4$ju failed on peer %7$d: retval = %5$d (expected %6$d)\n"
+                    ? "%1$s subop to %2$jx:%3$jx v%4$ju failed on osd %7$jd: retval = %5$d (expected %6$d)\n"
                     : "%1$s subop to %2$jx:%3$jx v%4$ju failed locally: retval = %5$d (expected %6$d)\n",
                 osd_op_names[opcode], subop->req.sec_rw.oid.inode, subop->req.sec_rw.oid.stripe, subop->req.sec_rw.version,
-                retval, expected, subop->peer_fd
+                retval, expected, peer_osd
             );
         }
         else
         {
             printf(
-                "%s subop failed on peer %d: retval = %d (expected %d)\n",
-                osd_op_names[opcode], subop->peer_fd, retval, expected
+                "%s subop failed on osd %jd: retval = %d (expected %d)\n",
+                osd_op_names[opcode], peer_osd, retval, expected
             );
         }
         subop->rmw_buf = NULL;
@@ -728,7 +744,7 @@ void osd_t::submit_primary_rollback_subops(osd_op_t *cur_op, const uint64_t* osd
     for (int role = 0; role < op_data->pg_size; role++)
     {
         if (osd_set[role] != 0 && !stripes[role].read_error &&
-            msgr.osd_peer_fds.find(osd_set[role]) != msgr.osd_peer_fds.end())
+            (osd_set[role] == this->osd_num || msgr.osd_peer_fds.find(osd_set[role]) != msgr.osd_peer_fds.end()))
         {
             n_subops++;
         }
@@ -745,7 +761,7 @@ void osd_t::submit_primary_rollback_subops(osd_op_t *cur_op, const uint64_t* osd
     for (int role = 0; role < op_data->pg_size; role++)
     {
         if (osd_set[role] != 0 && !stripes[role].read_error &&
-            msgr.osd_peer_fds.find(osd_set[role]) != msgr.osd_peer_fds.end())
+            (osd_set[role] == this->osd_num || msgr.osd_peer_fds.find(osd_set[role]) != msgr.osd_peer_fds.end()))
         {
             osd_op_t *subop = &op_data->subops[i];
             op_data->unstable_writes[i] = (obj_ver_id){

src/osd/osd_primary_write.cpp

@@ -260,11 +260,6 @@ resume_4:
     op_data->st = 4;
     return;
 resume_5:
-    if (op_data->scheme != POOL_SCHEME_REPLICATED)
-    {
-        // Remove version override just after the write, but before stabilizing
-        pg.ver_override.erase(op_data->oid);
-    }
     if (op_data->errors > 0)
     {
         // Handle ENOSPC/EDOM/ERANGE/EIO. If some subops fail, but others succeed,
@@ -276,7 +271,7 @@ resume_5:
         {
             if (op_data->scheme != POOL_SCHEME_REPLICATED)
             {
-                submit_primary_rollback_subops(cur_op, op_data->prev_set);
+                submit_primary_rollback_subops(cur_op, pg.cur_set.data());
 resume_11:
                 op_data->st = 11;
                 return;
@@ -287,15 +282,22 @@ resume_12:
             }
             else
             {
+                pg.ver_override.erase(op_data->oid);
                 mark_partial_write(pg, op_data->oid, op_data->object_state, op_data->stripes, true);
                 pg_cancel_write_queue(pg, cur_op, op_data->oid, op_data->errcode);
                 return;
             }
         }
+        pg.ver_override.erase(op_data->oid);
         deref_object_state(pg, &op_data->object_state, true);
         pg_cancel_write_queue(pg, cur_op, op_data->oid, op_data->errcode);
         return;
     }
+    if (op_data->scheme != POOL_SCHEME_REPLICATED)
+    {
+        // Remove version override just after the write, but before stabilizing
+        pg.ver_override.erase(op_data->oid);
+    }
     if (op_data->object_state)
     {
         // We must forget the unclean state of the object before deleting it

src/osd/osd_secondary.cpp

@@ -81,6 +81,12 @@ void osd_t::exec_secondary(osd_op_t *op)
 
 void osd_t::exec_secondary_real(osd_op_t *cur_op)
 {
+    if (cur_op->req.hdr.opcode == OSD_OP_SEC_LIST &&
+        (cur_op->req.sec_list.flags & OSD_LIST_PRIMARY))
+    {
+        continue_primary_list(cur_op);
+        return;
+    }
     if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
     {
         int n = cur_op->req.sec_read_bmp.len / sizeof(obj_ver_id);

src/test/CMakeLists.txt

@@ -12,6 +12,11 @@ target_link_libraries(stub_bench tcmalloc_minimal)
 add_executable(osd_test osd_test.cpp ../util/rw_blocking.cpp ../util/addr_util.cpp)
 target_link_libraries(osd_test tcmalloc_minimal)
 
+# bindiff
+add_executable(bindiff
+	bindiff.c
+)
+
 # stub_uring_osd
 add_executable(stub_uring_osd
 	stub_uring_osd.cpp

src/test/bindiff.c

@@ -0,0 +1,177 @@
+// Copyright (c) Vitaliy Filippov, 2004+
+// License: VNPL-1.1 (see README.md for details)
+
+#ifndef _LARGEFILE64_SOURCE
+#define _LARGEFILE64_SOURCE
+#endif
+
+#include <string.h>
+#include <sys/stat.h>
+#include <errno.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <stdint.h>
+#include <unistd.h>
+#include <fcntl.h>
+
+#define BUFSIZE 0x100000
+
+uint64_t filelength(int fd)
+{
+    struct stat st;
+    if (fstat(fd, &st) < 0)
+    {
+        fprintf(stderr, "fstat failed: %s\n", strerror(errno));
+        return 0;
+    }
+    if (st.st_size < 0)
+    {
+        return 0;
+    }
+    return (uint64_t)st.st_size;
+}
+
+size_t read_blocking(int fd, void *read_buf, size_t remaining)
+{
+    size_t done = 0;
+    while (done < remaining)
+    {
+        ssize_t r = read(fd, read_buf, remaining-done);
+        if (r <= 0)
+        {
+            if (!errno)
+            {
+                // EOF
+                return done;
+            }
+            else if (errno != EINTR && errno != EAGAIN && errno != EPIPE)
+            {
+                perror("read");
+                exit(1);
+            }
+            continue;
+        }
+        done += (size_t)r;
+        read_buf = (uint8_t*)read_buf + r;
+    }
+    return done;
+}
+
+size_t write_blocking(int fd, void *write_buf, size_t remaining)
+{
+    size_t done = 0;
+    while (done < remaining)
+    {
+        ssize_t r = write(fd, write_buf, remaining-done);
+        if (r < 0)
+        {
+            if (errno != EINTR && errno != EAGAIN && errno != EPIPE)
+            {
+                perror("write");
+                exit(1);
+            }
+            continue;
+        }
+        done += (size_t)r;
+        write_buf = (uint8_t*)write_buf + r;
+    }
+    return done;
+}
+
+int main(int narg, char *args[])
+{
+    int fd1 = -1, fd2 = -1;
+    uint8_t *buf1 = NULL, *buf2 = NULL;
+    uint64_t addr = 0, l1 = 0, l2 = 0, l = 0, diffl = 0;
+    size_t buf1_len = 0, buf2_len = 0, i = 0, j = 0, dl = 0;
+    int argoff = 0;
+    int nosource = 0;
+    fprintf(stderr, "VMX HexDiff v2.1\nLicense: GPLv3.0+, (c) 2005+, Vitaliy Filippov\n");
+    argoff = 1;
+    if (narg > argoff && strcmp(args[argoff], "-n") == 0)
+    {
+        nosource = 1;
+        argoff++;
+    }
+    if (narg < argoff+2)
+    {
+        fprintf(stderr, "USAGE: bindiff [-n] <file1> <file2>\n"
+            "This will create hex patch file1->file2 and write it to stdout.\n"
+            "[-n] = do not write file1 data in patch, only file2.\n");
+        return -1;
+    }
+    fd1 = open(args[argoff], O_RDONLY);
+    if (fd1 < 0)
+    {
+        fprintf(stderr, "Couldn't open %s: %s\n", args[argoff], strerror(errno));
+        return -1;
+    }
+    fd2 = open(args[argoff+1], O_RDONLY);
+    if (fd2 < 0)
+    {
+        fprintf(stderr, "Couldn't open %s: %s\n", args[argoff+1], strerror(errno));
+        close(fd1);
+        return -1;
+    }
+    l1 = filelength(fd1);
+    l2 = filelength(fd2);
+    if (l1 < l2)
+        l = l1;
+    else
+        l = l2;
+    addr = diffl = 0;
+    buf1 = malloc(BUFSIZE+1);
+    buf2 = malloc(BUFSIZE+1);
+    while ((buf1_len = read_blocking(fd1, buf1, BUFSIZE)) > 0 && (buf2_len = read_blocking(fd2, buf2, BUFSIZE)) > 0)
+    {
+        buf1[buf1_len] = buf2[buf2_len] = 0;
+        for (dl = 0, i = 0; i <= buf1_len && i <= buf2_len; i++, addr++)
+        {
+            if (buf1[i] != buf2[i])
+            {
+                dl++;
+            }
+            else if (dl)
+            {
+                printf("%08jX: ", addr-dl);
+                if (!nosource)
+                {
+                    for (j = i-dl; j < i; j++)
+                        printf("%02X", buf1[j]);
+                    printf(" ");
+                }
+                for (j = i-dl; j < i; j++)
+                    printf("%02X", buf2[j]);
+                printf("\n");
+                diffl += dl;
+                dl = 0;
+            }
+        }
+        addr--;
+    }
+    if (l1 < l2)
+    {
+        printf("%08zX: ", i);
+        while ((buf2_len = read_blocking(fd2, buf2, BUFSIZE)) > 0)
+        {
+            for (j = 0; j < buf2_len; j++, i++)
+                printf("%02X", buf2[j]);
+        }
+        printf("\n");
+    }
+    else if (l1 > l2)
+    {
+        printf("SIZE %08zX\n", l2);
+    }
+    if (diffl != 0 || l1 != l2)
+    {
+        fprintf(stderr, "Difference in %zu of %zu common bytes\n", diffl, l);
+        if (l1 != l2)
+            fprintf(stderr, "Length difference!\nFile \"%s\": %zu\nFile \"%s\": %zu\n", args [1], l1, args [2], l2);
+    }
+    else
+    {
+        fprintf(stderr, "Files are equal\n");
+    }
+    return 0;
+}

src/util/rw_blocking.cpp

@@ -10,7 +10,7 @@
 
 #include "rw_blocking.h"
 
-int read_blocking(int fd, void *read_buf, size_t remaining)
+size_t read_blocking(int fd, void *read_buf, size_t remaining)
 {
     size_t done = 0;
     while (done < remaining)
@@ -30,13 +30,13 @@ int read_blocking(int fd, void *read_buf, size_t remaining)
             }
             continue;
         }
-        done += r;
+        done += (size_t)r;
         read_buf = (uint8_t*)read_buf + r;
     }
     return done;
 }
 
-int write_blocking(int fd, void *write_buf, size_t remaining)
+size_t write_blocking(int fd, void *write_buf, size_t remaining)
 {
     size_t done = 0;
     while (done < remaining)
@@ -51,7 +51,7 @@ int write_blocking(int fd, void *write_buf, size_t remaining)
             }
             continue;
         }
-        done += r;
+        done += (size_t)r;
         write_buf = (uint8_t*)write_buf + r;
     }
     return done;

src/util/rw_blocking.h

@@ -6,8 +6,8 @@
 #include <unistd.h>
 #include <sys/uio.h>
 
-int read_blocking(int fd, void *read_buf, size_t remaining);
-int write_blocking(int fd, void *write_buf, size_t remaining);
+size_t read_blocking(int fd, void *read_buf, size_t remaining);
+size_t write_blocking(int fd, void *write_buf, size_t remaining);
 int readv_blocking(int fd, iovec *iov, int iovcnt);
 int writev_blocking(int fd, iovec *iov, int iovcnt);
 int sendv_blocking(int fd, iovec *iov, int iovcnt, int flags);

tests/run_tests.sh

@@ -36,6 +36,8 @@ SCHEME=ec ./test_snapshot.sh
 
 ./test_rm.sh
 
+./test_rm_degraded.sh
+
 ./test_snapshot_chain.sh
 SCHEME=ec ./test_snapshot_chain.sh
 

tests/test_heal.sh

@@ -60,6 +60,7 @@ qemu-img convert -S 4096 -p \
     -O raw ./testdata/bin/read.bin
 
 if ! diff -q ./testdata/bin/read.bin ./testdata/bin/mirror.bin; then
+    build/src/test/bindiff ./testdata/bin/read.bin ./testdata/bin/mirror.bin
     format_error Data lost during self-heal
 fi
 

tests/test_rm_degraded.sh

@@ -0,0 +1,44 @@
+#!/bin/bash -ex
+
+SCHEME=xor
+PG_COUNT=16
+PG_MINSIZE=2
+. `dirname $0`/run_3osds.sh
+
+build/src/cmd/vitastor-cli --etcd_address $ETCD_URL create -s 128M testimg
+
+LD_PRELOAD="build/src/client/libfio_vitastor.so" \
+    fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so -bs=4M -direct=1 -iodepth=1 \
+        -end_fsync=1 -fsync=1 -rw=write -etcd=$ETCD_URL -image=testimg -size=128M -cluster_log_level=10
+
+kill -9 $OSD3_PID
+$ETCDCTL del /vitastor/osd/state/3
+
+if build/src/cmd/vitastor-cli --etcd_address $ETCD_URL rm testimg --log_level 10 ; then
+    format_error "Delete should not be successful with inactive OSDs"
+fi
+
+if ! ( build/src/cmd/vitastor-cli --etcd_address $ETCD_URL ls | grep testimg | grep DEL ) ; then
+    format_error "Image should be marked as partially deleted"
+fi
+
+start_osd 3
+sleep 5
+
+# Now do the same but without del /vitastor/osd/state
+
+LD_PRELOAD="build/src/client/libfio_vitastor.so" \
+    fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so -bs=4M -direct=1 -iodepth=1 \
+        -end_fsync=1 -fsync=1 -rw=write -etcd=$ETCD_URL -image=testimg -size=128M -cluster_log_level=10
+
+kill -9 $OSD3_PID
+
+if build/src/cmd/vitastor-cli --etcd_address $ETCD_URL rm testimg --log_level 10 ; then
+    format_error "Delete should not be successful with inactive OSDs"
+fi
+
+if ! ( build/src/cmd/vitastor-cli --etcd_address $ETCD_URL ls | grep testimg | grep DEL ) ; then
+    format_error "Image should be marked as partially deleted"
+fi
+
+format_green OK

tests/test_snapshot_pool2.sh

@@ -21,7 +21,16 @@ LD_PRELOAD="build/src/client/libfio_vitastor.so" \
     fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so -bs=4k -direct=1 -iodepth=4 -end_fsync=1 -rw=randwrite -number_ios=32 \
         -etcd=$ETCD_URL -image=testchain -mirror_file=./testdata/bin/mirror.bin -buffer_pattern=0xabcd
 
-build/src/cmd/vitastor-cli --etcd_address $ETCD_URL dd iimg=testchain of=./testdata/bin/res.bin bs=128k iodepth=4
+# Create a second snapshot - there was a bug where snapshotted reads from another pool
+# were working only when the image and the snapshot were modified in the same revision
+# (i.e. there was only one snapshot)
+build/src/cmd/vitastor-cli --etcd_address $ETCD_URL snap-create testchain@snap2 -p testpool2
+
+LD_PRELOAD="build/src/client/libfio_vitastor.so" \
+    fio -thread -name=test -ioengine=build/src/client/libfio_vitastor.so -bs=4k -direct=1 -iodepth=4 -end_fsync=1 -rw=randwrite -number_ios=32 \
+        -etcd=$ETCD_URL -image=testchain -mirror_file=./testdata/bin/mirror.bin -buffer_pattern=0xabcd
+
+build/src/cmd/vitastor-cli --etcd_address $ETCD_URL dd iimg=testchain of=./testdata/bin/res.bin bs=128k iodepth=4 --log_level 10
 
 cmp ./testdata/bin/res.bin ./testdata/bin/mirror.bin